fix(workflows): fail a gate whose on_reject is not abort/skip/retry (#3888)

execute() reads `on_reject = config.get("on_reject", "abort")` and, in the
reject branch, handles only "abort" and "retry" before falling through to
its `# on_reject == "skip"` case. So any other value makes a REJECTED gate
report COMPLETED and the run walks straight past the review the gate
exists to enforce:

  on_reject='abort'  -> failed     "Gate rejected by user at step 'g'"
  on_reject='retry'  -> paused
  on_reject='skip'   -> completed  (by design)
  on_reject='Abort'  -> completed  <-- rejection silently discarded
  on_reject='fail'   -> completed  <-- same
  on_reject='stop'   -> completed  <-- same
  on_reject=None     -> completed  <-- same
  on_reject=5        -> completed  <-- same

Reachable by a capitalisation slip, a guessed verb, a non-string, or a
bare `on_reject:` — note `config.get(k, default)` does NOT substitute the
default for an explicit YAML null.

`validate` already rejects anything outside abort/skip/retry, but the
engine does not auto-validate before execute(). Fail loudly instead,
mirroring the `options` and `verdict_input` guards in the same method, and
placed before the non-TTY short-circuit so it surfaces in CI too.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/reference/workflows.md

@@ -623,6 +623,19 @@ pauses: the named stored input is reset to `""`. A later resume therefore
 prompts or pauses again until another verdict is supplied. Approve, abort, and
 skip outcomes leave the input unchanged.
 
+Because of that reset, a verdict input used with `on_reject: retry` must accept
+`""`. If it declares an `enum`, include the empty string — otherwise the reset
+value violates the input's own `enum` and the run can no longer be resumed with
+any input. `specify workflow add` reports this as a validation error.
+
+```yaml
+inputs:
+  spec_verdict:
+    type: string
+    enum: ["", approve, reject]
+    default: ""
+```
+
 ## FAQ
 
 ### What happens when a workflow hits a gate step?

src/specify_cli/bundler/services/adapters.py

@@ -18,7 +18,7 @@ from urllib.request import url2pathname
 from ..._assets import _locate_core_pack, _repo_root
 from ..._download_security import MAX_JSON_CATALOG_BYTES, read_response_limited
 from .. import BundlerError
-from ..lib.yamlio import loads_json
+from ..lib.yamlio import load_json, loads_json
 from ..models.catalog import CatalogSource
 from ..models.manifest import ComponentRef
 
@@ -145,13 +145,13 @@ def make_catalog_fetcher(*, allow_network: bool = True):
             path = _file_url_to_path(parsed)
             if not path.exists():
                 raise BundlerError(f"Catalog file not found: {path}")
-            return loads_json(path.read_text(encoding="utf-8"), origin=str(path))
+            return load_json(path)
 
         if scheme == "" or _is_windows_drive_path(url):
             path = Path(url)
             if not path.exists():
                 raise BundlerError(f"Catalog file not found: {path}")
-            return loads_json(path.read_text(encoding="utf-8"), origin=str(path))
+            return load_json(path)
 
         if scheme in ("http", "https"):
             if not allow_network:

src/specify_cli/bundler/services/packager.py

@@ -93,9 +93,11 @@ def build_bundle(
             # extraction, but collapse to two canonical modes (0755 when any
             # execute bit is set on the source, otherwise 0644) so identical
             # inputs yield a byte-for-byte identical artifact.
-            mode = 0o755 if file_path.stat().st_mode & 0o111 else 0o644
-            info.external_attr = mode << 16
-            archive.writestr(info, file_path.read_bytes())
+            with file_path.open("rb") as fh:
+                st = os.fstat(fh.fileno())
+                mode = 0o755 if st.st_mode & 0o111 else 0o644
+                info.external_attr = mode << 16
+                archive.writestr(info, fh.read())
 
     return BuildResult(artifact_path=artifact_path, file_count=len(files))
 

src/specify_cli/commands/bundle/__init__.py

@@ -119,6 +119,8 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N
             preset=None,
             integration=integration,
             integration_options=None,
+            extensions=None,
+            trust_extension_urls=False,
         )
     except typer.Exit as exc:
         if exc.exit_code:

src/specify_cli/commands/init.py

@@ -30,6 +30,145 @@ def _stdin_is_interactive() -> bool:
     return sys.stdin.isatty()
 
 
+def _ext_spec_is_url(ext_spec: str) -> bool:
+    """Return True when *ext_spec* is an http(s) URL rather than a name/path."""
+    from urllib.parse import urlparse
+
+    try:
+        return urlparse(ext_spec).scheme in ("http", "https")
+    except ValueError:
+        return False
+
+
+def _confirm_extension_url_trust(
+    url_specs: list[str], *, trust_override: bool
+) -> dict[str, bool]:
+    """Resolve trust for each URL-based extension before the Live display.
+
+    URL installs pull an arbitrary external extension, so they get the same
+    default-deny confirmation as ``extension add --from``. Returns a mapping of
+    ``url_spec -> approved``. With *trust_override* every URL is pre-approved.
+    In a non-interactive session without the override, every URL is denied
+    (the prompt cannot be answered), mirroring the default-deny posture.
+    """
+    from rich.markup import escape as _escape_markup
+    from rich.panel import Panel
+
+    approvals: dict[str, bool] = {}
+    interactive = _stdin_is_interactive()
+    for spec in url_specs:
+        if trust_override:
+            approvals[spec] = True
+            continue
+        if not interactive:
+            approvals[spec] = False
+            continue
+        console.print()
+        console.print(
+            Panel(
+                "[bold]You are installing an extension from an external URL that is not\n"
+                "listed in any of your configured extension catalogs.[/bold]\n\n"
+                f"URL: {_escape_markup(spec)}\n\n"
+                "Only install extensions from sources you trust.",
+                title="[bold yellow]⚠ Untrusted Source[/bold yellow]",
+                border_style="yellow",
+                padding=(1, 2),
+            )
+        )
+        console.print()
+        approvals[spec] = typer.confirm(
+            f"Install extension from {spec}?", default=False
+        )
+    return approvals
+
+
+def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_version: str) -> str:
+    """Install a single extension during ``specify init``.
+
+    Handles bundled extension names, local directory paths, and HTTPS URLs.
+    Returns a short status message on success.
+    Raises ``ValueError`` on failure so the caller can convert it to a
+    tracker error without aborting the entire init.
+    """
+    from urllib.parse import urlparse
+
+    from .._assets import _locate_bundled_extension
+    from ..extensions import ExtensionCatalog, ExtensionError, ExtensionManager
+    from ..extensions._commands import (
+        _resolve_catalog_extension,
+        install_extension_from_url,
+    )
+
+    manager = ExtensionManager(project_path)
+
+    # --- URL ---
+    parsed = urlparse(ext_spec)
+    if parsed.scheme in ("http", "https"):
+        try:
+            manifest = install_extension_from_url(
+                manager, project_path, ext_spec, speckit_version
+            )
+        except ExtensionError as exc:
+            raise ValueError(str(exc)) from exc
+        return f"{manifest.name} v{manifest.version} installed"
+
+    # --- Local path ---
+    if ext_spec.startswith(("./", "../", "/", "~/", ".\\", "..\\")) or Path(ext_spec).is_absolute():
+        source_path = Path(ext_spec).expanduser().resolve()
+        if not source_path.exists():
+            raise ValueError(f"Directory not found: {source_path}")
+        if not (source_path / "extension.yml").exists():
+            raise ValueError(f"No extension.yml found in {source_path}")
+        manifest = manager.install_from_directory(source_path, speckit_version)
+        return f"{manifest.name} v{manifest.version} installed"
+
+    # --- Bundled extension name or catalog ID ---
+    bundled_path = _locate_bundled_extension(ext_spec)
+    if bundled_path is not None:
+        if manager.registry.is_installed(ext_spec):
+            return "already installed"
+        manifest = manager.install_from_directory(bundled_path, speckit_version)
+        return f"{manifest.name} v{manifest.version} installed"
+
+    # Fall back to catalog
+    catalog = ExtensionCatalog(project_path)
+    ext_info, catalog_error = _resolve_catalog_extension(ext_spec, catalog, "add")
+    if catalog_error:
+        raise ValueError(f"Could not query extension catalog: {catalog_error}")
+    if not ext_info:
+        raise ValueError(f"Extension '{ext_spec}' not found in bundled extensions or catalog")
+
+    resolved_id = ext_info["id"]
+    if resolved_id != ext_spec:
+        bundled_path = _locate_bundled_extension(resolved_id)
+        if bundled_path is not None:
+            if manager.registry.is_installed(resolved_id):
+                return "already installed"
+            manifest = manager.install_from_directory(bundled_path, speckit_version)
+            return f"{manifest.name} v{manifest.version} installed"
+
+    if ext_info.get("bundled") and not ext_info.get("download_url"):
+        from ..extensions import REINSTALL_COMMAND
+
+        raise ValueError(
+            f"Extension '{resolved_id}' is bundled with spec-kit but not found in the installed package. "
+            f"Try reinstalling spec-kit: {REINSTALL_COMMAND}"
+        )
+
+    if not ext_info.get("_install_allowed", True):
+        catalog_name = ext_info.get("_catalog_name", "community")
+        raise ValueError(
+            f"Extension '{ext_spec}' is in the '{catalog_name}' catalog but installation is not allowed from that catalog"
+        )
+
+    zip_path = catalog.download_extension(resolved_id)
+    try:
+        manifest = manager.install_from_zip(zip_path, speckit_version)
+    finally:
+        zip_path.unlink(missing_ok=True)
+    return f"{manifest.name} v{manifest.version} installed"
+
+
 def ensure_constitution_from_template(
     project_path: Path, tracker: StepTracker | None = None
 ) -> None:
