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)'), config=typer.Option(None, '--config', '-c', help='Path to cluster YAML config to use for auto-deploy'), bake=typer.Option(False, '--bake', help='Snapshot resolved cluster.extensions.packages into a pinned ``colony-local:<hash>`` image instead of installing into a persistent overlay at container start. Slower up, faster steady-state, fully reproducible — recommended for production.'))

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)"),
    config: Optional[str] = typer.Option(None, "--config", "-c", help="Path to cluster YAML config to use for auto-deploy"),
    bake: bool = typer.Option(
        False, "--bake",
        help="Snapshot resolved cluster.extensions.packages into a pinned "
             "``colony-local:<hash>`` image instead of installing into a "
             "persistent overlay at container start. Slower up, faster steady"
             "-state, fully reproducible — recommended for production.",
    ),
):
    """Build Colony image and start Ray cluster + Redis."""
    deploy_config = DeployConfig(mode="k8s" if k8s else "compose")
    manager = DeploymentManager(deploy_config)

    try:
        services = _run(manager.up(
            build=not no_build,
            workers=workers,
            config_path=config,
            on_status=lambda msg: console.print(f"  [blue]{msg}[/blue]"),
            bake=bake,
        ))
    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)

image_info()

List polymathera-* packages installed in the running cluster.

Splits baked (in the runtime image) vs overlay (installed at container start from cluster.extensions.packages). Useful for debugging "why isn't my CPS entry showing up".

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

@app.command("image-info")
def image_info():
    """List polymathera-* packages installed in the running cluster.

    Splits ``baked`` (in the runtime image) vs ``overlay`` (installed at
    container start from ``cluster.extensions.packages``). Useful for
    debugging "why isn't my CPS entry showing up".
    """
    manager = DeploymentManager()
    try:
        info = _run(manager.image_info())
    except RuntimeError as e:
        console.print(f"[red]Error:[/red] {e}")
        raise typer.Exit(1)

    for kind in ("baked", "overlay"):
        lines = info.get(kind, [])
        header = f"[bold]{kind} ({len(lines)})[/bold]"
        console.print(header)
        if not lines:
            console.print("  [dim](none)[/dim]")
            continue
        for line in lines:
            console.print(f"  {line}")

image_build(config=typer.Option(None, '--config', '-c', help='Path to cluster YAML config'), bake=typer.Option(False, '--bake', help='Build a pinned ``colony-local:<hash>`` image with extensions baked in.'))

Build base + runtime images (and optionally bake) without bringing the cluster up.

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

@app.command("image-build")
def image_build(
    config: Optional[str] = typer.Option(None, "--config", "-c", help="Path to cluster YAML config"),
    bake: bool = typer.Option(
        False, "--bake",
        help="Build a pinned ``colony-local:<hash>`` image with extensions baked in.",
    ),
):
    """Build base + runtime images (and optionally bake) without bringing the cluster up."""
    manager = DeploymentManager()
    try:
        tag = _run(manager.image_build(
            config_path=config, bake=bake,
            on_status=lambda msg: console.print(f"  [blue]{msg}[/blue]"),
        ))
    except (RuntimeError, NotImplementedError) as e:
        console.print(f"[red]Error:[/red] {e}")
        raise typer.Exit(1)

    console.print(f"[green]Built:[/green] {tag}")

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)