i forgot too commit
All checks were successful
Enterprise AI Code Review / ai-review (pull_request) Successful in 38s

This commit is contained in:
2026-02-01 15:57:45 +01:00
parent 9a334e80be
commit d957120eb3
25 changed files with 5047 additions and 23 deletions

137
cli/README.md Normal file
View File

@@ -0,0 +1,137 @@
# Loyal Companion CLI
A quiet, terminal-based interface for conversations with Loyal Companion.
## Installation
```bash
# Install dependencies
pip install typer httpx rich
# Make CLI executable
chmod +x lc
```
## Quick Start
```bash
# Start the web server (required)
python3 run_web.py
# Start a conversation
./lc talk
```
## Commands
### `lc talk`
Start or resume a conversation.
```bash
lc talk # Resume default session
lc talk --new # Start fresh default session
lc talk -s work # Resume 'work' session
lc talk -s personal --new # Start fresh 'personal' session
```
### `lc history`
Show conversation history.
```bash
lc history # Show default session history
lc history -s work # Show 'work' session history
lc history -n 10 # Show last 10 messages
```
### `lc sessions`
List or manage sessions.
```bash
lc sessions # List all sessions
lc sessions -d work # Delete 'work' session
```
### `lc config-cmd`
Manage configuration.
```bash
lc config-cmd --show # Show current config
lc config-cmd --api-url http://localhost:8080 # Set API URL
lc config-cmd --email user@example.com # Set email
lc config-cmd --reset # Reset to defaults
```
### `lc auth`
Manage authentication.
```bash
lc auth # Show auth status
lc auth --logout # Clear stored token
```
### `lc health`
Check API health.
```bash
lc health # Check if API is reachable
```
## Configuration
Configuration is stored in `~/.lc/config.json`:
```json
{
"api_url": "http://127.0.0.1:8080",
"auth_token": "web:user@example.com",
"email": "user@example.com",
"show_mood": true,
"show_relationship": false
}
```
## Sessions
Sessions are stored in `~/.lc/sessions.json`:
- Multiple named sessions supported
- Sessions persist across CLI invocations
- Auto-save on exit
## Design Philosophy
The CLI embodies the "empty table at closing time" philosophy:
- **Quiet:** No spinners, no ASCII art, minimal output
- **Intentional:** Explicit commands, named sessions
- **Focused:** Text-first, no distractions
## Architecture
The CLI is a thin HTTP client that communicates with the Web API:
```
CLI (lc) → HTTP → Web API → ConversationGateway → Living AI Core
```
- Platform: `CLI`
- Intimacy: `HIGH` (via Web API)
- Transport: HTTP/REST
## Development
Run component tests:
```bash
python3 test_cli.py
```
## Dependencies
- `typer>=0.9.0` - CLI framework
- `httpx>=0.26.0` - HTTP client
- `rich>=13.7.0` - Terminal formatting (optional)
## Documentation
See [Phase 4 Complete](../docs/implementation/phase-4-complete.md) for full documentation.

