feat(cli): agnes status = workspace state; old health check moves to agnes diagnose system

cli/commands/diagnose.py

@@ -5,17 +5,24 @@ import json
 import typer
 
 from cli.client import api_get
+from cli.config import get_sync_state
 
 diagnose_app = typer.Typer(help="System diagnostics")
 
 
 @diagnose_app.callback(invoke_without_command=True)
 def diagnose(
+    ctx: typer.Context,
     symptom: str = typer.Option(None, "--symptom", help="Describe the problem"),
     component: str = typer.Option(None, "--component", help="Check specific component"),
     as_json: bool = typer.Option(False, "--json", help="Output as JSON"),
 ):
     """Run comprehensive system diagnostics. AI-agent friendly output."""
+    # If a subcommand was invoked (e.g. `agnes diagnose system`), defer to it
+    # rather than running the default whole-system diagnostic.
+    if ctx.invoked_subcommand is not None:
+        return
+
     checks = []
 
     # 1. API reachability
@@ -79,3 +86,67 @@ def diagnose(
             typer.echo("\nSuggested actions:")
             for a in actions:
                 typer.echo(f"  - {a}")
+
+
+@diagnose_app.command("system")
+def system_status(
+    local: bool = typer.Option(False, "--local", help="Show local-only status (no server)"),
+    as_json: bool = typer.Option(False, "--json", help="Output as JSON"),
+):
+    """Show server-side health status (was `agnes status` pre-clean-bootstrap).
+
+    Reports server reachability and per-service health. Use `agnes status` for
+    workspace-side state (initialized? data fresh?).
+    """
+    if local:
+        state = get_sync_state()
+        info = {
+            "mode": "local",
+            "tables_synced": len(state.get("tables", {})),
+            "last_sync": state.get("last_sync", "never"),
+            "tables": state.get("tables", {}),
+        }
+        if as_json:
+            typer.echo(json.dumps(info, indent=2))
+        else:
+            typer.echo("Mode: offline (local data)")
+            typer.echo(f"Tables synced: {info['tables_synced']}")
+            typer.echo(f"Last sync: {info['last_sync']}")
+        return
+
+    try:
+        # Minimal health ping first
+        resp = api_get("/api/health")
+        minimal = resp.json()
+        if minimal.get("status") != "ok":
+            if as_json:
+                typer.echo(json.dumps(minimal, indent=2))
+            else:
+                typer.echo(f"Status: {minimal.get('status', 'unknown')}")
+            return
+
+        # Detailed health (auth required) for service-level info
+        try:
+            resp = api_get("/api/health/detailed")
+            data = resp.json()
+        except Exception:
+            data = minimal
+
+        if as_json:
+            typer.echo(json.dumps(data, indent=2))
+        else:
+            typer.echo(f"Status: {data.get('status', 'unknown')}")
+            for name, check in data.get("services", {}).items():
+                s = check.get("status", "?")
+                detail = ""
+                if "tables" in check:
+                    detail = f" ({check['tables']} tables, {check.get('total_rows', 0)} rows)"
+                if "count" in check:
+                    detail = f" ({check['count']})"
+                if check.get("stale_tables"):
+                    detail += f" [stale: {', '.join(check['stale_tables'])}]"
+                typer.echo(f"  {name}: {s}{detail}")
+    except Exception as e:
+        typer.echo(f"Cannot reach server: {e}", err=True)
+        typer.echo("Use --local for offline status.")
+        raise typer.Exit(1)

cli/commands/status.py

@@ -1,70 +1,68 @@
-"""Status commands — agnes status."""
+"""`agnes status` — workspace status: initialized? data fresh? hooks active?
+
+Replaces the old `da analyst status` command. The previous server-health
+content (which used to live here) has moved to `agnes diagnose system`
+under the existing `agnes diagnose` group (Task 13).
+"""
+
+from __future__ import annotations
 
 import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
 
 import typer
 
-from cli.client import api_get
-from cli.config import get_sync_state
 
-status_app = typer.Typer(help="System status")
+_INIT_MARKER = "AI Data Analyst"
+
+
+status_app = typer.Typer(help="Show workspace status (initialized? data fresh? hooks active?)")
 
 
 @status_app.callback(invoke_without_command=True)
 def status(
-    local: bool = typer.Option(False, "--local", help="Show local-only status (no server)"),
-    as_json: bool = typer.Option(False, "--json", help="Output as JSON"),
+    as_json: bool = typer.Option(False, "--json", help="Machine-readable output"),
 ):
-    """Show system health and sync status."""
-    if local:
-        state = get_sync_state()
-        info = {
-            "mode": "local",
-            "tables_synced": len(state.get("tables", {})),
-            "last_sync": state.get("last_sync", "never"),
-            "tables": state.get("tables", {}),
-        }
-        if as_json:
-            typer.echo(json.dumps(info, indent=2))
-        else:
-            typer.echo(f"Mode: offline (local data)")
-            typer.echo(f"Tables synced: {info['tables_synced']}")
-            typer.echo(f"Last sync: {info['last_sync']}")
+    workspace = Path(os.environ.get("AGNES_LOCAL_DIR", ".")).resolve()
+
+    initialized = False
+    claude_md = workspace / "CLAUDE.md"
+    if claude_md.exists():
+        initialized = _INIT_MARKER in claude_md.read_text(encoding="utf-8")
+
+    parquet_dir = workspace / "server" / "parquet"
+    parquets = list(parquet_dir.glob("*.parquet")) if parquet_dir.exists() else []
+
+    db_path = workspace / "user" / "duckdb" / "analytics.duckdb"
+    last_synced = None
+    if db_path.exists():
+        last_synced = datetime.fromtimestamp(db_path.stat().st_mtime, tz=timezone.utc).isoformat()
+
+    sessions_dir = workspace / "user" / "sessions"
+    session_count = len(list(sessions_dir.glob("*.jsonl"))) if sessions_dir.exists() else 0
+
+    info = {
+        "workspace": str(workspace),
+        "initialized": initialized,
+        "parquet_tables": len(parquets),
+        "duckdb_exists": db_path.exists(),
+        "last_synced": last_synced,
+        "sessions_pending_upload": session_count,
+    }
+
+    if as_json:
+        typer.echo(json.dumps(info, indent=2))
         return
 
-    try:
-        # Minimal health ping first
-        resp = api_get("/api/health")
-        minimal = resp.json()
-        if minimal.get("status") != "ok":
-            if as_json:
-                typer.echo(json.dumps(minimal, indent=2))
-            else:
-                typer.echo(f"Status: {minimal.get('status', 'unknown')}")
-            return
+    typer.echo(f"Workspace : {workspace}")
+    typer.echo(f"Initialized: {'yes' if initialized else 'no'}")
+    typer.echo(f"Parquets  : {info['parquet_tables']}")
+    typer.echo(f"DuckDB    : {'yes' if info['duckdb_exists'] else 'no'}")
+    typer.echo(f"Last sync : {last_synced or 'never'}")
+    typer.echo(f"Pending uploads: {session_count} sessions")
 
-        # Detailed health (auth required) for service-level info
-        try:
-            resp = api_get("/api/health/detailed")
-            data = resp.json()
-        except Exception:
-            data = minimal
-
-        if as_json:
-            typer.echo(json.dumps(data, indent=2))
-        else:
-            typer.echo(f"Status: {data.get('status', 'unknown')}")
-            for name, check in data.get("services", {}).items():
-                s = check.get("status", "?")
-                detail = ""
-                if "tables" in check:
-                    detail = f" ({check['tables']} tables, {check.get('total_rows', 0)} rows)"
-                if "count" in check:
-                    detail = f" ({check['count']})"
-                if check.get("stale_tables"):
-                    detail += f" [stale: {', '.join(check['stale_tables'])}]"
-                typer.echo(f"  {name}: {s}{detail}")
-    except Exception as e:
-        typer.echo(f"Cannot reach server: {e}", err=True)
-        typer.echo("Use --local for offline status.")
-        raise typer.Exit(1)
+    if not initialized:
+        typer.echo("")
+        typer.echo("Run `agnes init --server-url <URL> --token <PAT>` to bootstrap.")

tests/test_cli.py

@@ -144,13 +144,17 @@ class TestAuth:
 
 
 class TestStatus:
+    """Legacy `--local` status content moved to `agnes diagnose system` per the
+    clean-bootstrap spec (Tasks 12 + 13). `agnes status` itself now reports
+    workspace state — see tests/test_cli_status.py."""
+
     def test_local_status_empty(self):
-        result = runner.invoke(app, ["status", "--local"])
+        result = runner.invoke(app, ["diagnose", "system", "--local"])
         assert result.exit_code == 0
         assert "Tables synced: 0" in result.output
 
     def test_local_status_json(self):
-        result = runner.invoke(app, ["status", "--local", "--json"])
+        result = runner.invoke(app, ["diagnose", "system", "--local", "--json"])
         assert result.exit_code == 0
         data = json.loads(result.output)
         assert data["mode"] == "local"

tests/test_cli_diagnose_system.py

@@ -0,0 +1,27 @@
+"""Tests for `agnes diagnose system` (former `agnes status` content)."""
+
+from typer.testing import CliRunner
+from cli.commands.diagnose import diagnose_app
+
+runner = CliRunner()
+
+
+def test_diagnose_system_help():
+    result = runner.invoke(diagnose_app, ["system", "--help"])
+    assert result.exit_code == 0
+
+
+def test_diagnose_help_lists_system():
+    """Top-level diagnose help should mention the `system` subcommand."""
+    result = runner.invoke(diagnose_app, ["--help"])
+    assert result.exit_code == 0
+    assert "system" in result.output
+
+
+def test_diagnose_default_still_works():
+    """`agnes diagnose` (no subcommand) should still produce its existing output —
+    we only added a sibling subcommand, didn't change the default."""
+    result = runner.invoke(diagnose_app, [])
+    # Either runs successfully or fails for unrelated reasons (no server etc).
+    # We just want to verify no traceback from the addition.
+    assert "Traceback" not in (result.output + (result.stderr or ""))

tests/test_cli_status.py

@@ -0,0 +1,45 @@
+"""Tests for agnes status (workspace status)."""
+
+import json
+from pathlib import Path
+from typer.testing import CliRunner
+
+from cli.commands.status import status_app
+
+runner = CliRunner()
+
+
+def test_status_uninitialized_workspace(tmp_path, monkeypatch):
+    """Empty folder → exit 0, output indicates uninitialized state."""
+    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
+    result = runner.invoke(status_app)
+    assert result.exit_code in (0, 1)
+    out = result.output.lower()
+    assert "no" in out  # "Initialized: no" or similar
+    assert "agnes init" in result.output  # hint to initialize
+
+
+def test_status_initialized_workspace(tmp_path, monkeypatch):
+    """A bootstrapped workspace → 'initialized: yes' and shows parquet count."""
+    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
+    (tmp_path / "CLAUDE.md").write_text("# AI Data Analyst\n")
+    (tmp_path / "user" / "duckdb").mkdir(parents=True)
+    (tmp_path / "user" / "duckdb" / "analytics.duckdb").touch()
+    (tmp_path / "server" / "parquet").mkdir(parents=True)
+    (tmp_path / "server" / "parquet" / "tbl1.parquet").touch()
+
+    result = runner.invoke(status_app)
+    assert result.exit_code == 0
+    out = result.output.lower()
+    assert "yes" in out  # "Initialized: yes"
+    assert "1" in result.output  # one parquet
+
+
+def test_status_json(tmp_path, monkeypatch):
+    """--json flag returns machine-readable output."""
+    monkeypatch.setenv("AGNES_LOCAL_DIR", str(tmp_path))
+    (tmp_path / "CLAUDE.md").write_text("# AI Data Analyst\n")
+    result = runner.invoke(status_app, ["--json"])
+    assert result.exit_code == 0
+    body = json.loads(result.output)
+    assert "workspace" in body and "initialized" in body