Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,7 @@ generic_automation_mcp/
│ ├── _mcp_names.py # MCP prefix detection
│ ├── _onboarding.py # First-run detection + guided menu
│ ├── _prompts.py # Orchestrator prompt builder
│ ├── _timed_input.py # timed_prompt() and status_line() CLI primitives
│ ├── _update.py # run_update_command(): first-class upgrade path for `autoskillit update`
│ ├── _update_checks.py # Unified startup update check: version/hook/source-drift signals, branch-aware dismissal
│ ├── _workspace.py # Workspace clean helpers
Expand Down
9 changes: 5 additions & 4 deletions src/autoskillit/cli/_cook.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,12 +115,13 @@ def cook(*, resume: bool = False, session_id: str | None = None) -> None:
print()

from autoskillit.cli._ansi import permissions_warning
from autoskillit.cli._init_helpers import _require_interactive_stdin
from autoskillit.cli._timed_input import timed_prompt

print(permissions_warning())
_require_interactive_stdin("autoskillit cook")
confirm = input("\nLaunch session? [Enter/n]: ").strip().lower()
if confirm in ("n", "no"):
confirm = timed_prompt(
"\nLaunch session? [Enter/n]", default="", timeout=120, label="autoskillit cook"
)
if confirm.lower() in ("n", "no"):
return

from autoskillit.cli._onboarding import is_first_run, run_onboarding_menu
Expand Down
27 changes: 20 additions & 7 deletions src/autoskillit/cli/_init_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,21 +73,25 @@ def _require_interactive_stdin(command_name: str) -> None:


def _prompt_recipe_choice() -> str:
_require_interactive_stdin("autoskillit order")
from autoskillit.cli._timed_input import timed_prompt

available = list_recipes(Path.cwd()).items
if not available:
print("No recipes found. Run 'autoskillit recipes list' to check.")
raise SystemExit(1)
print("Available recipes:")
for i, r in enumerate(available, 1):
print(f" {i}. {r.name}")
return input("Recipe name: ").strip()
return timed_prompt("Recipe name:", default="", timeout=120, label="autoskillit order")


def _prompt_test_command() -> list[str]:
_require_interactive_stdin("autoskillit init")
from autoskillit.cli._timed_input import timed_prompt

default = "task test-all"
answer = input(f"Test command [{default}]: ").strip()
answer = timed_prompt(
f"Test command [{default}]:", default=default, timeout=120, label="autoskillit init"
)
return (answer if answer else default).split()


Expand Down Expand Up @@ -116,12 +120,19 @@ def _prompt_github_repo() -> str | None:
_B, _C, _D, _G, _Y, _R = _colors()
detected = _detect_github_repo()

from autoskillit.cli._timed_input import timed_prompt

if detected:
print(f"\n {_Y}GitHub repo{_R} {_G}{detected}{_R} {_D}(detected from git remote){_R}")
value = input(f" {_D}Press Enter to confirm, or type a different repo:{_R} ").strip()
value = timed_prompt(
"Press Enter to confirm, or type a different repo:",
default=detected or "",
timeout=120,
label="init",
)
else:
print(f"\n {_Y}GitHub repo{_R} {_D}owner/repo, URL, or blank to skip{_R}")
value = input(f" {_D}Repository:{_R} ").strip()
value = timed_prompt("Repository:", default="", timeout=120, label="init")

if not value:
return detected
Expand Down Expand Up @@ -251,9 +262,11 @@ def _check_secret_scanning(project_dir: Path) -> _ScanResult:
" Recommended: add gitleaks to .pre-commit-config.yaml\n"
" before proceeding.\n"
)
from autoskillit.cli._timed_input import timed_prompt

print(" To bypass, type exactly:\n")
print(f" {_D}{_SECRET_SCAN_BYPASS_PHRASE}{_R}\n")
response = input(" > ").strip()
response = timed_prompt(">", default="", timeout=120, label="secret-scan-bypass")
if response != _SECRET_SCAN_BYPASS_PHRASE:
print(f"\n {_B}Aborted.{_R} Phrase did not match.")
return _ScanResult(False)
Expand Down
21 changes: 8 additions & 13 deletions src/autoskillit/cli/_onboarding.py
Original file line number Diff line number Diff line change
Expand Up @@ -152,12 +152,13 @@ def run_onboarding_menu(project_dir: Path, *, color: bool = True) -> str | None:
_Y = "\x1b[33m" if color else ""
_R = "\x1b[0m" if color else ""

from autoskillit.cli._timed_input import timed_prompt

print(f"\n{_B}It looks like this is your first time using AutoSkillit in this project.{_R}")
try:
ans = input("Would you like help getting started? [Y/n]: ").strip().lower()
except EOFError:
ans = ""
if ans in ("n", "no"):
ans = timed_prompt(
"Would you like help getting started? [Y/n]", default="", timeout=120, label="onboarding"
)
if ans.lower() in ("n", "no"):
mark_onboarded(project_dir)
return None

