feat: add linting, type checking, README badges, and Makefile

.github/workflows/ci.yml

@@ -1,6 +1,8 @@
 name: CI
 
 on:
+  push:
+    branches: [main]
   pull_request:
     branches: [main]
 
@@ -9,6 +11,45 @@ permissions:
   pull-requests: write
 
 jobs:
+  lint:
+    name: Lint & Format
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install ruff
+        run: pip install ruff>=0.8.0
+
+      - name: Ruff lint
+        run: ruff check .
+
+      - name: Ruff format check
+        run: ruff format --check .
+
+  type-check:
+    name: Type Check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Run mypy
+        run: mypy server/
+
   unit-tests:
     name: Unit Tests (Python ${{ matrix.python-version }})
     runs-on: ubuntu-latest

Makefile

@@ -0,0 +1,35 @@
+.DEFAULT_GOAL := help
+PYTHON := python
+PORT := 3000
+
+.PHONY: help up down test lint format typecheck check clean
+
+help: ## Show this help
+	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "  \033[36m%-15s\033[0m %s\n", $$1, $$2}'
+
+up: ## Start the server (port 3000)
+	uvicorn server.main:app --port $(PORT) --reload
+
+down: ## Stop the server
+	@pkill -f "uvicorn server.main:app" 2>/dev/null && echo "Server stopped" || echo "Server not running"
+
+test: ## Run all tests
+	$(PYTHON) -m pytest tests/ -v
+
+lint: ## Run ruff linter and format check
+	ruff check .
+	ruff format --check .
+
+format: ## Auto-format code with ruff
+	ruff check --fix .
+	ruff format .
+
+typecheck: ## Run mypy type checking
+	mypy server/
+
+check: lint typecheck test ## Run all checks (lint + typecheck + test)
+
+clean: ## Remove build artifacts and caches
+	rm -rf __pycache__ .pytest_cache .mypy_cache .ruff_cache reports/
+	find . -path ./.venv -prune -o -type d -name __pycache__ -print -exec rm -rf {} +
+	find . -path ./.venv -prune -o -type f -name '*.pyc' -print -delete

README.md

@@ -1,5 +1,11 @@
 # Claude Code Command Center
 