@@ -142,6 +281,16 @@ def register(app: typer.Typer) -> None:
             "--integration-options",
             help='Options for the integration (e.g. --integration-options="--commands-dir .myagent/cmds")',
         ),
+        extensions: list[str] | None = typer.Option(
+            None,
+            "--extension",
+            help="Install an extension during initialization (bundled name, local path, or HTTPS URL). Repeatable.",
+        ),
+        trust_extension_urls: bool = typer.Option(
+            False,
+            "--trust-extension-urls",
+            help="Pre-authorize installing extensions from external URLs without the interactive trust prompt (required for non-interactive URL installs).",
+        ),
     ):
         """
         Initialize a new Specify project.
@@ -174,6 +323,10 @@ def register(app: typer.Typer) -> None:
             specify init --here --integration gemini
             specify init my-project --integration generic --integration-options="--commands-dir .myagent/commands/"  # Bring your own agent; requires --commands-dir
             specify init my-project --integration claude --preset healthcare-compliance  # With preset
+            specify init my-project --integration copilot --extension git  # With bundled extension
+            specify init my-project --extension git --extension selftest  # Multiple extensions
+            specify init my-project --extension ./my-extensions/custom-ext  # Local path extension
+            specify init my-project --extension https://example.com/extensions/my-ext.zip --trust-extension-urls  # URL extension (non-interactive)
         """
         # Lazy imports to avoid circular dependency — __init__.py imports this module
         from .. import (
@@ -413,10 +566,31 @@ def register(app: typer.Typer) -> None:
             ("chmod", "Ensure scripts executable"),
             ("constitution", "Constitution setup"),
             ("workflow", "Install bundled workflow"),
-            ("final", "Finalize"),
         ]:
             tracker.add(key, label)
 
+        if extensions:
+            from rich.markup import escape as _escape_markup
+
+            for i, ext_spec in enumerate(extensions):
+                tracker.add(
+                    f"extension-{i}", f"Install extension: {_escape_markup(ext_spec)}"
+                )
+
+        tracker.add("final", "Finalize")
+
+        # Resolve trust for URL-based extensions BEFORE entering the Live
+        # display: the confirmation prompt cannot be shown/answered underneath
+        # the Rich Live spinner. URL installs are default-deny unless the user
+        # confirms interactively or passes --trust-extension-urls.
+        extension_url_approvals: dict[str, bool] = {}
+        if extensions:
+            url_specs = [e for e in extensions if _ext_spec_is_url(e)]
+            if url_specs:
+                extension_url_approvals = _confirm_extension_url_trust(
+                    url_specs, trust_override=trust_extension_urls
+                )
+
         # Disable transient mode on Windows: PowerShell 5.1's legacy console
         # hangs when Rich tries to restore cursor state via VT escape sequences.
         _transient = sys.platform != "win32"
@@ -626,6 +800,46 @@ def register(app: typer.Typer) -> None:
                             continuing="Continuing without the optional preset.",
                         )
 
+                # Install extensions specified via --extension
+                if extensions:
+                    from rich.markup import escape as _escape_markup
+
+                    from ..extensions._commands import _refresh_events_and_warn
+
+                    speckit_ver = get_speckit_version()
+                    any_extension_installed = False
+                    for i, ext_spec in enumerate(extensions):
+                        tracker.start(f"extension-{i}")
+                        # Skip URL extensions the user did not confirm as trusted
+                        # (default-deny; resolved before the Live display).
+                        if _ext_spec_is_url(ext_spec) and not extension_url_approvals.get(
+                            ext_spec, False
+                        ):
+                            tracker.error(
+                                f"extension-{i}",
+                                "skipped: untrusted URL not confirmed "
+                                "(use --trust-extension-urls)",
+                            )
+                            continue
+                        try:
+                            status_msg = _install_extension_during_init(
+                                project_path, ext_spec, speckit_ver
+                            )
+                            tracker.complete(f"extension-{i}", status_msg)
+                            any_extension_installed = True
+                        except Exception as ext_err:
+                            sanitized_ext = str(ext_err).replace("\n", " ").strip()
+                            tracker.error(
+                                f"extension-{i}",
+                                f"failed: {_escape_markup(sanitized_ext[:120])}",
+                            )
+
+                    # Refresh native event configuration once after the batch so
+                    # that an extension declaring ``events:`` has its hooks
+                    # activated, mirroring the ``extension add`` path.
+                    if any_extension_installed:
+                        _refresh_events_and_warn(project_path)
+
                 # Seed the constitution AFTER preset installation so that a
                 # preset-provided constitution-template (resolved via the
                 # priority stack) wins over the core template.