Expand All @@ -175,10 +176,7 @@ def run_onboarding_menu(project_dir: Path, *, color: bool = True) -> str | None:
print(f" {_Y}D{_R} — {_C}Write a custom recipe{_R} (runs /autoskillit:write-recipe)")
print(f" {_Y}E{_R} — {_C}Skip{_R} (start a normal session)")

try:
choice = input(f"\n{_B}[A/B/C/D/E]: {_R}").strip().upper()
except EOFError:
choice = ""
choice = timed_prompt("\n[A/B/C/D/E]", default="E", timeout=120, label="onboarding").upper()

# Wait up to 5s for gather_intel to complete, then shut down cleanly
try:
Expand All @@ -193,10 +191,7 @@ def run_onboarding_menu(project_dir: Path, *, color: bool = True) -> str | None:
return "/autoskillit:setup-project"

if choice == "B":
try:
ref = input("Issue URL or number: ").strip()
except EOFError:
ref = ""
ref = timed_prompt("Issue URL or number:", default="", timeout=120, label="onboarding")
if ref:
return f"/autoskillit:prepare-issue {ref}"
return "/autoskillit:setup-project"
Expand Down
89 changes: 89 additions & 0 deletions src/autoskillit/cli/_timed_input.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
"""Timed prompt primitive with TTY guard, ANSI formatting, and timeout.

Every user-facing ``input()`` in the CLI layer must go through
``timed_prompt()`` rather than calling ``input()`` directly. A structural
test (``test_input_tty_contracts.py``) enforces this invariant.

``status_line()`` is the "pre-flight feedback" primitive — it prints a
single status message before any blocking I/O so the user always sees
output immediately.
"""

from __future__ import annotations

import select
import sys

from autoskillit.cli._ansi import supports_color
from autoskillit.cli._init_helpers import _require_interactive_stdin
Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[critical] arch: _timed_input.py imports _require_interactive_stdin from _init_helpers.py at module level, while _init_helpers.py lazily imports timed_prompt from _timed_input.py. This creates a latent circular dependency. As the foundational prompt primitive, _timed_input.py must not depend on peer CLI modules. Inline the TTY guard or extract _require_interactive_stdin to a shared utility.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Investigated — this is intentional. _init_helpers.py has NO top-level import of _timed_input. All four timed_prompt imports are strictly inside function bodies (lazy): _prompt_recipe_choice, _prompt_test_command, _prompt_github_repo, _check_secret_scanning. Python's import machinery resolves these lazily only when those functions are called, not when _timed_input is first loaded. No circular import cycle exists at module load time.



def timed_prompt(
text: str,
*,
default: str = "",
timeout: int = 30,
label: str = "prompt",
) -> str:
"""Display a formatted prompt with a bounded wait.

Composes three invariants into one call:

1. **TTY guard** — calls ``_require_interactive_stdin(label)``.
2. **ANSI formatting** — bold prompt, dim timeout hint.
3. **Timeout** — ``select.select`` bounds the wait; returns *default*
on expiry instead of blocking forever.

Parameters
----------
text
The prompt text shown to the user (plain text — ANSI is applied
automatically based on ``supports_color()``).
default
Value returned when the user does not respond within *timeout*
seconds, or on ``EOFError``.
timeout
Maximum seconds to wait for input. ``0`` disables the timeout
(waits indefinitely — use only for prompts the user explicitly
initiated).
label
Human-readable label for the TTY-guard error message.
"""
_require_interactive_stdin(label)
Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[warning] defense: TTY guard fires unconditionally, including when timeout=0 (documented as "user explicitly initiated"). A user-initiated prompt could legitimately come from piped input. Should timeout=0 callers bypass the guard?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Investigated — this is intentional. The docstring documents timeout=0 as 'use only for prompts the user explicitly initiated', presupposing a TTY. The module-level invariant is that ALL timed_prompt calls require an interactive terminal. Zero call sites in the codebase use timeout=0, making this a theoretical concern about an untested path. The TTY guard is coherent with the design contract.


color = supports_color()
_B = "\x1b[1m" if color else ""
_D = "\x1b[2m" if color else ""
_R = "\x1b[0m" if color else ""

hint = f" {_D}(auto-continues in {timeout}s){_R}" if timeout else ""
formatted = f"{_B}{text}{_R}{hint} "
print(formatted, end="", flush=True)

if timeout:
try:
ready, _, _ = select.select([sys.stdin], [], [], timeout)
except (OSError, TypeError, ValueError):
# stdin lacks fileno() or fileno() returns non-int
# (e.g. pytest capture, piped input) — fall back to blocking input().
ready = [sys.stdin]
if not ready:
print(f"\n{_D}(timed out, continuing...){_R}", flush=True)
return default

try:
return input("").strip()
except EOFError:
return default


