CLI

colony-env

The colony-env CLI is built with Typer.

polymathera.colony.cli.deploy.cli

colony-env: CLI for managing Colony local test environments.

Usage

colony-env up [--workers N][--no-build] [--k8s] colony-env down [--k8s] colony-env status colony-env run --origin-url URL [--branch BRANCH][--commit SHA] [--config YAML] colony-env run --local-repo PATH [--branch BRANCH][--commit SHA] [--config YAML] colony-env doctor

up(workers=typer.Option(1, '--workers', '-w', help='Number of Ray workers'), no_build=typer.Option(False, '--no-build', help='Skip image build'), k8s=typer.Option(False, '--k8s', help='Use Kind + KubeRay (advanced)'))

Build Colony image and start Ray cluster + Redis.

Source code in src/polymathera/colony/cli/deploy/cli.py

@app.command()
def up(
    workers: int = typer.Option(1, "--workers", "-w", help="Number of Ray workers"),
    no_build: bool = typer.Option(False, "--no-build", help="Skip image build"),
    k8s: bool = typer.Option(False, "--k8s", help="Use Kind + KubeRay (advanced)"),
):
    """Build Colony image and start Ray cluster + Redis."""
    config = DeployConfig(mode="k8s" if k8s else "compose")
    manager = DeploymentManager(config)

    try:
        services = _run(manager.up(
            build=not no_build,
            workers=workers,
            on_status=lambda msg: console.print(f"  [blue]{msg}[/blue]"),
        ))
    except RuntimeError as e:
        console.print(f"[red]Error:[/red] {e}")
        raise typer.Exit(1)

    all_ok = all(s.status == ProviderStatus.RUNNING for s in services)

    for svc in services:
        icon = "[green]OK[/green]" if svc.status == ProviderStatus.RUNNING else "[red]FAIL[/red]"
        detail = ""
        if svc.details.get("dashboard"):
            detail = f"  dashboard: {svc.details['dashboard']}"
        elif svc.port:
            detail = f"  localhost:{svc.port}"
        if svc.details.get("replicas"):
            detail += f"  (x{svc.details['replicas']})"
        console.print(f"  {svc.name:16s} {icon}{detail}")

    if all_ok:
        console.print()
        console.print("[green]Ready![/green] Run your analysis:")
        console.print("  colony-env run --local-repo /path/to/codebase --config analysis.yaml")
    else:
        console.print()
        console.print("[red]Some services failed to start. Check 'docker compose logs'.[/red]")
        raise typer.Exit(1)

down(k8s=typer.Option(False, '--k8s', help='Use Kind + KubeRay (advanced)'))

Stop and remove all containers and resources.

Source code in src/polymathera/colony/cli/deploy/cli.py

@app.command()
def down(
    k8s: bool = typer.Option(False, "--k8s", help="Use Kind + KubeRay (advanced)"),
):
    """Stop and remove all containers and resources."""
    config = DeployConfig(mode="k8s" if k8s else "compose")
    manager = DeploymentManager(config)

    try:
        with console.status("[blue]Stopping colony environment..."):
            _run(manager.down())
    except RuntimeError as e:
        console.print(f"[red]Error:[/red] {e}")
        raise typer.Exit(1)

    console.print("[green]Colony environment stopped.[/green]")

status()

Show status of all running services.

Source code in src/polymathera/colony/cli/deploy/cli.py

@app.command()
def status():
    """Show status of all running services."""
    manager = DeploymentManager()
    services = _run(manager.status())

    if not services:
        console.print("No colony services running.")
        return

    table = Table(title="Colony Environment")
    table.add_column("Service", style="bold")
    table.add_column("Status")
    table.add_column("Details")

    for svc in services:
        if svc.status == ProviderStatus.RUNNING:
            status_str = "[green]running[/green]"
        elif svc.status == ProviderStatus.STOPPED:
            status_str = "[dim]stopped[/dim]"
        else:
            status_str = f"[red]{svc.status.value}[/red]"

        details = ", ".join(f"{k}={v}" for k, v in svc.details.items())
        table.add_row(svc.name, status_str, details)

    console.print(table)

run(origin_url=typer.Option(None, '--origin-url', help='Git repository URL (HTTPS) for the codebase to analyze.'), local_repo=typer.Option(None, '--local-repo', help='Path to a local git repository to analyze.'), branch=typer.Option('main', '--branch', help='Git branch to check out.'), commit=typer.Option('HEAD', '--commit', help='Git commit SHA (defaults to branch HEAD).'), config=typer.Option(None, '--config', '-c', help='Path to analysis YAML config'), k8s=typer.Option(False, '--k8s', help='Use Kind + KubeRay (advanced)'), verbose=typer.Option(False, '--verbose', '-v', help='Enable verbose polymath output'), dry_run=typer.Option(False, '--dry-run', help='Show planned hierarchy without running'))

Run polymath.py analysis inside the cluster.

Exactly one of --origin-url or --local-repo must be provided.

Source code in src/polymathera/colony/cli/deploy/cli.py

@app.command()
def run(
    origin_url: Optional[str] = typer.Option(
        None, "--origin-url",
        help="Git repository URL (HTTPS) for the codebase to analyze.",
    ),
    local_repo: Optional[str] = typer.Option(
        None, "--local-repo",
        help="Path to a local git repository to analyze.",
    ),
    branch: str = typer.Option("main", "--branch", help="Git branch to check out."),
    commit: str = typer.Option("HEAD", "--commit", help="Git commit SHA (defaults to branch HEAD)."),
    config: Optional[str] = typer.Option(None, "--config", "-c", help="Path to analysis YAML config"),
    k8s: bool = typer.Option(False, "--k8s", help="Use Kind + KubeRay (advanced)"),
    verbose: bool = typer.Option(False, "--verbose", "-v", help="Enable verbose polymath output"),
    dry_run: bool = typer.Option(False, "--dry-run", help="Show planned hierarchy without running"),
):
    """Run polymath.py analysis inside the cluster.

    Exactly one of --origin-url or --local-repo must be provided.
    """
    if origin_url and local_repo:
        console.print("[red]Error:[/red] --origin-url and --local-repo are mutually exclusive.")
        raise typer.Exit(1)
    if not origin_url and not local_repo:
        console.print("[red]Error:[/red] Provide --origin-url <URL> or --local-repo <path>.")
        raise typer.Exit(1)

    deploy_config = DeployConfig(mode="k8s" if k8s else "compose")
    manager = DeploymentManager(deploy_config)

    extra_args: list[str] = []
    if verbose:
        extra_args.append("--verbose")
    if dry_run:
        extra_args.append("--dry-run")

    try:
        exit_code = _run(manager.run(
            origin_url=origin_url,
            local_repo=local_repo,
            branch=branch,
            commit=commit,
            config_path=config,
            extra_args=extra_args or None,
        ))
    except (RuntimeError, FileNotFoundError) as e:
        console.print(f"[red]Error:[/red] {e}")
        raise typer.Exit(1)

    raise typer.Exit(exit_code)

dashboard(port=typer.Option(8080, '--port', '-p', help='Dashboard port (must match COLONY_DASHBOARD_UI_PORT)'))

Open the Colony web dashboard in the browser.

Source code in src/polymathera/colony/cli/deploy/cli.py

@app.command()
def dashboard(
    port: int = typer.Option(8080, "--port", "-p", help="Dashboard port (must match COLONY_DASHBOARD_UI_PORT)"),
):
    """Open the Colony web dashboard in the browser."""
    import webbrowser

    url = f"http://localhost:{port}"
    console.print(f"Opening dashboard at [blue]{url}[/blue]")
    webbrowser.open(url)

doctor()

Check prerequisites for running colony-env.

Source code in src/polymathera/colony/cli/deploy/cli.py

@app.command()
def doctor():
    """Check prerequisites for running colony-env."""
    manager = DeploymentManager()
    checks = _run(manager.doctor())

    all_ok = True
    for name, passed in checks.items():
        icon = "[green]OK[/green]" if passed else "[red]MISSING[/red]"
        console.print(f"  {name:20s} {icon}")
        if not passed:
            all_ok = False

    if all_ok:
        console.print()
        console.print("[green]All prerequisites met.[/green]")
    else:
        console.print()
        console.print("[red]Some prerequisites are missing. Install Docker and try again.[/red]")
        raise typer.Exit(1)