Add environment status API and CLI command

affinetes/api.py

@@ -551,4 +551,88 @@ def get_environment(env_id: str) -> Optional[EnvironmentWrapper]:
         >>> env = get_environment('affine-latest_1234567890')
     """
     registry = get_registry()
-    return registry.get(env_id)
+    return registry.get(env_id)
+
+
+def get_environment_status(
+    env_id: str,
+    include_stats: bool = True,
+) -> Optional[Dict[str, Any]]:
+    """
+    Get status information for a single environment.
+
+    The returned dictionary is safe to serialize as JSON and is designed
+    for both programmatic inspection and CLI usage.
+
+    Args:
+        env_id: Environment identifier as returned by list_active_environments().
+        include_stats: When True, include per-instance statistics for multi-instance pools
+                       (see EnvironmentWrapper.get_stats()).
+
+    Returns:
+        Dictionary with status information or None if the environment is not found.
+
+    Example:
+        >>> status = get_environment_status('affine-latest_1234567890')
+        >>> status['ready']
+        True
+    """
+    registry = get_registry()
+    env = registry.get(env_id)
+    if env is None:
+        return None
+
+    # Base metadata that is always available
+    status: Dict[str, Any] = {
+        "id": env_id,
+        "name": getattr(env, "name", env_id),
+        "ready": bool(env.is_ready()),
+    }
+
+    # Pool statistics (only for multi-instance pools)
+    stats: Optional[Dict[str, Any]] = None
+    try:
+        stats = env.get_stats()
+    except Exception as e:
+        # get_stats() is optional and should never break status inspection
+        logger.debug(f"get_stats() failed for environment '{env_id}': {e}")
+
+    status["is_pool"] = stats is not None
+
+    if include_stats and stats is not None:
+        status["stats"] = stats
+
+    return status
+
+
+def get_all_environment_statuses(
+    include_stats: bool = True,
+) -> List[Dict[str, Any]]:
+    """
+    Get status information for all active environments.
+
+    This is a thin convenience wrapper around list_active_environments()
+    and get_environment_status() and is intended for CLI commands and
+    monitoring tools.
+
+    Args:
+        include_stats: When True, include per-instance statistics for multi-instance pools.
+
+    Returns:
+        List of status dictionaries (see get_environment_status()).
+
+    Example:
+        >>> statuses = get_all_environment_statuses()
+        >>> [s['id'] for s in statuses]
+        ['affine-latest_1234567890', 'custom-env-pool-3']
+    """
+    registry = get_registry()
+    env_ids = registry.list_all()
+
+    statuses: List[Dict[str, Any]] = []
+    for env_id in env_ids:
+        status = get_environment_status(env_id, include_stats=include_stats)
+        if status is not None:
+            statuses.append(status)
+
+    return statuses

affinetes/cli/commands.py

@@ -6,9 +6,15 @@
 import os
 import sys
 from pathlib import Path
-from typing import Optional, Dict, Any
-
-from ..api import load_env, build_image_from_env, get_environment
+from typing import Optional, Dict, Any, List
+
+from ..api import (
+    load_env,
+    build_image_from_env,
+    get_environment,
+    get_all_environment_statuses,
+    get_environment_status,
+)
 from ..utils.logger import logger
 from .templates import (
     ACTOR_ENV_PY,
@@ -541,4 +547,105 @@ async def test_environment(
 
     except Exception as e:
         logger.error(f"Failed to run tests: {e}")
-        raise
+        raise
+
+
+def show_status(
+    name: Optional[str],
+    output: str = "table",
+    verbose: bool = False
+) -> None:
+    """
+    Display status information about active environments.
+
+    Args:
+        name: Optional specific environment ID to inspect. When omitted, all
+              active environments from the global registry are shown.
+        output: Output format: "table" (default) or "json".
+        verbose: When True, include per-instance details for pools in table mode.
+    """
+    if name:
+        statuses: List[Dict[str, Any]] = []
+        status = get_environment_status(name)
+        if not status:
+            logger.info(f"No active environment found with id '{name}'")
+            return
+        statuses.append(status)
+    else:
+        statuses = get_all_environment_statuses()
+
+    if not statuses:
+        logger.info("No active environments found")
+        return
+
+    if output == "json":
+        print(json.dumps(statuses, indent=2, ensure_ascii=False))
+        return
+
+    # Default: human-friendly table output
+    try:
+        from tabulate import tabulate  # type: ignore[import]
+
+        use_tabulate = True
+    except Exception:
+        use_tabulate = False
+
+    headers = ["ID", "Name", "Ready", "Pool", "Instances", "Requests"]
+    rows = []
+
+    for status in statuses:
+        stats = status.get("stats") if isinstance(status, dict) else None
+        is_pool = bool(status.get("is_pool")) if isinstance(status, dict) else False
+
+        total_instances = "-"
+        total_requests = "-"
+
+        if isinstance(stats, dict):
+            total_instances = stats.get("total_instances", "-")
+            total_requests = stats.get("total_requests", "-")
+
+        rows.append([
+            status.get("id"),
+            status.get("name"),
+            "yes" if status.get("ready") else "no",
+            "yes" if is_pool else "no",
+            total_instances,
+            total_requests,
+        ])
+
+    if use_tabulate:
+        table = tabulate(rows, headers=headers, tablefmt="github")
+        print(table)
+    else:
+        # Fallback formatting when tabulate is not available
+        col_widths = [len(h) for h in headers]
+        for row in rows:
+            for i, cell in enumerate(row):
+                col_widths[i] = max(col_widths[i], len(str(cell)))
+
+        def _format_row(values):
+            return "  ".join(str(v).ljust(col_widths[i]) for i, v in enumerate(values))
+
+        print(_format_row(headers))
+        print("-" * (sum(col_widths) + 2 * (len(headers) - 1)))
+        for row in rows:
+            print(_format_row(row))
+
+    if verbose:
+        # Print per-instance details for pools
+        print("\nInstance details:")
+        for status in statuses:
+            stats = status.get("stats") if isinstance(status, dict) else None
+            if not isinstance(stats, dict):
+                continue
+
+            instances = stats.get("instances") or []
+            if not instances:
+                continue
+
+            print(f"\nEnvironment: {status.get('name')} ({status.get('id')})")
+            for inst in instances:
+                host = inst.get("host", "unknown")
+                port = inst.get("port", "unknown")
+                reqs = inst.get("requests", 0)
+                print(f"  - {host}:{port} (requests: {reqs})")

affinetes/cli/main.py

@@ -8,7 +8,14 @@
 from dotenv import load_dotenv
 
 from ..utils.logger import logger
-from .commands import run_environment, call_method, build_and_push, init_environment, test_environment
+from .commands import (
+    run_environment,
+    call_method,
+    build_and_push,
+    init_environment,
+    test_environment,
+    show_status,
+)
 
 load_dotenv(override=True)
 
@@ -222,6 +229,27 @@ def create_parser() -> argparse.ArgumentParser:
         help='Timeout for each evaluation in seconds (default: 60)'
     )
 
+    # === status command ===
+    status_parser = subparsers.add_parser(
+        'status',
+        help='Show status of active environments'
+    )
+    status_parser.add_argument(
+        'name',
+        nargs='?',
+        help='Optional environment ID to inspect (default: show all)'
+    )
+    status_parser.add_argument(
+        '--json',
+        action='store_true',
+        help='Output status as JSON'
+    )
+    status_parser.add_argument(
+        '--verbose',
+        action='store_true',
+        help='Show per-instance details for pools'
+    )
+
     return parser
 
 
@@ -341,6 +369,14 @@ def main():
                 timeout=args.timeout
             ))
 
+        elif args.command == 'status':
+            output = "json" if args.json else "table"
+            show_status(
+                name=args.name,
+                output=output,
+                verbose=args.verbose,
+            )
+
         else:
             parser.print_help()
             sys.exit(1)

tests/test_api_status.py

@@ -0,0 +1,102 @@
+import types
+
+import pytest
+
+from affinetes import api
+
+
+class FakeWrapper:
+    def __init__(self, name, ready=True, stats=None):
+        self.name = name
+        self._ready = ready
+        self._stats = stats
+
+    def is_ready(self):
+        return self._ready
+
+    def get_stats(self):
+        return self._stats
+
+
+class FakeRegistry:
+    def __init__(self, mapping):
+        self._mapping = dict(mapping)
+
+    def get(self, env_id):
+        return self._mapping.get(env_id)
+
+    def list_all(self):
+        return list(self._mapping.keys())
+
+
+def test_get_environment_status_returns_none_for_missing_env(monkeypatch):
+    registry = FakeRegistry({})
+    monkeypatch.setattr(api, "get_registry", lambda: registry)
+
+    status = api.get_environment_status("missing-env")
+    assert status is None
+
+
+def test_get_environment_status_single_instance(monkeypatch):
+    wrapper = FakeWrapper("env-1", ready=True, stats=None)
+    registry = FakeRegistry({"env-1": wrapper})
+    monkeypatch.setattr(api, "get_registry", lambda: registry)
+
+    status = api.get_environment_status("env-1")
+    assert status is not None
+    assert status["id"] == "env-1"
+    assert status["name"] == "env-1"
+    assert status["ready"] is True
+    assert status["is_pool"] is False
+    assert "stats" not in status
+
+
+def test_get_environment_status_includes_stats_for_pools(monkeypatch):
+    stats = {
+        "total_instances": 3,
+        "total_requests": 10,
+        "instances": [
+            {"host": "h1", "port": 8000, "requests": 3},
+            {"host": "h2", "port": 8000, "requests": 7},
+        ],
+    }
+    wrapper = FakeWrapper("pool-1", ready=True, stats=stats)
+    registry = FakeRegistry({"pool-1": wrapper})
+    monkeypatch.setattr(api, "get_registry", lambda: registry)
+
+    status = api.get_environment_status("pool-1")
+    assert status is not None
+    assert status["id"] == "pool-1"
+    assert status["is_pool"] is True
+    assert status["stats"] == stats
+
+
+def test_get_environment_status_excludes_stats_when_flag_false(monkeypatch):
+    stats = {"total_instances": 2, "total_requests": 5, "instances": []}
+    wrapper = FakeWrapper("pool-2", ready=False, stats=stats)
+    registry = FakeRegistry({"pool-2": wrapper})
+    monkeypatch.setattr(api, "get_registry", lambda: registry)
+
+    status = api.get_environment_status("pool-2", include_stats=False)
+    assert status is not None
+    assert status["ready"] is False
+    assert status["is_pool"] is True
+    assert "stats" not in status
+
+
+def test_get_all_environment_statuses_aggregates_multiple_envs(monkeypatch):
+    w1 = FakeWrapper("env-1", ready=True, stats=None)
+    w2_stats = {"total_instances": 2, "total_requests": 4, "instances": []}
+    w2 = FakeWrapper("pool-1", ready=True, stats=w2_stats)
+
+    registry = FakeRegistry({"env-1": w1, "pool-1": w2})
+    monkeypatch.setattr(api, "get_registry", lambda: registry)
+
+    statuses = api.get_all_environment_statuses()
+    assert len(statuses) == 2
+
+    by_id = {s["id"]: s for s in statuses}
+    assert by_id["env-1"]["is_pool"] is False
+    assert by_id["pool-1"]["is_pool"] is True
+    assert by_id["pool-1"]["stats"] == w2_stats
+