src/specify_cli/extensions/_commands.py

@@ -95,6 +95,141 @@ def _refresh_events_and_warn(project_root: Path) -> None:
             console.print(f"    {key}: {_escape_markup(detail)}")
 
 
+def install_extension_from_url(
+    manager,
+    project_root: Path,
+    url: str,
+    speckit_version: str,
+    *,
+    priority: int = 10,
+    force: bool = False,
+):
+    """Download an archive from *url* and install it, reusing the hardened path.
+
+    Shares the same download hardening as ``extension add --from``:
+    HTTPS enforcement, the catalog's authenticated + redirect-guarded
+    ``_open_url`` fetch, a bounded (50 MiB) response read, archive-format
+    detection (ZIP or tar.gz/tgz), and a TOCTOU-safe transient download file
+    consumed directly by ``install_from_zip``.
+
+    Returns the installed manifest. Raises ``ExtensionError`` on any failure so
+    callers can present a uniform message without a second downloader.
+    """
+    import urllib.error
+
+    from . import ExtensionCatalog, ExtensionError
+
+    if not is_https_or_localhost_http(url):
+        raise ExtensionError(
+            "URL must use HTTPS (HTTP is only allowed for localhost)"
+        )
+
+    download_dir = _validate_safe_cache_dir(project_root)
+    archive_filename = f"extension-url-download-{uuid4().hex}.archive"
+    # Only used for diagnostic messages: the real archive is a transient inode
+    # (unlinked on POSIX, O_TEMPORARY on Windows) consumed via ``archive_file``
+    # below, so this path is never opened again.
+    archive_path = download_dir / archive_filename
+
+    try:
+        dl_catalog = ExtensionCatalog(project_root)
+        download_url = url
+        extra_headers = None
+        resolved_url = dl_catalog._resolve_github_release_asset_api_url(download_url)
+        if resolved_url:
+            download_url = resolved_url
+            extra_headers = {"Accept": "application/octet-stream"}
+
+        with dl_catalog._open_url(
+            download_url, timeout=60, extra_headers=extra_headers
+        ) as response:
+            archive_data = read_response_limited(
+                response,
+                error_type=ExtensionError,
+                label=f"extension {url}",
+            )
+            final_url = (
+                response.geturl() if hasattr(response, "geturl") else download_url
+            )
+            content_type = (
+                response.getheader("Content-Type")
+                if hasattr(response, "getheader")
+                else None
+            )
+    except urllib.error.URLError as exc:
+        raise ExtensionError(f"Failed to download from {url}: {exc}") from exc
+
+    download_fd = -1
+    download_file = None
+    try:
+        try:
+            download_fd = _safe_open_download_zip(
+                project_root, download_dir, archive_filename
+            )
+        except OSError as exc:
+            raise ExtensionError(
+                f"Could not safely create download file: {exc}"
+            ) from exc
+
+        try:
+            download_file = os.fdopen(download_fd, "w+b")
+            download_fd = -1
+            download_file.write(archive_data)
+            download_file.flush()
+            download_file.seek(0)
+        except OSError as exc:
+            raise ExtensionError(
+                f"Could not safely write download file: {exc}"
+            ) from exc
+
+        format_source = (
+            final_url
+            if archive_format_from_name(final_url) is not None
+            else url
+        )
+        try:
+            detect_archive_format(
+                archive_path,
+                archive_file=download_file,
+                source_name=format_source,
+                content_type=content_type,
+                error_type=ExtensionError,
+            )
+        except ExtensionError as exc:
+            raise ExtensionError(
+                f"{url} did not return a ZIP archive or tar.gz/tgz archive "
+                f"(got {len(archive_data)} bytes). This usually means the request "
+                "was not authenticated and a login/HTML page was returned. "
+                "Verify the URL and configured credentials."
+            ) from exc
+
+        # Consume the transient inode reserved above rather than reopening the
+        # cache pathname during extraction.
+        try:
+            return manager.install_from_zip(
+                archive_path,
+                speckit_version,
+                priority=priority,
+                force=force,
+                archive_file=download_file,
+            )
+        except OSError as exc:
+            raise ExtensionError(
+                f"Could not install extension from downloaded archive: {exc}"
+            ) from exc
+    finally:
+        if download_file is not None:
+            try:
+                download_file.close()
+            except OSError:
+                pass
+        elif download_fd >= 0:
+            try:
+                os.close(download_fd)
+            except OSError:
+                pass
+
+
 def _load_catalog_command_config(project_root: Path, config_path: Path) -> dict:
     """Load extension catalog CLI config with user-facing shape errors."""
     try:
@@ -810,134 +945,20 @@ def extension_add(
                 )
 
             elif from_url:
-                # Install from an archive URL.
-                import urllib.error
-
+                # Install from URL archive via the shared hardened downloader
+                # (HTTPS enforcement, authenticated redirect-guarded fetch,
+                # bounded read, archive-format detection, TOCTOU-safe transient
+                # archive). Same path used by ``specify init --extension <url>``.
                 console.print(f"Downloading from {safe_url}...")
+                manifest = install_extension_from_url(
+                    manager,
+                    project_root,
+                    from_url,
+                    speckit_version,
+                    priority=priority,
+                    force=force,
+                )
 
