Method execute

src/specify_cli/workflows/steps/gate/__init__.py:38–90

(self, config: dict[str, Any], context: StepContext)

    MAX_SHOW_FILE_LINES = 200

    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
        message = config.get("message", "Review required.")
        if isinstance(message, str) and "{{" in message:
            message = evaluate_expression(message, context)

        options = config.get("options", ["approve", "reject"])
        on_reject = config.get("on_reject", "abort")

        show_file = config.get("show_file")
        if isinstance(show_file, str) and "{{" in show_file:
            show_file = evaluate_expression(show_file, context)
        # ``evaluate_expression`` can return a non-string for a single
        # expression (e.g. a number from a prior step), and a literal
        # non-string is also possible; coerce so it is rendered rather
        # than silently skipped at the prompt.
        if show_file is not None:
            show_file = str(show_file)

        output = {
            "message": message,
            "options": options,
            "on_reject": on_reject,
            "show_file": show_file,
            "choice": None,
        }

        # Non-interactive: pause for later resume (the file is not read here)
        if not sys.stdin.isatty():
            return StepResult(status=StepStatus.PAUSED, output=output)

        # Interactive: prompt the user. ``show_file`` contents are folded
        # into the displayed message so the operator can review the
        # referenced material before choosing. Composing the prompt text
        # here keeps ``_prompt`` to its ``(message, options)`` contract, so
        # adding review material never widens the interactive seam.
        choice = self._prompt(self._compose_prompt(message, show_file), options)
        output["choice"] = choice

        if choice in ("reject", "abort"):
            if on_reject == "abort":
                output["aborted"] = True
                return StepResult(
                    status=StepStatus.FAILED,
                    output=output,
                    error=f"Gate rejected by user at step {config.get('id', '?')!r}",
                )
            if on_reject == "retry":
                # Pause so the next resume re-executes this gate
                return StepResult(status=StepStatus.PAUSED, output=output)
            # on_reject == "skip" → completed, downstream steps decide
            return StepResult(status=StepStatus.COMPLETED, output=output)

        return StepResult(status=StepStatus.COMPLETED, output=output)

    @classmethod
    def _compose_prompt(cls, message: object, show_file: str | None) -> str: