diff --git a/docs/reference/mcp.md b/docs/reference/mcp.md index b5665f9aa5..ddf1efa75d 100644 --- a/docs/reference/mcp.md +++ b/docs/reference/mcp.md @@ -25,6 +25,7 @@ read-only, local-only (no network access), and reject unexpected arguments. | `specify_version` | Return the installed CLI version, runtime, system, and feature capabilities. Takes no arguments. | | `specify_artifact_list` | Return one paginated page of a project's artifact inventory (`command`, `template`, `script`, `hook`) | | `specify_artifact_info` | Return one artifact and its complete composition stack | +| `specify_workflow_list` | List installed workflows and skipped-entry warnings in registry order | The artifact tools take an optional absolute `project_directory`; it defaults to the directory the server was launched from. @@ -50,7 +51,47 @@ tool error. | `specify_run_command` | Run one supported command through the real Specify CLI | The server does not expose mutations, installation or update operations, -workflows, confirmations, or access tiers. +workflow execution, confirmations, or access tiers. + +## Installed workflow list + +`specify_workflow_list` accepts an optional absolute `project_directory` +(default: the server launch directory) and `limit` (1–1000, default 100). +It reads only the installed workflow registry; it never loads custom steps, +fetches catalogs, repairs files, or executes workflows. + +The result has ordered `items` and a `truncated` boolean. Workflow items have +`kind: "workflow"`, `id`, `name`, `version`, nullable `description`, and `enabled`. +Malformed individual registry entries produce `kind: "warning"` items with +`code: "invalid_registry_entry"`, `workflow_id`, and `message` in their original +positions. The limit counts both workflows and warnings. `truncated: true` +explicitly indicates omitted entries; increase `limit` to retrieve more, up to +1000. There is no continuation cursor. The CLI retains its complete listing. + +Missing registry files produce an empty list. Corrupt or unreadable registry +files produce `workflow_registry_read_failed`; unsafe storage produces +`unsafe_workflow_storage`. Other expected errors are `invalid_project_directory` +and `not_a_spec_kit_project`. Errors contain `code`, `message`, +`details`, and `retryable` under `error`, with `isError: true`. +Unexpected failures use sanitized `internal_error`; invalid operation outcomes +use `invalid_operation_result`. Responses exceeding 1 MiB produce a retryable +`response_too_large` error instead of partial success; retry with a smaller +limit. Registry exception details are not exposed through MCP. + +Out-of-range `limit` values are rejected before the operation runs and produce +text-only input-validation errors with no structured content; see +[Input validation errors](#input-validation-errors). + +Workflow listing reads registry content through the selected project's +`FileHelper` policy owner, rejecting Windows junctions and unsupported reparse +points at the read boundary. Ordinary workflow registry symlinks remain +disallowed. Other workflow registry callers are not migrated by this tool. +These checks do not provide an OS sandbox or race-free filesystem confinement; +the registry is still parsed in full before the entry limit is applied. + +Operation outcomes exceeding the requested entry limit, or claiming truncation +without filling that limit, are rejected as `invalid_operation_result`. The +output schema also caps `items` at 1000. ## Input validation errors diff --git a/src/specify_cli/mcp_server/server.py b/src/specify_cli/mcp_server/server.py index c7b5b6dae9..708410c79a 100644 --- a/src/specify_cli/mcp_server/server.py +++ b/src/specify_cli/mcp_server/server.py @@ -10,6 +10,7 @@ from ..artifacts._mcp import register as register_artifacts from ..mcp_version import register as register_version +from ..workflows._mcp import register as register_workflows from .catalog import ( CommandAdapterError, CommandDescription, @@ -89,6 +90,7 @@ def specify_run_command(command: str) -> VersionResult: register_version(server) register_artifacts(server, launch_directory=server_launch_directory) + register_workflows(server, launch_directory=server_launch_directory) return server diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 8cd60c7f4b..3c380d9f5b 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -183,16 +183,13 @@ def _reject_unsafe_dir(path: Path, label: str) -> None: def _reject_unsafe_workflow_storage(project_root: Path) -> None: """Refuse symlinked workflow storage directories before workflow commands run.""" - _reject_unsafe_dir(project_root / ".specify", ".specify") - _reject_unsafe_dir(project_root / ".specify" / "workflows", ".specify/workflows") - _reject_unsafe_dir( - project_root / ".specify" / "workflows" / "runs", - ".specify/workflows/runs", - ) - _reject_unsafe_dir( - project_root / ".specify" / "workflows" / "overlays", - ".specify/workflows/overlays", - ) + from ._storage import WorkflowStorageError, validate_workflow_storage + + try: + validate_workflow_storage(project_root) + except WorkflowStorageError as exc: + err_console.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) from exc def _resolve_installed_workflow_ownership( diff --git a/src/specify_cli/workflows/_mcp.py b/src/specify_cli/workflows/_mcp.py index 4477c00d75..fd0f3d75d9 100644 --- a/src/specify_cli/workflows/_mcp.py +++ b/src/specify_cli/workflows/_mcp.py @@ -1,7 +1,7 @@ """Explicit MCP dispositions for the complete ``specify workflow`` hierarchy. -No workflow adapter is registered yet. These records describe the conservative -union of the current CLI paths, not a promise of a future MCP contract. +Only workflow.list has a shared operation and adapter. Unavailable records +describe the conservative union of CLI paths, not a future MCP contract. Effect boundaries worth retaining when adapters are introduced: @@ -23,8 +23,14 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import Path from typing import Literal +from mcp.server import MCPServer + +from ._operation_list import WORKFLOW_LIST_OPERATION +from .mcp_list import register as register_list + WorkflowDisposition = Literal["available", "unavailable", "excluded"] WorkflowCapability = Literal[ "local-read", "project-write", "execution", "self-modifying" @@ -101,12 +107,11 @@ class WorkflowToolInventory: operation_id="workflow.list", cli_path="specify workflow list", mcp_tool_name="specify_workflow_list", - disposition_reason=( - "WorkflowRegistry.list returns registry dictionaries rendered by the CLI; " - "no shared typed list outcome or MCP adapter." - ), - capabilities=_LOCAL_READ, - network_access="none", + disposition_reason=None, + disposition="available", + contract_version=WORKFLOW_LIST_OPERATION.contract_version, + capabilities=WORKFLOW_LIST_OPERATION.capabilities, + network_access=WORKFLOW_LIST_OPERATION.network_access, ), WorkflowToolInventory( operation_id="workflow.remove", @@ -375,3 +380,14 @@ class WorkflowToolInventory: network_access="none", ), ) + + +def register(server: MCPServer, *, launch_directory: Path) -> None: + """Register only the workflow list operation from its owning hierarchy.""" + record = next( + item for item in WORKFLOW_TOOLS if item.operation_id == "workflow.list" + ) + if record.disposition == "available": + register_list( + server, launch_directory=launch_directory, tool_name=record.mcp_tool_name + ) diff --git a/src/specify_cli/workflows/_operation_list.py b/src/specify_cli/workflows/_operation_list.py new file mode 100644 index 0000000000..3f0fa902d5 --- /dev/null +++ b/src/specify_cli/workflows/_operation_list.py @@ -0,0 +1,149 @@ +"""Shared, read-only application operation for installed workflow listing.""" + +from __future__ import annotations + +from dataclasses import dataclass +from itertools import islice +from pathlib import Path +import stat +from typing import Literal + +from .._file_policy import ProjectPolicyError, project_files +from ._storage import WorkflowStorageError, validate_workflow_storage +from .catalog import WorkflowRegistry + +WORKFLOW_LIST_MAX_LIMIT = 1000 + + +@dataclass(frozen=True) +class WorkflowListRequest: + project_directory: Path + # CLI retains its complete listing. MCP supplies an explicit bounded limit. + limit: int | None = None + + +@dataclass(frozen=True) +class WorkflowListEntry: + id: str + name: str + version: str + description: str | None + enabled: bool + kind: Literal["workflow"] = "workflow" + + +@dataclass(frozen=True) +class WorkflowListWarning: + workflow_id: str + message: str + code: Literal["invalid_registry_entry"] = "invalid_registry_entry" + kind: Literal["warning"] = "warning" + + +@dataclass(frozen=True) +class WorkflowListResult: + # Warnings stay in registry order, including an all-invalid registry. + items: tuple[WorkflowListEntry | WorkflowListWarning, ...] + truncated: bool = False + + +class WorkflowListError(Exception): + """Expected operation failure; adapters own presentation and sanitization.""" + + def __init__(self, code: str, message: str) -> None: + self.code = code + self.details: dict[str, str] = {} + self.retryable = False + super().__init__(message) + + +@dataclass(frozen=True) +class WorkflowListOperationDescriptor: + operation_id: Literal["workflow.list"] + contract_version: Literal["1"] + request_type: type[WorkflowListRequest] + result_type: type[WorkflowListResult] + warning_types: tuple[type[WorkflowListWarning], ...] + error_types: tuple[type[WorkflowListError], ...] + capabilities: frozenset[Literal["local-read"]] + network_access: Literal["none"] + + +WORKFLOW_LIST_OPERATION = WorkflowListOperationDescriptor( + operation_id="workflow.list", + contract_version="1", + request_type=WorkflowListRequest, + result_type=WorkflowListResult, + warning_types=(WorkflowListWarning,), + error_types=(WorkflowListError,), + capabilities=frozenset({"local-read"}), + network_access="none", +) + + +def list_workflows(request: WorkflowListRequest) -> WorkflowListResult: + """Read the registry without importing custom steps or repairing its contents.""" + root = request.project_directory + if not isinstance(root, Path) or not root.is_absolute(): + raise WorkflowListError( + "invalid_project_directory", "project_directory must be an absolute path" + ) + limit = request.limit + if limit is not None and ( + type(limit) is not int or not 1 <= limit <= WORKFLOW_LIST_MAX_LIMIT + ): + raise WorkflowListError( + "invalid_limit", f"limit must be between 1 and {WORKFLOW_LIST_MAX_LIMIT}" + ) + try: + # Unlike is_dir(), stat() preserves permission errors on Python 3.14. + try: + root_is_directory = stat.S_ISDIR(root.stat().st_mode) + except (FileNotFoundError, NotADirectoryError, ValueError): + root_is_directory = False + if not root_is_directory: + raise WorkflowListError( + "not_a_spec_kit_project", + "Not a Spec Kit project (no .specify/ directory)", + ) + files = project_files(root) + if not (root / ".specify").is_dir(): + raise WorkflowListError( + "not_a_spec_kit_project", + "Not a Spec Kit project (no .specify/ directory)", + ) + validate_workflow_storage(root) + except (ProjectPolicyError, WorkflowStorageError, OSError) as exc: + raise WorkflowListError("unsafe_workflow_storage", str(exc)) from exc + + try: + installed = WorkflowRegistry(root, read_files=files).list() + except OSError as exc: + raise WorkflowListError( + "workflow_registry_read_failed", f"Failed to read workflow registry: {exc}" + ) from exc + + items: list[WorkflowListEntry | WorkflowListWarning] = [] + selected = installed.items() if limit is None else islice(installed.items(), limit) + for workflow_id, metadata in selected: + if not isinstance(metadata, dict): + items.append( + WorkflowListWarning( + workflow_id=workflow_id, + message=f"Skipping corrupted registry entry '{workflow_id}'.", + ) + ) + continue + description = metadata.get("description", "") + items.append( + WorkflowListEntry( + id=workflow_id, + name=str(metadata.get("name", workflow_id)), + version=str(metadata.get("version", "?")), + description=str(description) if description else None, + enabled=bool(metadata.get("enabled", True)), + ) + ) + return WorkflowListResult( + items=tuple(items), truncated=limit is not None and len(installed) > limit + ) diff --git a/src/specify_cli/workflows/_storage.py b/src/specify_cli/workflows/_storage.py new file mode 100644 index 0000000000..f33f3daf03 --- /dev/null +++ b/src/specify_cli/workflows/_storage.py @@ -0,0 +1,22 @@ +"""Transport-neutral validation of existing workflow storage boundaries.""" + +from pathlib import Path + + +class WorkflowStorageError(ValueError): + """A workflow storage directory is symlinked or has the wrong type.""" + + +def validate_workflow_storage(project_root: Path) -> None: + """Check storage without creating directories or following child symlinks.""" + for suffix in ( + ".specify", + ".specify/workflows", + ".specify/workflows/runs", + ".specify/workflows/overlays", + ): + path = project_root / suffix + if path.is_symlink(): + raise WorkflowStorageError(f"Refusing to use symlinked {suffix} path") + if path.exists() and not path.is_dir(): + raise WorkflowStorageError(f"{suffix} path exists but is not a directory") diff --git a/src/specify_cli/workflows/catalog/_domain.py b/src/specify_cli/workflows/catalog/_domain.py index cc29a0b0ca..4aaf82532a 100644 --- a/src/specify_cli/workflows/catalog/_domain.py +++ b/src/specify_cli/workflows/catalog/_domain.py @@ -26,6 +26,7 @@ MAX_JSON_CATALOG_BYTES as MAX_JSON_CATALOG_BYTES, read_response_limited, ) +from ...file_helper import FileHelper, FileHelperError from .._catalog_cache import read_cache_json, write_cache_json logger = logging.getLogger(__name__) @@ -82,10 +83,15 @@ class WorkflowRegistry: REGISTRY_FILE = "workflow-registry.json" SCHEMA_VERSION = "1.0" - def __init__(self, project_root: Path) -> None: + def __init__( + self, project_root: Path, *, read_files: FileHelper | None = None + ) -> None: self.project_root = project_root self.workflows_dir = project_root / ".specify" / "workflows" self.registry_path = self.workflows_dir / self.REGISTRY_FILE + # Optional read-policy handoff for workflow.list. Existing callers and + # all mutation paths retain their established behavior. + self._read_files = read_files self.data = self._load() def _has_symlinked_parent(self) -> bool: @@ -116,14 +122,22 @@ def _load(self) -> dict[str, Any]: f"Refusing to read workflow registry at {self.registry_path}: " "a parent directory or the registry file itself is a symlink" ) - if self.registry_path.exists(): + if self._read_files is not None or self.registry_path.exists(): try: - with open(self.registry_path, encoding="utf-8") as f: - data = json.load(f) - except OSError as exc: - # The real data may still be intact on disk. Fail closed at - # construction rather than fabricating an empty registry that - # a read-only caller could mistake for "nothing installed." + if self._read_files is None: + with open(self.registry_path, encoding="utf-8") as f: + data = json.load(f) + else: + try: + text = self._read_files.read_text(self.registry_path) + except FileNotFoundError: + return default_registry + # Match the legacy text stream's universal-newline handling, + # including JSON error offsets for malformed CRLF files. + text = text.replace("\r\n", "\n").replace("\r", "\n") + data = json.loads(text) + except (OSError, FileHelperError) as exc: + # Do not fabricate an empty registry on access/policy failures. raise OSError( f"Failed to read workflow registry at {self.registry_path}: {exc}" ) from exc diff --git a/src/specify_cli/workflows/command_list.py b/src/specify_cli/workflows/command_list.py index 69cc3be395..d803a2f91b 100644 --- a/src/specify_cli/workflows/command_list.py +++ b/src/specify_cli/workflows/command_list.py @@ -3,34 +3,38 @@ from __future__ import annotations from . import _commands as cli +from . import _operation_list as operation @cli.workflow_app.command("list") def workflow_list(): """List installed workflows.""" project_root = cli._require_specify_project() - registry = cli._open_workflow_registry(project_root) - installed = registry.list() + try: + result = operation.list_workflows(operation.WorkflowListRequest(project_root)) + except operation.WorkflowListError as exc: + cli.console.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}") + raise cli.typer.Exit(1) from exc - if not installed: + if not result.items: cli.console.print("[yellow]No workflows installed.[/yellow]") cli.console.print("\nInstall a workflow with:") cli.console.print(" [cyan]specify workflow add [/cyan]") return cli.console.print("\n[bold cyan]Installed Workflows:[/bold cyan]\n") - for wf_id, wf_data in installed.items(): - safe_id = cli._escape_markup(wf_id) - if not isinstance(wf_data, dict): + for item in result.items: + if isinstance(item, operation.WorkflowListWarning): cli.console.print( - f" [yellow]Warning:[/yellow] Skipping corrupted registry entry '{safe_id}'.\n" + f" [yellow]Warning:[/yellow] {cli._escape_markup(item.message)}\n" ) continue - marker = "" if wf_data.get("enabled", True) else " [red]\\[disabled][/red]" - name = cli._escape_markup(str(wf_data.get("name", wf_id))) - version = cli._escape_markup(str(wf_data.get("version", "?"))) + safe_id = cli._escape_markup(item.id) + marker = "" if item.enabled else " [red]\\[disabled][/red]" + name = cli._escape_markup(item.name) + version = cli._escape_markup(item.version) cli.console.print(f" [bold]{name}[/bold] ({safe_id}) v{version}{marker}") - desc = wf_data.get("description", "") + desc = item.description if desc: cli.console.print(f" {cli._escape_markup(str(desc))}") cli.console.print() diff --git a/src/specify_cli/workflows/mcp_list.py b/src/specify_cli/workflows/mcp_list.py new file mode 100644 index 0000000000..9c5bcd5e53 --- /dev/null +++ b/src/specify_cli/workflows/mcp_list.py @@ -0,0 +1,230 @@ +"""Strict MCP projection of the shared installed-workflow list operation.""" + +from __future__ import annotations + +import logging +from collections.abc import Callable +from dataclasses import asdict +from pathlib import Path +from typing import Annotated, Any, Literal + +from mcp.server import MCPServer +from mcp.types import CallToolResult, TextContent, ToolAnnotations +from pydantic import BaseModel, ConfigDict, Field + +from . import _operation_list as operation + +logger = logging.getLogger(__name__) +_MAX_WIRE_RESPONSE_BYTES = 1024 * 1024 +_WIRE_ENVELOPE_RESERVE_BYTES = 1024 +WorkflowListLimit = Annotated[ + int, Field(strict=True, ge=1, le=operation.WORKFLOW_LIST_MAX_LIMIT) +] +WorkflowProjectDirectory = Annotated[str, Field(strict=True)] + + +class WorkflowListToolInput(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + project_directory: WorkflowProjectDirectory | None = None + limit: WorkflowListLimit = 100 + + +class WorkflowEntryResult(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + kind: Literal["workflow"] + id: str + name: str + version: str + description: str | None + enabled: bool + + +class WorkflowWarningResult(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + kind: Literal["warning"] + code: Literal["invalid_registry_entry"] + workflow_id: str + message: str + + +class WorkflowListToolResult(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + items: list[ + Annotated[ + WorkflowEntryResult | WorkflowWarningResult, Field(discriminator="kind") + ] + ] = Field(max_length=operation.WORKFLOW_LIST_MAX_LIMIT) + truncated: bool + + +WorkflowListCallResult = Annotated[CallToolResult, WorkflowListToolResult] +WorkflowListTool = Callable[[str | None, int], CallToolResult] + + +class _InvalidOperationResult(Exception): + """The shared outcome does not match the declared result contract.""" + + +def _convert_result( + result: operation.WorkflowListResult, + *, + requested_limit: int, +) -> WorkflowListToolResult: + try: + if not isinstance(result, operation.WorkflowListResult): + raise TypeError("Expected WorkflowListResult") + if len(result.items) > requested_limit: + raise ValueError("Operation exceeded requested limit") + if result.truncated and len(result.items) != requested_limit: + raise ValueError("Truncated result must fill the requested limit") + return WorkflowListToolResult.model_validate( + { + "items": [asdict(item) for item in result.items], + "truncated": result.truncated, + }, + strict=True, + ) + except (TypeError, ValueError, AttributeError) as exc: + raise _InvalidOperationResult from exc + + +def _tool_error(code: str, message: str, *, retryable: bool = False) -> CallToolResult: + return CallToolResult( + content=[TextContent(type="text", text=f"{code}: {message}")], + structured_content={ + "error": { + "code": code, + "message": message, + "details": {}, + "retryable": retryable, + } + }, + is_error=True, + ) + + +def _wire_response_size(result: CallToolResult) -> int: + return ( + len(result.model_dump_json(by_alias=True, exclude_none=True).encode("utf-8")) + + _WIRE_ENVELOPE_RESERVE_BYTES + ) + + +def _response_too_large() -> CallToolResult: + return _tool_error( + "response_too_large", + "Workflow list response exceeds the byte limit; retry with a smaller limit.", + retryable=True, + ) + + +def create_workflow_list_tool( + *, + launch_directory: Path, + max_wire_response_bytes: int = _MAX_WIRE_RESPONSE_BYTES, +) -> WorkflowListTool: + """Bind the default project once; never change process cwd for requests.""" + launch_directory = Path(launch_directory) + if not launch_directory.is_absolute(): + raise ValueError("MCP server launch directory must be absolute") + if _wire_response_size(_response_too_large()) > max_wire_response_bytes: + raise ValueError( + "MCP response-size limit cannot hold the bounded error response" + ) + + def specify_workflow_list( + project_directory: WorkflowProjectDirectory | None = None, + limit: WorkflowListLimit = 100, + ) -> WorkflowListCallResult: + """List installed workflows and skipped-entry warnings in registry order.""" + tool_input = WorkflowListToolInput.model_validate( + {"project_directory": project_directory, "limit": limit}, strict=True + ) + try: + result = operation.list_workflows( + operation.WorkflowListRequest( + project_directory=( + launch_directory + if tool_input.project_directory is None + else Path(tool_input.project_directory) + ), + limit=tool_input.limit, + ) + ) + payload = _convert_result(result, requested_limit=tool_input.limit) + response = CallToolResult( + content=[ + TextContent(type="text", text=payload.model_dump_json(indent=2)) + ], + structured_content=payload.model_dump(mode="json"), + is_error=False, + ) + except operation.WorkflowListError as exc: + # Registry/OS exception strings can contain paths or untrusted data. + # CLI retains its established diagnostics; MCP never exposes them. + messages = { + "invalid_project_directory": "project_directory must be an absolute path.", + "not_a_spec_kit_project": "The selected directory is not a Spec Kit project.", + "unsafe_workflow_storage": "The selected workflow storage is unsafe or unreadable.", + "workflow_registry_read_failed": "Unable to read the workflow registry.", + "invalid_limit": "The workflow list limit is invalid.", + } + response = _tool_error( + exc.code, messages.get(exc.code, "Unable to list workflows.") + ) + except _InvalidOperationResult: + response = _tool_error( + "invalid_operation_result", "The workflow list result is invalid." + ) + except Exception: + logger.exception("Unexpected failure in the workflow list MCP adapter.") + response = _tool_error( + "internal_error", "Unable to list installed workflows." + ) + if _wire_response_size(response) > max_wire_response_bytes: + return _response_too_large() + return response + + return specify_workflow_list + + +def _lookup_registered_tool(server: MCPServer, name: str) -> Any: + try: + return server._tool_manager.get_tool(name) + except Exception as exc: + raise RuntimeError( + "MCP SDK compatibility error: registered tool lookup failed" + ) from exc + + +def register( + server: MCPServer, + *, + launch_directory: Path, + tool_name: str = "specify_workflow_list", +) -> None: + """Register once, closing SDK argument validation as well as discovery.""" + if _lookup_registered_tool(server, tool_name) is not None: + raise ValueError(f"MCP tool name collision: {tool_name}") + server.add_tool( + create_workflow_list_tool(launch_directory=launch_directory), + name=tool_name, + description=( + "List installed workflows and warnings without execution or network access. " + "limit bounds registry entries, including warnings; truncated reports omitted entries." + ), + annotations=ToolAnnotations( + read_only_hint=True, idempotent_hint=True, open_world_hint=False + ), + structured_output=True, + ) + try: + tool = _lookup_registered_tool(server, tool_name) + argument_model = tool.fn_metadata.arg_model + argument_model.model_config["extra"] = "forbid" + argument_model.model_rebuild(force=True) + tool.parameters = argument_model.model_json_schema(by_alias=True) + except Exception as exc: + raise RuntimeError( + "MCP SDK compatibility error while closing workflow list arguments" + ) from exc diff --git a/tests/specify_cli/mcp_server/test_server.py b/tests/specify_cli/mcp_server/test_server.py index efea3b5885..a0998fce2b 100644 --- a/tests/specify_cli/mcp_server/test_server.py +++ b/tests/specify_cli/mcp_server/test_server.py @@ -33,6 +33,7 @@ def test_tool_discovery_exposes_first_class_and_transitional_tools(tmp_path): "specify_version", "specify_artifact_list", "specify_artifact_info", + "specify_workflow_list", ] schemas = {tool.name: tool.input_schema for tool in tools} assert schemas["specify_list_commands"]["properties"] == {} diff --git a/tests/specify_cli/mcp_server/test_stdio.py b/tests/specify_cli/mcp_server/test_stdio.py index 377ddb2cd8..bd1553bccc 100644 --- a/tests/specify_cli/mcp_server/test_stdio.py +++ b/tests/specify_cli/mcp_server/test_stdio.py @@ -11,7 +11,9 @@ _TEST_TIMEOUT_SECONDS = 30 -def test_real_stdio_server_initializes_discovers_and_calls_first_class_tools(): +def test_real_stdio_server_initializes_discovers_and_calls_first_class_tools(tmp_path): + (tmp_path / ".specify").mkdir() + async def exercise() -> None: repo_root = Path(__file__).resolve().parents[3] parameters = StdioServerParameters( @@ -34,6 +36,12 @@ async def exercise() -> None: tools = await session.list_tools() listed = await session.call_tool("specify_list_commands", {}) version = await session.call_tool("specify_version", {}) + workflows = await session.call_tool( + "specify_workflow_list", {"project_directory": str(tmp_path)} + ) + workflow_failure = await session.call_tool( + "specify_workflow_list", {"project_directory": "relative"} + ) artifacts = await session.call_tool("specify_artifact_list", {}) artifact = await session.call_tool( "specify_artifact_info", @@ -65,6 +73,7 @@ async def exercise() -> None: "specify_version", "specify_artifact_list", "specify_artifact_info", + "specify_workflow_list", ] assert listed.structured_content["commands"][0]["command"] == "version" assert set(version.structured_content) == { @@ -74,6 +83,13 @@ async def exercise() -> None: "features", } assert version.is_error is False + assert workflows.is_error is False + assert workflows.structured_content == {"items": [], "truncated": False} + assert workflow_failure.is_error is True + assert ( + workflow_failure.structured_content["error"]["code"] + == "invalid_project_directory" + ) assert artifacts.is_error is False assert isinstance(artifacts.structured_content["rows"], list) assert artifacts.structured_content["rows"] diff --git a/tests/specify_cli/workflows/test_command_list.py b/tests/specify_cli/workflows/test_command_list.py index 18e42cf5c1..404a1d6be1 100644 --- a/tests/specify_cli/workflows/test_command_list.py +++ b/tests/specify_cli/workflows/test_command_list.py @@ -4,11 +4,43 @@ import json import os -from pathlib import Path import pytest +@pytest.mark.parametrize("entries,expected", [ + ({}, "No workflows installed.\n\nInstall a workflow with:\n specify workflow add \n"), + ({"bad": None}, "\nInstalled Workflows:\n\n Warning: Skipping corrupted registry entry 'bad'.\n\n"), + ({"z": {"enabled": False}, "bad": 1, "a": {"name": "[Name]", "description": "[desc]"}}, + "\nInstalled Workflows:\n\n z (z) v? [disabled]\n\n Warning: Skipping corrupted registry entry 'bad'.\n\n [Name] (a) v?\n [desc]\n\n"), +]) +def test_list_preserves_complete_cli_text(tmp_path, monkeypatch, entries, expected): + from typer.testing import CliRunner + from specify_cli import app + + directory = tmp_path / ".specify" / "workflows" + directory.mkdir(parents=True) + (directory / "workflow-registry.json").write_text(json.dumps({"workflows": entries}), encoding="utf-8") + monkeypatch.setenv("SPECIFY_INIT_DIR", str(tmp_path)) + result = CliRunner().invoke(app, ["workflow", "list"]) + assert result.exit_code == 0, result.output + assert result.output == expected + + +def test_cli_does_not_adopt_mcp_entry_limit(tmp_path, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + directory = tmp_path / ".specify" / "workflows" + directory.mkdir(parents=True) + entries = {f"workflow-{i}": {} for i in range(1001)} + (directory / "workflow-registry.json").write_text(json.dumps({"workflows": entries}), encoding="utf-8") + monkeypatch.setenv("SPECIFY_INIT_DIR", str(tmp_path)) + result = CliRunner().invoke(app, ["workflow", "list"]) + assert result.exit_code == 0, result.output + assert "workflow-1000 (workflow-1000) v?" in result.output + + class TestWorkflowAddSymlinkGuard: @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") @@ -96,7 +128,7 @@ def test_list_unreadable_registry_fails_closed_with_clean_error( is exactly the fail-open state a caller could otherwise mistake for "safe to (re)install", overwriting real files. Covers the read/query boundary fix required at every WorkflowRegistry call site.""" - import builtins + from specify_cli.file_helper import FileHelper from typer.testing import CliRunner from specify_cli import app from specify_cli.workflows.catalog import WorkflowRegistry @@ -106,14 +138,14 @@ def test_list_unreadable_registry_fails_closed_with_clean_error( self._install_dev(runner, app, project_dir) registry_path = WorkflowRegistry(project_dir).registry_path.resolve() - real_open = builtins.open + real_read = FileHelper.read_bytes - def _raising_open(file, mode="r", *args, **kwargs): - if Path(file).resolve() == registry_path and "r" in mode: + def _raising_read(files, file): + if (files.root / file).resolve() == registry_path: raise OSError("simulated read failure") - return real_open(file, mode, *args, **kwargs) + return real_read(files, file) - monkeypatch.setattr(builtins, "open", _raising_open) + monkeypatch.setattr(FileHelper, "read_bytes", _raising_read) result = runner.invoke(app, ["workflow", "list"]) assert result.exit_code != 0 assert result.exception is None or isinstance(result.exception, SystemExit) diff --git a/tests/specify_cli/workflows/test_mcp.py b/tests/specify_cli/workflows/test_mcp.py index ff56767897..3136ba89c6 100644 --- a/tests/specify_cli/workflows/test_mcp.py +++ b/tests/specify_cli/workflows/test_mcp.py @@ -26,6 +26,8 @@ ) from specify_cli.mcp_server.server import create_server from specify_cli.workflows import _commands, _mcp +from specify_cli.workflows._operation_list import WORKFLOW_LIST_OPERATION +from specify_cli.workflows.mcp_list import WorkflowListToolInput, WorkflowListToolResult from specify_cli.workflows.step import catalog as step_catalog @@ -68,13 +70,7 @@ def _assert_inventory( cli_paths: Sequence[str], tools: Sequence[Tool], ) -> None: - """Validate accounting and fail closed for available workflow records. - - No shared workflow operation descriptors exist yet. Real available records - must fail closed until operation-specific assertions bind their descriptor, - version, effects and adapter schemas, as in the artifact inventory tests. - A matching name and object schemas alone cannot establish that binding. - """ + """Bind the one implemented contract; fail closed for all other operations.""" for label, values in ( ("operation IDs", [item.operation_id for item in records]), ("CLI paths", [item.cli_path for item in records]), @@ -127,9 +123,17 @@ def _assert_inventory( f"Available reason: {item.operation_id}" ) _assert_synthetic_discovery_shape(item, tools) - raise AssertionError( + assert item.operation_id == WORKFLOW_LIST_OPERATION.operation_id, ( f"No workflow operation contract evidence: {item.operation_id}" ) + assert item.contract_version == WORKFLOW_LIST_OPERATION.contract_version + assert item.capabilities == WORKFLOW_LIST_OPERATION.capabilities + assert item.network_access == WORKFLOW_LIST_OPERATION.network_access + tool = matching[0] + expected_input = WorkflowListToolInput.model_json_schema() + assert tool.input_schema["properties"] == expected_input["properties"] + assert tool.input_schema["additionalProperties"] is False + assert tool.output_schema == WorkflowListToolResult.model_json_schema() else: assert item.contract_version is None, ( f"Nonavailable tool has contract version: {item.operation_id}" @@ -397,13 +401,14 @@ def test_uninventoried_workflow_tool_is_rejected(available_record_and_tool): _assert_inventory([], [], [tool]) -def test_existing_server_and_legacy_catalog_are_unchanged(tmp_path): +def test_server_adds_only_workflow_list_and_preserves_legacy_catalog(tmp_path): tools = asyncio.run(create_server(launch_directory=tmp_path).list_tools()) assert Counter(tool.name for tool in tools) == Counter( { "specify_version": 1, "specify_artifact_list": 1, "specify_artifact_info": 1, + "specify_workflow_list": 1, "specify_list_commands": 1, "specify_describe_command": 1, "specify_run_command": 1, @@ -414,6 +419,11 @@ def test_existing_server_and_legacy_catalog_are_unchanged(tmp_path): with pytest.raises(CommandAdapterError) as exc: describe_command("workflow.run") assert exc.value.code == "unavailable_command" + assert [ + item.operation_id + for item in _mcp.WORKFLOW_TOOLS + if item.disposition == "available" + ] == ["workflow.list"] @pytest.mark.parametrize("operation_id", ["workflow.run", "workflow.resume"]) diff --git a/tests/specify_cli/workflows/test_mcp_list.py b/tests/specify_cli/workflows/test_mcp_list.py new file mode 100644 index 0000000000..44a0eda3b0 --- /dev/null +++ b/tests/specify_cli/workflows/test_mcp_list.py @@ -0,0 +1,423 @@ +"""Strict schemas, dispatch, bounds and discovery for workflow.list.""" + +import asyncio +import json +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from pathlib import Path + +import pytest +import anyio +from mcp import ClientSession +from mcp.server import MCPServer +from mcp.server.mcpserver.exceptions import ToolError +from mcp.shared.memory import create_client_server_memory_streams +from pydantic import ValidationError +from typer.testing import CliRunner + +from specify_cli import app +from specify_cli.mcp_server.server import create_server +from specify_cli.workflows import _mcp, _operation_list as operation, mcp_list + + +def _run(coro): + return asyncio.run(coro) + + +def _project(root, entries): + directory = root / ".specify" / "workflows" + directory.mkdir(parents=True) + (directory / "workflow-registry.json").write_text( + json.dumps({"workflows": entries}), encoding="utf-8" + ) + return root + + +def test_actual_discovery_schema_annotations_and_collision(tmp_path): + server = MCPServer(name="test") + _mcp.register(server, launch_directory=tmp_path) + tools = _run(server.list_tools()) + assert len(tools) == 1 + tool = tools[0] + assert tool.name == "specify_workflow_list" + assert ( + tool.input_schema["properties"] + == mcp_list.WorkflowListToolInput.model_json_schema()["properties"] + ) + assert tool.input_schema["additionalProperties"] is False + assert set(tool.input_schema["properties"]) == {"project_directory", "limit"} + assert tool.input_schema["properties"]["limit"] == { + "type": "integer", + "title": "Limit", + "default": 100, + "minimum": 1, + "maximum": 1000, + } + assert tool.output_schema == mcp_list.WorkflowListToolResult.model_json_schema() + assert set(tool.output_schema["properties"]) == {"items", "truncated"} + assert set(tool.output_schema["required"]) == {"items", "truncated"} + assert set(tool.output_schema["$defs"]["WorkflowEntryResult"]["properties"]) == { + "kind", + "id", + "name", + "version", + "description", + "enabled", + } + assert ( + tool.output_schema["$defs"]["WorkflowEntryResult"]["additionalProperties"] + is False + ) + assert tool.annotations.read_only_hint is True + assert tool.annotations.idempotent_hint is True + assert tool.annotations.open_world_hint is False + with pytest.raises(ValueError, match="collision"): + _mcp.register(server, launch_directory=tmp_path) + assert len(_run(server.list_tools())) == 1 + + +def test_shared_operation_used_by_both_adapters(tmp_path, monkeypatch): + (tmp_path / ".specify").mkdir() + monkeypatch.setenv("SPECIFY_INIT_DIR", str(tmp_path)) + requests = [] + result = operation.WorkflowListResult( + (operation.WorkflowListEntry("id", "Name", "1", None, True),) + ) + + def collect(request): + requests.append(request) + return result + + monkeypatch.setattr(operation, "list_workflows", collect) + cli = CliRunner().invoke(app, ["workflow", "list"]) + assert cli.exit_code == 0, cli.output + assert "Name (id) v1" in cli.output + server = create_server(launch_directory=tmp_path) + response = _run(server.call_tool("specify_workflow_list", {})) + assert response.is_error is False + assert response.structured_content["items"][0]["name"] == "Name" + assert requests == [ + operation.WorkflowListRequest(tmp_path), + operation.WorkflowListRequest(tmp_path, 100), + ] + + +@pytest.mark.parametrize( + "arguments", + [ + {"extra": True}, + {"limit": 0}, + {"limit": 1001}, + {"limit": True}, + {"limit": "1"}, + {"limit": 1.5}, + {"limit": None}, + {"project_directory": 4}, + ], +) +def test_sdk_rejects_invalid_input_before_operation(tmp_path, monkeypatch, arguments): + monkeypatch.setattr( + operation, "list_workflows", lambda *a: pytest.fail("Unexpected dispatch") + ) + # The SDK's direct server API raises ToolError; the protocol layer converts + # that into a text-only error response (exercised separately below). + with pytest.raises(ToolError): + _run( + create_server(launch_directory=tmp_path).call_tool( + "specify_workflow_list", arguments + ) + ) + + +def test_real_registry_result_warnings_and_limit(tmp_path): + _project(tmp_path, {"z": {"enabled": False}, "bad": [], "a": {}}) + server = create_server(launch_directory=tmp_path) + response = _run(server.call_tool("specify_workflow_list", {"limit": 2})) + assert response.is_error is False + data = response.structured_content + assert [item["kind"] for item in data["items"]] == ["workflow", "warning"] + assert data["items"][0]["enabled"] is False + assert data["items"][1]["code"] == "invalid_registry_entry" + assert data["truncated"] is True + assert json.loads(response.content[0].text) == data + + +@pytest.mark.parametrize("kind", ["missing", "file"]) +def test_invalid_project_roots_return_not_a_project(tmp_path, kind): + root = tmp_path / "not-a-project" + if kind == "file": + root.write_text("not a directory", encoding="utf-8") + server = create_server(launch_directory=tmp_path) + response = _run( + server.call_tool("specify_workflow_list", {"project_directory": str(root)}) + ) + assert response.is_error is True + assert response.structured_content["error"]["code"] == "not_a_spec_kit_project" + assert str(root) not in response.model_dump_json() + + +def test_absolute_project_path_with_embedded_nul_returns_not_a_project(tmp_path): + root = tmp_path / "invalid\x00project" + assert root.is_absolute() + response = _run( + create_server(launch_directory=tmp_path).call_tool( + "specify_workflow_list", {"project_directory": str(root)} + ) + ) + assert response.is_error is True + assert response.structured_content["error"]["code"] == "not_a_spec_kit_project" + + +@pytest.mark.parametrize("kind", ["policy", "storage"]) +def test_unsafe_project_storage_keeps_structured_error(tmp_path, kind): + root = _project(tmp_path / "project", {}) + if kind == "policy": + path = root / ".specify" / "init-options.json" + path.write_text('{"allow_symlinks": "yes"}', encoding="utf-8") + else: + path = root / ".specify" / "workflows" / "runs" + path.write_text("not a directory", encoding="utf-8") + response = _run( + create_server(launch_directory=tmp_path).call_tool( + "specify_workflow_list", {"project_directory": str(root)} + ) + ) + assert response.is_error is True + assert response.structured_content["error"]["code"] == "unsafe_workflow_storage" + assert str(root) not in response.model_dump_json() + + +def test_launch_directory_captured_and_explicit_projects_are_independent( + tmp_path, monkeypatch +): + first = _project(tmp_path / "first", {"one": {}}) + second = _project(tmp_path / "second", {"two": {}}) + tool = mcp_list.create_workflow_list_tool(launch_directory=first) + monkeypatch.chdir(second) + monkeypatch.setenv("SPECIFY_INIT_DIR", str(second)) + assert tool().structured_content["items"][0]["id"] == "one" + with ThreadPoolExecutor(max_workers=2) as pool: + results = list(pool.map(lambda root: tool(str(root), 100), [first, second])) + assert [r.structured_content["items"][0]["id"] for r in results] == ["one", "two"] + assert Path.cwd() == second + relative = tool("relative", 100) + assert relative.structured_content["error"]["code"] == "invalid_project_directory" + + +@pytest.mark.parametrize( + "error", + [ + operation.WorkflowListError( + "workflow_registry_read_failed", "secret /private/path" + ), + RuntimeError("secret /private/path"), + TypeError("secret /private/path"), + ], +) +def test_error_mapping_never_exposes_exception_details(tmp_path, monkeypatch, error): + def fail(request): + raise error + + monkeypatch.setattr(operation, "list_workflows", fail) + response = mcp_list.create_workflow_list_tool(launch_directory=tmp_path)() + assert response.is_error is True + code = ( + "workflow_registry_read_failed" + if isinstance(error, operation.WorkflowListError) + else "internal_error" + ) + assert response.structured_content["error"]["code"] == code + assert "secret" not in response.model_dump_json() + assert "/private/path" not in response.model_dump_json() + + +@pytest.mark.parametrize( + "result", + [ + None, + {"items": []}, + operation.WorkflowListResult(("bad",)), + operation.WorkflowListResult( + (operation.WorkflowListEntry("id", "n", "v", None, "yes"),) + ), + ], +) +def test_invalid_operation_outcomes_are_rejected(tmp_path, monkeypatch, result): + monkeypatch.setattr(operation, "list_workflows", lambda request: result) + response = mcp_list.create_workflow_list_tool(launch_directory=tmp_path)() + assert response.is_error is True + assert response.structured_content["error"]["code"] == "invalid_operation_result" + + +@pytest.mark.parametrize( + "limit,count,truncated", + [ + (1, 2, False), + (100, 101, True), + (1000, 1001, False), + (100, 0, True), + (100, 99, True), + (1000, 999, True), + ], +) +def test_operation_result_must_obey_requested_limit_and_truncation( + tmp_path, + monkeypatch, + limit, + count, + truncated, +): + items = tuple( + operation.WorkflowListEntry(str(i), "N", "1", None, True) for i in range(count) + ) + monkeypatch.setattr( + operation, + "list_workflows", + lambda request: operation.WorkflowListResult(items, truncated), + ) + response = mcp_list.create_workflow_list_tool(launch_directory=tmp_path)( + None, limit + ) + assert response.is_error is True + assert response.structured_content["error"]["code"] == "invalid_operation_result" + + +@pytest.mark.parametrize("limit", [100, 1000]) +@pytest.mark.parametrize("extra", [0, 1]) +def test_real_result_boundaries_preserve_order_and_warnings(tmp_path, limit, extra): + entries = {str(i): (None if i == 1 else {}) for i in range(limit + extra)} + _project(tmp_path, entries) + tool = mcp_list.create_workflow_list_tool(launch_directory=tmp_path) + response = tool() if limit == 100 else tool(None, limit) + assert response.is_error is False + payload = response.structured_content + assert len(payload["items"]) == limit + assert [ + item.get("id", item.get("workflow_id")) for item in payload["items"] + ] == list(entries)[:limit] + assert payload["items"][1]["kind"] == "warning" + assert payload["truncated"] is bool(extra) + + +def test_output_schema_enforces_static_entry_ceiling(): + schema = mcp_list.WorkflowListToolResult.model_json_schema() + assert schema["properties"]["items"]["maxItems"] == 1000 + row = { + "kind": "workflow", + "id": "id", + "name": "N", + "version": "1", + "description": None, + "enabled": True, + } + with pytest.raises(ValidationError): + mcp_list.WorkflowListToolResult.model_validate( + {"items": [row] * 1001, "truncated": False} + ) + + +def test_response_byte_budget_and_constructor_bounds(tmp_path, monkeypatch): + result = operation.WorkflowListResult( + (operation.WorkflowListEntry("id", "é" * 2000, "v", None, True),) + ) + monkeypatch.setattr(operation, "list_workflows", lambda request: result) + tool = mcp_list.create_workflow_list_tool( + launch_directory=tmp_path, max_wire_response_bytes=2000 + ) + response = tool() + assert response.structured_content["error"]["code"] == "response_too_large" + assert mcp_list._wire_response_size(response) <= 2000 + assert response.structured_content["error"]["retryable"] is True + with pytest.raises(ValueError, match="bounded error response"): + mcp_list.create_workflow_list_tool( + launch_directory=tmp_path, max_wire_response_bytes=10 + ) + with pytest.raises(ValueError, match="absolute"): + mcp_list.create_workflow_list_tool(launch_directory=Path("relative")) + + +@pytest.mark.parametrize( + "field,value", + [ + ("contract_version", "999-fabricated"), + ("network_access", "optional"), + ("capabilities", frozenset({"execution"})), + ], +) +def test_real_available_inventory_must_match_descriptor(tmp_path, field, value): + from .test_mcp import _assert_inventory, _workflow_cli_paths + + records = [ + replace(r, **{field: value}) if r.operation_id == "workflow.list" else r + for r in _mcp.WORKFLOW_TOOLS + ] + tools = _run(create_server(launch_directory=tmp_path).list_tools()) + with pytest.raises(AssertionError): + _assert_inventory(records, _workflow_cli_paths(), tools) + + +def test_output_schema_rejects_extra_fields_and_wrong_types(): + with pytest.raises(ValidationError): + mcp_list.WorkflowListToolResult.model_validate({"items": [], "truncated": "no"}) + with pytest.raises(ValidationError): + mcp_list.WorkflowListToolResult.model_validate( + {"items": [], "truncated": False, "extra": 1} + ) + + +@pytest.mark.parametrize("field", ["input_schema", "output_schema"]) +def test_correct_tool_name_with_unrelated_schema_fails_parity(tmp_path, field): + from .test_mcp import _assert_inventory, _workflow_cli_paths + + tools = _run(create_server(launch_directory=tmp_path).list_tools()) + tools = [ + tool.model_copy( + update={ + field: { + "type": "object", + "properties": {"unrelated": {"type": "string"}}, + } + } + ) + if tool.name == "specify_workflow_list" + else tool + for tool in tools + ] + with pytest.raises(AssertionError): + _assert_inventory(_mcp.WORKFLOW_TOOLS, _workflow_cli_paths(), tools) + + +def test_in_memory_protocol_success_expected_error_and_schema_error(tmp_path): + _project(tmp_path, {"installed": {}}) + + async def exercise(): + server = create_server(launch_directory=tmp_path) + async with ( + create_client_server_memory_streams() as (client, streams), + anyio.create_task_group() as tasks, + ): + tasks.start_soon( + server._lowlevel_server.run, + streams[0], + streams[1], + server._lowlevel_server.create_initialization_options(), + ) + async with ClientSession(*client) as session: + await session.initialize() + success = await session.call_tool("specify_workflow_list", {}) + error = await session.call_tool( + "specify_workflow_list", {"project_directory": "relative"} + ) + invalid = await session.call_tool( + "specify_workflow_list", {"unexpected": True} + ) + tasks.cancel_scope.cancel() + return success, error, invalid + + success, error, invalid = _run(exercise()) + assert success.is_error is False + assert success.structured_content["items"][0]["id"] == "installed" + assert error.is_error is True + assert error.structured_content["error"]["code"] == "invalid_project_directory" + assert invalid.is_error is True + assert invalid.structured_content is None diff --git a/tests/specify_cli/workflows/test_operation_list.py b/tests/specify_cli/workflows/test_operation_list.py new file mode 100644 index 0000000000..c2730dcb23 --- /dev/null +++ b/tests/specify_cli/workflows/test_operation_list.py @@ -0,0 +1,342 @@ +"""Transport-neutral installed-workflow contract and effect boundaries.""" + +import builtins +import errno +import io +import json +import os +import subprocess +from dataclasses import FrozenInstanceError +from pathlib import Path + +import pytest + +from specify_cli.workflows import _operation_list as operation +from specify_cli.file_helper import FileHelper + + +def _registry(root, entries): + directory = root / ".specify" / "workflows" + directory.mkdir(parents=True) + path = directory / "workflow-registry.json" + path.write_text(json.dumps({"workflows": entries}), encoding="utf-8") + return path + + +def test_registry_order_defaults_and_warnings(tmp_path): + _registry( + tmp_path, + { + "z": {"description": ["old", "metadata"], "enabled": 0}, + "broken[entry]": None, + "a": {"name": None, "version": 7}, + }, + ) + result = operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert result == operation.WorkflowListResult( + ( + operation.WorkflowListEntry("z", "z", "?", "['old', 'metadata']", False), + operation.WorkflowListWarning( + "broken[entry]", "Skipping corrupted registry entry 'broken[entry]'." + ), + operation.WorkflowListEntry("a", "None", "7", None, True), + ) + ) + with pytest.raises(FrozenInstanceError): + result.truncated = True + + +def test_missing_registry_is_empty_without_creating_storage(tmp_path): + (tmp_path / ".specify").mkdir() + assert operation.list_workflows(operation.WorkflowListRequest(tmp_path)).items == () + assert list((tmp_path / ".specify").iterdir()) == [] + + +@pytest.mark.parametrize("contents", ["{", "[]", '{"workflows": []}', "\udcff"]) +def test_corrupt_registry_fails_closed(tmp_path, contents): + path = _registry(tmp_path, {}) + path.write_bytes(contents.encode("utf-8", errors="surrogateescape")) + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert exc.value.code == "workflow_registry_read_failed" + + +@pytest.mark.parametrize("limit", [0, -1, 1001, True, "2", 1.5]) +def test_invalid_limit_is_rejected_before_io(tmp_path, monkeypatch, limit): + monkeypatch.setattr( + operation, + "project_files", + lambda *a: pytest.fail("IO before validation"), + ) + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path, limit)) + assert exc.value.code == "invalid_limit" + + +def test_limit_counts_warnings_and_explicitly_reports_omissions(tmp_path): + _registry(tmp_path, {"broken": 1, "ok": {}, "last": {}}) + request = operation.WorkflowListRequest(tmp_path, 2) + result = operation.list_workflows(request) + assert len(result.items) == 2 + assert result.truncated is True + assert isinstance(result.items[0], operation.WorkflowListWarning) + assert ( + operation.list_workflows(operation.WorkflowListRequest(tmp_path, 3)).truncated + is False + ) + assert ( + len(operation.list_workflows(operation.WorkflowListRequest(tmp_path)).items) + == 3 + ) + + +def test_relative_and_nonproject_directories_are_rejected(tmp_path): + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(Path("relative"))) + assert exc.value.code == "invalid_project_directory" + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert exc.value.code == "not_a_spec_kit_project" + + +@pytest.mark.parametrize("kind", ["missing", "file"]) +def test_invalid_project_roots_are_rejected_before_policy_loading( + tmp_path, monkeypatch, kind +): + root = tmp_path / "not-a-project" + if kind == "file": + root.write_text("not a directory", encoding="utf-8") + policy_calls = [] + original = operation.project_files + + def load_policy(path): + policy_calls.append(path) + return original(path) + + monkeypatch.setattr(operation, "project_files", load_policy) + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(root)) + assert exc.value.code == "not_a_spec_kit_project" + assert policy_calls == [] + + +@pytest.mark.parametrize("boundary", ["root", "policy", "storage"]) +def test_project_access_errors_remain_unsafe_storage(tmp_path, monkeypatch, boundary): + (tmp_path / ".specify").mkdir() + + def denied(*args, **kwargs): + raise PermissionError("private filesystem details") + + if boundary == "root": + original_stat = Path.stat + + def root_stat(path, *args, **kwargs): + if path == tmp_path: + return denied() + return original_stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", root_stat) + else: + name = "project_files" if boundary == "policy" else "validate_workflow_storage" + monkeypatch.setattr(operation, name, denied) + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert exc.value.code == "unsafe_workflow_storage" + + +def test_invalid_project_policy_remains_unsafe_storage(tmp_path): + (tmp_path / ".specify").mkdir() + (tmp_path / ".specify" / "init-options.json").write_text( + '{"allow_symlinks": "yes"}', encoding="utf-8" + ) + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert exc.value.code == "unsafe_workflow_storage" + + +@pytest.mark.parametrize( + "suffix", + [ + ".specify", + ".specify/workflows", + ".specify/workflows/runs", + ".specify/workflows/overlays", + ], +) +def test_nondirectory_storage_is_rejected(tmp_path, suffix): + target = tmp_path / suffix + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text("not a directory", encoding="utf-8") + with pytest.raises(operation.WorkflowStorageError) as storage_error: + operation.validate_workflow_storage(tmp_path) + assert str(storage_error.value) == f"{suffix} path exists but is not a directory" + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert exc.value.code == "unsafe_workflow_storage" + + +@pytest.mark.parametrize( + "suffix", + [ + ".specify", + ".specify/workflows", + ".specify/workflows/runs", + ".specify/workflows/overlays", + ".specify/workflows/workflow-registry.json", + ], +) +def test_symlinked_storage_is_rejected(tmp_path, suffix): + root = tmp_path / "project" + target = root / suffix + target.parent.mkdir(parents=True, exist_ok=True) + outside = tmp_path / "outside" + is_file = suffix.endswith(".json") + if is_file: + outside.write_text('{"workflows": {}}', encoding="utf-8") + else: + outside.mkdir() + try: + target.symlink_to(outside, target_is_directory=not is_file) + except OSError as exc: + if exc.errno in { + errno.EPERM, + errno.ENOSYS, + errno.ENOTSUP, + errno.EOPNOTSUPP, + } or getattr(exc, "winerror", None) in {50, 1314}: + pytest.skip(f"Host cannot create symlinks: {exc}") + raise + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(root)) + expected = "workflow_registry_read_failed" if is_file else "unsafe_workflow_storage" + assert exc.value.code == expected + + +def test_operation_has_no_writes_network_execution_or_rendering( + tmp_path, monkeypatch, capsys +): + from specify_cli.authentication import http + from specify_cli.integrations import installer + from specify_cli import workflows + + path = _registry(tmp_path, {"ok": {}}) + before = path.read_bytes() + original_open = builtins.open + original_io_open = io.open + + def read_only_open(file, mode="r", *args, **kwargs): + assert not any(flag in mode for flag in "wax+") + return original_open(file, mode, *args, **kwargs) + + def unexpected(*args, **kwargs): + pytest.fail("Listing must not write, fetch, import custom steps or execute") + + monkeypatch.setattr(builtins, "open", read_only_open) + + def read_only_io_open(file, mode="r", *args, **kwargs): + assert not any(flag in mode for flag in "wax+") + return original_io_open(file, mode, *args, **kwargs) + + monkeypatch.setattr(io, "open", read_only_io_open) + monkeypatch.setattr(Path, "mkdir", unexpected) + monkeypatch.setattr(Path, "write_text", unexpected) + monkeypatch.setattr(Path, "write_bytes", unexpected) + monkeypatch.setattr(operation.WorkflowRegistry, "save", unexpected) + monkeypatch.setattr(http, "open_url", unexpected) + monkeypatch.setattr(subprocess, "run", unexpected) + monkeypatch.setattr(workflows, "load_custom_steps", unexpected) + monkeypatch.setattr(installer, "load_installed_integrations", unexpected) + result = operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert len(result.items) == 1 + assert path.read_bytes() == before + assert capsys.readouterr() == ("", "") + + +def test_unreadable_registry_is_not_empty_success(tmp_path, monkeypatch): + path = _registry(tmp_path, {}) + original = FileHelper.read_bytes + + def denied(files, file): + if files.root / file == path: + raise PermissionError("private registry details") + return original(files, file) + + monkeypatch.setattr(FileHelper, "read_bytes", denied) + with pytest.raises(operation.WorkflowListError) as exc: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert exc.value.code == "workflow_registry_read_failed" + + +@pytest.mark.skipif(os.name != "nt", reason="Windows directory junction regression") +@pytest.mark.parametrize("registry_present", [False, True]) +def test_list_rejects_junction_at_registry_read_boundary( + tmp_path, monkeypatch, registry_present +): + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows import mcp_list + + root = tmp_path / "project" + (root / ".specify").mkdir(parents=True) + outside = tmp_path / "outside" + outside.mkdir() + registry = outside / "workflow-registry.json" + if registry_present: + registry.write_text('{"workflows": {"outside-secret": {}}}', encoding="utf-8") + before = registry.read_bytes() if registry_present else None + junction = root / ".specify" / "workflows" + created = subprocess.run( + ["cmd", "/c", "mklink", "/J", str(junction), str(outside)], + capture_output=True, + text=True, + check=False, + ) + assert created.returncode == 0, created.stderr + assert not junction.is_symlink() + opened = [] + original_open = io.open + + def track(file, *args, **kwargs): + if Path(file).name == "workflow-registry.json": + opened.append(file) + return original_open(file, *args, **kwargs) + + monkeypatch.setattr(io, "open", track) + with pytest.raises(operation.WorkflowListError) as error: + operation.list_workflows(operation.WorkflowListRequest(root)) + assert error.value.code == "workflow_registry_read_failed" + response = mcp_list.create_workflow_list_tool(launch_directory=root)() + assert response.is_error is True + assert ( + response.structured_content["error"]["code"] == "workflow_registry_read_failed" + ) + monkeypatch.setenv("SPECIFY_INIT_DIR", str(root)) + cli = CliRunner().invoke(app, ["workflow", "list"]) + assert cli.exit_code == 1 + assert "Failed to read workflow registry" in cli.output + assert "outside-secret" not in cli.output + assert opened == [] + monkeypatch.setattr(io, "open", original_open) + if registry_present: + assert registry.read_bytes() == before + else: + assert not registry.exists() + + +def test_unexpected_symlink_setup_errors_are_not_skipped(tmp_path, monkeypatch): + def missing_parent(*args, **kwargs): + raise FileNotFoundError(errno.ENOENT, "unexpected fixture setup failure") + + monkeypatch.setattr(Path, "symlink_to", missing_parent) + with pytest.raises(FileNotFoundError, match="unexpected fixture setup failure"): + test_symlinked_storage_is_rejected(tmp_path, ".specify/workflows") + + +def test_policy_reader_preserves_legacy_json_error_offsets(tmp_path): + path = _registry(tmp_path, {}) + path.write_bytes(b'{\r\n "workflows":\r\n broken\r\n}') + with pytest.raises(OSError) as legacy: + operation.WorkflowRegistry(tmp_path) + with pytest.raises(operation.WorkflowListError) as current: + operation.list_workflows(operation.WorkflowListRequest(tmp_path)) + assert str(current.value) == f"Failed to read workflow registry: {legacy.value}"