6
cli/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""Loyal Companion CLI client.
A quiet, terminal-based interface for conversations with Loyal Companion.
"""
__version__ = "1.0.0"

185
cli/client.py Normal file
View File

@@ -0,0 +1,185 @@
"""HTTP client for Loyal Companion Web API."""
from typing import Any
import httpx
class APIError(Exception):
"""API request error."""
pass
class LoyalCompanionClient:
"""HTTP client for Loyal Companion API."""
def __init__(self, base_url: str, auth_token: str | None = None):
"""Initialize client.
Args:
base_url: API base URL
auth_token: Optional authentication token
"""
self.base_url = base_url.rstrip("/")
self.auth_token = auth_token
self.client = httpx.Client(timeout=60.0)
def _get_headers(self) -> dict[str, str]:
"""Get request headers.
Returns:
dict: Request headers
"""
headers = {"Content-Type": "application/json"}
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
return headers
def request_token(self, email: str) -> dict[str, Any]:
"""Request an authentication token.
Args:
email: User email
Returns:
dict: Token response
Raises:
APIError: If request fails
"""
try:
response = self.client.post(
f"{self.base_url}/api/auth/token",
json={"email": email},
headers={"Content-Type": "application/json"},
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise APIError(f"Failed to request token: {e}")
def send_message(self, session_id: str, message: str) -> dict[str, Any]:
"""Send a chat message.
Args:
session_id: Session identifier
message: User message
Returns:
dict: Chat response with AI's reply and metadata
Raises:
APIError: If request fails
"""
try:
response = self.client.post(
f"{self.base_url}/api/chat",
json={"session_id": session_id, "message": message},
headers=self._get_headers(),
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
if hasattr(e, "response") and e.response is not None:
try:
error_detail = e.response.json().get("detail", str(e))
except Exception:
error_detail = str(e)
raise APIError(f"Chat request failed: {error_detail}")
raise APIError(f"Chat request failed: {e}")
def get_history(self, session_id: str, limit: int = 50) -> dict[str, Any]:
"""Get conversation history.
Args:
session_id: Session identifier
limit: Maximum number of messages
Returns:
dict: History response
Raises:
APIError: If request fails
"""
try:
response = self.client.get(
f"{self.base_url}/api/sessions/{session_id}/history",
params={"limit": limit},
headers=self._get_headers(),
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise APIError(f"Failed to get history: {e}")
def list_sessions(self) -> list[dict[str, Any]]:
"""List all user sessions.
Returns:
list: List of sessions
Raises:
APIError: If request fails
"""
try:
response = self.client.get(
f"{self.base_url}/api/sessions",
headers=self._get_headers(),
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise APIError(f"Failed to list sessions: {e}")
def delete_session(self, session_id: str) -> dict[str, Any]:
"""Delete a session.
Args:
session_id: Session identifier
Returns:
dict: Deletion response
Raises:
APIError: If request fails
"""
try:
response = self.client.delete(
f"{self.base_url}/api/sessions/{session_id}",
headers=self._get_headers(),
)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise APIError(f"Failed to delete session: {e}")
def health_check(self) -> dict[str, Any]:
"""Check API health.
Returns:
dict: Health status
Raises:
APIError: If request fails
"""
try:
response = self.client.get(f"{self.base_url}/api/health")
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise APIError(f"Health check failed: {e}")
def close(self) -> None:
"""Close the HTTP client."""
self.client.close()
def __enter__(self):
"""Context manager entry."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Context manager exit."""
self.close()

101
cli/config.py Normal file
View File

@@ -0,0 +1,101 @@
"""Configuration management for CLI."""
import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
@dataclass
class CLIConfig:
"""CLI configuration."""
# API settings
api_url: str = "http://127.0.0.1:8080"
auth_token: str | None = None
# User settings
email: str | None = None
allow_emoji: bool = False
# Session settings
default_session: str = "default"
auto_save: bool = True
# Display settings
show_mood: bool = True
show_relationship: bool = False
show_facts: bool = False
show_timestamps: bool = False
# Paths
config_dir: Path = field(default_factory=lambda: Path.home() / ".lc")
sessions_file: Path = field(init=False)
config_file: Path = field(init=False)
def __post_init__(self):
"""Initialize computed fields."""
self.sessions_file = self.config_dir / "sessions.json"
self.config_file = self.config_dir / "config.json"
@classmethod
def load(cls) -> "CLIConfig":
"""Load configuration from file.
Returns:
CLIConfig: Loaded configuration
"""
config = cls()
config.config_dir.mkdir(parents=True, exist_ok=True)
if config.config_file.exists():
try:
with open(config.config_file, "r") as f:
data = json.load(f)
# Update fields from loaded data
for key, value in data.items():
if hasattr(config, key):
setattr(config, key, value)
except (json.JSONDecodeError, IOError) as e:
print(f"Warning: Could not load config: {e}")
return config
def save(self) -> None:
"""Save configuration to file."""
self.config_dir.mkdir(parents=True, exist_ok=True)
data = {
"api_url": self.api_url,
"auth_token": self.auth_token,
"email": self.email,
"allow_emoji": self.allow_emoji,
"default_session": self.default_session,
"auto_save": self.auto_save,
"show_mood": self.show_mood,
"show_relationship": self.show_relationship,
"show_facts": self.show_facts,
"show_timestamps": self.show_timestamps,
}
with open(self.config_file, "w") as f:
json.dump(data, f, indent=2)
def get_api_url(self) -> str:
"""Get API URL, checking environment variables first.
Returns:
str: API URL
"""
return os.getenv("LOYAL_COMPANION_API_URL", self.api_url)
def get_auth_token(self) -> str | None:
"""Get auth token, checking environment variables first.
Returns:
str | None: Auth token or None
"""
return os.getenv("LOYAL_COMPANION_TOKEN", self.auth_token)

