7 Commits
Author SHA1 Message Date
rob a37a13f77a Keep run-once timeout flag distinct 2026-09-29 13:15:29 -03:00
rob a2f95b5dd9 Correct repository links and labels 2026-09-29 13:11:19 -03:00
rob 16be1c8d92 Add configurable donation options 2026-09-29 13:11:19 -03:00
robandClaude Opus 5.5 2e38c5d629 Prompt steps take a caller-supplied time limit (X-8a)
Prompt steps called call_provider without a timeout, so every AI call was cut at its
300 s default whatever the caller allowed. Now:
- PromptStep has an optional `timeout` (seconds) in its YAML;
- a universal --provider-timeout (tool wrappers, `cmdforge run`, `run-once`) sets the
  caller's limit, carried in ProviderExecutionPolicy so nested tools get it too;
- precedence: --provider-timeout, then the step's timeout, then 300 (unchanged default).

Requested by Orchestrated Discussions, whose work phases need attempts longer than 300 s.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-09-29 11:56:42 -03:00
rob e042cd752b Enforce canonical setup result contract in CLI and GUI 2026-09-14 22:37:12 -03:00
rob 4bd0a7cc02 feat(gui): reconcile provider install dialog with agent-harness-setup helper 2026-09-11 22:55:42 -03:00
rob 60cd44bcfb feat(providers): implement Packet S4 agent-harness-setup CLI wrapper and provider handoff 2026-09-11 22:31:08 -03:00
27 changed files with 2738 additions and 217 deletions
+1 -1
View File
@@ -83,7 +83,7 @@ cf = "cmdforge.cli.picker:main"
[project.urls]
Homepage = "https://cmdforge.brrd.tech"
Documentation = "https://cmdforge.brrd.tech/docs"
Repository = "https://gitea.brrd.tech/rob/CmdForge.git"
Repository = "https://gitea.brrd.tech/rob/CmdForge"
Issues = "https://gitea.brrd.tech/rob/CmdForge/issues"
Changelog = "https://gitea.brrd.tech/rob/CmdForge/src/branch/main/CHANGELOG.md"
+21
View File
@@ -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.
+6
View File
@@ -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.
+1
View File
@@ -0,0 +1 @@
"""Vendored setup-result contract; see README.md."""
+304
View File
@@ -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}"]
+927
View File
@@ -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))
+3
View File
@@ -97,6 +97,7 @@ def main():
p_run.add_argument("--data-classification", choices=["public", "internal", "private"], help="Input data classification")
p_run.add_argument("--require-model-identity", action="store_true")
p_run.add_argument("--require-model-digest", action="store_true")
p_run.add_argument("--provider-timeout", type=int, help="Seconds each AI provider call may take")
p_run.add_argument("--result-envelope", choices=["json"], help="Wrap output with verified execution provenance")
p_run.add_argument("tool_args", nargs=argparse.REMAINDER, help="Additional tool-specific arguments (use -- to separate)")
p_run.set_defaults(func=cmd_run)
@@ -190,6 +191,8 @@ def main():
# providers install
p_prov_install = providers_sub.add_parser("install", help="Interactive guide to install AI providers")
p_prov_install.add_argument("--harness", help="Specific harness ID to install")
p_prov_install.add_argument("--dry-run", action="store_true", help="Preview installation without executing")
p_prov_install.set_defaults(func=cmd_providers)
# providers discover
+205 -125
View File
@@ -1,12 +1,15 @@
"""Provider management commands."""
import json
import os
import shlex
import shutil
import subprocess
import sys
import tempfile
import uuid
from pathlib import Path
from ..providers import (
KNOWN_PROVIDER_CLIS,
PRESET_CHAINS,
Provider,
add_provider,
@@ -121,142 +124,219 @@ def cmd_providers(args):
return 0
def _cmd_providers_install(args):
"""Interactive guide to install AI providers."""
print("=" * 60)
print("CmdForge Provider Installation Guide")
print("=" * 60)
print()
def _find_setup_helper() -> list[str] | None:
"""Locate the agent-harness-setup CLI executable or python module."""
# 1. System/user PATH
path_bin = shutil.which("agent-harness-setup")
if path_bin:
return [path_bin]
# Check what's already installed
providers = load_providers()
installed_groups = set()
for p in providers:
if p.name.lower() == "mock" or p.type == "api":
continue
available, _ = _provider_status(p)
if available:
# Find which group this belongs to
for group, info in PROVIDER_INSTALL_INFO.items():
if p.name in info.get("variants", []):
installed_groups.add(group)
# 2. Current environment virtualenv bin
venv_bin = Path(sys.prefix) / "bin" / "agent-harness-setup"
if venv_bin.is_file() and os.access(venv_bin, os.X_OK):
return [str(venv_bin)]
# Show available provider groups
print("Available AI Provider Groups:\n")
options = []
for i, (key, info) in enumerate(PROVIDER_INSTALL_INFO.items(), 1):
status = "[INSTALLED]" if key in installed_groups else ""
print(f" {i}. {info['group']} {status}")
print(f" Cost: {info['cost']}")
print(f" Models: {', '.join(info['variants']) if info['variants'] else 'Custom'}")
print()
options.append(key)
# 3. User local bin
local_bin = Path.home() / ".local" / "bin" / "agent-harness-setup"
if local_bin.is_file() and os.access(local_bin, os.X_OK):
return [str(local_bin)]
print(" 0. Cancel")
print()
try:
choice = input("Select a provider to install (1-{}, or 0 to cancel): ".format(len(options)))
choice = int(choice)
except (ValueError, EOFError):
print("Cancelled.")
return 0
if choice == 0 or choice > len(options):
print("Cancelled.")
return 0
selected = options[choice - 1]
info = PROVIDER_INSTALL_INFO[selected]
print()
print("=" * 60)
print(f"Installing: {info['group']}")
print("=" * 60)
print()
print(f"Requirements: {info['requires']}")
if info["install_cmd"]:
print(f"Install command: {info['install_cmd']}")
else:
print("Install method: Manual download")
print(f"Post-install: {info['setup']}")
print()
if not info["install_cmd"]:
print("CmdForge cannot install this provider automatically.")
print(info["setup"])
return 0
try:
confirm = input("Run installation command? (y/N): ").strip().lower()
except EOFError:
confirm = "n"
if confirm == "y":
print()
print(f"Running: {info['install_cmd']}")
print("-" * 40)
result = subprocess.run(info['install_cmd'], shell=True)
print("-" * 40)
if result.returncode == 0:
# Refresh PATH to pick up newly installed tools
import os
new_paths = []
# Common install locations that might have been added
potential_paths = [
Path.home() / ".opencode" / "bin", # OpenCode
Path.home() / ".local" / "bin", # pip/pipx installs
Path("/usr/local/bin"), # Ollama, system installs
# 4. Workspace sibling repository
workspace_candidates = [
Path(__file__).resolve().parents[4] / "agent-harness-setup" / ".venv" / "bin" / "agent-harness-setup",
Path("/home/rob/PycharmProjects/agent-harness-setup/.venv/bin/agent-harness-setup"),
]
for cand in workspace_candidates:
if cand.is_file() and os.access(cand, os.X_OK):
return [str(cand)]
# Also try to get npm global bin path
# 5. Check if agent_harness_setup is importable via python
try:
npm_result = subprocess.run(
["npm", "bin", "-g"],
capture_output=True, text=True, timeout=5
)
if npm_result.returncode == 0:
npm_bin = npm_result.stdout.strip()
if npm_bin:
potential_paths.append(Path(npm_bin))
except (subprocess.TimeoutExpired, FileNotFoundError):
import agent_harness_setup # type: ignore # noqa: F401
return [sys.executable, "-m", "agent_harness_setup"]
except ImportError:
pass
current_path = os.environ.get("PATH", "")
for p in potential_paths:
if p.exists() and str(p) not in current_path:
new_paths.append(str(p))
return None
if new_paths:
os.environ["PATH"] = ":".join(new_paths) + ":" + current_path
def _get_provider_defaults_for_harness(harness_id: str, executable_path: str | None = None) -> dict:
"""Derive CmdForge provider configuration defaults for an installed harness."""
harness_map = {
"claude-code": "claude",
"claude": "claude",
"openai-codex": "codex",
"codex": "codex",
"agy": "agy",
"gemini-cli": "agy",
"opencode": "opencode",
"crush": "crush",
"ollama": "ollama",
}
cli_key = harness_map.get(harness_id, harness_id)
if cli_key in KNOWN_PROVIDER_CLIS:
base = dict(KNOWN_PROVIDER_CLIS[cli_key])
if executable_path:
old_cmd = base.get("command", "")
parts = shlex.split(old_cmd)
if parts:
base["command"] = " ".join([executable_path] + parts[1:])
else:
base["command"] = executable_path
return base
cmd = executable_path if executable_path else harness_id
return {
"name": harness_id,
"command": cmd,
"description": f"{harness_id} CLI provider",
"tags": ["cli"],
"locality": "remote",
"capabilities": ["text"],
"data_policy": "public",
}
def _cmd_providers_install(args):
"""Interactive guide to install AI harnesses and configure providers."""
helper_argv = _find_setup_helper()
if not helper_argv:
print("=" * 60)
print("CmdForge Provider Installation")
print("=" * 60)
print()
print("Notice: Agent Harness Setup Helper ('agent-harness-setup') was not")
print("found on your PATH or in active virtual environments.")
print()
print("To install the setup helper, run:")
print(" pip install -e /home/rob/PycharmProjects/agent-harness-setup")
print()
print("Or run the helper directly:")
print(" PYTHONPATH=/home/rob/PycharmProjects/agent-harness-setup/src python3 -m agent_harness_setup")
print()
print("Manual installation instructions are available in documentation:")
print(" https://docs.anthropic.com/en/docs/agents-and-tools/claude-code/overview")
return 0
# Prepare secure temporary result file and correlation request_id
temp_dir = tempfile.mkdtemp(prefix="cmdforge-harness-setup-")
try:
os.chmod(temp_dir, 0o700)
except OSError:
pass
result_file = Path(temp_dir) / "result.json"
req_id = str(uuid.uuid4())
cmd = list(helper_argv) + ["--result-file", str(result_file), "--request-id", req_id]
harness_arg = getattr(args, "harness", None)
if harness_arg:
cmd.extend(["--harness", harness_arg])
if getattr(args, "dry_run", False):
cmd.append("--dry-run")
try:
proc = subprocess.run(cmd)
helper_exit_code = proc.returncode
except Exception as e:
print(f"\nError executing agent-harness-setup: {e}")
shutil.rmtree(temp_dir, ignore_errors=True)
return 1
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)
if verdict == "unknown":
print("\nSetup outcome unknown: " + "; ".join(problems))
return helper_exit_code if helper_exit_code else 1
# Dry-run completion
if verdict == "dry_run":
return 0
# Cancellation
if helper_exit_code in (10, 130, 143) or (result_data and result_data.get("status") == "cancelled"):
print("\nSetup cancelled. No provider configuration was changed.")
return 0
# Failure
if helper_exit_code != 0 or (result_data and result_data.get("status") == "failed"):
diag = result_data.get("error_diagnostics") if result_data else None
if diag:
print(f"\nSetup failed: {diag}")
return helper_exit_code if helper_exit_code != 0 else 1
# Verified Success
if helper_exit_code == 0 and result_data and result_data.get("status") == "success":
harness_id = result_data.get("harness_id") or "unknown"
verification = result_data.get("verification") or {}
executable_path = verification.get("executable_path")
path_updates = result_data.get("path_updates") or []
# Advisory PATH updates
if path_updates:
cur_path = os.environ.get("PATH", "")
new_dirs = [
str(Path(os.path.expanduser(p)))
for p in path_updates
if Path(os.path.expanduser(p)).is_dir() and str(Path(os.path.expanduser(p))) not in cur_path
]
if new_dirs:
os.environ["PATH"] = ":".join(new_dirs) + ":" + cur_path
provider_defaults = _get_provider_defaults_for_harness(harness_id, executable_path)
provider_name = provider_defaults["name"]
print()
print("Installation completed!")
print("=" * 60)
print(f"Harness Setup Verified: {harness_id}")
print("=" * 60)
if executable_path:
print(f"Verified executable: {executable_path}")
if result_data.get("manual_steps_remaining"):
print("\nRemaining setup steps:")
for step in result_data["manual_steps_remaining"]:
print(f" * {step}")
existing = get_provider(provider_name)
if existing:
print(f"\nProvider '{provider_name}' is already configured in ~/.cmdforge/providers.yaml.")
print(f"Command: {existing.command}")
print(f"\nTest with: cmdforge providers test {provider_name}")
return 0
# Interactive confirmation prompt for provider configuration
print()
print("IMPORTANT: Refresh your shell PATH before continuing:")
print(" source ~/.bashrc")
print()
print(f"Next steps:")
print(f" 1. source ~/.bashrc (required!)")
print(f" 2. {info['setup']}")
if info.get('post_install_note'):
print(f" 3. {info['post_install_note']}")
print(f" 4. Test with: cmdforge providers test {selected}")
try:
prompt_resp = input(
f"Would you like to register '{provider_name}' as a model provider in ~/.cmdforge/providers.yaml? (Y/n): "
).strip().lower()
except (EOFError, KeyboardInterrupt):
prompt_resp = "n"
if prompt_resp in ("", "y", "yes"):
new_prov = Provider(
name=provider_name,
command=provider_defaults["command"],
description=provider_defaults.get("description", ""),
tags=provider_defaults.get("tags", []),
locality=provider_defaults.get("locality", "remote"),
capabilities=provider_defaults.get("capabilities", ["text"]),
data_policy=provider_defaults.get("data_policy", "public"),
)
add_provider(new_prov)
print(f"\nProvider '{provider_name}' successfully added to ~/.cmdforge/providers.yaml.")
print(f"Command: {new_prov.command}")
print(f"\nTest with: cmdforge providers test {provider_name}")
else:
print(f" 3. Test with: cmdforge providers test {info['variants'][0] if info['variants'] else selected}")
else:
print()
print(f"Installation failed (exit code {result.returncode})")
print("Try running the command manually:")
print(f" {info['install_cmd']}")
else:
print()
print("To install manually, run:")
print(f" {info['install_cmd']}")
print()
print(f"Then: {info['setup']}")
print("\nSkipped provider registration. You can add it later with:")
print(f" cmdforge providers add {provider_name} '{provider_defaults['command']}'")
return 0
return 0
+5
View File
@@ -81,6 +81,11 @@ def _execution_policy_from_args(*namespaces):
getattr(namespace, "require_model_digest", False)
for namespace in namespaces
),
provider_timeout=next((
getattr(namespace, "provider_timeout", None)
for namespace in namespaces
if getattr(namespace, "provider_timeout", None)
), None),
)
@@ -1,26 +1,37 @@
"""Provider installation wizard dialog."""
"""Provider installation wizard dialog.
Reconciled in Packet S6 to delegate harness installation to agent-harness-setup
via a dedicated terminal window while preserving the GUI configuration and
provider variants management workflows.
"""
import json
import os
import shutil
import subprocess
import tempfile
import uuid
from pathlib import Path
from typing import Optional
from PySide6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QListWidget, QListWidgetItem, QTextEdit, QProgressBar,
QMessageBox, QGroupBox, QWidget, QStackedWidget
)
from PySide6.QtCore import Qt, QThread, Signal, QProcess
from PySide6.QtCore import Qt, QThread, Signal
from PySide6.QtGui import QFont
from ...providers import load_providers, add_provider, Provider
from ...cli.provider_commands import _find_setup_helper
# Provider installation info (shared with CLI)
PROVIDER_INSTALL_INFO = {
# Supported AI provider metadata and variants (installation catalogue retired and
# delegated to agent-harness-setup)
SUPPORTED_PROVIDERS = {
"claude": {
"harness_id": "claude-code",
"group": "Anthropic Claude",
"install_cmd": "curl -fsSL https://claude.ai/install.sh | bash",
"requires": "curl and bash",
"setup": "Run 'claude' to sign in with a subscription or API key.",
"cost": "Paid subscription or API key",
"variants": [
@@ -31,29 +42,27 @@ PROVIDER_INSTALL_INFO = {
],
},
"codex": {
"harness_id": "codex",
"group": "OpenAI Codex",
"install_cmd": "npm install -g @openai/codex",
"requires": "Node.js 18+ and npm",
"setup": "Run 'codex' to sign in.",
"setup": "Run 'codex' to sign in. ChatGPT Plus/Pro includes Codex access.",
"cost": "Free tier or ChatGPT subscription, subject to account limits",
"variants": [
("codex", "codex exec -", "OpenAI Codex"),
],
},
"agy": {
"harness_id": "agy",
"group": "Google Antigravity",
"install_cmd": None,
"requires": "Google Antigravity desktop application",
"setup": "Install from https://antigravity.google/download, then run 'agy install' to configure your PATH.",
"cost": "Free tier available, subject to account limits",
"manual_only": True,
"variants": [
("agy", "agy -p", "Google Antigravity (automatic model selection)"),
],
},
"opencode": {
"harness_id": "opencode",
"group": "OpenCode",
"install_cmd": "curl -fsSL https://opencode.ai/install | bash",
"requires": "curl, bash",
"setup": "Run 'opencode' and use /connect to configure providers.",
"cost": "Free models available; paid providers require credentials",
"variants": [
@@ -64,9 +73,8 @@ PROVIDER_INSTALL_INFO = {
],
},
"crush": {
"harness_id": "crush",
"group": "Crush (Charm)",
"install_cmd": "npm install -g @charmland/crush",
"requires": "Node.js 18+ and npm",
"setup": "Run 'crush' to configure Hyper or your provider API keys.",
"cost": "Limited Hyper credits or your own provider billing",
"variants": [
@@ -74,9 +82,8 @@ PROVIDER_INSTALL_INFO = {
],
},
"ollama": {
"harness_id": "ollama",
"group": "Ollama (Local LLMs)",
"install_cmd": "curl -fsSL https://ollama.ai/install.sh | bash",
"requires": "curl, bash, 8GB+ RAM (GPU recommended)",
"setup": "Run 'ollama pull llama3.2' to download a model.",
"cost": "FREE (runs entirely on your machine)",
"variants": [
@@ -88,43 +95,111 @@ PROVIDER_INSTALL_INFO = {
},
}
# Retained alias for backwards compatibility
PROVIDER_INSTALL_INFO = SUPPORTED_PROVIDERS
class InstallWorker(QThread):
"""Background worker for running installation commands."""
def find_terminal_emulator() -> list[str] | None:
"""Find an available graphical terminal emulator command prefix."""
candidates = [
("x-terminal-emulator", ["-e"]),
("gnome-terminal", ["--"]),
("konsole", ["-e"]),
("xfce4-terminal", ["-e"]),
("kitty", []),
("alacritty", ["-e"]),
("xterm", ["-e"]),
]
for term, prefix in candidates:
bin_path = shutil.which(term)
if bin_path:
return [bin_path] + prefix
return None
class SetupTerminalWorker(QThread):
"""Background worker running agent-harness-setup in a dedicated terminal window."""
output = Signal(str)
finished = Signal(bool, str)
finished = Signal(bool, str, dict)
def __init__(self, command: str):
def __init__(
self,
helper_argv: list[str],
harness_id: str,
result_file: Path,
request_id: str,
terminal_launcher: list[str] | None = None,
):
super().__init__()
self.command = command
self.helper_argv = helper_argv
self.harness_id = harness_id
self.result_file = result_file
self.request_id = request_id
self.terminal_launcher = terminal_launcher
self._process: Optional[subprocess.Popen] = None
def run(self):
helper_cmd = list(self.helper_argv) + [
"--result-file", str(self.result_file),
"--request-id", self.request_id,
"--harness", self.harness_id,
]
if self.terminal_launcher:
full_cmd = list(self.terminal_launcher) + helper_cmd
else:
full_cmd = helper_cmd
try:
process = subprocess.Popen(
self.command,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
self._process = subprocess.Popen(
full_cmd,
stdout=subprocess.PIPE if not self.terminal_launcher else None,
stderr=subprocess.STDOUT if not self.terminal_launcher else None,
text=True,
bufsize=1
)
for line in process.stdout:
if not self.terminal_launcher and self._process.stdout:
for line in self._process.stdout:
self.output.emit(line)
process.wait()
exit_code = self._process.wait()
if process.returncode == 0:
self.finished.emit(True, "Installation completed successfully!")
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 {})
elif exit_code in (10, 130, 143) or status == "cancelled":
self.finished.emit(False, "Setup was cancelled by user.", result_data or {})
else:
self.finished.emit(False, f"Installation failed (exit code {process.returncode})")
diag = result_data.get("error_diagnostics") if result_data else None
msg = f"Setup failed: {diag}" if diag else f"Setup process exited with code {exit_code}."
self.finished.emit(False, msg, result_data or {})
except Exception as e:
self.finished.emit(False, f"Error: {e}")
self.finished.emit(False, f"Error launching setup helper: {e}", {})
def terminate_process(self):
"""Terminate the active setup process if running."""
if self._process and self._process.poll() is None:
try:
self._process.terminate()
except Exception:
pass
# Alias for backwards compatibility
InstallWorker = SetupTerminalWorker
class ProviderInstallDialog(QDialog):
"""Dialog for installing AI provider CLI tools."""
"""Dialog for installing AI provider CLI tools via agent-harness-setup."""
def __init__(self, parent=None):
super().__init__(parent)
@@ -132,8 +207,9 @@ class ProviderInstallDialog(QDialog):
self.setMinimumSize(700, 500)
self.resize(750, 550)
self._worker = None
self._selected_provider = None
self._worker: Optional[SetupTerminalWorker] = None
self._selected_provider: Optional[str] = None
self._temp_dir: Optional[str] = None
self._setup_ui()
self._load_providers()
@@ -172,7 +248,8 @@ class ProviderInstallDialog(QDialog):
layout.addWidget(title)
desc = QLabel(
"CmdForge works with various AI providers. Select one below to install it. "
"CmdForge works with various AI providers. Select one below to install and configure it. "
"Installation is performed interactively via the Agent Harness Setup helper. "
"Providers marked as [INSTALLED] are already available on your system."
)
desc.setWordWrap(True)
@@ -314,11 +391,8 @@ class ProviderInstallDialog(QDialog):
def _load_providers(self):
"""Load and display available providers."""
# Get currently configured providers
configured = {p.name for p in load_providers()}
# Check what's installed on the system
for key, info in PROVIDER_INSTALL_INFO.items():
self.provider_list.clear()
for key, info in SUPPORTED_PROVIDERS.items():
item = QListWidgetItem()
# Check if CLI tool is installed
@@ -337,9 +411,6 @@ class ProviderInstallDialog(QDialog):
def _check_installed(self, provider_key: str) -> bool:
"""Check if a provider's CLI tool is installed."""
info = PROVIDER_INSTALL_INFO[provider_key]
# Check common locations based on provider
if provider_key == "opencode":
opencode_path = Path.home() / ".opencode" / "bin" / "opencode"
return opencode_path.exists() or shutil.which("opencode") is not None
@@ -354,30 +425,29 @@ class ProviderInstallDialog(QDialog):
"""Handle provider selection."""
key = item.data(Qt.UserRole)
is_installed = item.data(Qt.UserRole + 1)
info = PROVIDER_INSTALL_INFO[key]
info = SUPPORTED_PROVIDERS[key]
self._selected_provider = key
self.btn_install.setEnabled(True)
# Update details
install_method = info['install_cmd'] or "Manual download"
is_manual = info.get("manual_only", False)
install_method = "Manual desktop setup" if is_manual else "Agent Harness Setup Helper (terminal)"
details = f"""<b>{info['group']}</b><br><br>
<b>Cost:</b> {info['cost']}<br>
<b>Requirements:</b> {info['requires']}<br>
<b>Install method:</b> <code>{install_method}</code><br>
<b>After install:</b> {info['setup']}<br><br>
<b>Available models:</b> {', '.join(v[0] for v in info['variants'])}"""
if is_installed:
details += "<br><br><span style='color: green;'>This provider is already installed. Click Install to reinstall or update.</span>"
details += "<br><br><span style='color: green;'>This provider is already installed. Click Reinstall to run setup again.</span>"
self.details_label.setText(details)
# Change button text based on installed status
if info["install_cmd"]:
self.btn_install.setText("Reinstall" if is_installed else "Install Selected")
else:
if is_manual:
self.btn_install.setText("Add Providers" if is_installed else "Setup Instructions")
else:
self.btn_install.setText("Reinstall" if is_installed else "Install Selected")
def _on_provider_double_clicked(self, item: QListWidgetItem):
"""Handle double-click to start install."""
@@ -389,9 +459,9 @@ class ProviderInstallDialog(QDialog):
if not self._selected_provider:
return
info = PROVIDER_INSTALL_INFO[self._selected_provider]
info = SUPPORTED_PROVIDERS[self._selected_provider]
if not info["install_cmd"]:
if info.get("manual_only"):
if self._check_installed(self._selected_provider):
self._populate_variants()
self.stack.setCurrentIndex(2)
@@ -399,30 +469,65 @@ class ProviderInstallDialog(QDialog):
QMessageBox.information(self, "Manual Setup", info["setup"])
return
helper_argv = _find_setup_helper()
if not helper_argv:
QMessageBox.warning(
self,
"Setup Helper Missing",
"Agent Harness Setup Helper ('agent-harness-setup') was not found on your PATH or in active environments.\n\n"
"Please install the helper or run manually:\n"
" pip install -e /home/rob/PycharmProjects/agent-harness-setup\n\n"
"Or run directly:\n"
" PYTHONPATH=/home/rob/PycharmProjects/agent-harness-setup/src python3 -m agent_harness_setup"
)
return
# Prepare secure temporary directory and result file
self._temp_dir = tempfile.mkdtemp(prefix="cmdforge-harness-gui-")
try:
os.chmod(self._temp_dir, 0o700)
except OSError:
pass
result_file = Path(self._temp_dir) / "result.json"
request_id = str(uuid.uuid4())
harness_id = info.get("harness_id", self._selected_provider)
# Switch to install page
self.stack.setCurrentIndex(1)
self.install_title.setText(f"Installing {info['group']}...")
self.install_cmd_label.setText(f"Command: {info['install_cmd']}")
self.install_cmd_label.setText(f"Helper: agent-harness-setup --harness {harness_id}")
self.output_text.clear()
self.progress.setRange(0, 0)
self.btn_back.setEnabled(False)
self.btn_next.setEnabled(False)
# Start installation
self._worker = InstallWorker(info['install_cmd'])
term_launcher = find_terminal_emulator()
if term_launcher:
self.output_text.append(f"Dedicated terminal window launched ({Path(term_launcher[0]).name}).")
self.output_text.append("Please complete the preflight and approval steps in that window.\n")
else:
self.output_text.append("No graphical terminal emulator detected; running in background process.\n")
self._worker = SetupTerminalWorker(
helper_argv=helper_argv,
harness_id=harness_id,
result_file=result_file,
request_id=request_id,
terminal_launcher=term_launcher,
)
self._worker.output.connect(self._on_install_output)
self._worker.finished.connect(self._on_install_finished)
self._worker.finished.connect(self._on_setup_finished)
self._worker.start()
def _on_install_output(self, text: str):
"""Handle installation output."""
self.output_text.append(text.rstrip())
# Auto-scroll to bottom
scrollbar = self.output_text.verticalScrollBar()
scrollbar.setValue(scrollbar.maximum())
def _on_install_finished(self, success: bool, message: str):
"""Handle installation completion."""
def _on_setup_finished(self, success: bool, message: str, result_data: dict):
"""Handle setup helper process completion."""
self.progress.setRange(0, 1)
self.progress.setValue(1 if success else 0)
self.btn_back.setEnabled(True)
@@ -430,18 +535,39 @@ class ProviderInstallDialog(QDialog):
if success:
self.install_title.setText("Installation Complete!")
self.output_text.append(f"\n{'='*50}\n{message}")
# Advisory PATH updates
path_updates = result_data.get("path_updates") or []
if path_updates:
cur_path = os.environ.get("PATH", "")
new_dirs = [
str(Path(os.path.expanduser(p)))
for p in path_updates
if Path(os.path.expanduser(p)).is_dir() and str(Path(os.path.expanduser(p))) not in cur_path
]
if new_dirs:
os.environ["PATH"] = ":".join(new_dirs) + ":" + cur_path
# Update installed status for provider item in list
for i in range(self.provider_list.count()):
item = self.provider_list.item(i)
if item.data(Qt.UserRole) == self._selected_provider:
item.setData(Qt.UserRole + 1, True)
info = SUPPORTED_PROVIDERS[self._selected_provider]
item.setText(f"{info['group']} [INSTALLED]")
item.setForeground(Qt.darkGreen)
break
self.btn_next.setEnabled(True)
# Populate variants page
self._populate_variants()
self.stack.setCurrentIndex(2)
else:
self.install_title.setText("Installation Failed")
self.output_text.append(f"\n{'='*50}\nERROR: {message}")
self.install_title.setText("Installation Incomplete")
self.output_text.append(f"\n{'='*50}\n{message}")
QMessageBox.warning(
self, "Installation Failed",
f"The installation did not complete successfully.\n\n{message}\n\n"
"You may need to install manually. Check the output for details."
self,
"Installation Incomplete",
f"{message}\n\nYou can inspect the logs or retry.",
)
def _populate_variants(self):
@@ -449,7 +575,7 @@ class ProviderInstallDialog(QDialog):
if not self._selected_provider:
return
info = PROVIDER_INSTALL_INFO[self._selected_provider]
info = SUPPORTED_PROVIDERS[self._selected_provider]
self.variants_list.clear()
# Get currently configured provider names
@@ -458,6 +584,7 @@ class ProviderInstallDialog(QDialog):
for name, command, description in info['variants']:
item = QListWidgetItem(f"{name} - {description}")
item.setData(Qt.UserRole, (name, command, description))
self.variants_list.addItem(item)
# Pre-select if not already configured
if name not in configured:
@@ -466,8 +593,6 @@ class ProviderInstallDialog(QDialog):
item.setForeground(Qt.gray)
item.setText(f"{name} - {description} [already configured]")
self.variants_list.addItem(item)
# Update next steps
self.next_steps_label.setText(
f"<b>Important:</b> Before using these providers, you need to authenticate:<br><br>"
@@ -497,3 +622,14 @@ class ProviderInstallDialog(QDialog):
)
self.accept()
def closeEvent(self, event):
"""Clean up worker and temporary resources on close."""
if self._worker and self._worker.isRunning():
self._worker.terminate_process()
self._worker.wait(2000)
if self._temp_dir and Path(self._temp_dir).is_dir():
shutil.rmtree(self._temp_dir, ignore_errors=True)
super().closeEvent(event)
+3
View File
@@ -426,8 +426,11 @@ class ProviderExecutionPolicy:
data_classification: str = "public"
require_model_identity: bool = False
require_model_digest: bool = False
provider_timeout: Optional[int] = None # the caller's limit in seconds; overrides a step's own timeout
def __post_init__(self) -> None:
if self.provider_timeout is not None and self.provider_timeout <= 0:
raise ValueError("provider_timeout must be a positive number of seconds")
if self.required_locality not in {None, "local", "remote"}:
raise ValueError("required_locality must be local, remote, or null")
if self.data_classification not in {"public", "internal", "private"}:
+7 -2
View File
@@ -559,6 +559,8 @@ def execute_prompt_step(
# Determine provider
provider = provider_override or substitute_variables(step.provider, variables, warn_non_scalar=verbose)
# How long the provider may take: the caller's --provider-timeout, else the step's timeout, else 300 s
step_timeout = effective_policy.provider_timeout or step.timeout or 300
# Build context from the inside out so the final order is:
# profile system prompt -> skills -> user prompt.
@@ -580,7 +582,7 @@ def execute_prompt_step(
result = mock_provider(prompt)
else:
result = call_provider(
provider, prompt, max_tokens=step.max_tokens,
provider, prompt, timeout=step_timeout, max_tokens=step.max_tokens,
execution_policy=effective_policy,
)
@@ -629,7 +631,7 @@ Please try again with valid JSON matching the schema exactly."""
result = mock_provider(current_prompt)
else:
result = call_provider(
provider, current_prompt, max_tokens=step.max_tokens,
provider, current_prompt, timeout=step_timeout, max_tokens=step.max_tokens,
execution_policy=effective_policy,
)
@@ -1186,6 +1188,8 @@ def create_argument_parser(tool: Tool) -> argparse.ArgumentParser:
parser.add_argument("--require-model-identity", action="store_true")
parser.add_argument("--require-model-digest", action="store_true")
parser.add_argument("--result-envelope", choices=["json"])
parser.add_argument("--provider-timeout", type=int,
help="Seconds each AI provider call may take (overrides the tool's own step timeouts)")
# Tool-specific flags from arguments
for arg in tool.arguments:
@@ -1315,6 +1319,7 @@ def main():
data_classification=args.data_classification or "public",
require_model_identity=args.require_model_identity,
require_model_digest=args.require_model_digest,
provider_timeout=args.provider_timeout,
)
provider_calls = []
+12
View File
@@ -188,11 +188,14 @@ class PromptStep:
max_tokens: Optional[int] = None # Max output tokens (provider-dependent)
skills: Optional[List[str]] = None # Skill names to enable for this step
fallback_policy: str = "allow" # "allow" | "deny"
timeout: Optional[int] = None # Seconds the provider may take (None: the default 300; --provider-timeout wins)
def __post_init__(self) -> None:
_validate_skill_selection(self.skills)
if self.fallback_policy not in {"allow", "deny"}:
raise ValueError("PromptStep fallback_policy must be allow or deny")
if self.timeout is not None and self.timeout <= 0:
raise ValueError("PromptStep timeout must be a positive number of seconds")
def to_dict(self) -> dict:
d = {
@@ -221,6 +224,8 @@ class PromptStep:
d["skills"] = self.skills
if self.fallback_policy != "allow":
d["fallback_policy"] = self.fallback_policy
if self.timeout is not None:
d["timeout"] = self.timeout
return d
@classmethod
@@ -233,6 +238,12 @@ class PromptStep:
raise ValueError("max_tokens must be an integer")
if max_tokens <= 0 or max_tokens > 1_000_000:
raise ValueError("max_tokens must be between 1 and 1000000")
timeout = data.get("timeout")
if timeout is not None:
try:
timeout = int(timeout)
except (TypeError, ValueError):
raise ValueError("timeout must be a whole number of seconds")
return cls(
prompt=data["prompt"],
@@ -248,6 +259,7 @@ class PromptStep:
max_tokens=max_tokens,
skills=data.get("skills"),
fallback_policy=data.get("fallback_policy", "allow"),
timeout=timeout,
)
+2
View File
@@ -29,6 +29,7 @@ from cmdforge.registry import app as registry_app
from . import web_bp
from .auth import login, register, logout
from .donations import load_donation_options
from .filters import register_filters
from .forum import forum_bp
from .sessions import SQLiteSessionInterface, cleanup_expired_sessions
@@ -48,6 +49,7 @@ def create_web_app() -> Flask:
app.config["SESSION_COOKIE_NAME"] = "cmdforge_session"
app.config["SESSION_COOKIE_SECURE"] = os.environ.get("CMDFORGE_ENV") == "production"
app.config["SHOW_ADS"] = os.environ.get("CMDFORGE_SHOW_ADS", "").lower() == "true"
app.config["DONATION_OPTIONS"] = load_donation_options(os.environ)
# CSRF token generator
app.config["CSRF_GENERATOR"] = lambda: secrets.token_urlsafe(32)
+69
View File
@@ -0,0 +1,69 @@
"""Validation and presentation helpers for public donation settings."""
from __future__ import annotations
import re
from dataclasses import dataclass
from typing import Mapping
from urllib.parse import urlencode, urlparse
_BASE58_BITCOIN_ADDRESS = re.compile(r"^[13][a-km-zA-HJ-NP-Z1-9]{25,34}$")
_BECH32_BITCOIN_ADDRESS = re.compile(r"^bc1[ac-hj-np-z02-9]{11,71}$")
_PAYPAL_HOSTS = {"paypal.com", "www.paypal.com", "paypal.me", "www.paypal.me"}
@dataclass(frozen=True)
class DonationOptions:
"""Public donation methods safe to render as links."""
bitcoin_address: str = ""
bitcoin_uri: str = ""
paypal_url: str = ""
@property
def enabled(self) -> bool:
return bool(self.bitcoin_address or self.paypal_url)
def _valid_bitcoin_address(value: str) -> bool:
if _BASE58_BITCOIN_ADDRESS.fullmatch(value):
return True
# Bech32 addresses may be all lower- or uppercase, but never mixed case.
if value.lower() != value and value.upper() != value:
return False
return bool(_BECH32_BITCOIN_ADDRESS.fullmatch(value.lower()))
def _valid_paypal_url(value: str) -> bool:
parsed = urlparse(value)
return (
parsed.scheme == "https"
and parsed.hostname is not None
and parsed.hostname.lower() in _PAYPAL_HOSTS
and not parsed.username
and not parsed.password
)
def load_donation_options(environment: Mapping[str, str]) -> DonationOptions:
"""Load public donation details, ignoring malformed environment values."""
bitcoin_address = environment.get("CMDFORGE_DONATION_BITCOIN_ADDRESS", "").strip()
if not _valid_bitcoin_address(bitcoin_address):
bitcoin_address = ""
paypal_url = environment.get("CMDFORGE_DONATION_PAYPAL_URL", "").strip()
if not _valid_paypal_url(paypal_url):
paypal_url = ""
bitcoin_uri = ""
if bitcoin_address:
query = urlencode({"label": "CmdForge", "message": "Support CmdForge development"})
bitcoin_uri = f"bitcoin:{bitcoin_address}?{query}"
return DonationOptions(
bitcoin_address=bitcoin_address,
bitcoin_uri=bitcoin_uri,
paypal_url=paypal_url,
)
+4 -1
View File
@@ -876,7 +876,10 @@ def philosophy():
@web_bp.route("/donate", endpoint="donate")
def donate():
return render_template("pages/donate.html")
return render_template(
"pages/donate.html",
donations=current_app.config["DONATION_OPTIONS"],
)
@web_bp.route("/privacy", endpoint="privacy")
@@ -9,8 +9,8 @@
Provider-agnostic, composable, and community-driven.
</p>
<div class="mt-4 flex space-x-4">
<a href="https://github.com/rob/cmdforge" target="_blank" rel="noopener noreferrer"
class="text-gray-400 hover:text-white transition-colors" aria-label="GitHub">
<a href="https://gitea.brrd.tech/rob/CmdForge" target="_blank" rel="noopener noreferrer"
class="text-gray-400 hover:text-white transition-colors" aria-label="CmdForge source on Gitea">
<svg class="w-6 h-6" fill="currentColor" viewBox="0 0 24 24">
<path fill-rule="evenodd" d="M12 2C6.477 2 2 6.484 2 12.017c0 4.425 2.865 8.18 6.839 9.504.5.092.682-.217.682-.483 0-.237-.008-.868-.013-1.703-2.782.605-3.369-1.343-3.369-1.343-.454-1.158-1.11-1.466-1.11-1.466-.908-.62.069-.608.069-.608 1.003.07 1.531 1.032 1.531 1.032.892 1.53 2.341 1.088 2.91.832.092-.647.35-1.088.636-1.338-2.22-.253-4.555-1.113-4.555-4.951 0-1.093.39-1.988 1.029-2.688-.103-.253-.446-1.272.098-2.65 0 0 .84-.27 2.75 1.026A9.564 9.564 0 0112 6.844c.85.004 1.705.115 2.504.337 1.909-1.296 2.747-1.027 2.747-1.027.546 1.379.202 2.398.1 2.651.64.7 1.028 1.595 1.028 2.688 0 3.848-2.339 4.695-4.566 4.943.359.309.678.92.678 1.855 0 1.338-.012 2.419-.012 2.747 0 .268.18.58.688.482A10.019 10.019 0 0022 12.017C22 6.484 17.522 2 12 2z" clip-rule="evenodd"/>
</svg>
+1 -1
View File
@@ -137,7 +137,7 @@
<svg class="w-5 h-5 mr-2" fill="currentColor" viewBox="0 0 24 24">
<path fill-rule="evenodd" d="M12 2C6.477 2 2 6.484 2 12.017c0 4.425 2.865 8.18 6.839 9.504.5.092.682-.217.682-.483 0-.237-.008-.868-.013-1.703-2.782.605-3.369-1.343-3.369-1.343-.454-1.158-1.11-1.466-1.11-1.466-.908-.62.069-.608.069-.608 1.003.07 1.531 1.032 1.531 1.032.892 1.53 2.341 1.088 2.91.832.092-.647.35-1.088.636-1.338-2.22-.253-4.555-1.113-4.555-4.951 0-1.093.39-1.988 1.029-2.688-.103-.253-.446-1.272.098-2.65 0 0 .84-.27 2.75 1.026A9.564 9.564 0 0112 6.844c.85.004 1.705.115 2.504.337 1.909-1.296 2.747-1.027 2.747-1.027.546 1.379.202 2.398.1 2.651.64.7 1.028 1.595 1.028 2.688 0 3.848-2.339 4.695-4.566 4.943.359.309.678.92.678 1.855 0 1.338-.012 2.419-.012 2.747 0 .268.18.58.688.482A10.019 10.019 0 0022 12.017C22 6.484 17.522 2 12 2z" clip-rule="evenodd"/>
</svg>
View on GitHub
View on Gitea
</a>
</div>
</div>
+2 -2
View File
@@ -122,7 +122,7 @@
</div>
</article>
<!-- Edit on GitHub -->
<!-- Edit on Gitea -->
<div class="mt-6 text-center">
<a href="https://gitea.brrd.tech/rob/CmdForge/edit/main/docs/{{ current_path }}.md"
target="_blank"
@@ -131,7 +131,7 @@
<svg class="w-4 h-4 mr-1" fill="currentColor" viewBox="0 0 20 20">
<path d="M13.586 3.586a2 2 0 112.828 2.828l-.793.793-2.828-2.828.793-.793zM11.379 5.793L3 14.172V17h2.828l8.38-8.379-2.83-2.828z"/>
</svg>
Edit this page on GitHub
Edit this page on Gitea
</a>
</div>
</main>
+83 -2
View File
@@ -2,7 +2,7 @@
{% block title %}Support CmdForge{% endblock %}
{% block meta_description %}Support CmdForge by contributing code, tools, or feedback.{% endblock %}
{% block meta_description %}Support the open-source CmdForge project with Bitcoin, PayPal, tools, code, or feedback.{% endblock %}
{% block content %}
<div class="bg-gray-50 min-h-screen">
@@ -11,13 +11,76 @@
<div class="max-w-4xl mx-auto px-4 sm:px-6 lg:px-8 py-12 text-center">
<h1 class="text-3xl font-bold text-gray-900">Support CmdForge</h1>
<p class="mt-4 text-lg text-gray-600">
CmdForge is a passion project. The best way to support it is to use it and share what you build.
Help keep an independent, open-source toolkit growing. Every contribution supports hosting,
maintenance, documentation, and the next generation of tools.
</p>
</div>
</div>
<div class="max-w-5xl mx-auto px-4 sm:px-6 lg:px-8 py-12">
{% if donations.enabled %}
<section class="mb-12" aria-labelledby="financial-support-heading">
<div class="text-center mb-8">
<h2 id="financial-support-heading" class="text-2xl font-bold text-gray-900">Make a Contribution</h2>
<p class="mt-3 text-gray-600">
Choose the method that works for you. Donations are voluntary and do not unlock product features.
</p>
</div>
<div class="grid grid-cols-1 {% if donations.bitcoin_address and donations.paypal_url %}md:grid-cols-2{% endif %} gap-6 max-w-4xl mx-auto">
{% if donations.bitcoin_address %}
<article class="bg-white rounded-xl border border-amber-200 p-7 shadow-sm">
<div class="flex items-center gap-3">
<div class="w-11 h-11 bg-amber-100 rounded-full flex items-center justify-center text-amber-700 font-bold text-xl">&#8383;</div>
<div>
<h3 class="text-xl font-semibold text-gray-900">Bitcoin</h3>
<p class="text-sm text-gray-500">Send BTC directly to the project wallet.</p>
</div>
</div>
<label for="bitcoin-address" class="block mt-6 text-sm font-medium text-gray-700">Bitcoin address</label>
<div class="mt-2 flex flex-col sm:flex-row gap-2">
<code id="bitcoin-address" class="min-w-0 flex-1 px-3 py-3 bg-gray-50 border border-gray-200 rounded-md text-xs text-gray-800 break-all select-all">{{ donations.bitcoin_address }}</code>
<button type="button" id="copy-bitcoin-address"
class="px-4 py-2 text-sm font-medium text-amber-800 border border-amber-300 rounded-md hover:bg-amber-50"
data-address="{{ donations.bitcoin_address }}">
Copy
</button>
</div>
<p id="copy-bitcoin-status" class="mt-2 text-sm text-green-700 min-h-5" role="status" aria-live="polite"></p>
<a href="{{ donations.bitcoin_uri }}"
class="mt-2 inline-flex items-center justify-center w-full px-5 py-3 text-sm font-medium text-white bg-amber-600 rounded-md hover:bg-amber-700">
Open Bitcoin Wallet
</a>
<p class="mt-3 text-xs text-gray-500">Always verify the full address in your wallet before sending.</p>
</article>
{% endif %}
{% if donations.paypal_url %}
<article class="bg-white rounded-xl border border-blue-200 p-7 shadow-sm flex flex-col">
<div class="flex items-center gap-3">
<div class="w-11 h-11 bg-blue-100 rounded-full flex items-center justify-center text-blue-700 font-bold text-xl">P</div>
<div>
<h3 class="text-xl font-semibold text-gray-900">PayPal</h3>
<p class="text-sm text-gray-500">Contribute with PayPal or a supported card.</p>
</div>
</div>
<p class="mt-6 text-gray-600 flex-1">
PayPal handles the payment securely on its own website. CmdForge never receives your payment credentials.
</p>
<a href="{{ donations.paypal_url }}" target="_blank" rel="noopener noreferrer"
class="mt-6 inline-flex items-center justify-center w-full px-5 py-3 text-sm font-medium text-white bg-blue-600 rounded-md hover:bg-blue-700">
Donate with PayPal
</a>
</article>
{% endif %}
</div>
</section>
{% endif %}
<!-- Ways to help -->
<div class="text-center mb-8">
<h2 class="text-2xl font-bold text-gray-900">Other Ways to Help</h2>
<p class="mt-3 text-gray-600">Money is never the only way to make a meaningful contribution.</p>
</div>
<section class="grid grid-cols-1 md:grid-cols-3 gap-6 mb-12">
<div class="bg-white rounded-lg border border-gray-200 p-6">
<div class="w-12 h-12 bg-indigo-100 rounded-lg flex items-center justify-center mb-4">
@@ -85,3 +148,21 @@
</div>
</div>
{% endblock %}
{% block scripts %}
{{ super() }}
{% if donations.bitcoin_address %}
<script>
document.getElementById('copy-bitcoin-address')?.addEventListener('click', async function () {
const status = document.getElementById('copy-bitcoin-status');
try {
await navigator.clipboard.writeText(this.dataset.address);
status.textContent = 'Address copied.';
this.textContent = 'Copied';
} catch (error) {
status.textContent = 'Select the address above and copy it manually.';
}
});
</script>
{% endif %}
{% endblock %}
@@ -122,7 +122,7 @@
</p>
<ul>
<li>Email: [email protected]</li>
<li>GitHub: <a href="https://gitea.brrd.tech/rob/CmdForge/issues">Open an issue</a></li>
<li>Gitea: <a href="https://gitea.brrd.tech/rob/CmdForge/issues">Open an issue</a></li>
</ul>
</div>
</div>
+1 -1
View File
@@ -150,7 +150,7 @@
</p>
<ul>
<li>Email: [email protected]</li>
<li>GitHub: <a href="https://gitea.brrd.tech/rob/CmdForge/issues">Open an issue</a></li>
<li>Gitea: <a href="https://gitea.brrd.tech/rob/CmdForge/issues">Open an issue</a></li>
</ul>
</div>
</div>
+50
View File
@@ -216,6 +216,20 @@ class TestRunOnceCommand:
assert call_provider.call_args.args[1] == "Summarize this\n\ndocument body"
def test_timeout_reaches_provider_call(self):
result = SimpleNamespace(success=True, text="ok", error=None)
with (
patch('sys.argv', [
'cmdforge', 'run-once', 'hello', '--provider', 'mock',
'--timeout', '145',
]),
patch('sys.stdin', StringIO()),
patch('cmdforge.providers.call_provider', return_value=result) as call_provider,
):
assert main() == 0
assert call_provider.call_args.kwargs["timeout"] == 145
def test_uses_configured_default_provider(self):
result = SimpleNamespace(success=True, text="ok", error=None)
config = SimpleNamespace(default_provider="configured-provider")
@@ -294,6 +308,42 @@ class TestRunOnceCommand:
assert call["fallback_used"] is False
class TestRunCommand:
def test_provider_timeout_reaches_prompt_step_provider_call(self):
from cmdforge.providers import ProviderResult
tool = Tool(
name="timeout-tool",
steps=[PromptStep(
prompt="Process {input}",
provider="mock",
output_var="result",
plain_text=True,
timeout=900,
)],
output="{result}",
)
stdin = MagicMock()
stdin.isatty.return_value = True
with (
patch('sys.argv', [
'cmdforge', 'run', 'timeout-tool', '--provider-timeout', '145',
]),
patch('sys.stdin', stdin),
patch(
'cmdforge.cli.tool_commands._resolve_existing_tool',
return_value=tool,
),
patch(
'cmdforge.runner.call_provider',
return_value=ProviderResult(text="ok", success=True),
) as call_provider,
):
assert main() == 0
assert call_provider.call_args.kwargs["timeout"] == 145
class TestCreateCommand:
"""Tests for 'cmdforge create' command."""
+462
View File
@@ -0,0 +1,462 @@
"""Acceptance tests for Packet S6: CmdForge GUI Provider Install Dialog reconciliation."""
import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
os.environ["QT_QPA_PLATFORM"] = "offscreen"
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QApplication, QMessageBox
# Ensure single application instance for tests
_qapp = QApplication.instance()
if _qapp is None:
_qapp = QApplication(["cmdforge-gui-tests", "-platform", "offscreen"])
from cmdforge.gui.dialogs.provider_install_dialog import (
PROVIDER_INSTALL_INFO,
SUPPORTED_PROVIDERS,
ProviderInstallDialog,
SetupTerminalWorker,
find_terminal_emulator,
)
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"
with patch("cmdforge.providers.PROVIDERS_FILE", providers_file):
save_providers([])
yield providers_file
def test_supported_providers_catalogue_reconciliation():
"""Verify that duplicate raw shell install commands and requirements are retired."""
assert "claude" in SUPPORTED_PROVIDERS
assert "codex" in SUPPORTED_PROVIDERS
assert "agy" in SUPPORTED_PROVIDERS
assert "opencode" in SUPPORTED_PROVIDERS
assert "crush" in SUPPORTED_PROVIDERS
assert "ollama" in SUPPORTED_PROVIDERS
# Ensure no entry has raw install_cmd shell scripts or requires strings
for name, info in SUPPORTED_PROVIDERS.items():
assert "install_cmd" not in info, f"Provider {name} still contains raw install_cmd"
assert "requires" not in info, f"Provider {name} still contains requires"
assert "group" in info
assert "setup" in info
assert "cost" in info
assert "variants" in info
assert len(info["variants"]) > 0
# Ensure backwards compatibility alias
assert PROVIDER_INSTALL_INFO is SUPPORTED_PROVIDERS
def test_find_terminal_emulator():
"""Verify terminal emulator detection logic."""
with patch("shutil.which", side_effect=lambda cmd: "/usr/bin/gnome-terminal" if cmd == "gnome-terminal" else None):
launcher = find_terminal_emulator()
assert launcher == ["/usr/bin/gnome-terminal", "--"]
with patch("shutil.which", side_effect=lambda cmd: "/usr/bin/x-terminal-emulator" if cmd == "x-terminal-emulator" else None):
launcher = find_terminal_emulator()
assert launcher == ["/usr/bin/x-terminal-emulator", "-e"]
with patch("shutil.which", return_value=None):
launcher = find_terminal_emulator()
assert launcher is None
def test_dialog_init_and_provider_selection(temp_providers_file):
"""Verify dialog initialization and detail rendering for automated and manual providers."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
assert dialog.provider_list.count() == len(SUPPORTED_PROVIDERS)
assert not dialog.btn_install.isEnabled()
# Select Claude (automated setup helper)
claude_item = None
for i in range(dialog.provider_list.count()):
item = dialog.provider_list.item(i)
if item.data(Qt.UserRole) == "claude":
claude_item = item
break
assert claude_item is not None
dialog._on_provider_selected(claude_item)
assert dialog._selected_provider == "claude"
assert dialog.btn_install.isEnabled()
assert dialog.btn_install.text() == "Install Selected"
assert "Agent Harness Setup Helper (terminal)" in dialog.details_label.text()
assert "Anthropic Claude" in dialog.details_label.text()
# Select AGY (manual setup instructions)
agy_item = None
for i in range(dialog.provider_list.count()):
item = dialog.provider_list.item(i)
if item.data(Qt.UserRole) == "agy":
agy_item = item
break
assert agy_item is not None
dialog._on_provider_selected(agy_item)
assert dialog._selected_provider == "agy"
assert dialog.btn_install.isEnabled()
assert dialog.btn_install.text() == "Setup Instructions"
assert "Manual desktop setup" in dialog.details_label.text()
dialog.close()
def test_start_install_missing_helper_shows_warning(temp_providers_file):
"""Verify missing setup helper displays warning and does not advance page."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
dialog._selected_provider = "claude"
with (
patch("cmdforge.gui.dialogs.provider_install_dialog._find_setup_helper", return_value=None),
patch.object(QMessageBox, "warning") as mock_warn,
):
dialog._start_install()
assert mock_warn.called
assert "Agent Harness Setup Helper ('agent-harness-setup') was not found" in mock_warn.call_args[0][2]
# Verify dialog stayed on Page 0 (Selection)
assert dialog.stack.currentIndex() == 0
dialog.close()
def test_setup_terminal_worker_success(tmp_path):
"""Test SetupTerminalWorker success path with valid result correlation."""
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",
"path_updates": ["/custom/bin"],
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
_write_completion(result_file, data)
def wait(self):
return 0
def poll(self):
return 0
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
terminal_launcher=["/usr/bin/xterm", "-e"],
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, res_dict = results[0]
assert success is True
assert "verified successfully" in msg
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"
data = {
"status": "success",
"request_id": "wrong-uuid",
"harness_id": "claude-code",
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
_write_completion(result_file, data)
def wait(self):
return 0
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id="expected-uuid",
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "does not match the expected" in msg
def test_setup_terminal_worker_incoherent_exit(tmp_path):
"""Test SetupTerminalWorker rejects exit code 1 with status 'success'."""
result_file = tmp_path / "result.json"
req_id = "test-uuid"
data = {
"status": "success",
"request_id": req_id,
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
_write_completion(result_file, data)
def wait(self):
return 1
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "exit code 1" in msg
def test_setup_terminal_worker_cancelled(tmp_path):
"""Test SetupTerminalWorker handles user cancellation cleanly."""
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,
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
_write_completion(result_file, data)
def wait(self):
return 10
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "cancelled" in msg.lower()
def test_setup_terminal_worker_failure_with_diagnostics(tmp_path):
"""Test SetupTerminalWorker handles failure with error diagnostics."""
result_file = tmp_path / "result.json"
req_id = "fail-uuid"
data = {
"status": "failed",
"request_id": req_id,
"error_diagnostics": "Missing npm dependency",
}
class MockProcess:
def __init__(self, *args, **kwargs):
self.stdout = None
_write_completion(result_file, data)
def wait(self):
return 1
worker = SetupTerminalWorker(
helper_argv=["/bin/agent-harness-setup"],
harness_id="claude-code",
result_file=result_file,
request_id=req_id,
)
results = []
worker.finished.connect(lambda s, m, d: results.append((s, m, d)))
with patch("subprocess.Popen", side_effect=MockProcess):
worker.run()
assert len(results) == 1
success, msg, _ = results[0]
assert success is False
assert "Missing npm dependency" in msg
def test_dialog_full_flow_success_to_variants_and_add(temp_providers_file, tmp_path):
"""Test full dialog flow: install succeeds, advances to variants, registers selected variants."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
dialog._selected_provider = "claude"
mock_worker = MagicMock()
with (
patch("cmdforge.gui.dialogs.provider_install_dialog._find_setup_helper", return_value=["/bin/agent-harness-setup"]),
patch("cmdforge.gui.dialogs.provider_install_dialog.SetupTerminalWorker", return_value=mock_worker),
):
dialog._start_install()
assert dialog.stack.currentIndex() == 1 # Install page
assert "Installing Anthropic Claude" in dialog.install_title.text()
# Create dummy directory for path update test
custom_bin = tmp_path / "bin"
custom_bin.mkdir(parents=True, exist_ok=True)
result_data = {
"status": "success",
"request_id": "any",
"harness_id": "claude-code",
"path_updates": [str(custom_bin)],
}
dialog._on_setup_finished(True, "Setup verified!", result_data)
assert str(custom_bin) in os.environ.get("PATH", "")
assert dialog.stack.currentIndex() == 2
assert dialog.variants_list.count() == len(SUPPORTED_PROVIDERS["claude"]["variants"])
selected_items = [dialog.variants_list.item(i) for i in range(dialog.variants_list.count()) if dialog.variants_list.item(i).isSelected()]
assert len(selected_items) > 0
with patch.object(QMessageBox, "information") as mock_info:
dialog._add_selected_variants()
assert mock_info.called
provider = get_provider("claude")
assert provider is not None
assert provider.command == "claude -p"
dialog.close()
def test_dialog_failure_flow_shows_warning(temp_providers_file):
"""Test dialog failure flow warns user and remains on install page."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
dialog._selected_provider = "claude"
mock_worker = MagicMock()
with (
patch("cmdforge.gui.dialogs.provider_install_dialog._find_setup_helper", return_value=["/bin/agent-harness-setup"]),
patch("cmdforge.gui.dialogs.provider_install_dialog.SetupTerminalWorker", return_value=mock_worker),
):
dialog._start_install()
with patch.object(QMessageBox, "warning") as mock_warn:
dialog._on_setup_finished(False, "Compilation error in installer", {})
assert mock_warn.called
assert "Compilation error in installer" in mock_warn.call_args[0][2]
assert dialog.stack.currentIndex() == 1
assert "Installation Incomplete" in dialog.install_title.text()
assert dialog.btn_back.isEnabled()
dialog.close()
def test_dialog_close_cleans_up_resources(temp_providers_file):
"""Test closing dialog terminates active worker and removes temporary directory."""
with patch("shutil.which", return_value=None):
dialog = ProviderInstallDialog()
temp_dir = tempfile.mkdtemp(prefix="test-cleanup-")
dialog._temp_dir = temp_dir
mock_worker = MagicMock()
mock_worker.isRunning.return_value = True
dialog._worker = mock_worker
dialog.close()
assert mock_worker.terminate_process.called
assert not Path(temp_dir).exists()
+233
View File
@@ -0,0 +1,233 @@
"""Tests for Packet S4: CmdForge CLI integration with agent-harness-setup."""
import json
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from cmdforge.cli import main
from cmdforge.cli.provider_commands import _find_setup_helper, _get_provider_defaults_for_harness
from cmdforge.providers import Provider, get_provider, load_providers, save_providers
@pytest.fixture
def temp_providers_file(tmp_path):
providers_file = tmp_path / ".cmdforge" / "providers.yaml"
with patch("cmdforge.providers.PROVIDERS_FILE", providers_file):
save_providers([])
yield providers_file
def _mock_helper_run(status: str, harness_id: str = "claude-code", exit_code: int = 0, diagnostics: str | None = None, executable_path: str = "/usr/local/bin/claude"):
"""Create a mock subprocess.run that writes a valid completion result to --result-file."""
def _run(cmd, *args, **kwargs):
# Extract --result-file and --request-id from cmd
res_idx = cmd.index("--result-file")
result_file_path = Path(cmd[res_idx + 1])
req_idx = cmd.index("--request-id")
request_id = cmd[req_idx + 1]
data = {
"schema_version": 1,
"helper_version": "0.1.0",
"request_id": request_id,
"timestamp": "2026-09-11T22:00:00Z",
"status": status,
"harness_id": harness_id,
"recipe_id": harness_id,
"recipe_version": "2026-09-10",
"install_method": "argv",
"installer_exit_code": 0 if status == "success" else None,
"verification": {
"verified": True,
"executable_path": executable_path,
"version_output": f"{harness_id} 1.0.0",
} if status == "success" else None,
"path_updates": ["/usr/local/bin"],
"auth_status": "setup_required",
"manual_steps_remaining": ["Run 'claude' to complete browser sign-in."],
"partial_changes": {
"observed_created": [executable_path] if status == "success" else [],
"observed_modified": [],
"unknown_changes": True,
},
"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"):
helper = _find_setup_helper()
assert helper == ["/usr/local/bin/agent-harness-setup"]
def test_get_provider_defaults_for_claude():
"""Verify provider defaults derived for claude-code."""
defaults = _get_provider_defaults_for_harness("claude-code", "/custom/bin/claude")
assert defaults["name"] == "claude"
assert defaults["command"] == "/custom/bin/claude -p"
assert "Claude" in defaults["description"]
def test_providers_install_verified_success_registers_provider(temp_providers_file, capsys):
"""Test verified installation prompts and registers provider in providers.yaml."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 0)) as mock_run,
patch("builtins.input", return_value="y"),
patch("sys.argv", ["cmdforge", "providers", "install", "--harness", "claude-code"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "Harness Setup Verified: claude-code" in captured.out
assert "Provider 'claude' successfully added" in captured.out
# Verify provider stored in providers.yaml
provider = get_provider("claude")
assert provider is not None
assert provider.command == "/usr/local/bin/claude -p"
assert "code" in provider.tags
# Verify subprocess invocation arguments
call_args = mock_run.call_args[0][0]
assert call_args[0] == "agent-harness-setup"
assert "--result-file" in call_args
assert "--request-id" in call_args
assert "--harness" in call_args
assert "claude-code" in call_args
def test_providers_install_declined_registration(temp_providers_file, capsys):
"""Test declining the provider registration prompt leaves providers.yaml unchanged."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 0)),
patch("builtins.input", return_value="n"),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "Harness Setup Verified: claude-code" in captured.out
assert "Skipped provider registration" in captured.out
assert get_provider("claude") is None
def test_providers_install_already_configured(temp_providers_file, capsys):
"""Test that existing provider is recognized and not re-registered."""
save_providers([Provider("claude", "existing-claude -p")])
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 0)),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "already configured in ~/.cmdforge/providers.yaml" in captured.out
assert get_provider("claude").command == "existing-claude -p"
def test_providers_install_cancelled(temp_providers_file, capsys):
"""Test cancellation exits cleanly without changing providers."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("cancelled", "claude-code", 10)),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "Setup cancelled. No provider configuration was changed." in captured.out
assert get_provider("claude") is None
def test_providers_install_execution_failure(temp_providers_file, capsys):
"""Test helper failure reports error diagnostics and does not register provider."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("failed", "claude-code", 1, "Missing node.js")),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 1
captured = capsys.readouterr()
assert "Setup failed: Missing node.js" in captured.out
assert get_provider("claude") is None
def test_providers_install_incoherent_result_rejected(temp_providers_file, capsys):
"""Test exit code 1 with status 'success' is rejected as incoherent."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("success", "claude-code", 1)),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 1
captured = capsys.readouterr()
assert "not coherent with exit code 1" in captured.out
assert get_provider("claude") is None
def test_providers_install_helper_not_found(temp_providers_file, capsys):
"""Test that missing helper prints actionable instructions without crashing."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=None),
patch("sys.argv", ["cmdforge", "providers", "install"]),
):
exit_code = main()
assert exit_code == 0
captured = capsys.readouterr()
assert "agent-harness-setup" in captured.out
assert "pip install" in captured.out
assert "PYTHONPATH=" in captured.out
def test_providers_install_dry_run(temp_providers_file, capsys):
"""Test --dry-run passes flag to helper and makes no provider changes."""
with (
patch("cmdforge.cli.provider_commands._find_setup_helper", return_value=["agent-harness-setup"]),
patch("subprocess.run", side_effect=_mock_helper_run("dry_run", "claude-code", 0)) as mock_run,
patch("sys.argv", ["cmdforge", "providers", "install", "--dry-run"]),
):
exit_code = main()
assert exit_code == 0
call_args = mock_run.call_args[0][0]
assert "--dry-run" in call_args
assert get_provider("claude") is None
+32
View File
@@ -1256,3 +1256,35 @@ class TestProviderExecutionPolicies:
assert success is False
resolver.assert_not_called()
class TestPromptStepTimeout:
"""X-8a: the provider's time limit comes from the caller or the step, not a fixed 300 s."""
@patch('cmdforge.runner.call_provider')
def test_default_stays_300_seconds(self, mock_call):
mock_call.return_value = ProviderResult(text="ok", success=True)
execute_prompt_step(PromptStep(prompt="x", provider="claude", output_var="o", plain_text=True), {})
assert mock_call.call_args.kwargs["timeout"] == 300
@patch('cmdforge.runner.call_provider')
def test_step_timeout_is_used(self, mock_call):
mock_call.return_value = ProviderResult(text="ok", success=True)
step = PromptStep(prompt="x", provider="claude", output_var="o", plain_text=True, timeout=1800)
execute_prompt_step(step, {})
assert mock_call.call_args.kwargs["timeout"] == 1800
@patch('cmdforge.runner.call_provider')
def test_callers_provider_timeout_overrides_the_step(self, mock_call):
from cmdforge.providers import ProviderExecutionPolicy
mock_call.return_value = ProviderResult(text="ok", success=True)
step = PromptStep(prompt="x", provider="claude", output_var="o", plain_text=True, timeout=1800)
execute_prompt_step(step, {}, execution_policy=ProviderExecutionPolicy(provider_timeout=120))
assert mock_call.call_args.kwargs["timeout"] == 120
def test_step_timeout_round_trips_and_is_validated(self):
step = PromptStep.from_dict({"prompt": "x", "provider": "claude", "output_var": "o", "timeout": "900"})
assert step.timeout == 900 and step.to_dict()["timeout"] == 900
assert "timeout" not in PromptStep(prompt="x", provider="claude", output_var="o").to_dict()
with pytest.raises(ValueError):
PromptStep(prompt="x", provider="claude", output_var="o", timeout=0)
+86
View File
@@ -0,0 +1,86 @@
"""Tests for safe, configuration-driven donation methods."""
from cmdforge.web.donations import load_donation_options
def _donation_page(tmp_path, monkeypatch, **environment):
monkeypatch.setenv("CMDFORGE_REGISTRY_DB", str(tmp_path / "registry.db"))
monkeypatch.delenv("CMDFORGE_DONATION_BITCOIN_ADDRESS", raising=False)
monkeypatch.delenv("CMDFORGE_DONATION_PAYPAL_URL", raising=False)
for name, value in environment.items():
monkeypatch.setenv(name, value)
from cmdforge.web.app import create_web_app
app = create_web_app()
app.config["TESTING"] = True
return app.test_client().get("/donate")
def test_loads_valid_bitcoin_and_paypal_options():
options = load_donation_options(
{
"CMDFORGE_DONATION_BITCOIN_ADDRESS": "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh",
"CMDFORGE_DONATION_PAYPAL_URL": "https://www.paypal.com/donate/?hosted_button_id=EXAMPLE",
}
)
assert options.enabled
assert options.bitcoin_address == "bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh"
assert options.bitcoin_uri.startswith("bitcoin:bc1qxy2")
assert "label=CmdForge" in options.bitcoin_uri
assert options.paypal_url.startswith("https://www.paypal.com/donate/")
def test_accepts_legacy_base58_bitcoin_address():
options = load_donation_options(
{"CMDFORGE_DONATION_BITCOIN_ADDRESS": "1BoatSLRHtKNngkdXEeobR76b53LETtpyT"}
)
assert options.bitcoin_address == "1BoatSLRHtKNngkdXEeobR76b53LETtpyT"
def test_rejects_malformed_or_unsafe_values():
options = load_donation_options(
{
"CMDFORGE_DONATION_BITCOIN_ADDRESS": "not-a-bitcoin-address",
"CMDFORGE_DONATION_PAYPAL_URL": "javascript:alert(1)",
}
)
assert not options.enabled
assert options.bitcoin_address == ""
assert options.bitcoin_uri == ""
assert options.paypal_url == ""
def test_rejects_paypal_lookalike_and_embedded_credentials():
assert not load_donation_options(
{"CMDFORGE_DONATION_PAYPAL_URL": "https://paypal.com.example.test/donate"}
).enabled
assert not load_donation_options(
{"CMDFORGE_DONATION_PAYPAL_URL": "https://user:[email protected]/donate"}
).enabled
def test_donation_page_renders_configured_methods(tmp_path, monkeypatch):
response = _donation_page(
tmp_path,
monkeypatch,
CMDFORGE_DONATION_BITCOIN_ADDRESS="bc1qxy2kgdygjrsqtzq2n0yrf2493p83kkfjhx0wlh",
CMDFORGE_DONATION_PAYPAL_URL="https://paypal.me/cmdforge",
)
assert response.status_code == 200
assert b"Make a Contribution" in response.data
assert b"Open Bitcoin Wallet" in response.data
assert b"Donate with PayPal" in response.data
assert b"https://paypal.me/cmdforge" in response.data
def test_donation_page_hides_payment_section_when_unconfigured(tmp_path, monkeypatch):
response = _donation_page(tmp_path, monkeypatch)
assert response.status_code == 200
assert b"Make a Contribution" not in response.data
assert b"Other Ways to Help" in response.data