diff --git a/llm/cli.py b/llm/cli.py index 2e11e2c8..b98ae6ba 100644 --- a/llm/cli.py +++ b/llm/cli.py @@ -2,6 +2,7 @@ import click from click_default_group import DefaultGroup from dataclasses import asdict +import datetime import io import json import os @@ -1346,6 +1347,46 @@ def keys_set(name, value): path.write_text(json.dumps(current, indent=2) + "\n") +def localtime_enabled_in_userdir(): + """Check if logs-localtime file exists in user config dir.""" + return (user_dir() / "logs-localtime").exists() + + +def format_datetime(dt_str, use_localtime): + """ + Convert UTC datetime string to display format. + + Args: + dt_str: ISO format datetime string (assumed to be UTC) + use_localtime: If True, convert to local timezone; if False, keep as UTC + + Returns: + Formatted datetime string without microseconds + """ + if not dt_str: + return "" + + try: + # Parse the ISO string - it's in UTC + dt_utc = datetime.datetime.fromisoformat(dt_str) + + # Ensure it's marked as UTC + if dt_utc.tzinfo is None: + dt_utc = dt_utc.replace(tzinfo=datetime.timezone.utc) + + # Convert to local timezone if requested + if use_localtime: + dt_display = dt_utc.astimezone() + else: + dt_display = dt_utc + + # Return formatted without microseconds + return dt_display.strftime("%Y-%m-%dT%H:%M:%S") + except (ValueError, TypeError): + # Fall back to returning the original string, removing microseconds + return dt_str.split(".")[0] if dt_str else "" + + @cli.group( cls=DefaultGroup, default="list", @@ -1477,6 +1518,14 @@ def logs_turn_off(): default=None, help="Number of entries to show - defaults to 3, use 0 for all", ) +@click.option( + "--nc", + "--conv-count", + "conv_count", + type=int, + default=None, + help="Number of unique conversations to show - all responses from each", +) @click.option( "-p", "--path", @@ -1570,8 +1619,16 @@ def logs_turn_off(): is_flag=True, help="Expand fragments to show their content", ) +@click.option( + "-L", + "--localtime", + is_flag=True, + default=None, + help="Display datetimes in localtime tz (default: utc), also check logs-localtime file in userdir", +) def logs_list( count, + conv_count, path, database, model, @@ -1598,8 +1655,15 @@ def logs_list( id_gte, json_output, expand, + localtime, ): "Show logged prompts and their responses" + # Resolve localtime flag: explicit flag takes precedence, else check config file + if localtime is None: + use_localtime = localtime_enabled_in_userdir() + else: + use_localtime = localtime + if database and not path: path = database path = pathlib.Path(path or logs_db_path()) @@ -1624,7 +1688,14 @@ def logs_list( ) raise click.ClickException("Cannot use --short and {} together".format(invalid)) - if response and not current_conversation and not conversation_id: + # Validate mutually exclusive options + if count is not None and conv_count is not None: + raise click.ClickException("Cannot use both -n/--count and --nc/--conv-count together") + + if conv_count is not None and conversation_id: + raise click.ClickException("Cannot use both --nc/--conv-count and --conversation/--cid together") + + if response and not current_conversation and not conversation_id and conv_count is None: current_conversation = True if current_conversation: @@ -1638,8 +1709,8 @@ def logs_list( # No conversations yet raise click.ClickException("No conversations found") - # For --conversation set limit 0, if not explicitly set - if count is None: + # Set defaults for count and conv_count + if count is None and conv_count is None: if conversation_id: count = 0 else: @@ -1679,6 +1750,32 @@ def logs_list( "id_gt": id_gt, "id_gte": id_gte, } + + # Handle --nc (conv_count) option + if conv_count is not None and conv_count > 0: + # Use a CTE to get the last N conversations, then join to get all their responses + cte = f""" +with recent_conversations as ( + select + conversation_id, + max(id) as last_response_id + from responses + where conversation_id is not null + group by conversation_id + order by last_response_id desc + limit {conv_count} +) +select +{LOGS_COLUMNS} +from + responses +left join schemas on responses.schema_id = schemas.id +left join conversations on responses.conversation_id = conversations.id +join recent_conversations rc on responses.conversation_id = rc.conversation_id{{extra_where}} +order by {{order_by}}""" + sql_format["cte"] = cte + # We need a different base SQL that uses the CTE + sql = cte if model_id: where_bits.append("responses.model = :model") if conversation_id: @@ -1956,6 +2053,9 @@ def logs_list( {k: v for k, v in attachment.items() if k != "response_id"} for attachment in attachments_by_id.get(row["id"], []) ] + # Add datetime field in appropriate timezone + if use_localtime and "datetime_utc" in row: + row["datetime"] = format_datetime(row["datetime_utc"], True) output = json.dumps(list(rows), indent=2) elif extract or extract_last: # Extract and return first code block @@ -2005,7 +2105,7 @@ def _display_fragments(fragments, title): attachments = attachments_by_id.get(row["id"]) obj = { "model": row["model"], - "datetime": row["datetime_utc"].split(".")[0], + "datetime": format_datetime(row["datetime_utc"], use_localtime), "conversation": cid, } if row["tool_calls"]: @@ -2051,7 +2151,7 @@ def _display_fragments(fragments, title): # Not short, output Markdown click.echo( "# {}{}\n{}".format( - row["datetime_utc"].split(".")[0], + format_datetime(row["datetime_utc"], use_localtime), ( " conversation: {} id: {}".format( row["conversation_id"], row["id"] diff --git a/tests/test_llm_logs.py b/tests/test_llm_logs.py index ef582feb..8f0342b4 100644 --- a/tests/test_llm_logs.py +++ b/tests/test_llm_logs.py @@ -89,6 +89,124 @@ def schema_log_path(user_path): id_re = re.compile(r"id: \w+") +@pytest.mark.parametrize( + "nc", + (None, 0, 1, 2), +) +def test_logs_conv_count(nc): + """Test that --nc returns all responses from N unique conversations""" + from llm.utils import monotonic_ulid + + runner = CliRunner() + with runner.isolated_filesystem(): + # Create a database with multiple conversations + log_path = "test_logs.db" + db = sqlite_utils.Database(log_path) + migrate(db) + start = datetime.datetime.now(datetime.timezone.utc) + + # Create 3 conversations with varying numbers of responses + # Conversation 1: 2 responses + # Conversation 2: 3 responses + # Conversation 3: 1 response + response_count = 0 + for conv_id in ["conv1", "conv2", "conv3"]: + responses_in_conv = 2 if conv_id == "conv1" else (3 if conv_id == "conv2" else 1) + for i in range(responses_in_conv): + db["responses"].insert( + { + "id": str(monotonic_ulid()).lower(), + "system": f"system-{conv_id}-{i}", + "prompt": f"prompt-{conv_id}-{i}", + "response": f"response-{conv_id}-{i}", + "model": "davinci", + "datetime_utc": (start + datetime.timedelta(seconds=response_count)).isoformat(), + "conversation_id": conv_id, + "input_tokens": 2, + "output_tokens": 5, + } + ) + response_count += 1 + + # Test without --nc (default should be 3 responses) + if nc is None: + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--json"], catch_exceptions=False) + assert result.exit_code == 0 + logs = json.loads(result.output) + assert len(logs) == 3 + return + + # Test with --nc 0 (should be same as no limit) + if nc == 0: + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--nc", "0", "--json"], catch_exceptions=False) + assert result.exit_code == 0 + logs = json.loads(result.output) + assert len(logs) == 6 # All 6 responses + return + + # Test with --nc 1 (should get all responses from latest conversation only) + if nc == 1: + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--nc", "1", "--json"], catch_exceptions=False) + assert result.exit_code == 0 + logs = json.loads(result.output) + # conv3 is the latest (most recent response), has 1 response + assert len(logs) == 1 + assert logs[0]["conversation_id"] == "conv3" + return + + # Test with --nc 2 (should get all responses from 2 latest conversations) + if nc == 2: + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--nc", "2", "--json"], catch_exceptions=False) + assert result.exit_code == 0 + logs = json.loads(result.output) + # conv2 (3 responses) and conv3 (1 response) are the 2 latest = 4 responses total + assert len(logs) == 4 + conv_ids = {log["conversation_id"] for log in logs} + assert conv_ids == {"conv2", "conv3"} + return + + +def test_logs_conv_count_mutually_exclusive(): + """Test that --nc and -n cannot be used together""" + runner = CliRunner() + with runner.isolated_filesystem(): + log_path = "test_logs.db" + db = sqlite_utils.Database(log_path) + migrate(db) + db["responses"].insert({ + "id": "test1", + "prompt": "test", + "response": "test", + "model": "test", + }) + + # Test --nc and -n together + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--nc", "1", "-n", "5"], catch_exceptions=False) + assert result.exit_code != 0 + assert "Cannot use both" in result.output + + +def test_logs_conv_count_with_conversation_id(): + """Test that --nc and --conversation cannot be used together""" + runner = CliRunner() + with runner.isolated_filesystem(): + log_path = "test_logs.db" + db = sqlite_utils.Database(log_path) + migrate(db) + db["responses"].insert({ + "id": "test1", + "prompt": "test", + "response": "test", + "model": "test", + "conversation_id": "conv1", + }) + + # Test --nc and --conversation together + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--nc", "1", "--conversation", "conv1"], catch_exceptions=False) + assert result.exit_code != 0 + assert "Cannot use both" in result.output + + @pytest.mark.parametrize("usage", (False, True)) def test_logs_text(log_path, usage): runner = CliRunner() @@ -976,3 +1094,158 @@ def test_logs_resolved_model(logs_db, mock_model, async_mock_model, async_): # And the rendered logs result3 = runner.invoke(cli, ["logs"]) assert "Model: **mock** (resolved: **resolved-mock**)" in result3.output + + +def test_logs_localtime_flag_markdown(log_path, monkeypatch): + """Test that -L/--localtime flag displays datetime in local timezone for markdown output""" + import os + from datetime import datetime, timezone + + runner = CliRunner() + + # Test with -L flag - get markdown output with converted time + result = runner.invoke(cli, ["logs", "-p", str(log_path), "-L", "-n", "1"], catch_exceptions=False) + assert result.exit_code == 0 + output = result.output + + # Should contain a header with datetime + lines = output.split('\n') + assert len(lines) > 0 + # First line should start with # and contain datetime + assert lines[0].startswith("# ") + + # The datetime should be in ISO format without microseconds + assert datetime_re.search(output), "Datetime format should match YYYY-MM-DDTHH:MM:SS" + + +def test_logs_localtime_flag_json(log_path): + """Test that -L/--localtime flag adds 'datetime' field in JSON output""" + runner = CliRunner() + + # Test with -L flag and JSON output + result = runner.invoke(cli, ["logs", "-p", str(log_path), "-L", "--json", "-n", "1"], catch_exceptions=False) + assert result.exit_code == 0 + logs = json.loads(result.output) + + assert len(logs) == 1 + # Should have both datetime_utc and datetime fields + assert "datetime_utc" in logs[0] + assert "datetime" in logs[0] + # datetime should not have microseconds + assert "." not in logs[0]["datetime"] + + +def test_logs_default_utc_json(log_path): + """Test that without -L flag, JSON output has datetime_utc, no datetime field added""" + runner = CliRunner() + + # Test without -L flag + result = runner.invoke(cli, ["logs", "-p", str(log_path), "--json", "-n", "1"], catch_exceptions=False) + assert result.exit_code == 0 + logs = json.loads(result.output) + + assert len(logs) == 1 + assert "datetime_utc" in logs[0] + # When localtime is disabled, datetime field is not added + # (datetime field is only added when use_localtime=True) + + +def test_logs_short_yaml_with_localtime(log_path): + """Test that -L flag works with -s/--short YAML output""" + runner = CliRunner() + + result = runner.invoke(cli, ["logs", "-p", str(log_path), "-s", "-L", "-n", "1"], catch_exceptions=False) + assert result.exit_code == 0 + output = result.output + + # Should be valid YAML + parsed = yaml.safe_load(output) + assert isinstance(parsed, list) + assert len(parsed) == 1 + assert "datetime" in parsed[0] + # Datetime should not have microseconds + assert "." not in parsed[0]["datetime"] + + +def test_logs_localtime_config_file(user_path, monkeypatch): + """Test that logs-localtime config file enables localtime by default""" + from llm import user_dir + + # Create a logs database + log_path = str(user_path / "logs.db") + db = sqlite_utils.Database(log_path) + migrate(db) + start = datetime.datetime.now(datetime.timezone.utc) + db["responses"].insert_all( + { + "id": str(monotonic_ulid()).lower(), + "system": "system", + "prompt": "prompt", + "response": "response", + "model": "davinci", + "datetime_utc": (start + datetime.timedelta(seconds=i)).isoformat(), + "conversation_id": "abc123", + } + for i in range(2) + ) + + # Create logs-localtime file in config dir using pathlib + config_dir = pathlib.Path(str(user_path)) + localtime_file = config_dir / "logs-localtime" + localtime_file.write_text("") + + # Mock user_dir to return our test directory + monkeypatch.setattr("llm.cli.user_dir", lambda: config_dir) + + runner = CliRunner() + result = runner.invoke(cli, ["logs", "-p", str(log_path), "-n", "1"], catch_exceptions=False) + assert result.exit_code == 0 + + # Should have a datetime in the output + assert datetime_re.search(result.output), "Should contain datetime in output" + + +def test_logs_format_datetime_for_display(): + """Test the format_datetime helper function directly""" + from llm.cli import format_datetime + + # Test UTC format (no microseconds, no timezone info) + utc_str = "2023-08-17T20:53:58.123456" + + # Format as UTC (should be same but without microseconds) + formatted_utc = format_datetime(utc_str, use_localtime=False) + assert formatted_utc == "2023-08-17T20:53:58" + + # Format as localtime (could be different depending on system timezone) + formatted_local = format_datetime(utc_str, use_localtime=True) + # Should be in ISO format without microseconds + assert "." not in formatted_local + assert "T" in formatted_local + assert len(formatted_local) == 19 # YYYY-MM-DDTHH:MM:SS + + # Test with None or empty string + assert format_datetime(None, False) == "" + assert format_datetime("", False) == "" + + +def test_logs_localtime_comparison_timezone_aware(): + """Test that UTC and localtime conversions are correct for a known datetime""" + from llm.cli import format_datetime + + # Use a known UTC time + utc_iso = "2023-01-15T12:00:00.000000" + + utc_formatted = format_datetime(utc_iso, use_localtime=False) + local_formatted = format_datetime(utc_iso, use_localtime=True) + + # Both should be valid ISO datetime strings + assert utc_formatted == "2023-01-15T12:00:00" + assert "." not in local_formatted + + # Parse them back to verify they work + dt_utc = datetime.datetime.fromisoformat(utc_formatted) + dt_local = datetime.datetime.fromisoformat(local_formatted) + + # Both should parse successfully + assert dt_utc.year == 2023 + assert dt_local.year == 2023