mirror of
https://github.com/github/spec-kit.git
synced 2026-07-03 12:28:06 +08:00
fix(workflows): render gate show_file contents in the interactive prompt (#2810)
* fix(workflows): render gate show_file contents in the interactive prompt The gate step read and recorded `show_file` but never displayed its contents at the interactive prompt, so the operator approved/rejected without seeing the referenced file. Render the file inside the prompt when stdin is a TTY, with a graceful notice for missing/unreadable files. Non-interactive PAUSED behaviour, exit codes, resume semantics, and no-`show_file` output are unchanged. Closes #2809. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(workflows): keep gate _prompt signature stable and harden show_file reads The gate prompt rendered show_file by passing it as a third positional argument to _prompt. A test that stubs _prompt with a two-argument lambda (test_gate_abort_still_halts_with_continue_on_error) then failed once the branch caught up to main, because the call site passed three arguments to the two-argument stub. Compose the show_file material into the displayed message in execute() and keep _prompt to its (message, options) contract. Display data no longer widens the interactive seam, so stubbing _prompt stays stable and future review material can be added without breaking callers. _prompt now renders a multi-line message inside the gate box. Also catch ValueError in _read_show_file so a path the OS rejects outright (e.g. an embedded NUL byte) degrades to a notice instead of crashing the prompt, matching the helper's stated contract. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(workflows): coerce gate prompt message to str before rendering The multi-line render loop split the message on newlines, which assumes a str. A non-string message (e.g. a YAML numeric literal) previously rendered fine through the old f-string and would now raise on .split. Coerce with str() to preserve that tolerance, and add a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(workflows): make gate stdin handling robust; tidy compose_prompt typing Address review feedback on the gate tests and helper: - Swap the gate module's sys.stdin for a fixed-isatty stub (shared _StubStdin / _force_gate_stdin helpers) instead of setattr on sys.stdin.isatty, which is not assignable under some pytest capture modes. This also forces the non-interactive tests to a non-TTY so they cannot block on input() when run in a real terminal. - The non-interactive show_file test now hard-fails if _read_show_file is called, proving the file is not read on the PAUSED path. - _compose_prompt accepts a non-string message (e.g. a YAML numeric literal) and always returns str via str(message), keeping its annotation and docstring accurate; the redundant coercion in _prompt is removed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(workflows): strip control chars from gate show_file; default tests non-TTY Address review feedback: - _read_show_file strips C0 control characters (except tab) from each line, so a show_file containing ANSI escape sequences (e.g. \x1b[2J) cannot clear the screen or spoof the prompt/options when rendered to a terminal. - Add an autouse fixture on TestGateStep that defaults every gate test to a non-TTY stdin, so no test can drop into the interactive prompt and block on input() when the suite runs under a real TTY. Interactive tests opt back in via _force_gate_stdin(tty=True); the now-redundant explicit non-TTY calls were removed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(workflows): localize gate stdin patch to the gate module's sys _force_gate_stdin rebinds the gate module's `sys` name to a stand-in whose stdin has a fixed isatty() and which delegates every other attribute to the real sys, instead of mutating the process-wide sys.stdin. This keeps the patch local to the gate module and leaves real stdin untouched. The gate abort test, which used the same process-wide swap, now shares the helper, so the pattern exists in exactly one place. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(workflows): sanitize the displayed gate show_file path, not just content Control characters were stripped from show_file *contents* but the path was still printed verbatim as the header (`f"{show_file}:"`) and echoed in the read-error notice, so a show_file path containing ANSI escapes could still inject terminal sequences. Centralize stripping in `_sanitize_for_display` and apply it to every show_file-derived string that reaches the terminal — the displayed path, each file line, and the error notice — while still opening the file with the original path. Add a test for path sanitization. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * refactor(workflows): inline control-char stripping, drop the helper Reuse the existing _CONTROL_CHARS regex directly at the three display sites instead of wrapping it in a one-line helper. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(workflows): also strip LF and C1 controls from gate show_file display The control-char class skipped LF (so an embedded newline in a show_file path could break the boxed layout) and the C1 range (so \x9b CSI and other 8-bit controls survived). Widen the class to [\x00-\x08\x0a-\x1f\x7f-\x9f] (still keeping tab). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Huy Do
@@ -2,12 +2,20 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus
|
||||
from specify_cli.workflows.expressions import evaluate_expression
|
||||
|
||||
#: Control characters except tab: C0 (incl. LF, so an embedded newline cannot
|
||||
#: break the boxed layout), DEL, and C1 (incl. ``\x9b`` CSI). Stripped from
|
||||
#: anything derived from a ``show_file`` before it is printed — the file's
|
||||
#: contents and the path itself — so neither can inject ANSI/terminal escapes.
|
||||
_CONTROL_CHARS = re.compile(r"[\x00-\x08\x0a-\x1f\x7f-\x9f]")
|
||||
|
||||
|
||||
class GateStep(StepBase):
|
||||
"""Interactive review gate.
|
||||
@@ -23,6 +31,10 @@ class GateStep(StepBase):
|
||||
|
||||
type_key = "gate"
|
||||
|
||||
#: Maximum number of ``show_file`` lines rendered at the prompt, so a
|
||||
#: large file cannot flood the terminal before the choice.
|
||||
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:
|
||||
@@ -32,8 +44,14 @@ class GateStep(StepBase):
|
||||
on_reject = config.get("on_reject", "abort")
|
||||
|
||||
show_file = config.get("show_file")
|
||||
if show_file and isinstance(show_file, str) and "{{" in 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,
|
||||
@@ -43,12 +61,16 @@ class GateStep(StepBase):
|
||||
"choice": None,
|
||||
}
|
||||
|
||||
# Non-interactive: pause for later resume
|
||||
# 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
|
||||
choice = self._prompt(message, options)
|
||||
# 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"):
|
||||
@@ -67,11 +89,38 @@ class GateStep(StepBase):
|
||||
|
||||
return StepResult(status=StepStatus.COMPLETED, output=output)
|
||||
|
||||
@classmethod
|
||||
def _compose_prompt(cls, message: object, show_file: str | None) -> str:
|
||||
"""Build the gate's display text.
|
||||
|
||||
``message`` may be a non-string (e.g. a YAML numeric literal that
|
||||
``execute`` does not coerce), so it is rendered through ``str``.
|
||||
When ``show_file`` names a file, its contents (read safely, see
|
||||
``_read_show_file``) are appended below the message so the operator
|
||||
can review the referenced material before choosing. Always returns a
|
||||
``str`` — possibly multi-line — for ``_prompt`` to render in the box.
|
||||
"""
|
||||
text = str(message)
|
||||
if not show_file:
|
||||
return text
|
||||
# The path is opened with the original value but displayed stripped,
|
||||
# so a path that itself contains escapes cannot spoof the terminal.
|
||||
header = f"{_CONTROL_CHARS.sub('', show_file)}:"
|
||||
body = "\n".join(
|
||||
[header, *(f" {line}" for line in cls._read_show_file(show_file))]
|
||||
)
|
||||
return f"{text}\n\n{body}"
|
||||
|
||||
@staticmethod
|
||||
def _prompt(message: str, options: list[str]) -> str:
|
||||
"""Display gate message and prompt for a choice."""
|
||||
"""Display the gate message and prompt for a choice.
|
||||
|
||||
``message`` may span multiple lines (e.g. when review material has
|
||||
been folded in); each line is rendered inside the gate box.
|
||||
"""
|
||||
print("\n ┌─ Gate ─────────────────────────────────────")
|
||||
print(f" │ {message}")
|
||||
for line in message.split("\n"):
|
||||
print(f" │ {line}" if line else " │")
|
||||
print(" │")
|
||||
for i, opt in enumerate(options, 1):
|
||||
print(f" │ [{i}] {opt}")
|
||||
@@ -90,6 +139,40 @@ class GateStep(StepBase):
|
||||
return next(o for o in options if o.lower() == raw.lower())
|
||||
print(f" Invalid choice. Enter 1-{len(options)} or an option name.")
|
||||
|
||||
@staticmethod
|
||||
def _read_show_file(show_file: str) -> list[str]:
|
||||
"""Return the lines of ``show_file`` for display.
|
||||
|
||||
Reads at most ``MAX_SHOW_FILE_LINES`` lines so a large file cannot
|
||||
flood the prompt, and returns a short notice instead of raising
|
||||
when the file is missing, undecodable, or names an invalid path,
|
||||
so a misconfigured ``show_file`` never breaks the interactive
|
||||
prompt. ``ValueError`` covers paths the OS rejects outright (e.g.
|
||||
an embedded NUL byte), which ``Path.open`` raises before any I/O.
|
||||
|
||||
Control characters are stripped from each line so file content
|
||||
cannot inject ANSI escape sequences into the terminal.
|
||||
"""
|
||||
lines: list[str] = []
|
||||
truncated = False
|
||||
try:
|
||||
with Path(show_file).open(encoding="utf-8") as handle:
|
||||
for line in handle:
|
||||
if len(lines) >= GateStep.MAX_SHOW_FILE_LINES:
|
||||
truncated = True
|
||||
break
|
||||
lines.append(_CONTROL_CHARS.sub("", line.rstrip("\n")))
|
||||
except (OSError, UnicodeDecodeError, ValueError) as exc:
|
||||
# ``exc`` echoes the (possibly hostile) path, so strip it too.
|
||||
return [_CONTROL_CHARS.sub("", f"(could not read file: {exc})")]
|
||||
if not lines and not truncated:
|
||||
return ["(file is empty)"]
|
||||
if truncated:
|
||||
lines.append(
|
||||
f"… (output truncated at {GateStep.MAX_SHOW_FILE_LINES} lines)"
|
||||
)
|
||||
return lines
|
||||
|
||||
def validate(self, config: dict[str, Any]) -> list[str]:
|
||||
errors = super().validate(config)
|
||||
if "message" not in config:
|
||||
|
||||
@@ -15,6 +15,7 @@ from __future__ import annotations
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
@@ -859,9 +860,55 @@ class TestShellStep:
|
||||
assert any("missing 'run'" in e for e in errors)
|
||||
|
||||
|
||||
class _StubStdin:
|
||||
"""Stdin stub exposing only a fixed ``isatty`` result.
|
||||
|
||||
A real ``TextIOWrapper.isatty`` is not assignable under some runners
|
||||
(e.g. pytest with capture disabled), so the gate tests force the value
|
||||
through this stub to stay deterministic regardless of how the suite is
|
||||
run.
|
||||
"""
|
||||
|
||||
def __init__(self, tty: bool):
|
||||
self._tty = tty
|
||||
|
||||
def isatty(self) -> bool:
|
||||
return self._tty
|
||||
|
||||
|
||||
class _FakeSys:
|
||||
"""Stand-in for the gate module's ``sys`` with a fixed-``isatty`` stdin.
|
||||
|
||||
Every other attribute delegates to the real ``sys``. Rebinding the gate
|
||||
module's ``sys`` name (rather than mutating the process-wide
|
||||
``sys.stdin``) keeps the patch local to the gate module and leaves the
|
||||
real stdin untouched.
|
||||
"""
|
||||
|
||||
def __init__(self, tty: bool):
|
||||
self.stdin = _StubStdin(tty)
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(sys, name)
|
||||
|
||||
|
||||
def _force_gate_stdin(monkeypatch, *, tty: bool):
|
||||
from specify_cli.workflows.steps import gate as gate_module
|
||||
|
||||
monkeypatch.setattr(gate_module, "sys", _FakeSys(tty=tty))
|
||||
|
||||
|
||||
class TestGateStep:
|
||||
"""Test the gate step type."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _non_tty_stdin_by_default(self, monkeypatch):
|
||||
# Default every gate test to a non-TTY stdin so none can drop into
|
||||
# the interactive prompt and block on input() when the suite runs
|
||||
# with a real TTY. Interactive tests opt back in with
|
||||
# _force_gate_stdin(monkeypatch, tty=True).
|
||||
_force_gate_stdin(monkeypatch, tty=False)
|
||||
|
||||
def test_execute_returns_paused(self):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
@@ -897,6 +944,174 @@ class TestGateStep:
|
||||
})
|
||||
assert any("on_reject" in e for e in errors)
|
||||
|
||||
def test_interactive_prompt_renders_show_file(self, tmp_path, monkeypatch, capsys):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
|
||||
review = tmp_path / "spec.md"
|
||||
review.write_text("LINE-ONE\nLINE-TWO\n", encoding="utf-8")
|
||||
|
||||
_force_gate_stdin(monkeypatch, tty=True)
|
||||
monkeypatch.setattr("builtins.input", lambda _prompt="": "1")
|
||||
|
||||
step = GateStep()
|
||||
config = {
|
||||
"id": "review",
|
||||
"message": "Review the spec.",
|
||||
"show_file": str(review),
|
||||
"options": ["approve", "reject"],
|
||||
}
|
||||
result = step.execute(config, StepContext())
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "LINE-ONE" in out and "LINE-TWO" in out
|
||||
assert str(review) in out
|
||||
assert result.status == StepStatus.COMPLETED
|
||||
assert result.output["choice"] == "approve"
|
||||
|
||||
def test_interactive_prompt_missing_show_file_does_not_crash(
|
||||
self, tmp_path, monkeypatch, capsys
|
||||
):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
|
||||
missing = tmp_path / "does-not-exist.md"
|
||||
|
||||
_force_gate_stdin(monkeypatch, tty=True)
|
||||
monkeypatch.setattr("builtins.input", lambda _prompt="": "1")
|
||||
|
||||
step = GateStep()
|
||||
config = {
|
||||
"id": "review",
|
||||
"message": "Review.",
|
||||
"show_file": str(missing),
|
||||
"options": ["approve", "reject"],
|
||||
}
|
||||
result = step.execute(config, StepContext())
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "could not read file" in out
|
||||
assert result.status == StepStatus.COMPLETED
|
||||
|
||||
def test_non_interactive_show_file_still_pauses_without_reading(
|
||||
self, tmp_path, monkeypatch
|
||||
):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
|
||||
review = tmp_path / "spec.md"
|
||||
review.write_text("CONTENT\n", encoding="utf-8")
|
||||
|
||||
# stdin defaults to non-TTY via the autouse fixture.
|
||||
# The non-interactive path must not read the file; hard-fail if it does.
|
||||
monkeypatch.setattr(
|
||||
GateStep,
|
||||
"_read_show_file",
|
||||
staticmethod(
|
||||
lambda _p: (_ for _ in ()).throw(
|
||||
AssertionError("show_file read on the non-interactive path")
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
step = GateStep()
|
||||
config = {
|
||||
"id": "review",
|
||||
"message": "Review.",
|
||||
"show_file": str(review),
|
||||
"options": ["approve", "reject"],
|
||||
}
|
||||
result = step.execute(config, StepContext())
|
||||
assert result.status == StepStatus.PAUSED
|
||||
assert result.output["show_file"] == str(review)
|
||||
|
||||
def test_read_show_file_empty(self, tmp_path):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
|
||||
empty = tmp_path / "empty.md"
|
||||
empty.write_text("", encoding="utf-8")
|
||||
assert GateStep._read_show_file(str(empty)) == ["(file is empty)"]
|
||||
|
||||
def test_read_show_file_truncates_large_file(self, tmp_path):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
|
||||
big = tmp_path / "big.md"
|
||||
big.write_text(
|
||||
"\n".join(f"line{i}" for i in range(GateStep.MAX_SHOW_FILE_LINES + 50)),
|
||||
encoding="utf-8",
|
||||
)
|
||||
rendered = GateStep._read_show_file(str(big))
|
||||
# MAX_SHOW_FILE_LINES content lines + one truncation notice line.
|
||||
assert len(rendered) == GateStep.MAX_SHOW_FILE_LINES + 1
|
||||
assert "truncated" in rendered[-1]
|
||||
|
||||
def test_read_show_file_invalid_path_does_not_raise(self):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
|
||||
# An embedded NUL byte makes the OS reject the path with ValueError
|
||||
# before any I/O; it must degrade to a notice, not crash the prompt.
|
||||
rendered = GateStep._read_show_file("bad\x00path.md")
|
||||
assert len(rendered) == 1
|
||||
assert rendered[0].startswith("(could not read file:")
|
||||
|
||||
def test_read_show_file_strips_control_chars(self, tmp_path):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
|
||||
# A file with ANSI/control bytes must not inject escapes into the
|
||||
# terminal; ESC and other C0 controls are stripped, tab is kept.
|
||||
f = tmp_path / "ansi.md"
|
||||
f.write_text("a\x1b[2Jb\tc\x07d\n", encoding="utf-8")
|
||||
rendered = GateStep._read_show_file(str(f))
|
||||
assert rendered == ["a[2Jb\tcd"]
|
||||
assert "\x1b" not in rendered[0] and "\x07" not in rendered[0]
|
||||
|
||||
def test_compose_prompt_sanitizes_show_file_path(self):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
|
||||
# The displayed path header (and the read-error notice it produces)
|
||||
# must not carry escapes even when the path string itself contains
|
||||
# control characters — ESC, LF, and C1 CSI (\x9b); the file is still
|
||||
# opened with the raw value.
|
||||
out = GateStep._compose_prompt("Review.", "ev\x1bil\x9b[2J\npath.md")
|
||||
assert "\x1b" not in out and "\x9b" not in out
|
||||
assert "evil[2Jpath.md:" in out
|
||||
|
||||
def test_interactive_non_string_message_renders(self, monkeypatch, capsys):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
|
||||
# A YAML numeric literal reaches the prompt as a non-string; it must
|
||||
# render rather than crash on the multi-line split.
|
||||
_force_gate_stdin(monkeypatch, tty=True)
|
||||
monkeypatch.setattr("builtins.input", lambda _prompt="": "1")
|
||||
|
||||
step = GateStep()
|
||||
config = {"id": "review", "message": 123, "options": ["approve", "reject"]}
|
||||
result = step.execute(config, StepContext())
|
||||
out = capsys.readouterr().out
|
||||
assert "123" in out
|
||||
assert result.status == StepStatus.COMPLETED
|
||||
|
||||
def test_templated_show_file_resolving_to_non_string_is_coerced(self):
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
|
||||
# A single-expression template can resolve to a non-string (e.g. a
|
||||
# number from a prior step); it must be coerced to str, not skipped.
|
||||
# stdin defaults to non-TTY via the autouse fixture, so the path
|
||||
# stays non-interactive (-> PAUSED) and cannot block on input.
|
||||
step = GateStep()
|
||||
ctx = StepContext(steps={"prev": {"output": {"ref": 123}}})
|
||||
config = {
|
||||
"id": "review",
|
||||
"message": "Review.",
|
||||
"show_file": "{{ steps.prev.output.ref }}",
|
||||
"options": ["approve", "reject"],
|
||||
}
|
||||
result = step.execute(config, ctx) # non-interactive -> PAUSED
|
||||
assert result.status == StepStatus.PAUSED
|
||||
assert result.output["show_file"] == "123"
|
||||
|
||||
|
||||
class TestIfThenStep:
|
||||
"""Test the if/then/else step type."""
|
||||
@@ -2622,19 +2837,11 @@ steps:
|
||||
from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
|
||||
from specify_cli.workflows.base import RunStatus
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
from specify_cli.workflows.steps import gate as gate_module
|
||||
|
||||
# Force the gate step into interactive mode and feed a "reject"
|
||||
# choice so the abort path actually runs in the test env
|
||||
# (default behaviour returns StepStatus.PAUSED when stdin is not a TTY).
|
||||
# Swap sys.stdin itself for a stub: setattr on the real
|
||||
# TextIOWrapper's `isatty` method is not assignable under some
|
||||
# runners (e.g. pytest with capture disabled).
|
||||
class _TTYStdin:
|
||||
def isatty(self) -> bool:
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(gate_module.sys, "stdin", _TTYStdin())
|
||||
# choice so the abort path actually runs in the test env (default
|
||||
# behaviour returns StepStatus.PAUSED when stdin is not a TTY).
|
||||
_force_gate_stdin(monkeypatch, tty=True)
|
||||
monkeypatch.setattr(
|
||||
GateStep, "_prompt", staticmethod(lambda _msg, _opts: "reject")
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user