+[![CI](https://github.com/amahpour/claude-code-command-center/actions/workflows/ci.yml/badge.svg)](https://github.com/amahpour/claude-code-command-center/actions/workflows/ci.yml)
+[![Python 3.12+](https://img.shields.io/badge/python-3.12%2B-blue.svg)](https://www.python.org/downloads/)
+[![License: MIT](https://img.shields.io/github/license/amahpour/claude-code-command-center)](LICENSE)
+[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
+[![mypy](https://img.shields.io/badge/type--checked-mypy-blue.svg)](https://mypy-lang.org/)
+
 A web-based command center for monitoring and managing multiple Claude Code sessions in real-time.
 
 ![Dashboard](docs/dashboard.png)

pyproject.toml

@@ -19,8 +19,32 @@ dev = [
     "pytest-asyncio>=0.21.0",
     "httpx>=0.24.0",
     "playwright>=1.40.0",
+    "ruff>=0.8.0",
+    "mypy>=1.13.0",
 ]
 
+[tool.ruff]
+target-version = "py312"
+line-length = 120
+
+[tool.ruff.lint]
+select = ["E", "F", "W", "I", "UP", "B", "SIM"]
+ignore = [
+    "SIM105",  # contextlib.suppress — stylistic
+    "SIM117",  # nested with — often clearer in tests
+    "SIM114",  # combine if branches — stylistic
+    "E501",    # line length — handled by formatter
+]
+
+[tool.mypy]
+python_version = "3.12"
+warn_unused_configs = true
+disallow_untyped_defs = false
+check_untyped_defs = true
+ignore_missing_imports = true
+no_implicit_reexport = false
+warn_return_any = false
+
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]

requirements.txt

@@ -9,3 +9,5 @@ pytest-asyncio>=0.21.0
 httpx>=0.24.0
 playwright>=1.40.0
 pytest-cov>=4.1.0
+ruff>=0.8.0
+mypy>=1.13.0

scripts/hook-handler.py

@@ -7,8 +7,8 @@
 
 import json
 import sys
-import urllib.request
 import urllib.error
+import urllib.request
 
 SERVER_URL = "http://localhost:3000/api/hooks"
 TIMEOUT = 5
@@ -38,8 +38,7 @@ def main():
         )
         urllib.request.urlopen(req, timeout=TIMEOUT)
 
-    except (json.JSONDecodeError, urllib.error.URLError, urllib.error.HTTPError,
-            OSError, TimeoutError, Exception):
+    except (json.JSONDecodeError, urllib.error.URLError, urllib.error.HTTPError, OSError, TimeoutError, Exception):
         # Never block Claude Code — exit silently on any error
         pass
 

server/db.py

@@ -2,7 +2,7 @@
 
 import json
 import os
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 import aiosqlite
 
@@ -106,7 +106,14 @@ async def _create_tables(db: aiosqlite.Connection):
     """)
 
     # Add columns that may not exist in older databases
-    for col, col_type in [("session_name", "TEXT"), ("effort_level", "TEXT"), ("ticket_id", "TEXT"), ("display_name", "TEXT"), ("display_name_locked", "INTEGER DEFAULT 0"), ("last_activity_preview", "TEXT")]:
+    for col, col_type in [
+        ("session_name", "TEXT"),
+        ("effort_level", "TEXT"),
+        ("ticket_id", "TEXT"),
+        ("display_name", "TEXT"),
+        ("display_name_locked", "INTEGER DEFAULT 0"),
+        ("last_activity_preview", "TEXT"),
+    ]:
         try:
             await db.execute(f"ALTER TABLE sessions ADD COLUMN {col} {col_type}")
         except Exception:
@@ -129,23 +136,23 @@ async def _create_tables(db: aiosqlite.Connection):
 
 # --- Session CRUD ---
 
+
 async def create_session(
     session_id: str,
     project_path: str | None = None,
     model: str | None = None,
     task_description: str | None = None,
-) -> dict:
+) -> dict | None:
     """Create a new session record."""
     db = await get_db()
-    now = datetime.now(timezone.utc).isoformat()
+    now = datetime.now(UTC).isoformat()
     project_name = os.path.basename(project_path) if project_path else None
 
     await db.execute(
         """INSERT INTO sessions (id, project_path, project_name, model,
            task_description, status, started_at, last_activity_at, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, 'idle', ?, ?, ?, ?)""",
-        (session_id, project_path, project_name, model,
-         task_description, now, now, now, now),
+        (session_id, project_path, project_name, model, task_description, now, now, now, now),
     )
     await db.commit()
     return await get_session(session_id)
@@ -157,13 +164,11 @@ async def update_session(session_id: str, **kwargs) -> dict | None:
     if not kwargs:
         return await get_session(session_id)
 
-    kwargs["updated_at"] = datetime.now(timezone.utc).isoformat()
+    kwargs["updated_at"] = datetime.now(UTC).isoformat()
     set_clause = ", ".join(f"{k} = ?" for k in kwargs)
     values = list(kwargs.values()) + [session_id]
 
-    await db.execute(
-        f"UPDATE sessions SET {set_clause} WHERE id = ?", values
-    )
+    await db.execute(f"UPDATE sessions SET {set_clause} WHERE id = ?", values)
     await db.commit()
     return await get_session(session_id)
 
@@ -224,6 +229,7 @@ async def get_completed_sessions(limit: int = 50, offset: int = 0) -> list[dict]
 
 # --- Events ---
 
+
 async def add_event(
     session_id: str,
     event_type: str,
@@ -239,7 +245,7 @@ async def add_event(
         (session_id, event_type, tool_name, payload_json),
     )
     await db.commit()
-    return cursor.lastrowid
+    return cursor.lastrowid or 0
 
 
 async def get_session_events(session_id: str, limit: int = 100) -> list[dict]:
@@ -258,6 +264,7 @@ async def get_session_events(session_id: str, limit: int = 100) -> list[dict]:
 
 # --- Transcripts ---
 
+
 async def add_transcript(
     session_id: str,
     role: str,
@@ -268,13 +275,13 @@ async def add_transcript(
 ) -> int:
     """Add a transcript entry and update FTS index."""
     db = await get_db()
-    ts = timestamp or datetime.now(timezone.utc).isoformat()
+    ts = timestamp or datetime.now(UTC).isoformat()
     cursor = await db.execute(
         """INSERT INTO transcripts (session_id, source_file, role, content, token_count, timestamp)
            VALUES (?, ?, ?, ?, ?, ?)""",
         (session_id, source_file, role, content, token_count, ts),
     )
-    rowid = cursor.lastrowid
+    rowid = cursor.lastrowid or 0
 
     # Update FTS index
     await db.execute(
@@ -285,9 +292,7 @@ async def add_transcript(
     return rowid
 
 
-async def get_session_transcripts(
-    session_id: str, limit: int = 200, offset: int = 0
-) -> list[dict]:
+async def get_session_transcripts(session_id: str, limit: int = 200, offset: int = 0) -> list[dict]:
     """Get the latest N transcripts for a session, returned in chronological order."""
     db = await get_db()
     # Subquery gets the latest `limit` rows, outer query re-orders ASC for display
@@ -322,6 +327,7 @@ async def search_transcripts(query: str, limit: int = 50) -> list[dict]:
 
 # --- Analytics ---
 
+
 async def get_analytics_summary() -> dict:
     """Get overall analytics summary."""
     db = await get_db()
@@ -337,7 +343,7 @@ async def get_analytics_summary() -> dict:
         FROM sessions
     """)
     row = await cursor.fetchone()
-    summary = dict(row)
+    summary = dict(row) if row else {}
 
     # Today's cost
     cursor = await db.execute("""
@@ -346,7 +352,8 @@ async def get_analytics_summary() -> dict:
         WHERE date(created_at) = date('now')
     """)
     today = await cursor.fetchone()
-    summary["today_cost"] = today["today_cost"]
+    if today:
+        summary["today_cost"] = today["today_cost"]
 
     return summary
 
@@ -373,6 +380,7 @@ async def get_analytics_daily(days: int = 30) -> list[dict]:
 
 # --- Settings ---
 
+
 async def get_setting(key: str) -> str | None:
     """Get a single setting value by key."""
     db = await get_db()

server/hooks.py

@@ -5,7 +5,7 @@
 import logging
 import os
 import subprocess
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from server import db
 
@@ -74,7 +74,7 @@ async def process_hook_event(event_data: dict) -> dict | None:
         logger.warning("Hook event missing session_id: %s", event_type)
         return None
 
-    now = datetime.now(timezone.utc).isoformat()
+    now = datetime.now(UTC).isoformat()
     project_path = event_data.get("cwd") or event_data.get("project_path")
     session = await db.get_session(session_id)
 
@@ -95,7 +95,7 @@ async def process_hook_event(event_data: dict) -> dict | None:
 
     # Always update project info from cwd if we have it and session is missing it
     base_updates: dict = {"last_activity_at": now}
-    if project_path and not session.get("project_name"):
+    if project_path and session and not session.get("project_name"):
         base_updates["project_path"] = project_path
         base_updates["project_name"] = project_path.rsplit("/", 1)[-1] if "/" in project_path else project_path
         git_branch = _extract_git_branch(project_path)
@@ -142,10 +142,7 @@ async def process_hook_event(event_data: dict) -> dict | None:
                 base_updates["context_usage_percent"] = (event_data["context_tokens"] / max_ctx) * 100
         session = await db.update_session(session_id, **base_updates)
 
-    elif event_type == "SubagentStart":
-        session = await db.update_session(session_id, **base_updates)
-
-    elif event_type == "SubagentStop":
+    elif event_type == "SubagentStart" or event_type == "SubagentStop":
         session = await db.update_session(session_id, **base_updates)
 
     elif event_type == "Notification":
@@ -172,7 +169,7 @@ async def _check_stale_sessions():
         try:
             await asyncio.sleep(60)
             sessions = await db.get_all_active_sessions()
-            now = datetime.now(timezone.utc)
+            now = datetime.now(UTC)
             for session in sessions:
                 if session["status"] in ("completed", "stale"):
                     continue
@@ -182,7 +179,7 @@ async def _check_stale_sessions():
                 try:
                     last_dt = datetime.fromisoformat(last_activity.replace("Z", "+00:00"))
                     if last_dt.tzinfo is None:
-                        last_dt = last_dt.replace(tzinfo=timezone.utc)
+                        last_dt = last_dt.replace(tzinfo=UTC)
                     diff = (now - last_dt).total_seconds()
                     if diff > 300:  # 5 minutes
                         updated = await db.update_session(session["id"], status="stale")