248
cli/formatters.py Normal file
View File

@@ -0,0 +1,248 @@
"""Terminal formatting for CLI responses."""
from datetime import datetime
from typing import Any
try:
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from rich.text import Text
RICH_AVAILABLE = True
except ImportError:
RICH_AVAILABLE = False
class ResponseFormatter:
"""Formats API responses for terminal display."""
def __init__(
self,
show_mood: bool = True,
show_relationship: bool = False,
show_facts: bool = False,
show_timestamps: bool = False,
use_rich: bool = True,
):
"""Initialize formatter.
Args:
show_mood: Show mood information
show_relationship: Show relationship information
show_facts: Show extracted facts
show_timestamps: Show timestamps
use_rich: Use rich formatting (if available)
"""
self.show_mood = show_mood
self.show_relationship = show_relationship
self.show_facts = show_facts
self.show_timestamps = show_timestamps
self.use_rich = use_rich and RICH_AVAILABLE
if self.use_rich:
self.console = Console()
def format_message(self, role: str, content: str, timestamp: str | None = None) -> str:
"""Format a chat message.
Args:
role: Message role (user/assistant)
content: Message content
timestamp: Optional timestamp
Returns:
str: Formatted message
"""
if self.use_rich:
return self._format_message_rich(role, content, timestamp)
return self._format_message_plain(role, content, timestamp)
def _format_message_plain(self, role: str, content: str, timestamp: str | None = None) -> str:
"""Format message in plain text.
Args:
role: Message role
content: Message content
timestamp: Optional timestamp
Returns:
str: Formatted message
"""
prefix = "You" if role == "user" else "Bartender"
lines = [f"{prefix}: {content}"]
if timestamp and self.show_timestamps:
try:
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
time_str = dt.strftime("%H:%M:%S")
lines.append(f" [{time_str}]")
except Exception:
pass
return "\n".join(lines)
def _format_message_rich(self, role: str, content: str, timestamp: str | None = None) -> None:
"""Format message using rich.
Args:
role: Message role
content: Message content
timestamp: Optional timestamp
"""
if role == "user":
style = "bold cyan"
prefix = "You"
else:
style = "bold green"
prefix = "Bartender"
text = Text()
text.append(f"{prefix}: ", style=style)
text.append(content)
if timestamp and self.show_timestamps:
try:
dt = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
time_str = dt.strftime("%H:%M:%S")
text.append(f"\n [{time_str}]", style="dim")
except Exception:
pass
self.console.print(text)
def format_response(self, response: dict[str, Any]) -> str:
"""Format a chat response with metadata.
Args:
response: API response
Returns:
str: Formatted response
"""
if self.use_rich:
return self._format_response_rich(response)
return self._format_response_plain(response)
def _format_response_plain(self, response: dict[str, Any]) -> str:
"""Format response in plain text.
Args:
response: API response
Returns:
str: Formatted response
"""
lines = [f"Bartender: {response['response']}"]
# Add metadata
metadata = []
if self.show_mood and response.get("mood"):
mood = response["mood"]
metadata.append(f"Mood: {mood['label']}")
if self.show_relationship and response.get("relationship"):
rel = response["relationship"]
metadata.append(f"Relationship: {rel['level']} ({rel['score']})")
if self.show_facts and response.get("extracted_facts"):
facts = response["extracted_facts"]
if facts:
metadata.append(f"Facts learned: {len(facts)}")
if metadata:
lines.append(" " + " | ".join(metadata))
return "\n".join(lines)
def _format_response_rich(self, response: dict[str, Any]) -> None:
"""Format response using rich.
Args:
response: API response
"""
# Main response
text = Text()
text.append("Bartender: ", style="bold green")
text.append(response["response"])
self.console.print(text)
# Metadata panel
metadata_lines = []
if self.show_mood and response.get("mood"):
mood = response["mood"]
mood_line = Text()
mood_line.append("Mood: ", style="dim")
mood_line.append(mood["label"], style="yellow")
mood_line.append(
f" (v:{mood['valence']:.1f}, a:{mood['arousal']:.1f}, i:{mood['intensity']:.1f})",
style="dim",
)
metadata_lines.append(mood_line)
if self.show_relationship and response.get("relationship"):
rel = response["relationship"]
rel_line = Text()
rel_line.append("Relationship: ", style="dim")
rel_line.append(f"{rel['level']} ({rel['score']})", style="cyan")
rel_line.append(f" | {rel['interactions_count']} interactions", style="dim")
metadata_lines.append(rel_line)
if self.show_facts and response.get("extracted_facts"):
facts = response["extracted_facts"]
if facts:
facts_line = Text()
facts_line.append("Facts learned: ", style="dim")
facts_line.append(f"{len(facts)}", style="magenta")
metadata_lines.append(facts_line)
if metadata_lines:
self.console.print()
for line in metadata_lines:
self.console.print(" ", line)
def format_history_message(self, message: dict[str, Any]) -> str:
"""Format a history message.
Args:
message: History message
Returns:
str: Formatted message
"""
return self.format_message(message["role"], message["content"], message.get("timestamp"))
def print_error(self, message: str) -> None:
"""Print an error message.
Args:
message: Error message
"""
if self.use_rich:
self.console.print(f"[bold red]Error:[/bold red] {message}")
else:
print(f"Error: {message}")
def print_info(self, message: str) -> None:
"""Print an info message.
Args:
message: Info message
"""
if self.use_rich:
self.console.print(f"[dim]{message}[/dim]")
else:
print(message)
def print_success(self, message: str) -> None:
"""Print a success message.
Args:
message: Success message
"""
if self.use_rich:
self.console.print(f"[bold green]✓[/bold green] {message}")
else:
print(f"{message}")

