diff --git a/src/crewplane/adapters/invokers/cli.py b/src/crewplane/adapters/invokers/cli.py index 1f886ae..6b83003 100644 --- a/src/crewplane/adapters/invokers/cli.py +++ b/src/crewplane/adapters/invokers/cli.py @@ -2,7 +2,9 @@ import os import shutil -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterator, Mapping +from dataclasses import dataclass +from functools import cache from pathlib import Path from crewplane.architecture.contracts import ( @@ -12,13 +14,34 @@ JsonObject, ProviderKind, ) -from crewplane.core.config import Config +from crewplane.core.config import AgentConfig, Config from crewplane.core.workflow.models import WorkflowPlan from crewplane.runtime.agent.invoker import PlannedAgentInvoker from .cli_invoker import build_cli_invocation_plan, build_cli_log_presentation +from .cli_invoker.env_command import EnvCommandContext, parse_env_command_context from .cli_invoker.reasoning import validate_reasoning_request +_PLATFORM_ENV_EXECUTABLES = (Path("/bin/env"), Path("/usr/bin/env")) + + +@dataclass(frozen=True, slots=True) +class _ExecutableRequirement: + """Executable lookup parameters for one required CLI command.""" + + executable: str + base_dir: Path + search_path: str | None = None + + +@dataclass(frozen=True, slots=True) +class _ReasoningValidationTarget: + """Configured workflow provider requiring reasoning validation.""" + + location: str + agent_config: AgentConfig + requested_reasoning: str + def collect_cli_availability_errors( workflow: WorkflowPlan, @@ -28,26 +51,26 @@ def collect_cli_availability_errors( ) -> list[str]: """Collect missing executable errors for configured CLI providers.""" - executable_lookup = shutil.which if which_fn is None else which_fn + executable_lookup = cache(shutil.which if which_fn is None else which_fn) executable_base_dir = Path.cwd() if project_root is None else project_root - missing_cli_locations: dict[str, list[str]] = {} + missing_cli_locations: dict[tuple[str, str], list[str]] = {} for node in workflow.nodes: for provider in node.providers: agent_config = config.agents.get(provider.provider) if agent_config is None: continue - cli_executable = agent_config.cli_cmd[0] - if _cli_executable_available( - cli_executable, - executable_lookup, + for requirement in _required_cli_executables( + agent_config.cli_cmd, executable_base_dir, + executable_lookup, ): - continue - location = f"workflow '{workflow.name}' -> node '{node.id}'" - missing_cli_locations.setdefault(provider.provider, []).append( - f"{location} (CLI: {cli_executable})" - ) - return _format_missing_cli_errors(config, missing_cli_locations) + if _cli_executable_available(requirement, executable_lookup): + continue + location = f"workflow '{workflow.name}' -> node '{node.id}'" + missing_cli_locations.setdefault( + (provider.provider, requirement.executable), [] + ).append(f"{location} (CLI: {requirement.executable})") + return _format_missing_cli_errors(missing_cli_locations) def collect_cli_reasoning_errors( @@ -56,7 +79,26 @@ def collect_cli_reasoning_errors( environment: Mapping[str, str] | None = None, working_directory: Path | None = None, ) -> list[str]: + """Collect reasoning validation errors for configured workflow providers.""" + errors: list[str] = [] + for target in _reasoning_validation_targets(workflow, config): + try: + validate_reasoning_request( + target.agent_config, + target.requested_reasoning, + environment, + working_directory, + ) + except ValueError as exc: + errors.append(f"{target.location}: {exc}") + return errors + + +def _reasoning_validation_targets( + workflow: WorkflowPlan, + config: Config, +) -> Iterator[_ReasoningValidationTarget]: for node in workflow.nodes: for provider in node.providers: if provider.reasoning is None: @@ -64,19 +106,14 @@ def collect_cli_reasoning_errors( agent_config = config.agents.get(provider.provider) if agent_config is None: continue - try: - validate_reasoning_request( - agent_config, - provider.reasoning, - environment, - working_directory, - ) - except ValueError as exc: - errors.append( + yield _ReasoningValidationTarget( + location=( f"workflow '{workflow.name}' -> node '{node.id}' -> provider " - f"'{provider.provider}': {exc}" - ) - return errors + f"'{provider.provider}'" + ), + agent_config=agent_config, + requested_reasoning=provider.reasoning, + ) def collect_cli_model_arg_warnings(config: Config) -> list[str]: @@ -93,13 +130,13 @@ def collect_cli_model_arg_warnings(config: Config) -> list[str]: def _format_missing_cli_errors( - config: Config, - missing_cli_locations: dict[str, list[str]], + missing_cli_locations: dict[tuple[str, str], list[str]], ) -> list[str]: errors: list[str] = [] - for provider_name, locations in sorted(missing_cli_locations.items()): + for (provider_name, cli_executable), locations in sorted( + missing_cli_locations.items() + ): unique_locations = sorted(set(locations)) - cli_executable = config.agents[provider_name].cli_cmd[0] availability_message = ( "not found or not executable" if Path(cli_executable).is_absolute() @@ -113,17 +150,134 @@ def _format_missing_cli_errors( return errors -def _cli_executable_available( - executable: str, +def _required_cli_executables( + cli_command: list[str], + executable_base_dir: Path, + executable_lookup: Callable[[str], str | None], +) -> tuple[_ExecutableRequirement, ...]: + wrapper = _ExecutableRequirement( + executable=cli_command[0], + base_dir=executable_base_dir, + ) + if not _is_platform_env_wrapper(wrapper, executable_lookup): + return (wrapper,) + + wrapped = _env_wrapped_executable_requirement(cli_command, executable_base_dir) + if wrapped is None or wrapped == wrapper: + return (wrapper,) + return wrapper, wrapped + + +def _is_platform_env_wrapper( + requirement: _ExecutableRequirement, executable_lookup: Callable[[str], str | None], +) -> bool: + wrapper_path = Path(requirement.executable) + if wrapper_path.name != "env": + return False + resolved_wrapper = ( + str(requirement.base_dir / wrapper_path) + if not wrapper_path.is_absolute() + and _contains_path_separator(requirement.executable) + else executable_lookup(requirement.executable) + ) + return _is_platform_env_executable(resolved_wrapper) + + +def _env_wrapped_executable_requirement( + cli_command: list[str], executable_base_dir: Path, +) -> _ExecutableRequirement | None: + try: + command_context = parse_env_command_context( + cli_command, + inherited_value=os.environ.get("PATH"), + tracked_environment_name="PATH", + ) + except ValueError: + return None + if command_context.command_executable is None: + return None + return _ExecutableRequirement( + executable=command_context.command_executable, + base_dir=_env_command_base_dir(command_context, executable_base_dir), + search_path=_env_command_search_path(command_context), + ) + + +def _is_platform_env_executable(resolved_executable: str | None) -> bool: + if resolved_executable is None: + return False + executable_path = Path(resolved_executable).resolve(strict=False) + return any( + executable_path == platform_path.resolve(strict=False) + for platform_path in _PLATFORM_ENV_EXECUTABLES + ) + + +def _env_command_base_dir( + command_context: EnvCommandContext, + executable_base_dir: Path, +) -> Path: + configured_directory = command_context.command_working_directory + if configured_directory is None: + return executable_base_dir + working_directory = Path(configured_directory) + if working_directory.is_absolute(): + return working_directory + return executable_base_dir / working_directory + + +def _env_command_search_path(command_context: EnvCommandContext) -> str | None: + if command_context.command_search_path is not None: + return command_context.command_search_path + tracked_search_path = command_context.tracked_environment_value + if ( + command_context.tracked_environment_changed + or command_context.command_working_directory is not None + ): + return os.defpath if tracked_search_path is None else tracked_search_path + if tracked_search_path is not None and _has_relative_search_path_entry( + tracked_search_path + ): + return tracked_search_path + return None + + +def _has_relative_search_path_entry(search_path: str) -> bool: + return any(not Path(entry).is_absolute() for entry in search_path.split(os.pathsep)) + + +def _cli_executable_available( + requirement: _ExecutableRequirement, + executable_lookup: Callable[[str], str | None], ) -> bool: - executable_path = Path(executable) + executable_path = Path(requirement.executable) if executable_path.is_absolute(): return _is_executable_file(executable_path) - if _contains_path_separator(executable): - return _is_executable_file(executable_base_dir / executable_path) - return executable_lookup(executable) is not None + if _contains_path_separator(requirement.executable): + return _is_executable_file(requirement.base_dir / executable_path) + if requirement.search_path is None: + return executable_lookup(requirement.executable) is not None + return _search_path_executable_available( + requirement.executable, + requirement.search_path, + requirement.base_dir, + ) + + +def _search_path_executable_available( + executable: str, + search_path: str, + executable_base_dir: Path, +) -> bool: + for entry in search_path.split(os.pathsep): + directory = Path(entry) if entry else Path() + if not directory.is_absolute(): + directory = executable_base_dir / directory + if _is_executable_file(directory / executable): + return True + return False def _is_executable_file(path: Path) -> bool: diff --git a/src/crewplane/adapters/invokers/cli_invoker/claude_json.py b/src/crewplane/adapters/invokers/cli_invoker/claude_json.py index 47eddc8..09c6e43 100644 --- a/src/crewplane/adapters/invokers/cli_invoker/claude_json.py +++ b/src/crewplane/adapters/invokers/cli_invoker/claude_json.py @@ -6,17 +6,20 @@ "read_claude_model_usage", ] -import json from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path -from typing import TextIO from crewplane.architecture.contracts import ( CommandResult, OutputExtractionResult, ) +from .claude_json_parser import ( + ClaudeJsonParseError, + parse_claude_model_usage, + parse_claude_result, +) from .streaming import ( new_owned_output_file, path_has_non_whitespace_text, @@ -25,32 +28,28 @@ stream_source, ) -_SIMPLE_JSON_ESCAPES = { - '"': '"', - "\\": "\\", - "/": "/", - "b": "\b", - "f": "\f", - "n": "\n", - "r": "\r", - "t": "\t", -} + +@dataclass(frozen=True) +class ClaudeJsonDocument: + """Incrementally parsed Claude result and optional model-usage payload.""" + + result_path: Path | None + result_char_count: int + model_usage: object | None + error: str | None = None def extract_claude_output( result: CommandResult, - max_captured_usage_bytes: int, + max_captured_usage_bytes: int, # noqa: ARG001 - Stable parser facade contract. ) -> OutputExtractionResult: """Extract Claude's result string into an owned temporary output file.""" - extraction = _extract_claude_document( - result, - max_captured_usage_bytes=max_captured_usage_bytes, - ) + extraction = _extract_claude_document(result) if extraction.error is not None: return _malformed_output() if extraction.result_path is None: return _missing_output() - if not path_has_non_whitespace_text(extraction.result_path): + if not _path_has_output(extraction.result_path): remove_owned_path(extraction.result_path) return _missing_output() return OutputExtractionResult( @@ -62,38 +61,41 @@ def extract_claude_output( ) -@dataclass(frozen=True) -class ClaudeJsonDocument: - """Incrementally parsed Claude result and optional model-usage payload.""" +def _path_has_output(path: Path) -> bool: + try: + return path_has_non_whitespace_text(path) + except BaseException: + remove_owned_path(path) + raise - result_path: Path | None - result_char_count: int - model_usage: object | None - error: str | None = None +def _extract_claude_document(result: CommandResult) -> ClaudeJsonDocument: + document = _parse_result_source(stdout_source(result)) + if document.error is not None or document.result_path is not None: + return document + stderr_source = stream_source(result.stderr_text, result.stderr_path) + if stderr_source is None: + return document + return _parse_result_source(stderr_source) -def _extract_claude_document( - result: CommandResult, - max_captured_usage_bytes: int, -) -> ClaudeJsonDocument: - document = _parse_claude_source( - stdout_source(result), - capture_result=True, - parse_result=True, - parse_model_usage=False, - max_captured_usage_bytes=max_captured_usage_bytes, - ) - if document.error is None and document.result_path is None: - stderr_source = stream_source(result.stderr_text, result.stderr_path) - if stderr_source is not None: - document = _parse_claude_source( - stderr_source, - capture_result=True, - parse_result=True, - parse_model_usage=False, - max_captured_usage_bytes=max_captured_usage_bytes, - ) - return document + +def _parse_result_source(source: Iterable[str] | None) -> ClaudeJsonDocument: + if source is None: + return ClaudeJsonDocument(None, 0, None) + output_path = new_owned_output_file() + retain_output = False + try: + char_count = parse_claude_result(source, output_path) + if char_count is None: + return ClaudeJsonDocument(None, 0, None) + document = ClaudeJsonDocument(output_path, char_count, None) + retain_output = True + return document + except ClaudeJsonParseError: + return _malformed_document() + finally: + if not retain_output: + remove_owned_path(output_path) def read_claude_model_usage( @@ -104,316 +106,16 @@ def read_claude_model_usage( source = stdout_source(result) if source is None: source = stream_source(result.stderr_text, result.stderr_path) - document = _parse_claude_source( - source, - capture_result=False, - parse_result=False, - parse_model_usage=True, - max_captured_usage_bytes=max_captured_usage_bytes, - ) - return document.model_usage, document.error - - -def _parse_claude_source( - source: Iterable[str] | None, - capture_result: bool, - parse_result: bool, - parse_model_usage: bool, - max_captured_usage_bytes: int, -) -> ClaudeJsonDocument: if source is None: - return ClaudeJsonDocument(None, 0, None) - output_path = new_owned_output_file() if capture_result else None - parser = _ClaudeJsonParser( - source, - output_path, - parse_result=parse_result, - parse_model_usage=parse_model_usage, - max_captured_usage_bytes=max_captured_usage_bytes, - ) + return None, None try: - document = parser.parse() - except _ClaudeJsonParseError: - remove_owned_path(output_path) - return ClaudeJsonDocument(None, 0, None, "Malformed Claude JSON output.") - if output_path is not None and document.result_path is None: - remove_owned_path(output_path) - return document - - -class _ClaudeJsonParseError(ValueError): - pass - - -class _JsonCharCursor: - def __init__(self, chunks: Iterable[str]) -> None: - self._chunks = iter(chunks) - self._current = "" - self._index = 0 - self._pushback: list[str] = [] - - def read(self) -> str | None: - if self._pushback: - return self._pushback.pop() - while self._index >= len(self._current): - self._current = next(self._chunks, "") - self._index = 0 - if not self._current: - return None - char = self._current[self._index] - self._index += 1 - return char - - def push(self, chars: Iterable[str]) -> None: - self._pushback.extend(reversed(tuple(chars))) - - -class _ClaudeJsonParser: - def __init__( - self, - chunks: Iterable[str], - output_path: Path | None, - parse_result: bool, - parse_model_usage: bool, - max_captured_usage_bytes: int, - ) -> None: - self._cursor = _JsonCharCursor(chunks) - self._output_path = output_path - self._parse_result = parse_result - self._parse_model_usage = parse_model_usage - self._max_captured_usage_bytes = max_captured_usage_bytes - self.result_seen = False - self.result_char_count = 0 - self.model_usage: object | None = None - self._capture_overflow = False - self._captured_size = 0 - - def parse(self) -> ClaudeJsonDocument: - self._skip_whitespace() - self._expect("{") - self._skip_whitespace() - if self._consume_object_end(): - return ClaudeJsonDocument(None, 0, None) - while True: - key = self._read_string() - self._skip_whitespace() - self._expect(":") - self._parse_member_value(key) - self._skip_whitespace() - separator = self._read_required() - if separator == "}": - break - if separator != ",": - raise _ClaudeJsonParseError("Expected object separator.") - self._skip_whitespace() - self._skip_trailing_whitespace() - if self._capture_overflow: - raise _ClaudeJsonParseError("Captured Claude usage payload is too large.") - return ClaudeJsonDocument( - result_path=self._output_path if self.result_seen else None, - result_char_count=self.result_char_count, - model_usage=self.model_usage, - ) - - def _parse_member_value(self, key: str) -> None: - self._skip_whitespace() - if key == "result" and self._parse_result: - self._read_result_value() - return - if key == "modelUsage" and self._parse_model_usage: - self.model_usage = self._read_captured_value() - return - self._skip_value() - - def _read_result_value(self) -> None: - if self._peek() != '"': - self._skip_value() - raise _ClaudeJsonParseError("Claude result must be a JSON string.") - self.result_seen = True - if self._output_path is None: - self._stream_string(None) - return - with self._output_path.open("w", encoding="utf-8") as handle: - self.result_char_count = self._stream_string(handle) - - def _read_captured_value(self) -> object | None: - captured: list[str] = [] - self._skip_value(captured) - try: - decoded: object = json.loads("".join(captured)) - except json.JSONDecodeError as exc: - raise _ClaudeJsonParseError("Malformed Claude modelUsage payload.") from exc - return decoded - - def _skip_value(self, captured: list[str] | None = None) -> None: - self._skip_whitespace(captured) - char = self._peek() - if char is None: - raise _ClaudeJsonParseError("Unexpected end of JSON value.") - if char == '"': - self._skip_string(captured) - return - if char == "{": - self._skip_bracketed_value("{", "}", captured) - return - if char == "[": - self._skip_bracketed_value("[", "]", captured) - return - self._skip_scalar(captured) - - def _skip_bracketed_value( - self, - opener: str, - closer: str, - captured: list[str] | None, - ) -> None: - self._expect(opener, captured) - stack = [closer] - while stack: - char = self._read_required() - self._capture(captured, char) - if char == '"': - self._skip_string_tail(captured) - elif char == "{": - stack.append("}") - elif char == "[": - stack.append("]") - elif char == stack[-1]: - stack.pop() - - def _skip_scalar(self, captured: list[str] | None) -> None: - while True: - char = self._read_required() - if char in {",", "}", "]"}: - self._cursor.push((char,)) - return - self._capture(captured, char) - - def _skip_string(self, captured: list[str] | None) -> None: - self._expect('"', captured) - self._skip_string_tail(captured) - - def _skip_string_tail(self, captured: list[str] | None) -> None: - while True: - char = self._read_required() - self._capture(captured, char) - if char == '"': - return - if char != "\\": - continue - escaped = self._read_required() - self._capture(captured, escaped) - if escaped == "u": - self._capture(captured, self._read_required()) - self._capture(captured, self._read_required()) - self._capture(captured, self._read_required()) - self._capture(captured, self._read_required()) - - def _read_string(self) -> str: - self._expect('"') - chars: list[str] = [] - while True: - char = self._read_required() - if char == '"': - return "".join(chars) - if char == "\\": - char = self._read_escape() - chars.append(char) - - def _stream_string(self, sink: TextIO | None) -> int: - self._expect('"') - count = 0 - while True: - char = self._read_required() - if char == '"': - return count - if char == "\\": - char = self._read_escape() - if sink is not None: - sink.write(char) - count += len(char) - - def _read_escape(self) -> str: - escaped = self._read_required() - if escaped == "u": - return self._read_unicode_escape() - - replacement = _SIMPLE_JSON_ESCAPES.get(escaped) - if replacement is None: - raise _ClaudeJsonParseError("Invalid JSON string escape.") - return replacement - - def _read_unicode_escape(self) -> str: - value = self._read_hex_codepoint() - if 0xD800 <= value <= 0xDBFF: - next_chars = [self._read_required(), self._read_required()] - if next_chars == ["\\", "u"]: - low = self._read_hex_codepoint() - if 0xDC00 <= low <= 0xDFFF: - combined = 0x10000 + ((value - 0xD800) << 10) + (low - 0xDC00) - return chr(combined) - self._cursor.push(next_chars) - return "\ufffd" - if 0xDC00 <= value <= 0xDFFF: - return "\ufffd" - return chr(value) - - def _read_hex_codepoint(self) -> int: - chars = [self._read_required() for _ in range(4)] - if any(char not in "0123456789abcdefABCDEF" for char in chars): - raise _ClaudeJsonParseError("Invalid unicode escape.") - return int("".join(chars), 16) - - def _consume_object_end(self) -> bool: - if self._peek() != "}": - return False - self._expect("}") - self._skip_trailing_whitespace() - return True - - def _skip_whitespace(self, captured: list[str] | None = None) -> None: - while True: - char = self._read_required() - if not char.isspace(): - self._cursor.push((char,)) - return - self._capture(captured, char) - - def _skip_trailing_whitespace(self) -> None: - while True: - char = self._cursor.read() - if char is None: - return - if not char.isspace(): - raise _ClaudeJsonParseError("Unexpected trailing JSON data.") - - def _peek(self) -> str | None: - char = self._cursor.read() - if char is not None: - self._cursor.push((char,)) - return char - - def _expect(self, expected: str, captured: list[str] | None = None) -> None: - char = self._read_required() - if char != expected: - raise _ClaudeJsonParseError(f"Expected {expected!r}.") - self._capture(captured, char) + return parse_claude_model_usage(source, max_captured_usage_bytes), None + except ClaudeJsonParseError: + return None, "Malformed Claude JSON output." - def _read_required(self) -> str: - char = self._cursor.read() - if char is None: - raise _ClaudeJsonParseError("Unexpected end of JSON input.") - return char - def _capture(self, captured: list[str] | None, char: str) -> None: - if captured is None: - return - char_bytes = len(char.encode("utf-8")) - if self._captured_size + char_bytes > self._max_captured_usage_bytes: - self._capture_overflow = True - return - captured.append(char) - self._captured_size += char_bytes +def _malformed_document() -> ClaudeJsonDocument: + return ClaudeJsonDocument(None, 0, None, "Malformed Claude JSON output.") def _missing_output() -> OutputExtractionResult: diff --git a/src/crewplane/adapters/invokers/cli_invoker/claude_json_parser.py b/src/crewplane/adapters/invokers/cli_invoker/claude_json_parser.py new file mode 100644 index 0000000..c3926cd --- /dev/null +++ b/src/crewplane/adapters/invokers/cli_invoker/claude_json_parser.py @@ -0,0 +1,457 @@ +from __future__ import annotations + +__all__ = [ + "ClaudeJsonParseError", + "parse_claude_model_usage", + "parse_claude_result", +] + +import json +from collections.abc import Iterable, Iterator +from dataclasses import dataclass +from enum import Enum, StrEnum, auto +from pathlib import Path +from typing import Literal, TextIO + +from .json_number import ( + JSON_NUMBER_TERMINAL_STATES, + JsonNumberError, + JsonNumberState, + advance_json_number, + start_json_number, +) + + +class ClaudeJsonParseError(ValueError): + """Raised when Claude output is malformed for the requested parse mode.""" + + +def parse_claude_result(chunks: Iterable[str], output_path: Path) -> int | None: + """Stream Claude's result field and return its decoded character count.""" + selection = _ResultSelection(output_path) + _ClaudeJsonParser(chunks, selection).parse() + return selection.char_count + + +def parse_claude_model_usage( + chunks: Iterable[str], + max_captured_bytes: int, +) -> object | None: + """Decode Claude's bounded modelUsage field when present.""" + selection = _ModelUsageSelection(max_captured_bytes) + _ClaudeJsonParser(chunks, selection).parse() + return selection.value + + +type _CaptureBuffer = list[str] | None + +_SIMPLE_JSON_ESCAPES = { + '"': '"', + "\\": "\\", + "/": "/", + "b": "\b", + "f": "\f", + "n": "\n", + "r": "\r", + "t": "\t", +} +_JSON_WHITESPACE = frozenset(" \t\r\n") +_JSON_SCALAR_END = _JSON_WHITESPACE | frozenset(",}]") +_JSON_LITERAL_SUFFIXES = {"f": "alse", "n": "ull", "t": "rue"} + + +class _JsonContainer(StrEnum): + OBJECT = "{" + ARRAY = "[" + + @property + def closer(self) -> Literal["}", "]"]: + return "}" if self is _JsonContainer.OBJECT else "]" + + +class _JsonContainerPhase(Enum): + ITEM_OR_END = auto() + ITEM = auto() + SEPARATOR = auto() + + +@dataclass(slots=True) +class _JsonContainerFrame: + container: _JsonContainer + phase: _JsonContainerPhase = _JsonContainerPhase.ITEM_OR_END + + +@dataclass(slots=True) +class _ResultSelection: + output_path: Path + char_count: int | None = None + + +@dataclass(slots=True) +class _ModelUsageSelection: + max_captured_bytes: int + value: object | None = None + + +type _ClaudeJsonSelection = _ResultSelection | _ModelUsageSelection + + +class _JsonCharCursor: + __slots__ = ("_chunks", "_current", "_index", "_pushback") + + def __init__(self, chunks: Iterable[str]) -> None: + self._chunks = iter(chunks) + self._current = "" + self._index = 0 + self._pushback: list[str] = [] + + def read(self) -> str | None: + if self._pushback: + return self._pushback.pop() + while self._index >= len(self._current): + chunk = next(self._chunks, None) + if chunk is None: + return None + if not chunk: + continue + self._current = chunk + self._index = 0 + char = self._current[self._index] + self._index += 1 + return char + + def peek(self) -> str | None: + char = self.read() + if char is not None: + self.unread((char,)) + return char + + def unread(self, chars: Iterable[str]) -> None: + self._pushback.extend(reversed(tuple(chars))) + + +class _ClaudeJsonParser: + def __init__( + self, + chunks: Iterable[str], + selection: _ClaudeJsonSelection, + ) -> None: + self._cursor = _JsonCharCursor(chunks) + self._selection = selection + self._max_captured_bytes = ( + selection.max_captured_bytes + if isinstance(selection, _ModelUsageSelection) + else 0 + ) + self._captured_size = 0 + + def parse(self) -> None: + self._skip_whitespace() + self._expect("{") + self._parse_members() + self._skip_trailing_whitespace() + + def _parse_members(self) -> None: + self._skip_whitespace() + if self._consume_object_end(): + return + while True: + self._parse_member() + self._skip_whitespace() + if not self._read_has_next_item("}", None): + return + self._skip_whitespace() + + def _parse_member(self) -> None: + key = self._read_string() + self._skip_whitespace() + self._expect(":") + self._parse_member_value(key) + + def _parse_member_value(self, key: str) -> None: + self._skip_whitespace() + if key == "result" and isinstance(self._selection, _ResultSelection): + self._selection.char_count = self._read_result_value( + self._selection.output_path + ) + return + if key == "modelUsage" and isinstance(self._selection, _ModelUsageSelection): + self._selection.value = self._read_captured_value() + return + self._skip_value() + + def _read_result_value(self, output_path: Path) -> int: + if self._cursor.peek() != '"': + self._skip_value() + raise ClaudeJsonParseError("Claude result must be a JSON string.") + with output_path.open("w", encoding="utf-8") as handle: + return self._stream_string(handle) + + def _read_captured_value(self) -> object | None: + captured: list[str] = [] + self._skip_value(captured) + try: + decoded: object = json.loads("".join(captured)) + except (ValueError, RecursionError) as exc: + raise ClaudeJsonParseError("Malformed Claude modelUsage payload.") from exc + return decoded + + def _skip_value(self, captured: _CaptureBuffer = None) -> None: + container = self._consume_value_start(captured) + if container is not None: + self._skip_container_tail(container, captured) + + def _consume_value_start( + self, + captured: _CaptureBuffer, + ) -> _JsonContainer | None: + self._skip_whitespace(captured) + char = self._cursor.peek() + if char is None: + raise ClaudeJsonParseError("Unexpected end of JSON value.") + if char == '"': + self._skip_string(captured) + return None + if char == _JsonContainer.OBJECT: + self._expect(_JsonContainer.OBJECT, captured) + return _JsonContainer.OBJECT + if char == _JsonContainer.ARRAY: + self._expect(_JsonContainer.ARRAY, captured) + return _JsonContainer.ARRAY + self._skip_scalar(captured) + return None + + def _skip_container_tail( + self, + container: _JsonContainer, + captured: _CaptureBuffer, + ) -> None: + frames = [_JsonContainerFrame(container)] + while frames: + self._advance_container(frames, captured) + + def _advance_container( + self, + frames: list[_JsonContainerFrame], + captured: _CaptureBuffer, + ) -> None: + frame = frames[-1] + self._skip_whitespace(captured) + if self._consume_empty_container(frame, captured): + frames.pop() + return + if frame.phase is not _JsonContainerPhase.SEPARATOR: + nested = self._skip_container_item(frame, captured) + frame.phase = _JsonContainerPhase.SEPARATOR + if nested is not None: + frames.append(_JsonContainerFrame(nested)) + return + if self._read_has_next_item(frame.container.closer, captured): + frame.phase = _JsonContainerPhase.ITEM + else: + frames.pop() + + def _consume_empty_container( + self, + frame: _JsonContainerFrame, + captured: _CaptureBuffer, + ) -> bool: + if ( + frame.phase is not _JsonContainerPhase.ITEM_OR_END + or self._cursor.peek() != frame.container.closer + ): + return False + self._expect(frame.container.closer, captured) + return True + + def _skip_container_item( + self, + frame: _JsonContainerFrame, + captured: _CaptureBuffer, + ) -> _JsonContainer | None: + if frame.container is _JsonContainer.OBJECT: + self._skip_string(captured) + self._skip_whitespace(captured) + self._expect(":", captured) + return self._consume_value_start(captured) + + def _read_has_next_item( + self, + closer: str, + captured: _CaptureBuffer, + ) -> bool: + separator = self._read_required(captured) + if separator == closer: + return False + if separator != ",": + raise ClaudeJsonParseError("Expected JSON value separator.") + return True + + def _skip_scalar(self, captured: _CaptureBuffer) -> None: + first = self._read_required(captured) + literal_suffix = _JSON_LITERAL_SUFFIXES.get(first) + if literal_suffix is not None: + self._skip_literal_suffix(literal_suffix, captured) + return + self._skip_number(first, captured) + + def _skip_literal_suffix( + self, + suffix: str, + captured: _CaptureBuffer, + ) -> None: + for expected in suffix: + if self._read_required(captured) != expected: + raise ClaudeJsonParseError("Invalid JSON scalar value.") + self._require_scalar_end() + + def _skip_number(self, first: str, captured: _CaptureBuffer) -> None: + try: + state = start_json_number(first) + state = self._consume_number_tail(state, captured) + except JsonNumberError as exc: + raise ClaudeJsonParseError("Invalid JSON scalar value.") from exc + if state not in JSON_NUMBER_TERMINAL_STATES: + raise ClaudeJsonParseError("Invalid JSON scalar value.") + + def _consume_number_tail( + self, + state: JsonNumberState, + captured: _CaptureBuffer, + ) -> JsonNumberState: + while True: + char = self._cursor.read() + if char is None: + return state + if char in _JSON_SCALAR_END: + self._cursor.unread((char,)) + return state + self._capture(captured, char) + state = advance_json_number(state, char) + + def _require_scalar_end(self) -> None: + char = self._cursor.peek() + if char is not None and char not in _JSON_SCALAR_END: + raise ClaudeJsonParseError("Invalid JSON scalar value.") + + def _skip_string(self, captured: _CaptureBuffer) -> None: + self._expect('"', captured) + while True: + char = self._read_required(captured) + if char == '"': + return + self._validate_string_char(char) + if char == "\\": + self._skip_raw_escape(captured) + + def _skip_raw_escape(self, captured: _CaptureBuffer) -> None: + escaped = self._read_required(captured) + if escaped in _SIMPLE_JSON_ESCAPES: + return + if escaped != "u": + raise ClaudeJsonParseError("Invalid JSON string escape.") + self._read_hex_quad(captured) + + def _read_string(self) -> str: + return "".join(self._iter_decoded_string_chars()) + + def _stream_string(self, sink: TextIO) -> int: + count = 0 + for char in self._iter_decoded_string_chars(): + sink.write(char) + count += len(char) + return count + + def _iter_decoded_string_chars(self) -> Iterator[str]: + self._expect('"') + while True: + char = self._read_required() + if char == '"': + return + self._validate_string_char(char) + yield self._read_escape() if char == "\\" else char + + @staticmethod + def _validate_string_char(char: str) -> None: + if ord(char) < 0x20: + raise ClaudeJsonParseError("Unescaped control character.") + + def _read_escape(self) -> str: + escaped = self._read_required() + if escaped == "u": + return self._read_unicode_escape() + replacement = _SIMPLE_JSON_ESCAPES.get(escaped) + if replacement is None: + raise ClaudeJsonParseError("Invalid JSON string escape.") + return replacement + + def _read_unicode_escape(self) -> str: + value = int(self._read_hex_quad(), 16) + if 0xD800 <= value <= 0xDBFF: + return self._read_surrogate_pair(value) + if 0xDC00 <= value <= 0xDFFF: + return "\ufffd" + return chr(value) + + def _read_surrogate_pair(self, high: int) -> str: + prefix = [self._read_required(), self._read_required()] + if prefix != ["\\", "u"]: + self._cursor.unread(prefix) + return "\ufffd" + low_chars = self._read_hex_quad() + low = int(low_chars, 16) + if 0xDC00 <= low <= 0xDFFF: + combined = 0x10000 + ((high - 0xD800) << 10) + (low - 0xDC00) + return chr(combined) + self._cursor.unread((*prefix, *low_chars)) + return "\ufffd" + + def _read_hex_quad(self, captured: _CaptureBuffer = None) -> str: + chars = [self._read_required(captured) for _ in range(4)] + if any(char not in "0123456789abcdefABCDEF" for char in chars): + raise ClaudeJsonParseError("Invalid unicode escape.") + return "".join(chars) + + def _consume_object_end(self) -> bool: + if self._cursor.peek() != "}": + return False + self._expect("}") + return True + + def _skip_whitespace(self, captured: _CaptureBuffer = None) -> None: + while True: + char = self._read_required() + if char not in _JSON_WHITESPACE: + self._cursor.unread((char,)) + return + self._capture(captured, char) + + def _skip_trailing_whitespace(self) -> None: + while True: + char = self._cursor.read() + if char is None: + return + if char not in _JSON_WHITESPACE: + raise ClaudeJsonParseError("Unexpected trailing JSON data.") + + def _expect(self, expected: str, captured: _CaptureBuffer = None) -> None: + char = self._read_required() + if char != expected: + raise ClaudeJsonParseError(f"Expected {expected!r}.") + self._capture(captured, char) + + def _read_required(self, captured: _CaptureBuffer = None) -> str: + char = self._cursor.read() + if char is None: + raise ClaudeJsonParseError("Unexpected end of JSON input.") + self._capture(captured, char) + return char + + def _capture(self, captured: _CaptureBuffer, char: str) -> None: + if captured is None: + return + char_bytes = len(char.encode("utf-8")) + if self._captured_size + char_bytes > self._max_captured_bytes: + raise ClaudeJsonParseError("Captured Claude usage payload is too large.") + captured.append(char) + self._captured_size += char_bytes diff --git a/src/crewplane/adapters/invokers/cli_invoker/env_command.py b/src/crewplane/adapters/invokers/cli_invoker/env_command.py index 731fac1..8247402 100644 --- a/src/crewplane/adapters/invokers/cli_invoker/env_command.py +++ b/src/crewplane/adapters/invokers/cli_invoker/env_command.py @@ -6,163 +6,218 @@ from dataclasses import dataclass from pathlib import Path +_LONG_PASSTHROUGH_OPTIONS = frozenset( + { + "--block-signal", + "--debug", + "--default-signal", + "--ignore-signal", + "--list-signal-handling", + "--null", + } +) +_SHORT_PASSTHROUGH_OPTIONS = frozenset({"0", "v"}) +_SHORT_VALUE_OPTIONS = frozenset({"C", "P", "S", "a", "u"}) + @dataclass(frozen=True, slots=True) class EnvCommandContext: """Effective command arguments and tracked value after an env prefix.""" + command_executable: str | None command_arguments: tuple[str, ...] - tracked_environment_value: str + tracked_environment_value: str | None + tracked_environment_changed: bool + command_search_path: str | None + command_working_directory: str | None + command_working_directory_option: str | None def parse_env_command_context( tokens: Sequence[str], - inherited_value: str, + inherited_value: str | None, tracked_environment_name: str, ) -> EnvCommandContext: """Interpret supported env options without mutating the supplied tokens.""" - if not tokens: - return EnvCommandContext((), inherited_value) - if Path(tokens[0]).name != "env": - return EnvCommandContext(tuple(tokens[1:]), inherited_value) - - effective_value = inherited_value - index = 1 - while index < len(tokens): - token = tokens[index] - if token == "--": - index += 1 - break - if token == "-": - effective_value = "" - index += 1 - continue - if token.startswith("--"): - index, effective_value = _consume_long_option( - tokens, - index, - effective_value, - tracked_environment_name, - ) - continue - if token.startswith("-"): - index, effective_value = _consume_short_options( - tokens, - index, - effective_value, - tracked_environment_name, - ) - continue - break - - while index < len(tokens): - key, separator, value = tokens[index].partition("=") - if not separator: - break - if key == tracked_environment_name: - effective_value = value - index += 1 - command_arguments = tuple(tokens[index + 1 :]) if index < len(tokens) else () - return EnvCommandContext(command_arguments, effective_value) - - -def _consume_long_option( + if not tokens or Path(tokens[0]).name != "env": + return _unwrapped_command_context(tokens, inherited_value) + return _EnvCommandParser( + tokens, + inherited_value, + tracked_environment_name, + ).parse() + + +def _unwrapped_command_context( tokens: Sequence[str], - index: int, - effective_value: str, - tracked_environment_name: str, -) -> tuple[int, str]: - token = tokens[index] - option, separator, inline_value = token.partition("=") - if option == "--ignore-environment" and not separator: - return index + 1, "" - if option == "--unset": - value, next_index = _option_value( - tokens, - index, - inline_value if separator else None, - option, - ) - if value == tracked_environment_name: - effective_value = "" - return next_index, effective_value - if option == "--chdir": - raise ValueError( - "--chdir cannot be combined with a workflow reasoning request." - ) - if option == "--argv0": - _, next_index = _option_value( - tokens, - index, - inline_value if separator else None, - option, - ) - return next_index, effective_value - if option == "--split-string": - raise ValueError( - "--split-string cannot be combined with a workflow reasoning request." - ) - if option in { - "--block-signal", - "--debug", - "--default-signal", - "--ignore-signal", - "--list-signal-handling", - "--null", - }: - return index + 1, effective_value - raise ValueError( - f"Cannot validate env option {token!r} with a workflow reasoning request." + inherited_value: str | None, +) -> EnvCommandContext: + executable = tokens[0] if tokens else None + arguments = tuple(tokens[1:]) if tokens else () + return EnvCommandContext( + command_executable=executable, + command_arguments=arguments, + tracked_environment_value=inherited_value, + tracked_environment_changed=False, + command_search_path=None, + command_working_directory=None, + command_working_directory_option=None, ) -def _consume_short_options( - tokens: Sequence[str], - index: int, - effective_value: str, - tracked_environment_name: str, -) -> tuple[int, str]: - cluster = tokens[index][1:] - option_index = 0 - while option_index < len(cluster): - option = cluster[option_index] - if option == "i": - effective_value = "" - option_index += 1 - continue - if option in {"0", "v"}: - option_index += 1 - continue - if option not in {"C", "P", "S", "a", "u"}: - raise ValueError( - f"Cannot validate env option '-{option}' " - "with a workflow reasoning request." - ) - inline_value = cluster[option_index + 1 :] or None - value, next_index = _option_value( - tokens, - index, - inline_value, - f"-{option}", - ) - if option in {"C", "S"}: - raise ValueError( - f"-{option} cannot be combined with a workflow reasoning request." - ) - if option == "u" and value == tracked_environment_name: - effective_value = "" - return next_index, effective_value - return index + 1, effective_value +class _EnvCommandParser: + def __init__( + self, + tokens: Sequence[str], + inherited_value: str | None, + tracked_environment_name: str, + ) -> None: + self._tokens = tokens + self._tracked_environment_name = tracked_environment_name + self._tracked_environment_value = inherited_value + self._tracked_environment_changed = False + self._command_search_path: str | None = None + self._command_working_directory: str | None = None + self._command_working_directory_option: str | None = None + self._index = 1 + def parse(self) -> EnvCommandContext: + self._consume_options() + self._consume_environment_reset() + self._consume_assignments() + return self._build_context() -def _option_value( - tokens: Sequence[str], - index: int, - inline_value: str | None, - option: str, -) -> tuple[str, int]: - if inline_value is not None: - return inline_value, index + 1 - if index + 1 >= len(tokens): - raise ValueError(f"{option} requires a value.") - return tokens[index + 1], index + 2 + def _consume_options(self) -> None: + while self._index < len(self._tokens): + token = self._tokens[self._index] + if token == "--": + self._index += 1 + return + if token == "-" or not token.startswith("-"): + return + if token.startswith("--"): + self._consume_long_option(token) + else: + self._consume_short_option_cluster(token) + + def _consume_long_option(self, token: str) -> None: + option, separator, inline_value = token.partition("=") + value = inline_value if separator else None + match option: + case "--ignore-environment" if not separator: + self._set_tracked_environment_value(None) + self._index += 1 + case "--unset" | "--chdir" | "--argv0": + self._consume_long_value_option(option, value) + case "--split-string": + raise ValueError( + "--split-string cannot be combined with a workflow reasoning " + "request." + ) + case _ if option in _LONG_PASSTHROUGH_OPTIONS: + self._index += 1 + case _: + raise ValueError( + f"Cannot validate env option {token!r} with a workflow " + "reasoning request." + ) + + def _consume_long_value_option( + self, + option: str, + inline_value: str | None, + ) -> None: + value = self._take_option_value(inline_value, option) + if option == "--unset" and value == self._tracked_environment_name: + self._set_tracked_environment_value(None) + elif option == "--chdir": + self._set_working_directory(value, option) + + def _consume_short_option_cluster(self, token: str) -> None: + cluster = token[1:] + for option_index, option in enumerate(cluster): + if option == "i": + self._set_tracked_environment_value(None) + continue + if option in _SHORT_PASSTHROUGH_OPTIONS: + continue + if option not in _SHORT_VALUE_OPTIONS: + raise ValueError( + f"Cannot validate env option '-{option}' " + "with a workflow reasoning request." + ) + inline_value = cluster[option_index + 1 :] or None + self._consume_short_value_option(option, inline_value) + return + self._index += 1 + + def _consume_short_value_option( + self, + option: str, + inline_value: str | None, + ) -> None: + option_name = f"-{option}" + value = self._take_option_value(inline_value, option_name) + match option: + case "S": + raise ValueError( + f"{option_name} cannot be combined with a workflow reasoning " + "request." + ) + case "u" if value == self._tracked_environment_name: + self._set_tracked_environment_value(None) + case "P": + self._command_search_path = value + case "C": + self._set_working_directory(value, option_name) + + def _take_option_value( + self, + inline_value: str | None, + option: str, + ) -> str: + if inline_value is not None: + self._index += 1 + return inline_value + value_index = self._index + 1 + if value_index >= len(self._tokens): + raise ValueError(f"{option} requires a value.") + self._index = value_index + 1 + return self._tokens[value_index] + + def _consume_environment_reset(self) -> None: + if self._index >= len(self._tokens) or self._tokens[self._index] != "-": + return + self._set_tracked_environment_value(None) + self._index += 1 + + def _consume_assignments(self) -> None: + while self._index < len(self._tokens): + key, separator, value = self._tokens[self._index].partition("=") + if not separator: + return + if key == self._tracked_environment_name: + self._set_tracked_environment_value(value) + self._index += 1 + + def _set_tracked_environment_value(self, value: str | None) -> None: + self._tracked_environment_value = value + self._tracked_environment_changed = True + + def _set_working_directory(self, value: str, option: str) -> None: + self._command_working_directory = value + self._command_working_directory_option = option + + def _build_context(self) -> EnvCommandContext: + command = self._tokens[self._index :] + executable = command[0] if command else None + arguments = tuple(command[1:]) + return EnvCommandContext( + command_executable=executable, + command_arguments=arguments, + tracked_environment_value=self._tracked_environment_value, + tracked_environment_changed=self._tracked_environment_changed, + command_search_path=self._command_search_path, + command_working_directory=self._command_working_directory, + command_working_directory_option=self._command_working_directory_option, + ) diff --git a/src/crewplane/adapters/invokers/cli_invoker/json_number.py b/src/crewplane/adapters/invokers/cli_invoker/json_number.py new file mode 100644 index 0000000..7920889 --- /dev/null +++ b/src/crewplane/adapters/invokers/cli_invoker/json_number.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +__all__ = [ + "JSON_NUMBER_TERMINAL_STATES", + "JsonNumberError", + "JsonNumberState", + "advance_json_number", + "start_json_number", +] + +from typing import Literal + + +class JsonNumberError(ValueError): + """Raised when a character violates the JSON number grammar.""" + + +type JsonNumberState = Literal[ + "start", + "sign", + "zero", + "integer", + "fraction_start", + "fraction", + "exponent_start", + "exponent_sign", + "exponent", +] +type _JsonNumberToken = Literal[ + "zero", + "digit", + "decimal_point", + "exponent_marker", + "plus", + "minus", +] + +JSON_NUMBER_TERMINAL_STATES: frozenset[JsonNumberState] = frozenset( + {"zero", "integer", "fraction", "exponent"} +) +_REPEATING_STATES: frozenset[JsonNumberState] = frozenset( + {"integer", "fraction", "exponent"} +) +_TOKENS: dict[str, _JsonNumberToken] = { + ".": "decimal_point", + "E": "exponent_marker", + "e": "exponent_marker", + "+": "plus", + "-": "minus", +} +_TRANSITIONS: dict[ + JsonNumberState, + dict[_JsonNumberToken, JsonNumberState], +] = { + "start": {"minus": "sign", "zero": "zero", "digit": "integer"}, + "sign": {"zero": "zero", "digit": "integer"}, + "zero": { + "decimal_point": "fraction_start", + "exponent_marker": "exponent_start", + }, + "integer": { + "decimal_point": "fraction_start", + "exponent_marker": "exponent_start", + }, + "fraction_start": {"zero": "fraction", "digit": "fraction"}, + "fraction": {"exponent_marker": "exponent_start"}, + "exponent_start": { + "zero": "exponent", + "digit": "exponent", + "plus": "exponent_sign", + "minus": "exponent_sign", + }, + "exponent_sign": {"zero": "exponent", "digit": "exponent"}, + "exponent": {}, +} + + +def start_json_number(first: str) -> JsonNumberState: + """Start scanning a JSON number with its first character.""" + if first == "-": + return "sign" + if first == "0": + return "zero" + if first in "123456789": + return "integer" + raise JsonNumberError("Invalid JSON number.") + + +def advance_json_number(state: JsonNumberState, char: str) -> JsonNumberState: + """Advance a JSON number state with one ASCII character.""" + if state in _REPEATING_STATES and char in "0123456789": + return state + token = _classify_char(char) + if token is not None: + next_state = _TRANSITIONS[state].get(token) + if next_state is not None: + return next_state + raise JsonNumberError("Invalid JSON number.") + + +def _classify_char(char: str) -> _JsonNumberToken | None: + if char == "0": + return "zero" + if char in "123456789": + return "digit" + return _TOKENS.get(char) diff --git a/src/crewplane/adapters/invokers/cli_invoker/reasoning.py b/src/crewplane/adapters/invokers/cli_invoker/reasoning.py index 5f6ebb9..d9ba3a8 100644 --- a/src/crewplane/adapters/invokers/cli_invoker/reasoning.py +++ b/src/crewplane/adapters/invokers/cli_invoker/reasoning.py @@ -39,6 +39,11 @@ def validate_reasoning_request( ) cli_arguments = cli_context.command_arguments effective_reasoning_environment = cli_context.tracked_environment_value + if cli_context.command_working_directory_option is not None: + raise ValueError( + f"{cli_context.command_working_directory_option} cannot be combined " + "with a workflow reasoning request." + ) _reject_cli_command_terminator(cli_arguments) if provider_kind == ProviderKind.CODEX: _reject_codex_reasoning_conflict(cli_arguments) @@ -46,7 +51,7 @@ def validate_reasoning_request( return _reject_claude_reasoning_conflict(cli_arguments, working_directory) _reject_claude_reasoning_conflict(config.extra_args, working_directory) - if effective_reasoning_environment.strip(): + if effective_reasoning_environment and effective_reasoning_environment.strip(): raise ValueError( f"{CLAUDE_REASONING_ENV} conflicts with the workflow reasoning request." ) diff --git a/tests/integration/architecture/test_source_size_and_hygiene.py b/tests/integration/architecture/test_source_size_and_hygiene.py index c0c8687..4e6d435 100644 --- a/tests/integration/architecture/test_source_size_and_hygiene.py +++ b/tests/integration/architecture/test_source_size_and_hygiene.py @@ -7,10 +7,24 @@ import pytest from tests.integration.architecture.static_checks import ( + SRC_ROOT, text_rule_files, walk_ast, ) +MAX_CLAUDE_JSON_MODULE_LINES = 500 +CLAUDE_JSON_MODULES = tuple( + SRC_ROOT / "crewplane" / "adapters" / "invokers" / "cli_invoker" / filename + for filename in ("claude_json.py", "claude_json_parser.py", "json_number.py") +) + + +@pytest.mark.parametrize("module", CLAUDE_JSON_MODULES, ids=lambda path: path.name) +def test_claude_json_modules_remain_reviewable(module: Path) -> None: + line_count = len(module.read_text(encoding="utf-8").splitlines()) + + assert line_count <= MAX_CLAUDE_JSON_MODULE_LINES + def test_ast_walker_does_not_depend_on_mutable_stdlib_walk_helpers( monkeypatch, diff --git a/tests/unit/adapters/invokers/cli_invoker/test_claude_json_parser.py b/tests/unit/adapters/invokers/cli_invoker/test_claude_json_parser.py new file mode 100644 index 0000000..b37ec2a --- /dev/null +++ b/tests/unit/adapters/invokers/cli_invoker/test_claude_json_parser.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from crewplane.adapters.invokers.cli_invoker import ( + claude_json, + claude_json_parser, +) +from crewplane.adapters.invokers.cli_invoker.claude_json_parser import ( + parse_claude_result, +) +from crewplane.architecture.contracts import CommandResult + + +@pytest.mark.parametrize( + ("encoded_result", "expected_result", "expected_count"), + [ + ("\\uD83D\\uDE00", "😀", 1), + ("\\uD800", "\ufffd", 1), + ("\\uDC00", "\ufffd", 1), + ("\\uD800\\u0041", "\ufffdA", 2), + ], +) +def test_result_parser_decodes_surrogate_escapes( + tmp_path: Path, + encoded_result: str, + expected_result: str, + expected_count: int, +) -> None: + output_path = tmp_path / "result.txt" + + char_count = parse_claude_result( + (f'{{"result":"{encoded_result}"}}',), + output_path, + ) + + assert output_path.read_text(encoding="utf-8") == expected_result + assert char_count == expected_count + + +def test_result_parser_accepts_empty_intermediate_chunks(tmp_path: Path) -> None: + output_path = tmp_path / "result.txt" + + char_count = parse_claude_result( + ("", '{"res', "", 'ult":"o', "", 'k"}', ""), + output_path, + ) + + assert char_count == 2 + assert output_path.read_text(encoding="utf-8") == "ok" + + +def test_usage_parser_normalizes_integer_conversion_limits() -> None: + oversized_integer = "1" * 5_000 + + usage, error = claude_json.read_claude_model_usage( + CommandResult(0, f'{{"modelUsage":{oversized_integer}}}', ""), + 10_000, + ) + + assert usage is None + assert error == "Malformed Claude JSON output." + + +@pytest.mark.parametrize( + ("model_usage", "byte_limit", "expected_usage"), + [ + ('"é"', 4, "é"), + ('"é"', 3, None), + ('"\\u00e9"', 8, "é"), + ('"\\u00e9"', 7, None), + ], +) +def test_usage_parser_counts_raw_utf8_capture_bytes( + model_usage: str, + byte_limit: int, + expected_usage: object | None, +) -> None: + usage, error = claude_json.read_claude_model_usage( + CommandResult(0, f'{{"modelUsage":{model_usage}}}', ""), + byte_limit, + ) + + assert usage == expected_usage + assert (error is None) is (expected_usage is not None) + + +def test_usage_parser_applies_capture_limit_across_duplicate_fields() -> None: + result = CommandResult(0, '{"modelUsage":1,"modelUsage":2}', "") + + within_limit = claude_json.read_claude_model_usage(result, 2) + over_limit = claude_json.read_claude_model_usage(result, 1) + + assert within_limit == (2, None) + assert over_limit == (None, "Malformed Claude JSON output.") + + +def test_usage_parser_normalizes_decoder_recursion_errors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def raise_recursion_error(payload: str) -> object: + assert payload == "{}" + raise RecursionError + + monkeypatch.setattr(claude_json_parser.json, "loads", raise_recursion_error) + + usage, error = claude_json.read_claude_model_usage( + CommandResult(0, '{"modelUsage":{}}', ""), + 10, + ) + + assert usage is None + assert error == "Malformed Claude JSON output." diff --git a/tests/unit/adapters/invokers/cli_invoker/test_env_command.py b/tests/unit/adapters/invokers/cli_invoker/test_env_command.py new file mode 100644 index 0000000..45dc2d8 --- /dev/null +++ b/tests/unit/adapters/invokers/cli_invoker/test_env_command.py @@ -0,0 +1,206 @@ +from __future__ import annotations + +import pytest + +from crewplane.adapters.invokers.cli_invoker.env_command import ( + EnvCommandContext, + parse_env_command_context, +) + +_TRACKED_ENVIRONMENT = "TRACKED_ENVIRONMENT" + + +@pytest.mark.parametrize( + ("tokens", "expected"), + [ + pytest.param( + [], + EnvCommandContext( + command_executable=None, + command_arguments=(), + tracked_environment_value="inherited", + tracked_environment_changed=False, + command_search_path=None, + command_working_directory=None, + command_working_directory_option=None, + ), + id="empty-command", + ), + pytest.param( + ["codex", "exec"], + EnvCommandContext( + command_executable="codex", + command_arguments=("exec",), + tracked_environment_value="inherited", + tracked_environment_changed=False, + command_search_path=None, + command_working_directory=None, + command_working_directory_option=None, + ), + id="direct-command", + ), + ], +) +def test_parse_env_command_context_preserves_unwrapped_commands( + tokens: list[str], + expected: EnvCommandContext, +) -> None: + assert ( + parse_env_command_context(tokens, "inherited", _TRACKED_ENVIRONMENT) == expected + ) + + +def test_parse_env_command_context_consumes_assignments_before_command() -> None: + context = parse_env_command_context( + [ + "env", + "OTHER=value", + f"{_TRACKED_ENVIRONMENT}=first", + f"{_TRACKED_ENVIRONMENT}=second", + "claude", + f"{_TRACKED_ENVIRONMENT}=argument", + ], + "inherited", + _TRACKED_ENVIRONMENT, + ) + + assert context == EnvCommandContext( + command_executable="claude", + command_arguments=(f"{_TRACKED_ENVIRONMENT}=argument",), + tracked_environment_value="second", + tracked_environment_changed=True, + command_search_path=None, + command_working_directory=None, + command_working_directory_option=None, + ) + + +def test_parse_env_command_context_applies_short_option_state() -> None: + context = parse_env_command_context( + [ + "env", + "-ivPfirst-bin", + "-P", + "second-bin", + "-Cfirst-worktree", + "-C", + "second-worktree", + f"{_TRACKED_ENVIRONMENT}=configured", + "provider", + ], + "inherited", + _TRACKED_ENVIRONMENT, + ) + + assert context == EnvCommandContext( + command_executable="provider", + command_arguments=(), + tracked_environment_value="configured", + tracked_environment_changed=True, + command_search_path="second-bin", + command_working_directory="second-worktree", + command_working_directory_option="-C", + ) + + +def test_parse_env_command_context_applies_long_option_state() -> None: + context = parse_env_command_context( + [ + "/usr/bin/env", + "--ignore-environment", + "--unset=OTHER", + "--argv0=alias", + "--chdir=first-worktree", + "--chdir", + "second-worktree", + "--debug", + f"{_TRACKED_ENVIRONMENT}=configured", + "provider", + ], + "inherited", + _TRACKED_ENVIRONMENT, + ) + + assert context == EnvCommandContext( + command_executable="provider", + command_arguments=(), + tracked_environment_value="configured", + tracked_environment_changed=True, + command_search_path=None, + command_working_directory="second-worktree", + command_working_directory_option="--chdir", + ) + + +def test_parse_env_command_context_consumes_reset_after_option_terminator() -> None: + context = parse_env_command_context( + ["env", "--", "-", f"{_TRACKED_ENVIRONMENT}=configured", "provider"], + "inherited", + _TRACKED_ENVIRONMENT, + ) + + assert context == EnvCommandContext( + command_executable="provider", + command_arguments=(), + tracked_environment_value="configured", + tracked_environment_changed=True, + command_search_path=None, + command_working_directory=None, + command_working_directory_option=None, + ) + + +def test_parse_env_command_context_consumes_option_looking_values() -> None: + context = parse_env_command_context( + ["env", "--argv0", "--provider-alias", "provider"], + "inherited", + _TRACKED_ENVIRONMENT, + ) + + assert context.command_executable == "provider" + assert context.command_arguments == () + + +@pytest.mark.parametrize( + ("tokens", "message"), + [ + pytest.param( + ["env", "--unknown", "provider"], + "Cannot validate env option '--unknown' with a workflow reasoning request.", + id="unknown-long-option", + ), + pytest.param( + ["env", "-vx", "provider"], + "Cannot validate env option '-x' with a workflow reasoning request.", + id="unknown-short-option", + ), + pytest.param( + ["env", "--split-string=value", "provider"], + "--split-string cannot be combined with a workflow reasoning request.", + id="split-string", + ), + pytest.param( + ["env", "-S"], + "-S requires a value.", + id="split-string-missing-value", + ), + pytest.param( + ["env", "-Svalue", "provider"], + "-S cannot be combined with a workflow reasoning request.", + id="short-split-string", + ), + pytest.param( + ["env", "--unset"], + "--unset requires a value.", + id="unset-missing-value", + ), + ], +) +def test_parse_env_command_context_preserves_validation_errors( + tokens: list[str], + message: str, +) -> None: + with pytest.raises(ValueError) as exc_info: + parse_env_command_context(tokens, "inherited", _TRACKED_ENVIRONMENT) + + assert str(exc_info.value) == message diff --git a/tests/unit/adapters/invokers/cli_invoker/test_machine_json.py b/tests/unit/adapters/invokers/cli_invoker/test_machine_json.py index 8f8ea09..98d0389 100644 --- a/tests/unit/adapters/invokers/cli_invoker/test_machine_json.py +++ b/tests/unit/adapters/invokers/cli_invoker/test_machine_json.py @@ -1,11 +1,15 @@ from __future__ import annotations import json +from collections.abc import Iterable from pathlib import Path import pytest -from crewplane.adapters.invokers.cli_invoker import machine_json +from crewplane.adapters.invokers.cli_invoker import ( + claude_json, + machine_json, +) from crewplane.adapters.invokers.cli_invoker.machine_json import ( extract_claude_output, extract_codex_output, @@ -75,6 +79,40 @@ def test_claude_output_extractor_uses_stderr_when_stdout_is_empty() -> None: extracted.output_path.unlink(missing_ok=True) +def test_claude_output_extractor_uses_stderr_after_valid_missing_stdout() -> None: + extracted = extract_claude_output( + CommandResult(0, "{}", '{"result":"stderr response"}'), + None, + ) + + assert extracted.output_extraction_status == "success" + assert extracted.output_path is not None + try: + assert extracted.output_path.read_text(encoding="utf-8") == "stderr response" + finally: + extracted.output_path.unlink(missing_ok=True) + + +@pytest.mark.parametrize( + ("stdout_text", "expected_status"), + [ + ('{"result":', "malformed"), + ('{"result":" "}', "missing"), + ], +) +def test_claude_output_extractor_does_not_fall_back_after_selected_stdout_result( + stdout_text: str, + expected_status: str, +) -> None: + extracted = extract_claude_output( + CommandResult(0, stdout_text, '{"result":"stderr response"}'), + None, + ) + + assert extracted.output_extraction_status == expected_status + assert extracted.output_path is None + + @pytest.mark.parametrize( "stdout_text", [ @@ -82,6 +120,13 @@ def test_claude_output_extractor_uses_stderr_when_stdout_is_empty() -> None: '{"result":123}', '{"result":"bad\\q"}', '{"result":"bad\\uZZZZ"}', + '{"result":"accepted","metadata":invalid}', + '{"result":"accepted","metadata":"bad\\q"}', + '{"result":"accepted","metadata":{"ok":true "bad":false}}', + '{"result":"accepted","metadata":[1 2]}', + '{"result":"accepted","metadata":[1,]}', + '{"result":"accepted","metadata":{"item":1,}}', + '{"result":"accepted","metadata":{1:"bad"}}', '{"result":"unterminated', '{"result":"ok"} trailing', '{"result":"ok" "other":1}', @@ -114,6 +159,93 @@ def test_claude_output_extractor_decodes_escaped_and_nested_values() -> None: extracted.output_path.unlink(missing_ok=True) +def test_claude_output_extractor_uses_last_duplicate_result() -> None: + extracted = extract_claude_output( + CommandResult(0, '{"result":"first","result":"last"}', ""), + None, + ) + + assert extracted.output_extraction_status == "success" + assert extracted.output_path is not None + try: + assert extracted.output_path.read_text(encoding="utf-8") == "last" + assert extracted.output_char_count == 4 + finally: + extracted.output_path.unlink(missing_ok=True) + + +def test_claude_output_extractor_removes_owned_file_after_unexpected_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + output_path = tmp_path / "owned-result.txt" + + def create_output_file() -> Path: + output_path.touch() + return output_path + + def raise_read_error(chunks: Iterable[str], selected_path: Path) -> int | None: + assert list(chunks) + assert selected_path == output_path + raise OSError("read failed") + + monkeypatch.setattr(claude_json, "new_owned_output_file", create_output_file) + monkeypatch.setattr(claude_json, "parse_claude_result", raise_read_error) + + with pytest.raises(OSError, match="read failed"): + claude_json.extract_claude_output( + CommandResult(0, '{"result":"ignored"}', ""), + 1, + ) + + assert not output_path.exists() + + +def test_claude_output_extractor_removes_owned_file_after_scan_error( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + output_path = tmp_path / "owned-result.txt" + + def create_output_file() -> Path: + output_path.touch() + return output_path + + def raise_scan_error(selected_path: Path) -> bool: + assert selected_path == output_path + raise OSError("scan failed") + + monkeypatch.setattr(claude_json, "new_owned_output_file", create_output_file) + monkeypatch.setattr( + claude_json, + "path_has_non_whitespace_text", + raise_scan_error, + ) + + with pytest.raises(OSError, match="scan failed"): + claude_json.extract_claude_output( + CommandResult(0, '{"result":"ignored"}', ""), + 1, + ) + + assert not output_path.exists() + + +def test_claude_output_extractor_accepts_mixed_empty_nested_values() -> None: + stdout_text = ( + '{"ignored":[{},[],{"items":[null,true,false,{"nested":[]}]}],"result":"ok"}' + ) + + extracted = extract_claude_output(CommandResult(0, stdout_text, ""), None) + + assert extracted.output_extraction_status == "success" + assert extracted.output_path is not None + try: + assert extracted.output_path.read_text(encoding="utf-8") == "ok" + finally: + extracted.output_path.unlink(missing_ok=True) + + def test_claude_output_extractor_accepts_deeply_nested_ignored_value() -> None: depth = 1_200 stdout_text = '{"ignored":' + "[" * depth + "0" + "]" * depth + ',"result":"ok"}' @@ -129,6 +261,60 @@ def test_claude_output_extractor_accepts_deeply_nested_ignored_value() -> None: extracted.output_path.unlink(missing_ok=True) +@pytest.mark.parametrize( + "ignored_value", + [ + "null", + "true", + "false", + "0", + "-0", + "1234567890", + "-12.5", + "6.022e23", + "1E-9", + ], +) +def test_claude_output_extractor_accepts_valid_ignored_scalars( + ignored_value: str, +) -> None: + extracted = extract_claude_output( + CommandResult(0, f'{{"ignored":{ignored_value},"result":"ok"}}', ""), + None, + ) + + assert extracted.output_extraction_status == "success" + assert extracted.output_path is not None + extracted.output_path.unlink(missing_ok=True) + + +@pytest.mark.parametrize( + "ignored_value", + ["Null", "tru", "falsehood", "+1", "01", "1.", ".1", "1e", "1e+"], +) +def test_claude_output_extractor_rejects_invalid_ignored_scalars( + ignored_value: str, +) -> None: + extracted = extract_claude_output( + CommandResult(0, f'{{"result":"ok","ignored":{ignored_value}}}', ""), + None, + ) + + assert extracted.output_extraction_status == "malformed" + + +def test_claude_output_extractor_streams_large_ignored_number() -> None: + ignored_value = "1" * 1_000_000 + extracted = extract_claude_output( + CommandResult(0, f'{{"ignored":{ignored_value},"result":"ok"}}', ""), + None, + ) + + assert extracted.output_extraction_status == "success" + assert extracted.output_path is not None + extracted.output_path.unlink(missing_ok=True) + + def test_claude_usage_parser_reports_missing_and_malformed_payloads() -> None: missing = machine_json.read_claude_model_usage(CommandResult(0, "", "")) malformed = machine_json.read_claude_model_usage( @@ -153,11 +339,24 @@ def test_claude_usage_parser_uses_stderr_when_stdout_is_empty() -> None: assert usage == {"model": {"inputTokens": 4}} +def test_claude_usage_parser_does_not_fall_back_after_valid_stdout() -> None: + usage, error = machine_json.read_claude_model_usage( + CommandResult( + 0, + "{}", + '{"modelUsage":{"model":{"inputTokens":4}}}', + ) + ) + + assert usage is None + assert error is None + + def test_claude_usage_parser_bounds_captured_model_usage(monkeypatch) -> None: monkeypatch.setattr(machine_json, "MAX_CAPTURED_CLAUDE_USAGE_BYTES", 1) usage, error = machine_json.read_claude_model_usage( - CommandResult(0, '{"modelUsage":{"model":{"inputTokens":1}}}', "") + CommandResult(0, '{"modelUsage":12}', "") ) assert usage is None diff --git a/tests/unit/core/workflow_validation/test_workflow_validation_provider_and_budget.py b/tests/unit/core/workflow_validation/test_workflow_validation_provider_and_budget.py index b29b169..40c0dcd 100644 --- a/tests/unit/core/workflow_validation/test_workflow_validation_provider_and_budget.py +++ b/tests/unit/core/workflow_validation/test_workflow_validation_provider_and_budget.py @@ -1,7 +1,10 @@ +import shutil import stat import tempfile import unittest +from collections.abc import Callable from pathlib import Path +from unittest.mock import patch from crewplane.adapters.invokers.cli import collect_cli_availability_errors from crewplane.core.config import AgentConfig, Config, Settings @@ -85,6 +88,251 @@ def test_cli_adapter_validation_reports_missing_cli(self) -> None: self.assertEqual(len(errors), 1) self.assertIn("ghost-agent", errors[0]) + def test_cli_adapter_validation_reports_missing_env_wrapped_cli(self) -> None: + workflow = WorkflowPlan( + name="Workflow", + nodes=[ + WorkflowNode( + id="node.a", + mode="parallel", + prompt_segments=[ + PromptSegment(role=PromptSegmentRole.SHARED, content="p") + ], + providers=[ProviderSpec(provider="wrapped-agent")], + ) + ], + ) + missing_executable = "crewplane-provider-that-does-not-exist" + config = Config( + version=SCHEMA_VERSION, + agents={ + "wrapped-agent": AgentConfig( + cli_cmd=["env", "CREWPLANE_REPRO=1", missing_executable], + default_model="x", + ), + }, + ) + + probed_executables: list[str] = [] + + def executable_lookup(executable: str) -> str | None: + probed_executables.append(executable) + return _platform_env_executable(executable) + + errors = collect_cli_availability_errors( + workflow, + config, + which_fn=executable_lookup, + ) + + self.assertEqual(probed_executables, ["env", missing_executable]) + self.assertEqual(len(errors), 1) + self.assertIn(f"CLI '{missing_executable}' not found in PATH", errors[0]) + self.assertIn("wrapped-agent", errors[0]) + + def test_cli_adapter_validation_reports_missing_path_qualified_env_cli( + self, + ) -> None: + missing_executable = "crewplane-provider-that-does-not-exist" + + errors = _collect_wrapped_cli_errors( + ["/usr/bin/env", missing_executable], + Path.cwd(), + _platform_env_executable, + ) + + self.assertEqual(len(errors), 1) + self.assertIn(f"CLI '{missing_executable}' not found in PATH", errors[0]) + + def test_cli_adapter_validation_resolves_path_qualified_custom_env_from_project_root( + self, + ) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "env") + + def launcher_lookup(executable: str) -> str | None: + return "/usr/bin/env" if executable == "./env" else None + + errors = _collect_wrapped_cli_errors( + ["./env", "serve"], + project_root, + launcher_lookup, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_preserves_path_resolved_custom_env( + self, + ) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "env") + + with patch.dict("os.environ", {"PATH": str(project_root)}): + errors = _collect_wrapped_cli_errors( + ["env", "serve"], + project_root, + shutil.which, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_uses_env_path_assignment(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "configured-bin" / "custom-provider") + + errors = _collect_wrapped_cli_errors( + ["env", "PATH=configured-bin", "custom-provider"], + project_root, + _platform_env_executable, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_uses_env_reset_after_option_terminator( + self, + ) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "configured-bin" / "custom-provider") + + errors = _collect_wrapped_cli_errors( + [ + "env", + "--", + "-", + "PATH=configured-bin", + "custom-provider", + ], + project_root, + _platform_env_executable, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_treats_env_terminator_after_reset_as_command( + self, + ) -> None: + errors = _collect_wrapped_cli_errors( + ["env", "-", "--"], + Path.cwd(), + _platform_env_executable, + ) + + self.assertEqual(len(errors), 1) + self.assertIn("CLI '--' not found in PATH", errors[0]) + + def test_cli_adapter_validation_resolves_inherited_relative_env_path_from_project( + self, + ) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "inherited-bin" / "custom-provider") + + with patch.dict("os.environ", {"PATH": "inherited-bin"}): + errors = _collect_wrapped_cli_errors( + ["env", "custom-provider"], + project_root, + _platform_env_executable, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_uses_env_explicit_search_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "explicit-bin" / "custom-provider") + + errors = _collect_wrapped_cli_errors( + [ + "env", + "-Pexplicit-bin", + "PATH=missing-bin", + "custom-provider", + ], + project_root, + _platform_env_executable, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_reports_missing_cli_after_env_chdir(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + (project_root / "work").mkdir() + + errors = _collect_wrapped_cli_errors( + ["env", "-C", "work", "missing-provider"], + project_root, + _platform_env_executable, + ) + + self.assertEqual(len(errors), 1) + self.assertIn("CLI 'missing-provider' not found in PATH", errors[0]) + + def test_cli_adapter_validation_uses_env_chdir_for_relative_cli(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "work" / "bin" / "custom-provider") + + errors = _collect_wrapped_cli_errors( + ["env", "--chdir=work", "./bin/custom-provider"], + project_root, + _platform_env_executable, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_uses_env_chdir_for_relative_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + _write_executable(project_root / "work" / "bin" / "custom-provider") + + errors = _collect_wrapped_cli_errors( + ["env", "-Cwork", "PATH=bin", "custom-provider"], + project_root, + _platform_env_executable, + ) + + self.assertEqual(errors, []) + + def test_cli_adapter_validation_rejects_provider_outside_env_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + (project_root / "configured-bin").mkdir() + + def executable_lookup(executable: str) -> str | None: + if executable == "env": + return _platform_env_executable(executable) + return f"/parent/{executable}" + + errors = _collect_wrapped_cli_errors( + ["env", "PATH=configured-bin", "parent-provider"], + project_root, + executable_lookup, + ) + + self.assertEqual(len(errors), 1) + self.assertIn("CLI 'parent-provider' not found in PATH", errors[0]) + + def test_cli_adapter_validation_rechecks_env_when_wrapped_with_custom_path( + self, + ) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + project_root = Path(tmp_dir) + (project_root / "configured-bin").mkdir() + + errors = _collect_wrapped_cli_errors( + ["env", "PATH=configured-bin", "env"], + project_root, + _platform_env_executable, + ) + + self.assertEqual(len(errors), 1) + self.assertIn("CLI 'env' not found in PATH", errors[0]) + def test_cli_adapter_validation_checks_relative_path_executable(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: project_root = Path(tmp_dir) @@ -303,3 +551,47 @@ def test_topological_waves_preserve_frontmatter_order(self) -> None: def _missing_executable(executable: str) -> str | None: # noqa: ARG001 return None + + +def _platform_env_executable(executable: str) -> str | None: + if executable not in {"env", "/usr/bin/env"}: + return None + return "/usr/bin/env" + + +def _write_executable(path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + path.chmod(path.stat().st_mode | stat.S_IXUSR) + + +def _collect_wrapped_cli_errors( + cli_cmd: list[str], + project_root: Path, + which_fn: Callable[[str], str | None], +) -> list[str]: + workflow = WorkflowPlan( + name="Workflow", + nodes=[ + WorkflowNode( + id="node.a", + mode="parallel", + prompt_segments=[ + PromptSegment(role=PromptSegmentRole.SHARED, content="p") + ], + providers=[ProviderSpec(provider="wrapped-agent")], + ) + ], + ) + config = Config( + version=SCHEMA_VERSION, + agents={ + "wrapped-agent": AgentConfig(cli_cmd=cli_cmd, default_model="x"), + }, + ) + return collect_cli_availability_errors( + workflow, + config, + which_fn=which_fn, + project_root=project_root, + ) diff --git a/tests/unit/runtime/agent/test_reasoning_arguments.py b/tests/unit/runtime/agent/test_reasoning_arguments.py index deabb9f..55e3b4b 100644 --- a/tests/unit/runtime/agent/test_reasoning_arguments.py +++ b/tests/unit/runtime/agent/test_reasoning_arguments.py @@ -239,6 +239,11 @@ def test_env_prefix_can_remove_inherited_reasoning_or_use_safe_options( "Cannot validate env option '--unknown'", id="unknown-long", ), + pytest.param( + ["env", "--path=bin", "claude"], + "Cannot validate env option '--path=bin'", + id="unsupported-path-long", + ), pytest.param( ["env", "--unset"], "--unset requires a value",