-                download_dir = _validate_safe_cache_dir(project_root)
-                archive_filename = f"extension-url-download-{uuid4().hex}.archive"
-                # Only used for diagnostic messages: the real archive is a
-                # transient inode (unlinked on POSIX, O_TEMPORARY on Windows)
-                # consumed via ``archive_file`` below, so this path is never
-                # opened again.
-                archive_path = download_dir / archive_filename
-
-                try:
-                    # Use the catalog's authenticated fetch so configured
-                    # credentials (incl. GitHub Enterprise Server) are applied
-                    # and GHES release-asset URLs resolve via /api/v3 — keeping
-                    # --from consistent with catalog-based installs.
-                    dl_catalog = ExtensionCatalog(project_root)
-                    download_url = from_url
-                    extra_headers = None
-                    resolved_url = dl_catalog._resolve_github_release_asset_api_url(download_url)
-                    if resolved_url:
-                        download_url = resolved_url
-                        extra_headers = {"Accept": "application/octet-stream"}
-
-                    with dl_catalog._open_url(
-                        download_url, timeout=60, extra_headers=extra_headers
-                    ) as response:
-                        archive_data = read_response_limited(
-                            response,
-                            error_type=ExtensionError,
-                            label=f"extension {from_url}",
-                        )
-                        final_url = (
-                            response.geturl()
-                            if hasattr(response, "geturl")
-                            else download_url
-                        )
-                        content_type = (
-                            response.getheader("Content-Type")
-                            if hasattr(response, "getheader")
-                            else None
-                        )
-
-                    download_fd = -1
-                    download_file = None
-                    try:
-                        try:
-                            download_fd = _safe_open_download_zip(
-                                project_root, download_dir, archive_filename
-                            )
-                        except OSError as exc:
-                            console.print(
-                                "[red]Error:[/red] Could not safely create download file: "
-                                f"{_escape_markup(str(exc))}"
-                            )
-                            raise typer.Exit(1)
-
-                        try:
-                            download_file = os.fdopen(download_fd, "w+b")
-                            download_fd = -1
-                            download_file.write(archive_data)
-                            download_file.flush()
-                            download_file.seek(0)
-                        except OSError as exc:
-                            console.print(
-                                "[red]Error:[/red] Could not safely write download file: "
-                                f"{_escape_markup(str(exc))}"
-                            )
-                            raise typer.Exit(1)
-
-                        format_source = (
-                            final_url
-                            if archive_format_from_name(final_url) is not None
-                            else from_url
-                        )
-                        try:
-                            detect_archive_format(
-                                archive_path,
-                                archive_file=download_file,
-                                source_name=format_source,
-                                content_type=content_type,
-                                error_type=ExtensionError,
-                            )
-                        except ExtensionError:
-                            console.print(
-                                f"[red]Error:[/red] {safe_url} did not return a ZIP archive "
-                                "or tar.gz/tgz archive "
-                                f"(got {len(archive_data)} bytes). This usually means "
-                                "the request was not authenticated and a login/HTML page was "
-                                "returned. Verify the URL and configured credentials."
-                            )
-                            raise typer.Exit(1)
-
-                        # Consume the transient inode reserved above rather
-                        # than reopening the cache pathname during extraction.
-                        try:
-                            manifest = manager.install_from_zip(
-                                archive_path,
-                                speckit_version,
-                                priority=priority,
-                                force=force,
-                                archive_file=download_file,
-                            )
-                        except OSError as exc:
-                            console.print(
-                                "[red]Error:[/red] Could not install extension from downloaded archive: "
-                                f"{_escape_markup(str(exc))}"
-                            )
-                            raise typer.Exit(1)
-                    finally:
-                        if download_file is not None:
-                            try:
-                                download_file.close()
-                            except OSError:
-                                pass
-                        elif download_fd >= 0:
-                            try:
-                                os.close(download_fd)
-                            except OSError:
-                                pass
-                except urllib.error.URLError as e:
-                    console.print(
-                        f"[red]Error:[/red] Failed to download from {safe_url}: "
-                        f"{_escape_markup(str(e))}"
-                    )
-                    raise typer.Exit(1)
             else:
                 # Try bundled extensions first (shipped with spec-kit)
                 bundled_path = _locate_bundled_extension(extension)

src/specify_cli/presets/__init__.py

@@ -299,6 +299,12 @@ class PresetManifest:
                 f"(expected {self.SCHEMA_VERSION})"
             )
 
+        for section in ("preset", "requires", "provides"):
+            if not isinstance(self.data[section], dict):
+                raise PresetValidationError(
+                    f"Invalid {section}: expected a mapping"
+                )
+
         # Validate preset metadata
         pack = self.data["preset"]
         for field in ["id", "name", "version", "description"]:

src/specify_cli/workflows/catalog.py

@@ -514,7 +514,7 @@ class WorkflowCatalog:
                     cached = json.load(f)
                 if isinstance(cached, dict):
                     return cached
-            except (json.JSONDecodeError, OSError):
+            except (UnicodeDecodeError, json.JSONDecodeError, OSError):
                 # Ignore invalid/unreadable cache and fall back to fetching from source.
                 pass
 
@@ -1210,7 +1210,7 @@ class StepCatalog:
                     cached = json.load(f)
                 if isinstance(cached, dict):
                     return cached
-            except (json.JSONDecodeError, OSError):
+            except (UnicodeDecodeError, json.JSONDecodeError, OSError):
                 # Ignore invalid/unreadable cache and fall back to fetching from source.
                 pass
 

src/specify_cli/workflows/engine.py

@@ -308,15 +308,16 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
         errors.append("Workflow has no steps defined.")
 
     seen_ids: set[str] = set()
-    # ``input_names`` is the set of declared workflow input names — used by
-    # ``_validate_steps`` to cross-reference gate ``verdict_input`` bindings.
-    # ``None`` means the inputs block itself is malformed (already reported
-    # above); the cross-check is then disabled so one authoring mistake does
-    # not cascade into N spurious "undeclared" errors.
-    input_names: set[str] | None = (
-        set(definition.inputs) if isinstance(definition.inputs, dict) else None
+    # ``input_defs`` maps declared workflow input names to their definitions —
+    # used by ``_validate_steps`` to cross-reference gate ``verdict_input``
+    # bindings (both that the name exists and that its ``enum`` permits the
+    # reset sentinel). ``None`` means the inputs block itself is malformed
+    # (already reported above); the cross-check is then disabled so one
+    # authoring mistake does not cascade into N spurious "undeclared" errors.
+    input_defs: dict[str, Any] | None = (
+        dict(definition.inputs) if isinstance(definition.inputs, dict) else None
     )
-    _validate_steps(definition.steps, seen_ids, errors, input_names)
+    _validate_steps(definition.steps, seen_ids, errors, input_defs)
 
     return errors
 
@@ -325,15 +326,15 @@ def _validate_steps(
     steps: list[dict[str, Any]],
     seen_ids: set[str],
     errors: list[str],
-    input_names: set[str] | None = None,
+    input_defs: dict[str, Any] | None = None,
     inside_fan_out: bool = False,
 ) -> None:
     """Recursively validate a list of steps.
 
-    ``input_names`` is the set of declared workflow input names (or ``None``
-    when the inputs block is malformed). ``inside_fan_out`` is threaded
-    through nested control-flow steps so gate verdict bindings can be rejected
-    anywhere inside a fan-out template.
+    ``input_defs`` maps declared workflow input names to their definitions (or
+    is ``None`` when the inputs block is malformed). ``inside_fan_out`` is
+    threaded through nested control-flow steps so gate verdict bindings can be
+    rejected anywhere inside a fan-out template.
     """
     from . import STEP_REGISTRY
 
@@ -440,11 +441,39 @@ def _validate_steps(
                         f"Gate step {step_id!r}: 'verdict_input' is not "
                         "supported inside fan-out templates."
                     )
-                elif input_names is not None and verdict_input not in input_names:
+                elif input_defs is not None and verdict_input not in input_defs:
                     errors.append(
                         f"Gate step {step_id!r}: 'verdict_input' references "
                         f"undeclared input {verdict_input!r}."
                     )
+                elif input_defs is not None:
+                    # ``on_reject: retry`` resets the bound input to "" before
+                    # pausing, and every later resume re-resolves the persisted
+                    # inputs through ``_coerce_input``. If the input declares an
+                    # ``enum`` that omits "", that reset value is instantly
+                    # illegal: the run pauses fine, but the next resume that
+                    # supplies any input raises "value '' not in allowed
+                    # values", and no verdict can be routed through the gate
+                    # again. Require the enum to admit the sentinel so the
+                    # retry cycle the field advertises is actually reachable.
+                    verdict_def = input_defs.get(verdict_input)
+                    enum_values = (
+                        verdict_def.get("enum")
+                        if isinstance(verdict_def, dict)
+                        else None
+                    )
+                    if (
+                        step_config.get("on_reject") == "retry"
+                        and isinstance(enum_values, list)
+                        and "" not in enum_values
+                    ):
+                        errors.append(
+                            f"Gate step {step_id!r}: on_reject='retry' resets "
+                            f"verdict input {verdict_input!r} to '' when the "
+                            f"gate is rejected, but that input's 'enum' does "
+                            f"not allow ''. Add '' to the enum or use "
+                            f"on_reject='abort'/'skip'."
+                        )
 
         # Recursively validate nested steps
         for nested_key in ("then", "else", "steps"):
@@ -454,7 +483,7 @@ def _validate_steps(
                     nested,
                     seen_ids,
                     errors,
-                    input_names,
+                    input_defs,
                     inside_fan_out=inside_fan_out,
                 )
 
@@ -467,7 +496,7 @@ def _validate_steps(
                         case_steps,
                         seen_ids,
                         errors,
-                        input_names,
+                        input_defs,
                         inside_fan_out=inside_fan_out,
                     )
 
@@ -478,7 +507,7 @@ def _validate_steps(
                 default,
                 seen_ids,
                 errors,
-                input_names,
+                input_defs,
                 inside_fan_out=inside_fan_out,
             )
 
@@ -491,7 +520,7 @@ def _validate_steps(
                 [fan_step],
                 set(),
                 fan_errors,
-                input_names,
+                input_defs,
                 inside_fan_out=True,
             )
             errors.extend(fan_errors)

src/specify_cli/workflows/steps/fan_in/__init__.py

@@ -20,9 +20,31 @@ class FanInStep(StepBase):
 
     def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         wait_for = config.get("wait_for", [])
-        output_config = config.get("output") or {}
-        if not isinstance(output_config, dict):
+        output_config = config.get("output")
+        if output_config is None:
             output_config = {}
+        elif not isinstance(output_config, dict):
+            # ``validate`` rejects a non-mapping ``output`` and its comment says
+            # why: "execute() silently coerces a non-mapping output to {}, so the
+            # author's declared aggregation keys would vanish with no error."
+            # The engine does not auto-validate before ``execute``, so on an
+            # unvalidated run that is exactly what happened -- and ``x or {}``
+            # masked the falsy shapes ([], false, 0, '') before the isinstance
+            # check even ran. Every declared key vanished while the step still
+            # reported COMPLETED, so downstream ``steps.<id>.output.<key>``
+            # resolved to None and interpolated as "": the same "silent empty
+            # result + COMPLETED" wiring bug the ``wait_for`` guard below
+            # rejects. Fail loudly with validate()'s own message instead. An
+            # explicit ``output:`` (YAML null) stays valid, matching validate.
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Fan-in step {config.get('id', '?')!r}: 'output' must be a "
+                    f"mapping of key -> expression, got "
+                    f"{type(output_config).__name__}."
+                ),
+                output={"results": []},
+            )
 
         # The engine does not auto-validate step config, so an unvalidated run
         # with a non-list ``wait_for`` reaches here raw. Iterating it then