362
cli/main.py Normal file
View File

@@ -0,0 +1,362 @@
"""Loyal Companion CLI - Main entry point."""
import sys
from pathlib import Path
import typer
from typing_extensions import Annotated
from cli.client import APIError, LoyalCompanionClient
from cli.config import CLIConfig
from cli.formatters import ResponseFormatter
from cli.session import SessionManager
app = typer.Typer(
name="loyal-companion",
help="Loyal Companion CLI - A quiet, terminal-based interface for conversations.",
add_completion=False,
)
def _ensure_authenticated(config: CLIConfig) -> tuple[CLIConfig, str]:
"""Ensure user is authenticated.
Args:
config: CLI configuration
Returns:
tuple: (config, auth_token)
Raises:
typer.Exit: If authentication fails
"""
auth_token = config.get_auth_token()
if not auth_token:
# Need to authenticate
email = config.email
if not email:
email = typer.prompt("Email address")
config.email = email
config.save()
# Request token
try:
client = LoyalCompanionClient(config.get_api_url())
response = client.request_token(email)
auth_token = response.get("token")
if not auth_token:
typer.echo(f"Error: {response.get('message', 'No token received')}", err=True)
raise typer.Exit(1)
# Save token
config.auth_token = auth_token
config.save()
typer.echo(f"Authenticated as {email}")
except APIError as e:
typer.echo(f"Authentication failed: {e}", err=True)
raise typer.Exit(1)
return config, auth_token
@app.command()
def talk(
session_name: Annotated[str, typer.Option("--session", "-s", help="Session name")] = "default",
new: Annotated[bool, typer.Option("--new", "-n", help="Start a new session")] = False,
show_mood: Annotated[
bool, typer.Option("--mood/--no-mood", help="Show mood information")
] = True,
show_relationship: Annotated[
bool, typer.Option("--relationship/--no-relationship", help="Show relationship info")
] = False,
):
"""Start or resume a conversation.
Examples:
lc talk # Resume default session
lc talk --new # Start fresh default session
lc talk -s work # Resume 'work' session
lc talk -s personal --new # Start fresh 'personal' session
"""
# Load config
config = CLIConfig.load()
# Ensure authenticated
config, auth_token = _ensure_authenticated(config)
# Initialize client
client = LoyalCompanionClient(config.get_api_url(), auth_token)
# Initialize session manager
session_manager = SessionManager(config.sessions_file)
# Get or create session
if new:
# Delete old session if exists
session_manager.delete_session(session_name)
session = session_manager.get_or_create_session(session_name)
# Initialize formatter
formatter = ResponseFormatter(
show_mood=show_mood,
show_relationship=show_relationship,
show_facts=config.show_facts,
show_timestamps=config.show_timestamps,
)
# Print welcome message
formatter.print_info("Bartender is here.")
if session.message_count > 0:
formatter.print_info(
f"Resuming session '{session.name}' ({session.message_count} messages)"
)
formatter.print_info("Type your message and press Enter. Press Ctrl+D to end.\n")
# Conversation loop
try:
while True:
# Get user input
try:
user_message = typer.prompt("You", prompt_suffix=": ")
except (EOFError, KeyboardInterrupt):
# User pressed Ctrl+D or Ctrl+C
break
if not user_message.strip():
continue
# Send message
try:
response = client.send_message(session.session_id, user_message)
# Format and display response
formatter.format_response(response)
print() # Empty line for spacing
# Update session
session_manager.update_last_active(session.name)
except APIError as e:
formatter.print_error(str(e))
continue
except KeyboardInterrupt:
pass
# Goodbye message
print() # Empty line
formatter.print_info("Session saved.")
client.close()
@app.command()
def history(
session_name: Annotated[str, typer.Option("--session", "-s", help="Session name")] = "default",
limit: Annotated[int, typer.Option("--limit", "-n", help="Number of messages")] = 50,
):
"""Show conversation history for a session.
Examples:
lc history # Show default session history
lc history -s work # Show 'work' session history
lc history -n 10 # Show last 10 messages
"""
# Load config
config = CLIConfig.load()
# Ensure authenticated
config, auth_token = _ensure_authenticated(config)
# Initialize client
client = LoyalCompanionClient(config.get_api_url(), auth_token)
# Initialize session manager
session_manager = SessionManager(config.sessions_file)
# Get session
session = session_manager.get_session(session_name)
if not session:
typer.echo(f"Session '{session_name}' not found", err=True)
raise typer.Exit(1)
# Get history
try:
response = client.get_history(session.session_id, limit)
messages = response.get("messages", [])
if not messages:
typer.echo("No messages in this session yet.")
raise typer.Exit(0)
# Format and display
formatter = ResponseFormatter(
show_timestamps=True,
use_rich=True,
)
typer.echo(f"History for session '{session.name}' ({len(messages)} messages):\n")
for message in messages:
formatter.format_history_message(message)
print() # Spacing between messages
except APIError as e:
typer.echo(f"Failed to get history: {e}", err=True)
raise typer.Exit(1)
client.close()
@app.command()
def sessions(
delete: Annotated[str | None, typer.Option("--delete", "-d", help="Delete a session")] = None,
):
"""List all sessions or delete a specific session.
Examples:
lc sessions # List all sessions
lc sessions -d work # Delete 'work' session
"""
# Load config
config = CLIConfig.load()
# Initialize session manager
session_manager = SessionManager(config.sessions_file)
if delete:
# Delete session
if session_manager.delete_session(delete):
typer.echo(f"Deleted session '{delete}'")
else:
typer.echo(f"Session '{delete}' not found", err=True)
raise typer.Exit(1)
return
# List sessions
all_sessions = session_manager.list_sessions()
if not all_sessions:
typer.echo("No sessions found.")
return
typer.echo(f"Found {len(all_sessions)} session(s):\n")
for session in all_sessions:
typer.echo(f" {session.name}")
typer.echo(f" Created: {session.created_at}")
typer.echo(f" Last active: {session.last_active}")
typer.echo(f" Messages: {session.message_count}")
typer.echo()
@app.command()
def config_cmd(
show: Annotated[bool, typer.Option("--show", help="Show current configuration")] = False,
set_api_url: Annotated[str | None, typer.Option("--api-url", help="Set API URL")] = None,
set_email: Annotated[str | None, typer.Option("--email", help="Set email")] = None,
reset: Annotated[bool, typer.Option("--reset", help="Reset configuration")] = False,
):
"""Manage CLI configuration.
Examples:
lc config --show # Show current config
lc config --api-url http://localhost:8080 # Set API URL
lc config --email user@example.com # Set email
lc config --reset # Reset to defaults
"""
config = CLIConfig.load()
if reset:
# Delete config file
if config.config_file.exists():
config.config_file.unlink()
typer.echo("Configuration reset to defaults")
return
if set_api_url:
config.api_url = set_api_url
config.save()
typer.echo(f"API URL set to: {set_api_url}")
if set_email:
config.email = set_email
# Clear token when email changes
config.auth_token = None
config.save()
typer.echo(f"Email set to: {set_email}")
typer.echo("(Auth token cleared - you'll need to re-authenticate)")
if show or (not set_api_url and not set_email and not reset):
# Show config
typer.echo("Current configuration:\n")
typer.echo(f" API URL: {config.get_api_url()}")
typer.echo(f" Email: {config.email or '(not set)'}")
typer.echo(f" Authenticated: {'Yes' if config.get_auth_token() else 'No'}")
typer.echo(f" Config file: {config.config_file}")
typer.echo(f" Sessions file: {config.sessions_file}")
@app.command()
def auth(
logout: Annotated[bool, typer.Option("--logout", help="Clear authentication")] = False,
):
"""Manage authentication.
Examples:
lc auth # Show auth status
lc auth --logout # Clear stored token
"""
config = CLIConfig.load()
if logout:
config.auth_token = None
config.save()
typer.echo("Authentication cleared")
return
# Show auth status
if config.get_auth_token():
typer.echo(f"Authenticated as: {config.email}")
else:
typer.echo("Not authenticated")
typer.echo("Run 'lc talk' to authenticate")
@app.command()
def health():
"""Check API health status.
Examples:
lc health # Check if API is reachable
"""
config = CLIConfig.load()
try:
client = LoyalCompanionClient(config.get_api_url())
response = client.health_check()
typer.echo(f"API Status: {response.get('status', 'unknown')}")
typer.echo(f"Platform: {response.get('platform', 'unknown')}")
typer.echo(f"Version: {response.get('version', 'unknown')}")
except APIError as e:
typer.echo(f"Health check failed: {e}", err=True)
raise typer.Exit(1)
def main():
"""Main entry point."""
app()
if __name__ == "__main__":
main()