def status_line(message: str) -> None:
"""Print a single-line status message with appropriate formatting.

This is the "pre-flight feedback" primitive: call it before any
blocking I/O so the terminal is never silent.
"""
color = supports_color()
_D = "\x1b[2m" if color else ""
_R = "\x1b[0m" if color else ""
print(f"{_D}{message}{_R}", flush=True)
23 changes: 17 additions & 6 deletions src/autoskillit/cli/_update_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@
_DISMISS_FILE = "update_check.json"
_STABLE_DISMISS_WINDOW = timedelta(days=7)
_DEV_DISMISS_WINDOW = timedelta(hours=12)
_SNOOZE_WINDOW = timedelta(hours=1)

# Disposable on-disk cache for GitHub API GETs. This is intentionally a plain
# JSON dict (not ``write_versioned_json``) — it is a transient performance
Expand Down Expand Up @@ -639,7 +638,7 @@ def run_update_checks(home: Path | None = None) -> None:
InstallType.UNKNOWN,
InstallType.LOCAL_PATH,
InstallType.LOCAL_EDITABLE,
):
) and not os.environ.get("AUTOSKILLIT_FORCE_UPDATE_CHECK"):
return

import autoskillit as _pkg
Expand All @@ -649,6 +648,11 @@ def run_update_checks(home: Path | None = None) -> None:
window = dismissal_window(info)
state = _read_dismiss_state(_home)

# Pre-flight feedback: ensure the terminal is never silent during network I/O
from autoskillit.cli._timed_input import status_line

status_line("Checking for updates...")

# Gather signals
raw_signals: list[Signal | None] = [
_binary_signal(info, _home, current),
Expand All @@ -673,12 +677,19 @@ def run_update_checks(home: Path | None = None) -> None:
return

# Consolidated prompt
from autoskillit.cli._timed_input import timed_prompt

_TIMEOUT_SENTINEL = "__timeout__"
Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[warning] cohesion: _TIMEOUT_SENTINEL string sentinel is asymmetric with all other timed_prompt call sites that treat the returned default as a valid answer. If timeout-vs-empty-input distinction is needed here, consider adding a return-type flag to timed_prompt itself rather than leaking sentinel detection into callers.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Investigated — this is intentional. The _TIMEOUT_SENTINEL implements a deliberate 3-way branch: (1) timeout → sentinel returned → early return WITHOUT writing dismissal, so the prompt reappears next invocation; (2) empty/y/yes → update runs; (3) n/no → dismissal record written. Simplifying to default='' would conflate timeout with user pressing Enter (the Y path), triggering an unwanted update. Commit 9b7dbb5 explicitly documents this contract.

bullet_lines = "\n".join(f" - {s.message}" for s in firing_signals)
prompt_text = f"\nAutoSkillit has updates available:\n{bullet_lines}\nUpdate now? [Y/n] "
print(prompt_text, end="", flush=True)
answer = input("").strip().lower()
prompt_text = f"\nAutoSkillit has updates available:\n{bullet_lines}\nUpdate now? [Y/n]"
answer = timed_prompt(prompt_text, default=_TIMEOUT_SENTINEL, timeout=30, label="update check")

# On timeout: proceed to app() without writing a dismissal record so the
# prompt reappears on the next invocation.
if answer == _TIMEOUT_SENTINEL:
return

if answer in ("", "y", "yes"):
if answer.lower() in ("", "y", "yes"):
_run_update_sequence(info, current, _home, state, _skip_env)
return

Expand Down
24 changes: 16 additions & 8 deletions src/autoskillit/cli/_workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,12 +74,16 @@ async def run_workspace_clean(
print()

if not force:
from autoskillit.cli._init_helpers import _require_interactive_stdin
from autoskillit.cli._timed_input import timed_prompt

_require_interactive_stdin("autoskillit workspace clean")
suffix = "ies" if len(stale) != 1 else "y"
answer = input(f"Remove {len(stale)} director{suffix}? [y/N]: ")
if answer.strip().lower() != "y":
answer = timed_prompt(
f"Remove {len(stale)} director{suffix}? [y/N]",
default="n",
timeout=120,
label="autoskillit workspace clean",
)
if answer.lower() != "y":
print("Aborted.")
return

Expand Down Expand Up @@ -148,12 +152,16 @@ def _safe_mtime(p: Path) -> float | None:
print()

if not force:
from autoskillit.cli._init_helpers import _require_interactive_stdin
from autoskillit.cli._timed_input import timed_prompt

_require_interactive_stdin("autoskillit workspace clean")
suffix = "ies" if len(stale_wts) != 1 else "y"
answer = input(f"Remove {len(stale_wts)} worktree director{suffix}? [y/N]: ")
if answer.strip().lower() != "y":
answer = timed_prompt(
f"Remove {len(stale_wts)} worktree director{suffix}? [y/N]",
default="n",
timeout=120,
label="autoskillit workspace clean",
)
if answer.lower() != "y":
print("Aborted.")
return

Expand Down
Loading
Loading