diff --git a/src/timmy/cli.py b/src/timmy/cli.py index b7feb380..db975273 100644 --- a/src/timmy/cli.py +++ b/src/timmy/cli.py @@ -1,3 +1,4 @@ +import asyncio import logging import subprocess import sys @@ -143,7 +144,9 @@ def think( @app.command() def chat( - message: list[str] = typer.Argument(..., help="Message to send to Timmy"), + message: list[str] = typer.Argument( + ..., help="Message to send (multiple words are joined automatically)" + ), backend: str | None = _BACKEND_OPTION, model_size: str | None = _MODEL_SIZE_OPTION, new_session: bool = typer.Option( @@ -172,11 +175,26 @@ def chat( Use --autonomous for non-interactive contexts (scripts, dev loops). Tool calls are checked against config/allowlist.yaml — allowlisted operations execute automatically, everything else is safely rejected. - """ - full_message = " ".join(message) + Read from stdin by passing "-" as the message or piping input. + """ import uuid + # Join multiple arguments into a single message string + message_str = " ".join(message) + + # Handle stdin input if "-" is passed or stdin is not a tty + if message_str == "-" or not _is_interactive(): + try: + stdin_content = sys.stdin.read().strip() + except (KeyboardInterrupt, EOFError): + stdin_content = "" + if stdin_content: + message_str = stdin_content + elif message_str == "-": + typer.echo("No input provided via stdin.", err=True) + raise typer.Exit(1) + if session_id is not None: pass # use the provided value elif new_session: @@ -186,7 +204,7 @@ def chat( timmy = create_timmy(backend=backend, model_size=model_size, session_id=session_id) # Use agent.run() so we can intercept paused runs for tool confirmation. - run_output = timmy.run(full_message, stream=False, session_id=session_id) + run_output = timmy.run(message_str, stream=False, session_id=session_id) # Handle paused runs — dangerous tools need user approval run_output = _handle_tool_confirmation(timmy, run_output, session_id, autonomous=autonomous) @@ -199,6 +217,61 @@ def chat( typer.echo(_clean_response(content)) +@app.command() +def repl( + backend: str | None = _BACKEND_OPTION, + model_size: str | None = _MODEL_SIZE_OPTION, + session_id: str | None = typer.Option( + None, + "--session-id", + help="Use a specific session ID for this conversation", + ), +): + """Start an interactive REPL session with Timmy. + + Keeps the agent warm between messages. Conversation history is persisted + across invocations. Use Ctrl+C or Ctrl+D to exit gracefully. + """ + from timmy.session import chat + + if session_id is None: + session_id = _CLI_SESSION_ID + + typer.echo(typer.style("Timmy REPL", bold=True)) + typer.echo("Type your messages below. Use Ctrl+C or Ctrl+D to exit.") + typer.echo() + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + while True: + try: + user_input = input("> ") + except (KeyboardInterrupt, EOFError): + typer.echo() + typer.echo("Goodbye!") + break + + user_input = user_input.strip() + if not user_input: + continue + + if user_input.lower() in ("exit", "quit", "q"): + typer.echo("Goodbye!") + break + + try: + response = loop.run_until_complete(chat(user_input, session_id=session_id)) + if response: + typer.echo(response) + typer.echo() + except Exception as exc: + typer.echo(f"Error: {exc}", err=True) + finally: + loop.close() + + @app.command() def status( backend: str | None = _BACKEND_OPTION, diff --git a/tests/timmy/test_cli.py b/tests/timmy/test_cli.py index e6e1679e..f0690483 100644 --- a/tests/timmy/test_cli.py +++ b/tests/timmy/test_cli.py @@ -141,6 +141,180 @@ def test_chat_cleans_response(): assert "The answer is 2." in result.output +# --------------------------------------------------------------------------- +# chat command — multi-word messages +# --------------------------------------------------------------------------- + + +def test_chat_multiword_without_quotes(): + """chat should accept multiple arguments without requiring quotes.""" + mock_run_output = MagicMock() + mock_run_output.content = "Response" + mock_run_output.status = "COMPLETED" + mock_run_output.active_requirements = [] + + mock_timmy = MagicMock() + mock_timmy.run.return_value = mock_run_output + + with patch("timmy.cli.create_timmy", return_value=mock_timmy): + result = runner.invoke(app, ["chat", "hello", "world", "how", "are", "you"]) + + mock_timmy.run.assert_called_once_with( + "hello world how are you", stream=False, session_id=_CLI_SESSION_ID + ) + assert result.exit_code == 0 + + +# --------------------------------------------------------------------------- +# chat command — stdin support +# --------------------------------------------------------------------------- + + +def test_chat_stdin_with_dash(): + """chat - should read message from stdin when "-" is passed.""" + mock_run_output = MagicMock() + mock_run_output.content = "Stdin response" + mock_run_output.status = "COMPLETED" + mock_run_output.active_requirements = [] + + mock_timmy = MagicMock() + mock_timmy.run.return_value = mock_run_output + + with patch("timmy.cli.create_timmy", return_value=mock_timmy): + result = runner.invoke(app, ["chat", "-"], input="Hello from stdin") + + mock_timmy.run.assert_called_once_with( + "Hello from stdin", stream=False, session_id=_CLI_SESSION_ID + ) + assert result.exit_code == 0 + assert "Stdin response" in result.output + + +def test_chat_stdin_piped(): + """chat should read from stdin when piped (non-tty).""" + mock_run_output = MagicMock() + mock_run_output.content = "Piped response" + mock_run_output.status = "COMPLETED" + mock_run_output.active_requirements = [] + + mock_timmy = MagicMock() + mock_timmy.run.return_value = mock_run_output + + with ( + patch("timmy.cli.create_timmy", return_value=mock_timmy), + patch("timmy.cli._is_interactive", return_value=False), + ): + result = runner.invoke(app, ["chat", "ignored"], input="Piped message") + + mock_timmy.run.assert_called_once_with( + "Piped message", stream=False, session_id=_CLI_SESSION_ID + ) + assert result.exit_code == 0 + + +def test_chat_stdin_empty_exits_with_error(): + """chat - should exit with error when stdin is empty.""" + with patch("timmy.cli._is_interactive", return_value=False): + result = runner.invoke(app, ["chat", "-"], input="") + + assert result.exit_code == 1 + assert "No input provided" in result.output or "No input provided" in str(result.output) + + +# --------------------------------------------------------------------------- +# repl command +# --------------------------------------------------------------------------- + + +def test_repl_exits_on_ctrl_c(): + """repl should exit gracefully on Ctrl+C (KeyboardInterrupt).""" + with patch("builtins.input", side_effect=KeyboardInterrupt): + result = runner.invoke(app, ["repl"]) + + assert result.exit_code == 0 + assert "Goodbye" in result.output + + +def test_repl_exits_on_ctrl_d(): + """repl should exit gracefully on Ctrl+D (EOFError).""" + with patch("builtins.input", side_effect=EOFError): + result = runner.invoke(app, ["repl"]) + + assert result.exit_code == 0 + assert "Goodbye" in result.output + + +def test_repl_exits_on_quit_command(): + """repl should exit when user types 'quit'.""" + with patch("builtins.input", return_value="quit"): + result = runner.invoke(app, ["repl"]) + + assert result.exit_code == 0 + assert "Goodbye" in result.output + + +def test_repl_exits_on_exit_command(): + """repl should exit when user types 'exit'.""" + with patch("builtins.input", return_value="exit"): + result = runner.invoke(app, ["repl"]) + + assert result.exit_code == 0 + assert "Goodbye" in result.output + + +def test_repl_sends_message_to_chat(): + """repl should send user input to session.chat().""" + with ( + patch("builtins.input", side_effect=["Hello Timmy", "exit"]), + patch("timmy.session.chat") as mock_chat, + ): + mock_chat.return_value = "Hi there!" + result = runner.invoke(app, ["repl"]) + + mock_chat.assert_called() + assert result.exit_code == 0 + + +def test_repl_skips_empty_input(): + """repl should skip empty input lines.""" + with ( + patch("builtins.input", side_effect=["", "", "hello", "exit"]), + patch("timmy.session.chat") as mock_chat, + ): + mock_chat.return_value = "Response" + result = runner.invoke(app, ["repl"]) + + # chat should only be called once (for "hello"), empty lines are skipped, exit breaks + assert mock_chat.call_count == 1 + + +def test_repl_uses_custom_session_id(): + """repl --session-id should use the provided session ID.""" + with ( + patch("builtins.input", side_effect=["hello", "exit"]), + patch("timmy.session.chat") as mock_chat, + ): + mock_chat.return_value = "Response" + runner.invoke(app, ["repl", "--session-id", "custom-session"]) + + # Check that chat was called with the custom session_id + calls = mock_chat.call_args_list + assert any(call.kwargs.get("session_id") == "custom-session" for call in calls) + + +def test_repl_handles_chat_errors(): + """repl should handle errors from chat() gracefully.""" + with ( + patch("builtins.input", side_effect=["hello", "exit"]), + patch("timmy.session.chat") as mock_chat, + ): + mock_chat.side_effect = Exception("Chat error") + result = runner.invoke(app, ["repl"]) + + assert result.exit_code == 0 + assert "Error" in result.output + + # --------------------------------------------------------------------------- # Tool confirmation gate # ---------------------------------------------------------------------------