src/specify_cli/workflows/steps/gate/__init__.py

@@ -75,6 +75,31 @@ class GateStep(StepBase):
                 },
             )
 
+        # ``validate`` rejects an ``on_reject`` outside abort/skip/retry, but the
+        # engine does not auto-validate before ``execute``. The reject branch
+        # below handles only "abort" and "retry" and then falls through to its
+        # ``on_reject == "skip"`` case, so on an unvalidated run any other value
+        # makes a REJECTED gate report COMPLETED and the run walks straight past
+        # the review the gate exists to enforce. Reachable by a capitalisation
+        # slip ("Abort"), a guessed verb ("fail", "stop"), a non-string, or the
+        # ``None`` that a bare ``on_reject:`` yields -- note ``config.get(k,
+        # default)`` does NOT substitute the default for an explicit null. Fail
+        # loudly instead, mirroring the ``options``/``verdict_input`` guards here.
+        if on_reject not in ("abort", "skip", "retry"):
+            return StepResult(
+                status=StepStatus.FAILED,
+                error=(
+                    f"Gate step {config.get('id', '?')!r}: 'on_reject' must be "
+                    f"'abort', 'skip', or 'retry', got {on_reject!r}."
+                ),
+                output={
+                    "message": message,
+                    "options": options,
+                    "on_reject": on_reject,
+                    "choice": None,
+                },
+            )
+
         if has_verdict_input and (
             not isinstance(verdict_input, str) or not verdict_input
         ):

tests/integrations/test_cli.py

@@ -2372,3 +2372,279 @@ def test_refresh_shared_templates_preserves_recovered_user_file(tmp_path):
 
     # Recovered user content must survive (fail-before: replaced by bundled body).
     assert user_file.read_text(encoding="utf-8") == "# USER CUSTOM CONTENT\n"