164
cli/session.py Normal file
View File

@@ -0,0 +1,164 @@
"""Session management for CLI."""
import json
import secrets
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path
from typing import Any
@dataclass
class SessionData:
"""Local session data."""
session_id: str
name: str
created_at: str
last_active: str
message_count: int = 0
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary.
Returns:
dict: Session data as dictionary
"""
return asdict(self)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "SessionData":
"""Create from dictionary.
Args:
data: Dictionary data
Returns:
SessionData: Session instance
"""
return cls(**data)
class SessionManager:
"""Manages local CLI sessions."""
def __init__(self, sessions_file: Path):
"""Initialize session manager.
Args:
sessions_file: Path to sessions file
"""
self.sessions_file = sessions_file
self.sessions: dict[str, SessionData] = {}
self._load()
def _load(self) -> None:
"""Load sessions from file."""
if self.sessions_file.exists():
try:
with open(self.sessions_file, "r") as f:
data = json.load(f)
self.sessions = {
name: SessionData.from_dict(session_data) for name, session_data in data.items()
}
except (json.JSONDecodeError, IOError) as e:
print(f"Warning: Could not load sessions: {e}")
self.sessions = {}
def _save(self) -> None:
"""Save sessions to file."""
self.sessions_file.parent.mkdir(parents=True, exist_ok=True)
data = {name: session.to_dict() for name, session in self.sessions.items()}
with open(self.sessions_file, "w") as f:
json.dump(data, f, indent=2)
def create_session(self, name: str = "default") -> SessionData:
"""Create or get a session.
Args:
name: Session name
Returns:
SessionData: Created or existing session
"""
if name in self.sessions:
return self.sessions[name]
# Generate unique session ID
session_id = f"cli_{name}_{secrets.token_hex(8)}"
now = datetime.utcnow().isoformat()
session = SessionData(
session_id=session_id,
name=name,
created_at=now,
last_active=now,
message_count=0,
)
self.sessions[name] = session
self._save()
return session
def get_session(self, name: str) -> SessionData | None:
"""Get a session by name.
Args:
name: Session name
Returns:
SessionData | None: Session or None if not found
"""
return self.sessions.get(name)
def get_or_create_session(self, name: str = "default") -> SessionData:
"""Get or create a session.
Args:
name: Session name
Returns:
SessionData: Session
"""
session = self.get_session(name)
if session:
return session
return self.create_session(name)
def update_last_active(self, name: str) -> None:
"""Update session's last active time.
Args:
name: Session name
"""
if name in self.sessions:
self.sessions[name].last_active = datetime.utcnow().isoformat()
self.sessions[name].message_count += 1
self._save()
def list_sessions(self) -> list[SessionData]:
"""List all sessions.
Returns:
list[SessionData]: All sessions
"""
return sorted(self.sessions.values(), key=lambda s: s.last_active, reverse=True)
def delete_session(self, name: str) -> bool:
"""Delete a session.
Args:
name: Session name
Returns:
bool: True if deleted, False if not found
"""
if name in self.sessions:
del self.sessions[name]
self._save()
return True
return False