"""
Main CLI entry point for Continuity plugin.
"""
from pathlib import Path
from typing import Optional
import click
from rich.console import Console
from rich.panel import Panel
from .._version import __version__
from ..config import ContinuityConfig
from ..core.mailbox import ContinuityCore
from .commands import register_commands
console = Console()
[docs]
class ContinuityCLI:
"""Main CLI application."""
def __init__(self, config: Optional[ContinuityConfig] = None) -> None:
# Use provided config or load defaults
self.config = config if config is not None else ContinuityConfig.load()
# Don't initialize core immediately - defer until needed
self._core: Optional[ContinuityCore] = None
self._base_path: Optional[Path] = None
@property
def core(self) -> ContinuityCore:
"""Lazy-load the core when needed."""
if self._core is None:
base_path = self._get_effective_base_path()
# Check if base path has .continuity directory
project_continuity = base_path / ".continuity"
if not project_continuity.exists():
from ..cli.commands import UserError
raise UserError(
f"No continuity project found at {base_path}",
"Run 'continuity init' to initialize a project first",
)
self._core = ContinuityCore(base_path, config=self.config)
return self._core
def _get_effective_base_path(self) -> Path:
"""Get the effective base path, preferring project-specific paths."""
try:
current_dir = Path.cwd()
except (OSError, FileNotFoundError) as err:
# If we can't get cwd, we can't find a project
from ..cli.commands import UserError
raise UserError(
"Cannot determine current directory",
"Please navigate to a continuity project directory",
) from err
# Check for project-specific .continuity folder
project_continuity = current_dir / ".continuity"
if project_continuity.exists():
return current_dir
# Walk up the directory tree looking for .continuity
for parent in current_dir.parents:
project_continuity = parent / ".continuity"
if project_continuity.exists():
return parent
# No project found - fail immediately
from ..cli.commands import UserError
raise UserError(
"No continuity project found",
"Run 'continuity init' to initialize a project first",
)
[docs]
def hello(self, name: str = "World") -> None:
"""Say hello - simple direct implementation."""
result = self.core.say_hello(name)
console.print(
Panel(
f"[green]{result['message']}[/green]",
title="👋 Continuity Hello",
subtitle="[dim]Saved to history[/dim]",
)
)
# Show history count
history = self.core.get_hello_history()
console.print(
f"\n[dim]Continuity maintained: {len(history)} greetings logged[/dim]"
)
[docs]
class CustomGroup(click.Group):
"""Custom Click group that shows our cached help for --help."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Add --usage flag to show brief help
self.params.append(
click.Option(
["-u", "--usage"], is_flag=True, help="Show brief usage summary"
)
)
[docs]
def invoke(self, ctx):
"""Check for --usage flag before normal invocation."""
if ctx.params.get("usage"):
# Show brief usage and exit
try:
from ._help_cache import get_quick_help
click.echo(get_quick_help())
except ImportError:
click.echo("Usage: continuity <command> [options]")
click.echo("Try 'continuity --help' for full reference")
ctx.exit()
return super().invoke(ctx)
[docs]
def get_help(self, ctx):
"""Override to show our full cached help."""
try:
from ._help_cache import get_full_help
return get_full_help()
except ImportError:
# Fallback to default Click help
return super().get_help(ctx)
[docs]
def set_user_override(ctx, param, value):
"""Set user override in environment if provided."""
if value:
import os
# Set environment variable for this session
os.environ["CONTINUITY_SESSION_USER"] = value
return value
@click.group(cls=CustomGroup, context_settings={"help_option_names": ["-h", "--help"]})
@click.version_option(version=__version__, prog_name="continuity")
@click.option(
"--config", "-c", type=click.Path(exists=True), help="Configuration file path"
)
@click.option(
"--user",
callback=set_user_override,
expose_value=False,
help="Override user role for this session (human, agent, claude, ai, or username)",
)
@click.pass_context
def cli(ctx, config, usage=None):
"""Continuity - Async communication for Claude Code.
Simple async communication system for AI development workflows.
"""
ctx.ensure_object(dict)
try:
from pathlib import Path
config_path_obj = Path(config) if config else None
app_config = ContinuityConfig.load(config_path=config_path_obj)
ctx.obj = ContinuityCLI(config=app_config)
except Exception as e:
console.print(f"[red]Error initializing Continuity: {e}[/red]")
console.print(
"[yellow]Try running 'continuity status' to diagnose issues[/yellow]"
)
ctx.obj = None
# Register all commands
register_commands(cli)
[docs]
def main() -> None:
"""Main entry point."""
try:
cli()
except KeyboardInterrupt:
console.print("\n[yellow]Interrupted by user[/yellow]")
except Exception as e:
console.print(f"\n[red]Unexpected error: {e}[/red]")
console.print("[dim]Use --help for usage information[/dim]")
if __name__ == "__main__":
main()