perplexity/Timmy-time-dashboard
Archived
1
0
Fork 0
forked from Rockachopa/Timmy-time-dashboard
Code Issues Pull Requests 1 Activity

feat: REPL mode, stdin support, multi-word fix for CLI (#26)

Browse Source
This commit is contained in:
Kimi Agent
2026-03-14 19:25:10 -04:00
parent 9134ce2f71
commit 65e5e7786f
2 changed files with 251 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
src/timmy/cli.py
View File

@@ -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,

174
tests/timmy/test_cli.py
View File

@@ -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
# ---------------------------------------------------------------------------