+
+
+class TestExtensionFlag:
+    """Tests for the --extension flag on specify init."""
+
+    def _run_init(self, tmp_path, args, project_name="ext-test"):
+        from unittest.mock import patch
+        from typer.testing import CliRunner
+        from specify_cli import app
+
+        project = tmp_path / project_name
+        project.mkdir(exist_ok=True)
+        old_cwd = os.getcwd()
+        try:
+            os.chdir(project)
+            runner = CliRunner()
+            # Patch get_speckit_version to return a stable (non-dev) version so that
+            # the extension compatibility check (SpecifierSet(">=0.2.0")) passes.
+            with patch(
+                "specify_cli.commands.init.get_speckit_version",
+                return_value="0.8.2",
+            ):
+                result = runner.invoke(app, [
+                    "init", "--here",
+                    "--integration", "copilot",
+                    "--script", "sh",
+                    "--ignore-agent-tools",
+                ] + args, catch_exceptions=False)
+        finally:
+            os.chdir(old_cwd)
+        return project, result
+
+    def test_bundled_extension_installed(self, tmp_path):
+        """--extension git installs the bundled git extension."""
+        project, result = self._run_init(tmp_path, ["--extension", "git"], project_name="ext-bundled")
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+
+        ext_dir = project / ".specify" / "extensions" / "git"
+        assert ext_dir.exists(), "git extension directory not found"
+        assert (ext_dir / "extension.yml").exists(), "extension.yml not found"
+
+        # Tracker should show extension step as done
+        normalized = _normalize_cli_output(result.output)
+        assert "Install extension: git" in normalized
+
+    def test_multiple_extensions_installed(self, tmp_path):
+        """--extension can be specified multiple times."""
+        project, result = self._run_init(
+            tmp_path,
+            ["--extension", "git", "--extension", "selftest"],
+            project_name="ext-multi",
+        )
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+
+        ext_dir_git = project / ".specify" / "extensions" / "git"
+        ext_dir_selftest = project / ".specify" / "extensions" / "selftest"
+        assert ext_dir_git.exists(), "git extension not installed"
+        assert ext_dir_selftest.exists(), "selftest extension not installed"
+
+    def test_local_path_extension_installed(self, tmp_path):
+        """--extension /abs/path installs from a local absolute directory path."""
+        from specify_cli import _locate_bundled_extension
+
+        # Use the bundled git extension directory as our "local" extension source
+        bundled_git = _locate_bundled_extension("git")
+        assert bundled_git is not None, "bundled git extension not found; cannot run test"
+
+        # Pass the absolute path directly (starts with "/")
+        project, result = self._run_init(
+            tmp_path,
+            ["--extension", str(bundled_git)],
+            project_name="ext-local",
+        )
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+
+        ext_dir = project / ".specify" / "extensions" / "git"
+        assert ext_dir.exists(), "extension from local path not installed"
+
+    def test_unknown_extension_shows_error_in_tracker(self, tmp_path):
+        """An unknown extension name records a tracker error but does not abort init."""
+        project, result = self._run_init(
+            tmp_path,
+            ["--extension", "nonexistent-xyz-ext"],
+            project_name="ext-unknown",
+        )
+
+        assert result.exit_code == 0, "init should not abort on unknown extension"
+        normalized = _normalize_cli_output(result.output)
+        assert "failed" in normalized.lower(), "expected 'failed' for unknown extension"
+
+    def test_extension_flag_works_with_preset(self, tmp_path):
+        """--extension and --preset can be combined."""
+        project, result = self._run_init(
+            tmp_path,
+            ["--extension", "git", "--preset", "lean"],
+            project_name="ext-preset",
+        )
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+
+        ext_dir = project / ".specify" / "extensions" / "git"
+        assert ext_dir.exists(), "git extension not installed alongside preset"
+
+    @staticmethod
+    def _zip_bytes_from_dir(source_dir):
+        """Build in-memory ZIP bytes from an extension directory (yml at root)."""
+        import io
+        import zipfile
+
+        buf = io.BytesIO()
+        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
+            for path in sorted(source_dir.rglob("*")):
+                if path.is_file():
+                    zf.write(path, arcname=str(path.relative_to(source_dir)))
+        return buf.getvalue()
+
+    def test_url_extension_rejects_non_https(self, tmp_path):
+        """A non-HTTPS URL is rejected before any download; init is not aborted."""
+        project, result = self._run_init(
+            tmp_path,
+            ["--extension", "http://example.com/ext.zip", "--trust-extension-urls"],
+            project_name="ext-http",
+        )
+
+        assert result.exit_code == 0, "init should not abort on a rejected URL"
+        normalized = _normalize_cli_output(result.output)
+        assert "failed" in normalized.lower()
+        # No extension directory should have been created for the bad URL.
+        assert not (project / ".specify" / "extensions" / "ext").exists()
+
+    def test_url_extension_skipped_without_trust(self, tmp_path):
+        """Non-interactive URL install without --trust-extension-urls is denied."""
+        from unittest.mock import patch
+
+        with patch(
+            "specify_cli.commands.init._stdin_is_interactive", return_value=False
+        ), patch("specify_cli.authentication.http.open_url") as mock_open:
+            project, result = self._run_init(
+                tmp_path,
+                ["--extension", "https://example.com/git.zip"],
+                project_name="ext-url-denied",
+            )
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+        # Default-deny: no download attempted, nothing installed.
+        mock_open.assert_not_called()
+        normalized = _normalize_cli_output(result.output)
+        assert "untrusted url" in normalized.lower()
+        assert not (project / ".specify" / "extensions" / "git").exists()
+
+    def test_url_extension_interactive_confirm_installs(self, tmp_path):
+        """An interactive 'yes' to the trust prompt allows the URL install."""
+        import io
+
+        from unittest.mock import patch
+
+        from specify_cli import _locate_bundled_extension
+
+        bundled_git = _locate_bundled_extension("git")
+        assert bundled_git is not None, "bundled git extension not found"
+        zip_bytes = self._zip_bytes_from_dir(bundled_git)
+
+        class FakeResponse(io.BytesIO):
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+        def _cache_dir_stand_in(project_root):
+            d = project_root / ".specify" / "extensions" / ".cache" / "downloads"
+            d.mkdir(parents=True, exist_ok=True)
+            return d
+
+        def _open_download_zip(project_root, download_dir, zip_filename):
+            target = download_dir / zip_filename
+            o_temporary = getattr(os, "O_TEMPORARY", 0)
+            if o_temporary:
+                return os.open(
+                    target, os.O_RDWR | os.O_CREAT | os.O_EXCL | o_temporary, 0o600
+                )
+            fd = os.open(target, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
+            try:
+                os.unlink(target)
+            except OSError:
+                os.close(fd)
+                raise
+            return fd
+
+        with patch(
+            "specify_cli.commands.init._stdin_is_interactive", return_value=True
+        ), patch("typer.confirm", return_value=True), patch(
+            "specify_cli.authentication.http.open_url",
+            return_value=FakeResponse(zip_bytes),
+        ), patch(
+            "specify_cli.extensions._commands._validate_safe_cache_dir",
+            side_effect=_cache_dir_stand_in,
+        ), patch(
+            "specify_cli.extensions._commands._safe_open_download_zip",
+            side_effect=_open_download_zip,
+        ):
+            project, result = self._run_init(
+                tmp_path,
+                ["--extension", "https://example.com/git.zip"],
+                project_name="ext-url-confirm",
+            )
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+        assert (project / ".specify" / "extensions" / "git").exists()
+
+    def test_url_extension_installs_zip(self, tmp_path):
+        """A successful HTTPS ZIP download installs via the shared hardened path."""
+        import io
+
+        from unittest.mock import patch
+
+        from specify_cli import _locate_bundled_extension
+
+        bundled_git = _locate_bundled_extension("git")
+        assert bundled_git is not None, "bundled git extension not found"
+        zip_bytes = self._zip_bytes_from_dir(bundled_git)
+
+        class FakeResponse(io.BytesIO):
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc, tb):
+                return False
+
+        def _cache_dir_stand_in(project_root):
+            d = project_root / ".specify" / "extensions" / ".cache" / "downloads"
+            d.mkdir(parents=True, exist_ok=True)
+            return d
+
+        def _open_download_zip(project_root, download_dir, zip_filename):
+            target = download_dir / zip_filename
+            o_temporary = getattr(os, "O_TEMPORARY", 0)
+            if o_temporary:
+                return os.open(
+                    target, os.O_RDWR | os.O_CREAT | os.O_EXCL | o_temporary, 0o600
+                )
+            fd = os.open(target, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
+            try:
+                os.unlink(target)
+            except OSError:
+                os.close(fd)
+                raise
+            return fd
+
+        with patch(
+            "specify_cli.authentication.http.open_url",
+            return_value=FakeResponse(zip_bytes),
+        ), patch(
+            "specify_cli.extensions._commands._validate_safe_cache_dir",
+            side_effect=_cache_dir_stand_in,
+        ), patch(
+            "specify_cli.extensions._commands._safe_open_download_zip",
+            side_effect=_open_download_zip,
+        ):
+            project, result = self._run_init(
+                tmp_path,
+                ["--extension", "https://example.com/git.zip", "--trust-extension-urls"],
+                project_name="ext-url",
+            )
+
+        assert result.exit_code == 0, f"init failed:\n{result.output}"
+        ext_dir = project / ".specify" / "extensions" / "git"
+        assert ext_dir.exists(), "extension from URL not installed"
+        assert (ext_dir / "extension.yml").exists()
+        # Transient download archive must not linger in the cache.
+        cache_dir = project / ".specify" / "extensions" / ".cache" / "downloads"
+        leftover = list(cache_dir.glob("*.zip")) if cache_dir.exists() else []
+        assert not leftover, f"download cache not cleaned: {leftover}"

tests/test_extensions.py

@@ -5003,9 +5003,9 @@ class TestExtensionCatalog:
         catalog = self._make_catalog(temp_dir)
 
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(
+        mock_response.read.side_effect = io.BytesIO(json.dumps(
             {"schema_version": "1.0", "extensions": {}}
-        ).encode()
+        ).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         mock_response.geturl.return_value = "http://evil.test/catalog.json"
@@ -5051,9 +5051,9 @@ class TestExtensionCatalog:
 
         catalog = self._make_catalog(temp_dir)
         mock_response = MagicMock()
-        mock_response.read.return_value = json.dumps(
+        mock_response.read.side_effect = io.BytesIO(json.dumps(
             {"schema_version": "1.0", "extensions": {}}
-        ).encode()
+        ).encode()).read
         mock_response.__enter__ = lambda s: s
         mock_response.__exit__ = MagicMock(return_value=False)
         mock_response.geturl.return_value = "http://evil.test/catalog.json"

tests/test_presets.py

@@ -198,6 +198,25 @@ class TestPresetManifest:
             with pytest.raises(PresetValidationError, match="YAML mapping"):
                 PresetManifest(manifest_path)
 
+    @pytest.mark.parametrize("section", ["preset", "requires", "provides"])
+    @pytest.mark.parametrize("bad_value", [None, [], "text"])
+    def test_required_section_not_mapping_raises_validation_error(
+        self, temp_dir, valid_pack_data, section, bad_value
+    ):
+        """Required manifest sections reject null, list, and scalar values."""
+        valid_pack_data[section] = bad_value
+        manifest_path = temp_dir / "preset.yml"
+        manifest_path.write_text(
+            yaml.safe_dump(valid_pack_data),
+            encoding="utf-8",
+        )
+
+        with pytest.raises(
+            PresetValidationError,
+            match=rf"Invalid {section}: expected a mapping",
+        ):
+            PresetManifest(manifest_path)
+
     @pytest.mark.parametrize(
         "bad",
         [

tests/test_workflows.py

@@ -2552,6 +2552,36 @@ steps:
         })
         assert any("on_reject" in e for e in errors)
 
+    @pytest.mark.parametrize(
+        "bad_on_reject", ["Abort", "fail", "stop", "SKIP", None, 5, ["abort"]]
+    )
+    def test_execute_invalid_on_reject_fails_loudly(self, bad_on_reject):
+        """An unrecognised ``on_reject`` must not silently complete a rejection.
+
+        ``validate`` rejects anything outside abort/skip/retry, but the engine
+        does not auto-validate before ``execute``. The reject branch handles only
+        "abort" and "retry", then falls through to its ``"skip"`` case — so a
+        REJECTED gate reported COMPLETED and the run continued past the review
+        the gate exists to enforce. Reachable by a capitalisation slip, a guessed
+        verb, a non-string, or a bare ``on_reject:`` (which yields None, since
+        ``config.get(k, default)`` does not replace an explicit null).
+        """
+        from specify_cli.workflows.steps.gate import GateStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        result = GateStep().execute(
+            {
+                "id": "review",
+                "message": "Review the spec.",
+                "options": ["approve", "reject"],
+                "on_reject": bad_on_reject,
+                "verdict_input": "spec_verdict",
+            },
+            StepContext(inputs={"spec_verdict": "reject"}),
+        )
+        assert result.status == StepStatus.FAILED
+        assert "'on_reject' must be" in (result.error or "")
+
     def test_validate_non_string_options_does_not_raise(self):
         """Non-string options with on_reject=abort/retry must be REPORTED as an
         error, not crash: the reject-choice check calls o.lower() on each option,
@@ -3629,6 +3659,44 @@ class TestFanInStep:
         assert "'wait_for' must be a list" in (result.error or "")
         assert result.output["results"] == []
 
+    @pytest.mark.parametrize(
+        "bad_output", [[], False, 0, "", ["a"], "oops", 5]
+    )
+    def test_execute_non_mapping_output_fails_loudly(self, bad_output):
+        """A non-mapping ``output`` must fail the step, not drop every key.
+
+        ``validate`` rejects it and says why: "execute() silently coerces a
+        non-mapping output to {}, so the author's declared aggregation keys would
+        vanish with no error." The engine does not auto-validate before
+        ``execute``, so that is exactly what happened — and ``x or {}`` masked
+        the falsy shapes (``[]``, ``false``, ``0``, ``''``) before the isinstance
+        check even ran. The step still returned COMPLETED, so downstream
+        ``steps.<id>.output.<key>`` resolved to None and interpolated as "".
+        """
+        from specify_cli.workflows.steps.fan_in import FanInStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = FanInStep()
+        ctx = StepContext(steps={"a": {"output": {"x": 1}}})
+        result = step.execute(
+            {"id": "collect", "wait_for": ["a"], "output": bad_output}, ctx
+        )
+        assert result.status == StepStatus.FAILED
+        assert "'output' must be a mapping" in (result.error or "")
+        assert result.output["results"] == []
+
+    def test_execute_explicit_null_output_stays_valid(self):
+        """An explicit ``output:`` (YAML null) is valid, matching ``validate``."""
+        from specify_cli.workflows.steps.fan_in import FanInStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        step = FanInStep()
+        ctx = StepContext(steps={"a": {"output": {"x": 1}}})
+        result = step.execute(
+            {"id": "collect", "wait_for": ["a"], "output": None}, ctx
+        )
+        assert result.status == StepStatus.COMPLETED
+
     @pytest.mark.parametrize("bad_entry", [["a", "b"], {"a": 1}, 123, None])
     def test_execute_non_string_wait_for_entry_fails_loudly(self, bad_entry):
         """A ``wait_for`` list with a non-string entry must fail the step, not
@@ -4673,6 +4741,149 @@ steps:
         # No undeclared-input error (123 is not a string, so cross-check skips)
         assert not any("undeclared input" in e for e in errors)
 
+    def test_retry_verdict_enum_must_allow_reset_sentinel(self):
+        # on_reject: retry resets the bound input to "" before pausing, and
+        # every resume re-resolves persisted inputs through _coerce_input. An
+        # enum that omits "" makes that reset value instantly illegal, so the
+        # next resume supplying any input dies with "value '' not in allowed
+        # values" and no verdict can reach the gate again.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    enum: [approve, reject]
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+""")
+        assert any(
+            "on_reject='retry' resets verdict input 'spec_verdict'" in e
+            for e in errors
+        ), errors
+
+    def test_retry_verdict_enum_including_sentinel_passes(self):
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    enum: ["", approve, reject]
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+""")
+        assert not any("on_reject='retry'" in e for e in errors), errors
+
+    def test_verdict_enum_without_sentinel_passes_when_not_retry(self):
+        # abort/skip never reset the input, so the enum need not admit "".
+        for on_reject in ("abort", "skip"):
+            errors = self._errors(f"""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    enum: [approve, reject]
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: {on_reject}
+    verdict_input: spec_verdict
+""")
+            assert not any("on_reject='retry'" in e for e in errors), (
+                on_reject,
+                errors,
+            )
+
+    def test_retry_verdict_without_enum_passes(self):
+        # No enum means _coerce_input accepts "" — the documented shape.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+inputs:
+  spec_verdict:
+    type: string
+    default: ""
+steps:
+  - id: review
+    type: gate
+    message: "Review?"
+    options: [approve, reject]
+    on_reject: retry
+    verdict_input: spec_verdict
+""")
+        assert not any("on_reject='retry'" in e for e in errors), errors
+
+    def test_retry_verdict_enum_wedge_is_reachable_end_to_end(self, tmp_path):
+        """The validation error above guards a real, unrecoverable run state.
+
+        Without the guard this workflow installs and runs fine, then wedges:
+        the retry reset writes "" into the persisted inputs, and the next
+        resume that supplies *any* input re-resolves them and dies on the
+        enum. Only a resume with no inputs at all still works, so the bound
+        verdict can never be delivered.
+        """
+        import pytest
+        import yaml as _yaml
+
+        from specify_cli.workflows.engine import WorkflowEngine
+
+        definition_data = {
+            "schema_version": "1.0",
+            "workflow": {"id": "wf", "name": "WF", "version": "1.0.0"},
+            "inputs": {
+                "spec_verdict": {"type": "string", "enum": ["approve", "reject"]},
+                "note": {"type": "string", "default": "a"},
+            },
+            "steps": [
+                {
+                    "id": "review",
+                    "type": "gate",
+                    "message": "Review?",
+                    "options": ["approve", "reject"],
+                    "on_reject": "retry",
+                    "verdict_input": "spec_verdict",
+                }
+            ],
+        }
+        wf_dir = tmp_path / ".specify" / "workflows" / "wf"
+        wf_dir.mkdir(parents=True)
+        (wf_dir / "workflow.yml").write_text(
+            _yaml.safe_dump(definition_data), encoding="utf-8"
+        )
+
+        engine = WorkflowEngine(tmp_path)
+        definition = engine.load_workflow("wf")
+        state = engine.execute(definition, inputs={"spec_verdict": "reject"})
+        assert state.status.value == "paused"
+        # The retry reset persisted a value the input's own enum forbids.
+        assert state.inputs["spec_verdict"] == ""
+
+        with pytest.raises(ValueError, match="not in allowed values"):
+            engine.resume(state.run_id, inputs={"note": "b"})
+
     def test_verdict_input_in_switch_case(self):
         # Recursion coverage: bad reference inside a switch case must surface.
         errors = self._errors("""
@@ -7249,6 +7460,59 @@ class TestWorkflowCatalog:
 
         assert catalog._fetch_single_catalog(entry) == payload
 
+    @pytest.mark.parametrize("catalog_type", ["workflow", "step"])
+    def test_non_utf8_cached_catalog_is_refetched(
+        self, project_dir, monkeypatch, catalog_type
+    ):
+        import io
+
+        from specify_cli.authentication import http as auth_http
+        from specify_cli.workflows.catalog import (
+            StepCatalog,
+            StepCatalogEntry,
+            WorkflowCatalog,
+            WorkflowCatalogEntry,
+        )
+
+        catalog_cls = WorkflowCatalog if catalog_type == "workflow" else StepCatalog
+        entry_cls = (
+            WorkflowCatalogEntry
+            if catalog_type == "workflow"
+            else StepCatalogEntry
+        )
+        payload_key = "workflows" if catalog_type == "workflow" else "steps"
+        url = f"https://example.com/{catalog_type}.json"
+        catalog = catalog_cls(project_dir)
+        cache_path, metadata_path = catalog._get_cache_paths(url)
+        cache_path.parent.mkdir(parents=True, exist_ok=True)
+        cache_path.write_bytes(b"\xff\xfe")
+        metadata_path.write_text(
+            json.dumps({"fetched_at": 4_102_444_800}),
+            encoding="utf-8",
+        )
+        payload = {"schema_version": "1.0", payload_key: {}}
+
+        class _FakeResponse(io.BytesIO):
+            def geturl(self):
+                return url
+
+        monkeypatch.setattr(
+            auth_http,
+            "open_url",
+            lambda url, timeout=30, redirect_validator=None: _FakeResponse(
+                json.dumps(payload).encode("utf-8")
+            ),
+        )
+        entry = entry_cls(
+            url=url,
+            name="test",
+            priority=1,
+            install_allowed=True,
+        )
+
+        assert catalog._fetch_single_catalog(entry) == payload
+        assert json.loads(cache_path.read_text(encoding="utf-8")) == payload
+
     def test_non_mapping_stale_workflow_catalog_is_rejected(
         self, project_dir, monkeypatch
     ):

tests/unit/test_bundler_adapters.py

@@ -104,6 +104,18 @@ def test_fetch_rejects_malformed_source_url_cleanly(url):
         fetcher(_source(url))
 
 
+@pytest.mark.parametrize("use_file_url", [False, True], ids=["path", "file-url"])
+def test_local_catalog_decode_errors_are_wrapped(tmp_path, use_file_url):
+    catalog_path = tmp_path / "catalog.json"
+    catalog_path.write_bytes(b"\xff\xfe")
+    url = catalog_path.as_uri() if use_file_url else str(catalog_path)
+
+    fetcher = adapters.make_catalog_fetcher(allow_network=False)
+
+    with pytest.raises(BundlerError, match="Could not read"):
+        fetcher(_source(url))
+
+
 def test_builtin_community_catalog_fetches_repository_catalog_online(monkeypatch):
     captured: dict = {}
 

tests/unit/test_bundler_packager.py

@@ -207,4 +207,30 @@ def test_executable_bit_preserved_in_artifact(tmp_path: Path):
         }
     # Executable source -> 0755; plain text files -> 0644.
     assert modes["scripts/hook.sh"] == 0o755
+
+
+def test_toctou_stat_read_consistency(tmp_path: Path):
+    """Regression: stat() and read() must use the same file descriptor.
+
+    The old implementation called file_path.stat() then file_path.read_bytes()
+    as separate syscalls. Between the two, another process could replace the
+    file. The fix opens the file once and uses os.fstat() + fh.read() on the
+    same handle. This test verifies the archived bytes and mode are consistent.
+    """
+    bundle = _make_bundle(tmp_path / "b")
+    target = bundle / "assets" / "data.bin"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_bytes(b"\x00\x01\x02\x03")
+    target.chmod(0o644)
+
+    result = build_bundle(bundle, output_dir=tmp_path / "out")
+    with zipfile.ZipFile(result.artifact_path) as archive:
+        content = archive.read("assets/data.bin")
+        modes = {
+            info.filename: (info.external_attr >> 16) & 0o777
+            for info in archive.infolist()
+        }
+
+    assert content == b"\x00\x01\x02\x03"
+    assert modes["assets/data.bin"] == 0o644
     assert modes["README.md"] == 0o644