update

src/tai/cli.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 from time import perf_counter
 from typing import Annotated
 
@@ -318,7 +319,14 @@ async def _interactive_loop(
 
     while True:
         try:
-            command = console.input("\n[bold cyan]tai[/bold cyan][dim] >[/dim] ").strip()
+            if _stdin_is_tty():
+                command = console.input("\n[bold cyan]tai[/bold cyan][dim] >[/dim] ").strip()
+            else:
+                line = sys.stdin.readline()  # non-TTY / piped mode
+                if not line:
+                    return
+                command = line.strip()
+                console.print(f"\n[bold cyan]tai[/bold cyan][dim] >[/dim] {command}")
         except (EOFError, KeyboardInterrupt):
             console.print("\n[yellow]Exiting interactive mode.[/yellow]")
             if logger is not None:
@@ -570,6 +578,10 @@ def _run_analysis(
         raise typer.Exit(code=1) from exc
 
 
+def _stdin_is_tty() -> bool:
+    return sys.stdin.isatty()
+
+
 def _estimate_tokens(text: str) -> int:
     """Rough token estimate for metrics and tuning; assumes ~4 chars/token."""
     return max(1, len(text) // 4)

tests/test_cli.py

@@ -166,6 +166,7 @@ def test_interactive_collect_then_quit(monkeypatch) -> None:  # type: ignore[no-
 
     monkeypatch.setattr("tai.cli.collect_from_plan", fake_collect_from_plan)
     monkeypatch.setattr("tai.cli.console.input", lambda _prompt: next(commands))
+    monkeypatch.setattr("tai.cli._stdin_is_tty", lambda: True)
 
     runner = CliRunner()
     result = runner.invoke(
@@ -183,7 +184,7 @@ def test_interactive_collect_then_quit(monkeypatch) -> None:  # type: ignore[no-
 
     assert result.exit_code == 0
     assert "ask questions directly" in result.stdout.lower()
-    assert "Collection complete" in result.stdout
+    assert "collection complete" in result.stdout.lower()
     assert "Bye." in result.stdout
 
 
@@ -213,6 +214,7 @@ def test_interactive_unknown_command_prints_hint(monkeypatch) -> None:  # type:
         lambda *_args, **_kwargs: iter(["Check logs."]),
     )
     monkeypatch.setattr("tai.cli.console.input", lambda _prompt: next(commands))
+    monkeypatch.setattr("tai.cli._stdin_is_tty", lambda: True)
 
     runner = CliRunner()
     result = runner.invoke(
@@ -257,6 +259,7 @@ def test_interactive_prints_rag_fallback_notice_on_index_failure(monkeypatch) ->
     monkeypatch.setattr("tai.cli._try_embed_report", lambda *_args: (None, "embed failed", 1.0))
     monkeypatch.setattr("tai.cli.AIClient.stream", lambda *_args, **_kwargs: iter(["Check logs."]))
     monkeypatch.setattr("tai.cli.console.input", lambda _prompt: next(commands))
+    monkeypatch.setattr("tai.cli._stdin_is_tty", lambda: True)
 
     runner = CliRunner()
     result = runner.invoke(
@@ -309,6 +312,7 @@ def test_interactive_rag_debug_prints_retrieval_scores(monkeypatch) -> None:  #
     monkeypatch.setattr("tai.cli.AIClient.embed", lambda *_args, **_kwargs: [1.0, 0.0])
     monkeypatch.setattr("tai.cli.AIClient.stream", lambda *_args, **_kwargs: iter(["Check logs."]))
     monkeypatch.setattr("tai.cli.console.input", lambda _prompt: next(commands))
+    monkeypatch.setattr("tai.cli._stdin_is_tty", lambda: True)
 
     runner = CliRunner()
     result = runner.invoke(