diff --git a/docs/superpowers/plans/2026-05-25-opus-agent-plan.md b/docs/superpowers/plans/2026-05-25-opus-agent-plan.md new file mode 100644 index 0000000..933e62e --- /dev/null +++ b/docs/superpowers/plans/2026-05-25-opus-agent-plan.md @@ -0,0 +1,2905 @@ +# OpusAgent Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build a two-service system (FastAPI + Node.js worker) that proxies requests to Claude models via Claude Code SDK, using an existing Pro/Max subscription. + +**Architecture:** FastAPI service handles REST API, authentication, rate limiting, and request queuing via SQLite. A Node.js sidecar worker processes requests sequentially using the `@anthropic-ai/claude-code` SDK. Communication between services is internal HTTP over Docker network. + +**Tech Stack:** Python 3.11+, FastAPI, aiosqlite, Pydantic v2, slowapi, httpx, Node.js 20, @anthropic-ai/claude-code SDK, Docker + +**Spec:** `docs/superpowers/specs/2026-05-25-opus-agent-design.md` + +--- + +## File Map + +### Python Backend (`src/backend/`) + +| File | Responsibility | +|------|----------------| +| `config.py` | Pydantic Settings loaded from `.env` | +| `main.py` | FastAPI app, lifespan (init DB + start queue loop), router assembly | +| `api/dependencies.py` | API key verification, rate limiter setup | +| `api/routes/health.py` | `GET /api/health` | +| `api/routes/topics.py` | Topics CRUD + `POST /api/topics/{id}/summarize` | +| `api/routes/requests.py` | Requests CRUD | +| `api/routes/sessions.py` | Sessions list + delete | +| `models/api.py` | Pydantic models for all request/response bodies, envelope helpers | +| `db/database.py` | SQLite connection, schema DDL, init, all query functions | +| `services/worker_client.py` | Async HTTP client wrapping calls to Node.js worker | +| `services/queue.py` | Background queue loop: pick pending → call worker → update DB | + +### Node.js Worker (`worker/`) + +| File | Responsibility | +|------|----------------| +| `index.js` | HTTP server with `/process`, `/summarize`, `/health` endpoints | +| `package.json` | Dependencies: express, @anthropic-ai/claude-code | +| `Dockerfile` | Node.js 20 image with Claude Code SDK | + +### Tests (`tests/`) + +| File | Covers | +|------|--------| +| `conftest.py` | Shared fixtures: test DB, httpx client, env setup | +| `test_api_topics.py` | Topics CRUD endpoints | +| `test_api_requests.py` | Requests CRUD endpoints | +| `test_api_sessions.py` | Sessions endpoints | +| `test_services.py` | Queue processor, worker client, model resolution | + +### Root + +| File | Responsibility | +|------|----------------| +| `pyproject.toml` | Dependencies, scripts, metadata | +| `.python-version` | Python version for uv | +| `.env.example` | Template configuration | +| `.gitignore` | Standard Python + Node + Docker ignores | +| `Dockerfile` | Python backend image | +| `docker-compose.yml` | Both services + volumes | + +--- + +## Task 1: Project Scaffold + +Create the directory structure, dependency manifest, and configuration files. + +**Files:** +- Create: `pyproject.toml` +- Create: `.python-version` +- Create: `.env.example` +- Create: `.gitignore` +- Create: all `__init__.py` files + +- [ ] **Step 1: Create pyproject.toml** + +```toml +[project] +name = "opus-agent" +version = "0.1.0" +description = "Self-hosted proxy to Claude models via Claude Code SDK" +requires-python = ">=3.11" +dependencies = [ + "fastapi>=0.115", + "uvicorn[standard]>=0.34", + "pydantic>=2.0", + "pydantic-settings>=2.0", + "python-dotenv>=1.0", + "aiosqlite>=0.20", + "httpx>=0.27", + "slowapi>=0.1.9", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "pytest-asyncio>=0.24", +] + +[project.scripts] +start = "uvicorn src.backend.main:app --host 0.0.0.0 --port 8000" +dev = "uvicorn src.backend.main:app --reload --host 0.0.0.0 --port 8000" +``` + +- [ ] **Step 2: Create .python-version** + +``` +3.11 +``` + +- [ ] **Step 3: Create .env.example** + +``` +# Server +API_HOST=0.0.0.0 +API_PORT=8000 + +# Sicurezza +API_KEY= + +# Rate Limiting +RATE_LIMIT_PER_MINUTE=10 +RATE_LIMIT_PER_HOUR=100 + +# Worker +WORKER_URL=http://claude-worker:3001 + +# Database +DB_PATH=/data/opus-agent.db + +# Claude Code +CLAUDE_MODEL=claude-sonnet-4-6 +``` + +- [ ] **Step 4: Create .gitignore** + +```gitignore +__pycache__/ +*.py[cod] +*.egg-info/ +dist/ +build/ +.venv/ +.env +!.env.example +.vscode/ +.idea/ +.DS_Store +Thumbs.db +node_modules/ +*.log +``` + +- [ ] **Step 5: Create directory structure with __init__.py files** + +```bash +mkdir -p src/backend/api/routes src/backend/models src/backend/db src/backend/services tests worker +touch src/__init__.py src/backend/__init__.py src/backend/api/__init__.py src/backend/api/routes/__init__.py src/backend/models/__init__.py src/backend/db/__init__.py src/backend/services/__init__.py tests/__init__.py +``` + +- [ ] **Step 6: Install dependencies and create .env for local dev** + +```bash +uv sync --all-groups +cp .env.example .env +# Edit .env: set API_KEY=dev-key-12345, DB_PATH=./data/opus-agent.db +mkdir -p data +``` + +- [ ] **Step 7: Commit** + +```bash +git add pyproject.toml .python-version .env.example .gitignore src/ tests/ +git commit -m "feat: project scaffold with dependencies and directory structure" +``` + +--- + +## Task 2: Configuration + +**Files:** +- Create: `src/backend/config.py` +- Create: `tests/conftest.py` +- Create: `tests/test_services.py` + +- [ ] **Step 1: Write the failing test** + +Create `tests/conftest.py` with env setup (this must be at the top, before any app imports): + +```python +import os +import tempfile + +_test_dir = tempfile.mkdtemp() + +os.environ.setdefault("API_KEY", "test-key-12345") +os.environ.setdefault("DB_PATH", os.path.join(_test_dir, "test.db")) +os.environ.setdefault("WORKER_URL", "http://localhost:3001") +os.environ.setdefault("CLAUDE_MODEL", "claude-sonnet-4-6") +os.environ.setdefault("RATE_LIMIT_PER_MINUTE", "100") +os.environ.setdefault("RATE_LIMIT_PER_HOUR", "1000") +os.environ.setdefault("API_HOST", "0.0.0.0") +os.environ.setdefault("API_PORT", "8000") +``` + +Create `tests/test_services.py`: + +```python +from src.backend.config import settings + + +def test_settings_loads_api_key(): + assert settings.api_key == "test-key-12345" + + +def test_settings_loads_db_path(): + assert "test.db" in settings.db_path + + +def test_settings_loads_worker_url(): + assert settings.worker_url == "http://localhost:3001" + + +def test_settings_loads_claude_model(): + assert settings.claude_model == "claude-sonnet-4-6" + + +def test_settings_loads_rate_limits(): + assert settings.rate_limit_per_minute == 100 + assert settings.rate_limit_per_hour == 1000 +``` + +- [ ] **Step 2: Run test to verify it fails** + +```bash +uv run pytest tests/test_services.py -v +``` + +Expected: FAIL with `ModuleNotFoundError: No module named 'src.backend.config'` + +- [ ] **Step 3: Write config.py** + +```python +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + api_host: str = "0.0.0.0" + api_port: int = 8000 + + api_key: str + + rate_limit_per_minute: int = 10 + rate_limit_per_hour: int = 100 + + worker_url: str = "http://claude-worker:3001" + + db_path: str = "/data/opus-agent.db" + + claude_model: str = "claude-sonnet-4-6" + + model_config = { + "env_file": ".env", + "env_file_encoding": "utf-8", + } + + +settings = Settings() +``` + +- [ ] **Step 4: Run test to verify it passes** + +```bash +uv run pytest tests/test_services.py -v +``` + +Expected: all 5 tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/backend/config.py tests/conftest.py tests/test_services.py +git commit -m "feat: add Pydantic Settings configuration" +``` + +--- + +## Task 3: Database Layer + +**Files:** +- Create: `src/backend/db/database.py` +- Modify: `tests/test_services.py` +- Modify: `tests/conftest.py` + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_services.py`: + +```python +import pytest + + +@pytest.mark.asyncio +async def test_init_db_creates_tables(): + from src.backend.db.database import init_db, get_db + + await init_db() + db = await get_db() + try: + cursor = await db.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" + ) + tables = [row[0] for row in await cursor.fetchall()] + assert "topics" in tables + assert "requests" in tables + finally: + await db.close() + + +@pytest.mark.asyncio +async def test_foreign_keys_enabled(): + from src.backend.db.database import init_db, get_db + + await init_db() + db = await get_db() + try: + cursor = await db.execute("PRAGMA foreign_keys") + row = await cursor.fetchone() + assert row[0] == 1 + finally: + await db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +```bash +uv run pytest tests/test_services.py::test_init_db_creates_tables -v +``` + +Expected: FAIL with `ModuleNotFoundError: No module named 'src.backend.db.database'` + +- [ ] **Step 3: Write database.py** + +```python +import aiosqlite + +from src.backend.config import settings + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS topics ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL UNIQUE, + system_prompt TEXT NOT NULL, + summary TEXT DEFAULT '', + model TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS requests ( + id TEXT PRIMARY KEY, + session_id TEXT, + topic_id TEXT NOT NULL REFERENCES topics(id), + prompt TEXT NOT NULL, + model TEXT, + summarize INTEGER DEFAULT 0, + status TEXT NOT NULL DEFAULT 'pending', + result TEXT, + error TEXT, + created_at TEXT NOT NULL, + completed_at TEXT +); + +CREATE INDEX IF NOT EXISTS idx_requests_status ON requests(status); +CREATE INDEX IF NOT EXISTS idx_requests_topic_id ON requests(topic_id); +CREATE INDEX IF NOT EXISTS idx_requests_session_id ON requests(session_id); +""" + + +async def get_db() -> aiosqlite.Connection: + db = await aiosqlite.connect(settings.db_path) + db.row_factory = aiosqlite.Row + await db.execute("PRAGMA journal_mode=WAL") + await db.execute("PRAGMA foreign_keys=ON") + return db + + +async def init_db(): + db = await get_db() + try: + await db.executescript(SCHEMA) + await db.commit() + finally: + await db.close() +``` + +- [ ] **Step 4: Run test to verify it passes** + +```bash +uv run pytest tests/test_services.py -v +``` + +Expected: all tests PASS + +- [ ] **Step 5: Add DB init fixture to conftest.py** + +Append to `tests/conftest.py`: + +```python +import pytest + + +@pytest.fixture(autouse=True) +async def setup_db(): + from src.backend.db.database import init_db, get_db + + await init_db() + yield + db = await get_db() + try: + await db.execute("DELETE FROM requests") + await db.execute("DELETE FROM topics") + await db.commit() + finally: + await db.close() +``` + +- [ ] **Step 6: Commit** + +```bash +git add src/backend/db/database.py tests/conftest.py tests/test_services.py +git commit -m "feat: add SQLite database layer with schema init" +``` + +--- + +## Task 4: Pydantic API Models + +**Files:** +- Create: `src/backend/models/api.py` +- Modify: `tests/test_services.py` + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_services.py`: + +```python +def test_topic_create_model_validates(): + from src.backend.models.api import TopicCreate + + topic = TopicCreate( + name="coding", + system_prompt="You are an expert coder.", + summary="Coding questions", + model="claude-sonnet-4-6", + ) + assert topic.name == "coding" + assert topic.model == "claude-sonnet-4-6" + + +def test_topic_create_model_optional_fields(): + from src.backend.models.api import TopicCreate + + topic = TopicCreate(name="coding", system_prompt="You are an expert coder.") + assert topic.summary == "" + assert topic.model is None + + +def test_request_create_model_validates(): + from src.backend.models.api import RequestCreate + + req = RequestCreate( + prompt="How to do X?", + topic_id="550e8400-e29b-41d4-a716-446655440000", + ) + assert req.session_id is None + assert req.model is None + assert req.summarize is False + + +def test_model_validation_rejects_invalid(): + from src.backend.models.api import TopicCreate + + topic = TopicCreate( + name="test", system_prompt="test", model="gpt-4" + ) + # model validation: must start with "claude-" or be None + # gpt-4 should fail + from pydantic import ValidationError + import pytest as pt + + with pt.raises(ValidationError): + TopicCreate(name="test", system_prompt="test", model="gpt-4") + + +def test_envelope_ok(): + from src.backend.models.api import ok + + result = ok({"key": "value"}) + assert result == {"success": True, "data": {"key": "value"}, "error": None} + + +def test_envelope_fail(): + from src.backend.models.api import fail + + result = fail("NOT_FOUND", "Resource not found") + assert result["success"] is False + assert result["error"]["code"] == "NOT_FOUND" +``` + +- [ ] **Step 2: Run tests to verify they fail** + +```bash +uv run pytest tests/test_services.py::test_topic_create_model_validates -v +``` + +Expected: FAIL with `ModuleNotFoundError` + +- [ ] **Step 3: Write models/api.py** + +```python +from pydantic import BaseModel, field_validator +from typing import Any + + +class ApiError(BaseModel): + code: str + message: str + + +class ApiResponse(BaseModel): + success: bool + data: Any | None = None + error: ApiError | None = None + + +def ok(data: Any) -> dict: + return {"success": True, "data": data, "error": None} + + +def fail(code: str, message: str) -> dict: + return { + "success": False, + "data": None, + "error": {"code": code, "message": message}, + } + + +def _validate_model(v: str | None) -> str | None: + if v is not None and not v.startswith("claude-"): + raise ValueError("Model must start with 'claude-' or be null") + return v + + +class TopicCreate(BaseModel): + name: str + system_prompt: str + summary: str = "" + model: str | None = None + + validate_model = field_validator("model")(_validate_model) + + +class TopicUpdate(BaseModel): + name: str | None = None + system_prompt: str | None = None + summary: str | None = None + model: str | None = None + + validate_model = field_validator("model")(_validate_model) + + +class RequestCreate(BaseModel): + prompt: str + topic_id: str + session_id: str | None = None + model: str | None = None + summarize: bool = False + + validate_model = field_validator("model")(_validate_model) +``` + +- [ ] **Step 4: Fix test — model validation test needs correction** + +The test `test_model_validation_rejects_invalid` defines `topic` before the `with raises` block. Fix it: + +```python +def test_model_validation_rejects_invalid(): + from src.backend.models.api import TopicCreate + from pydantic import ValidationError + + with pytest.raises(ValidationError): + TopicCreate(name="test", system_prompt="test", model="gpt-4") +``` + +(Remove the duplicate line before the `with` block.) + +- [ ] **Step 5: Run tests to verify they pass** + +```bash +uv run pytest tests/test_services.py -v +``` + +Expected: all tests PASS + +- [ ] **Step 6: Commit** + +```bash +git add src/backend/models/api.py tests/test_services.py +git commit -m "feat: add Pydantic API models with envelope helpers" +``` + +--- + +## Task 5: API Key Authentication + +**Files:** +- Create: `src/backend/api/dependencies.py` +- Create: `tests/test_api_topics.py` (starter with auth tests) + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_api_topics.py`: + +```python +import pytest +from httpx import AsyncClient, ASGITransport + + +@pytest.fixture +def app(): + from src.backend.main import app + + return app + + +@pytest.fixture +async def client(app): + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-Api-Key": "test-key-12345"}, + ) as ac: + yield ac + + +@pytest.fixture +async def client_no_auth(app): + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + yield ac + + +@pytest.mark.asyncio +async def test_request_without_api_key_returns_403(client_no_auth): + resp = await client_no_auth.get("/api/health") + assert resp.status_code == 403 + + +@pytest.mark.asyncio +async def test_request_with_wrong_api_key_returns_403(app): + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-Api-Key": "wrong-key"}, + ) as ac: + resp = await ac.get("/api/health") + assert resp.status_code == 403 +``` + +- [ ] **Step 2: Run test to verify it fails** + +```bash +uv run pytest tests/test_api_topics.py::test_request_without_api_key_returns_403 -v +``` + +Expected: FAIL (main.py doesn't exist yet) + +- [ ] **Step 3: Write dependencies.py** + +```python +from fastapi import Header, HTTPException + +from src.backend.config import settings + + +async def verify_api_key(x_api_key: str = Header(...)): + if x_api_key != settings.api_key: + raise HTTPException(status_code=403, detail="API key non valida") +``` + +- [ ] **Step 4: Create minimal main.py to make auth tests pass** + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from src.backend.db.database import init_db + + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + yield + + +app = FastAPI(title="OpusAgent", version="0.1.0", lifespan=lifespan) +``` + +- [ ] **Step 5: Create minimal health route** + +Create `src/backend/api/routes/health.py`: + +```python +from fastapi import APIRouter, Depends + +from src.backend.api.dependencies import verify_api_key +from src.backend.models.api import ok + +router = APIRouter(dependencies=[Depends(verify_api_key)]) + + +@router.get("/health", summary="Health check", tags=["system"]) +async def health(): + return ok({"status": "running"}) +``` + +- [ ] **Step 6: Register health router in main.py** + +Update `src/backend/main.py`: + +```python +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from src.backend.api.routes.health import router as health_router +from src.backend.db.database import init_db + + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + yield + + +app = FastAPI(title="OpusAgent", version="0.1.0", lifespan=lifespan) + +app.include_router(health_router, prefix="/api") +``` + +- [ ] **Step 7: Run tests to verify they pass** + +```bash +uv run pytest tests/test_api_topics.py -v +``` + +Expected: all 2 auth tests PASS + +- [ ] **Step 8: Commit** + +```bash +git add src/backend/api/dependencies.py src/backend/api/routes/health.py src/backend/main.py tests/test_api_topics.py +git commit -m "feat: add API key authentication and health endpoint" +``` + +--- + +## Task 6: Topics CRUD + +**Files:** +- Modify: `src/backend/db/database.py` (add topic query functions) +- Create: `src/backend/api/routes/topics.py` +- Modify: `src/backend/main.py` (register topics router) +- Modify: `tests/test_api_topics.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_api_topics.py`: + +```python +@pytest.mark.asyncio +async def test_create_topic(client): + resp = await client.post( + "/api/topics", + json={ + "name": "coding", + "system_prompt": "You are an expert coder.", + "summary": "Coding topic", + "model": "claude-sonnet-4-6", + }, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["success"] is True + assert data["data"]["name"] == "coding" + assert data["data"]["model"] == "claude-sonnet-4-6" + assert "id" in data["data"] + + +@pytest.mark.asyncio +async def test_create_topic_minimal(client): + resp = await client.post( + "/api/topics", + json={"name": "analysis", "system_prompt": "You analyze data."}, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["data"]["summary"] == "" + assert data["data"]["model"] is None + + +@pytest.mark.asyncio +async def test_create_topic_duplicate_name_fails(client): + await client.post( + "/api/topics", + json={"name": "coding", "system_prompt": "Prompt 1"}, + ) + resp = await client.post( + "/api/topics", + json={"name": "coding", "system_prompt": "Prompt 2"}, + ) + assert resp.status_code == 409 + + +@pytest.mark.asyncio +async def test_list_topics(client): + await client.post( + "/api/topics", json={"name": "a", "system_prompt": "A"} + ) + await client.post( + "/api/topics", json={"name": "b", "system_prompt": "B"} + ) + resp = await client.get("/api/topics") + assert resp.status_code == 200 + data = resp.json() + assert len(data["data"]) == 2 + + +@pytest.mark.asyncio +async def test_get_topic(client): + create_resp = await client.post( + "/api/topics", json={"name": "coding", "system_prompt": "Code"} + ) + topic_id = create_resp.json()["data"]["id"] + + resp = await client.get(f"/api/topics/{topic_id}") + assert resp.status_code == 200 + assert resp.json()["data"]["name"] == "coding" + + +@pytest.mark.asyncio +async def test_get_topic_not_found(client): + resp = await client.get("/api/topics/nonexistent-id") + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_update_topic(client): + create_resp = await client.post( + "/api/topics", json={"name": "coding", "system_prompt": "Old"} + ) + topic_id = create_resp.json()["data"]["id"] + + resp = await client.put( + f"/api/topics/{topic_id}", + json={"system_prompt": "New prompt"}, + ) + assert resp.status_code == 200 + assert resp.json()["data"]["system_prompt"] == "New prompt" + assert resp.json()["data"]["name"] == "coding" + + +@pytest.mark.asyncio +async def test_delete_topic(client): + create_resp = await client.post( + "/api/topics", json={"name": "coding", "system_prompt": "Code"} + ) + topic_id = create_resp.json()["data"]["id"] + + resp = await client.delete(f"/api/topics/{topic_id}") + assert resp.status_code == 200 + + resp = await client.get(f"/api/topics/{topic_id}") + assert resp.status_code == 404 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +```bash +uv run pytest tests/test_api_topics.py::test_create_topic -v +``` + +Expected: FAIL (404 — no topics route registered) + +- [ ] **Step 3: Add topic query functions to database.py** + +Append to `src/backend/db/database.py`: + +```python +import uuid +from datetime import datetime, timezone + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _new_id() -> str: + return str(uuid.uuid4()) + + +async def create_topic( + db: aiosqlite.Connection, + name: str, + system_prompt: str, + summary: str = "", + model: str | None = None, +) -> dict: + topic_id = _new_id() + now = _now() + await db.execute( + "INSERT INTO topics (id, name, system_prompt, summary, model, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (topic_id, name, system_prompt, summary, model, now, now), + ) + await db.commit() + return { + "id": topic_id, + "name": name, + "system_prompt": system_prompt, + "summary": summary, + "model": model, + "created_at": now, + "updated_at": now, + } + + +async def list_topics(db: aiosqlite.Connection) -> list[dict]: + cursor = await db.execute("SELECT * FROM topics ORDER BY created_at DESC") + rows = await cursor.fetchall() + return [dict(row) for row in rows] + + +async def get_topic(db: aiosqlite.Connection, topic_id: str) -> dict | None: + cursor = await db.execute("SELECT * FROM topics WHERE id = ?", (topic_id,)) + row = await cursor.fetchone() + return dict(row) if row else None + + +async def update_topic( + db: aiosqlite.Connection, topic_id: str, **fields +) -> dict | None: + existing = await get_topic(db, topic_id) + if not existing: + return None + fields = {k: v for k, v in fields.items() if v is not None} + if not fields: + return existing + fields["updated_at"] = _now() + set_clause = ", ".join(f"{k} = ?" for k in fields) + values = list(fields.values()) + [topic_id] + await db.execute( + f"UPDATE topics SET {set_clause} WHERE id = ?", values + ) + await db.commit() + return await get_topic(db, topic_id) + + +async def delete_topic(db: aiosqlite.Connection, topic_id: str) -> bool: + cursor = await db.execute( + "SELECT COUNT(*) FROM requests WHERE topic_id = ? AND status IN ('pending', 'processing')", + (topic_id,), + ) + row = await cursor.fetchone() + if row[0] > 0: + return False + cursor = await db.execute("DELETE FROM topics WHERE id = ?", (topic_id,)) + await db.commit() + return cursor.rowcount > 0 +``` + +- [ ] **Step 4: Create routes/topics.py** + +```python +from fastapi import APIRouter, Depends +from starlette.responses import JSONResponse + +from src.backend.api.dependencies import verify_api_key +from src.backend.db.database import ( + get_db, + create_topic, + list_topics, + get_topic, + update_topic, + delete_topic, +) +from src.backend.models.api import TopicCreate, TopicUpdate, ok, fail + +router = APIRouter( + prefix="/topics", + tags=["topics"], + dependencies=[Depends(verify_api_key)], +) + + +async def _get_db(): + db = await get_db() + try: + yield db + finally: + await db.close() + + +@router.post("", summary="Create topic") +async def create(body: TopicCreate, db=Depends(_get_db)): + try: + topic = await create_topic( + db, + name=body.name, + system_prompt=body.system_prompt, + summary=body.summary, + model=body.model, + ) + except Exception as e: + if "UNIQUE constraint failed" in str(e): + return JSONResponse( + status_code=409, + content=fail("DUPLICATE", f"Topic '{body.name}' already exists"), + ) + raise + return JSONResponse(status_code=201, content=ok(topic)) + + +@router.get("", summary="List topics") +async def list_all(db=Depends(_get_db)): + topics = await list_topics(db) + return ok(topics) + + +@router.get("/{topic_id}", summary="Get topic") +async def get_one(topic_id: str, db=Depends(_get_db)): + topic = await get_topic(db, topic_id) + if not topic: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found"), + ) + return ok(topic) + + +@router.put("/{topic_id}", summary="Update topic") +async def update(topic_id: str, body: TopicUpdate, db=Depends(_get_db)): + topic = await update_topic( + db, + topic_id, + name=body.name, + system_prompt=body.system_prompt, + summary=body.summary, + model=body.model, + ) + if not topic: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found"), + ) + return ok(topic) + + +@router.delete("/{topic_id}", summary="Delete topic") +async def delete(topic_id: str, db=Depends(_get_db)): + deleted = await delete_topic(db, topic_id) + if not deleted: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found or has active requests"), + ) + return ok({"deleted": True}) +``` + +- [ ] **Step 5: Register topics router in main.py** + +Add to `src/backend/main.py` imports: + +```python +from src.backend.api.routes.topics import router as topics_router +``` + +Add after health router: + +```python +app.include_router(topics_router, prefix="/api") +``` + +- [ ] **Step 6: Run tests to verify they pass** + +```bash +uv run pytest tests/test_api_topics.py -v +``` + +Expected: all 10 tests PASS + +- [ ] **Step 7: Commit** + +```bash +git add src/backend/db/database.py src/backend/api/routes/topics.py src/backend/main.py tests/test_api_topics.py +git commit -m "feat: add topics CRUD endpoints" +``` + +--- + +## Task 7: Requests CRUD + +**Files:** +- Modify: `src/backend/db/database.py` (add request query functions) +- Create: `src/backend/api/routes/requests.py` +- Modify: `src/backend/main.py` +- Create: `tests/test_api_requests.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_api_requests.py`: + +```python +import pytest +from httpx import AsyncClient, ASGITransport + + +@pytest.fixture +def app(): + from src.backend.main import app + + return app + + +@pytest.fixture +async def client(app): + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-Api-Key": "test-key-12345"}, + ) as ac: + yield ac + + +@pytest.fixture +async def topic_id(client): + resp = await client.post( + "/api/topics", + json={"name": "test-topic", "system_prompt": "You are helpful."}, + ) + return resp.json()["data"]["id"] + + +@pytest.mark.asyncio +async def test_create_request(client, topic_id): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + assert resp.status_code == 202 + data = resp.json() + assert data["success"] is True + assert data["data"]["status"] == "pending" + assert "id" in data["data"] + + +@pytest.mark.asyncio +async def test_create_request_with_new_session(client, topic_id): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id, "session_id": "new"}, + ) + assert resp.status_code == 202 + data = resp.json() + assert data["data"]["session_id"] is not None + assert data["data"]["session_id"] != "new" + + +@pytest.mark.asyncio +async def test_create_request_invalid_topic_fails(client): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": "nonexistent"}, + ) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_get_request(client, topic_id): + create_resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + req_id = create_resp.json()["data"]["id"] + + resp = await client.get(f"/api/requests/{req_id}") + assert resp.status_code == 200 + assert resp.json()["data"]["prompt"] == "Hello" + + +@pytest.mark.asyncio +async def test_list_requests_with_filters(client, topic_id): + await client.post( + "/api/requests", + json={"prompt": "First", "topic_id": topic_id}, + ) + await client.post( + "/api/requests", + json={"prompt": "Second", "topic_id": topic_id}, + ) + + resp = await client.get("/api/requests") + assert len(resp.json()["data"]) == 2 + + resp = await client.get(f"/api/requests?topic_id={topic_id}") + assert len(resp.json()["data"]) == 2 + + resp = await client.get("/api/requests?status=completed") + assert len(resp.json()["data"]) == 0 + + +@pytest.mark.asyncio +async def test_delete_pending_request(client, topic_id): + create_resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + req_id = create_resp.json()["data"]["id"] + + resp = await client.delete(f"/api/requests/{req_id}") + assert resp.status_code == 200 + + +@pytest.mark.asyncio +async def test_delete_nonexistent_request_fails(client): + resp = await client.delete("/api/requests/nonexistent") + assert resp.status_code == 404 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +```bash +uv run pytest tests/test_api_requests.py::test_create_request -v +``` + +Expected: FAIL (404 — no requests route) + +- [ ] **Step 3: Add request query functions to database.py** + +Append to `src/backend/db/database.py`: + +```python +async def create_request( + db: aiosqlite.Connection, + prompt: str, + topic_id: str, + session_id: str | None = None, + model: str | None = None, + summarize: bool = False, +) -> dict: + topic = await get_topic(db, topic_id) + if not topic: + return None + + request_id = _new_id() + now = _now() + + if session_id == "new": + session_id = _new_id() + + await db.execute( + "INSERT INTO requests (id, session_id, topic_id, prompt, model, summarize, status, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)", + (request_id, session_id, topic_id, prompt, model, int(summarize), now), + ) + await db.commit() + return { + "id": request_id, + "session_id": session_id, + "topic_id": topic_id, + "prompt": prompt, + "model": model, + "summarize": bool(summarize), + "status": "pending", + "result": None, + "error": None, + "created_at": now, + "completed_at": None, + } + + +async def get_request(db: aiosqlite.Connection, request_id: str) -> dict | None: + cursor = await db.execute("SELECT * FROM requests WHERE id = ?", (request_id,)) + row = await cursor.fetchone() + if not row: + return None + d = dict(row) + d["summarize"] = bool(d["summarize"]) + return d + + +async def list_requests( + db: aiosqlite.Connection, + status: str | None = None, + topic_id: str | None = None, + session_id: str | None = None, +) -> list[dict]: + query = "SELECT * FROM requests WHERE 1=1" + params = [] + if status: + query += " AND status = ?" + params.append(status) + if topic_id: + query += " AND topic_id = ?" + params.append(topic_id) + if session_id: + query += " AND session_id = ?" + params.append(session_id) + query += " ORDER BY created_at DESC" + cursor = await db.execute(query, params) + rows = await cursor.fetchall() + result = [] + for row in rows: + d = dict(row) + d["summarize"] = bool(d["summarize"]) + result.append(d) + return result + + +async def delete_request(db: aiosqlite.Connection, request_id: str) -> bool: + cursor = await db.execute( + "DELETE FROM requests WHERE id = ? AND status = 'pending'", + (request_id,), + ) + await db.commit() + return cursor.rowcount > 0 + + +async def pick_next_pending(db: aiosqlite.Connection) -> dict | None: + cursor = await db.execute( + "SELECT * FROM requests WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1" + ) + row = await cursor.fetchone() + if not row: + return None + d = dict(row) + d["summarize"] = bool(d["summarize"]) + await db.execute( + "UPDATE requests SET status = 'processing' WHERE id = ?", (d["id"],) + ) + await db.commit() + return d + + +async def complete_request( + db: aiosqlite.Connection, + request_id: str, + result: str, + session_id: str | None = None, +) -> None: + now = _now() + if session_id: + await db.execute( + "UPDATE requests SET status = 'completed', result = ?, completed_at = ?, session_id = ? WHERE id = ?", + (result, now, session_id, request_id), + ) + else: + await db.execute( + "UPDATE requests SET status = 'completed', result = ?, completed_at = ? WHERE id = ?", + (result, now, request_id), + ) + await db.commit() + + +async def fail_request( + db: aiosqlite.Connection, request_id: str, error: str +) -> None: + now = _now() + await db.execute( + "UPDATE requests SET status = 'failed', error = ?, completed_at = ? WHERE id = ?", + (error, now, request_id), + ) + await db.commit() + + +async def append_topic_summary( + db: aiosqlite.Connection, topic_id: str, summary_line: str +) -> None: + now = _now() + date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") + line = f"[{date_str}] {summary_line}" + await db.execute( + "UPDATE topics SET summary = CASE WHEN summary = '' THEN ? ELSE summary || char(10) || ? END, updated_at = ? WHERE id = ?", + (line, line, now, topic_id), + ) + await db.commit() +``` + +- [ ] **Step 4: Create routes/requests.py** + +```python +from fastapi import APIRouter, Depends, Query +from starlette.responses import JSONResponse + +from src.backend.api.dependencies import verify_api_key +from src.backend.db.database import ( + get_db, + create_request, + get_request, + list_requests, + delete_request, +) +from src.backend.models.api import RequestCreate, ok, fail + +router = APIRouter( + prefix="/requests", + tags=["requests"], + dependencies=[Depends(verify_api_key)], +) + + +async def _get_db(): + db = await get_db() + try: + yield db + finally: + await db.close() + + +@router.post("", summary="Create request") +async def create(body: RequestCreate, db=Depends(_get_db)): + req = await create_request( + db, + prompt=body.prompt, + topic_id=body.topic_id, + session_id=body.session_id, + model=body.model, + summarize=body.summarize, + ) + if req is None: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found"), + ) + return JSONResponse( + status_code=202, + content=ok({ + "id": req["id"], + "session_id": req["session_id"], + "status": req["status"], + "created_at": req["created_at"], + }), + ) + + +@router.get("", summary="List requests") +async def list_all( + status: str | None = Query(None), + topic_id: str | None = Query(None), + session_id: str | None = Query(None), + db=Depends(_get_db), +): + reqs = await list_requests(db, status=status, topic_id=topic_id, session_id=session_id) + return ok(reqs) + + +@router.get("/{request_id}", summary="Get request") +async def get_one(request_id: str, db=Depends(_get_db)): + req = await get_request(db, request_id) + if not req: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Request not found"), + ) + return ok(req) + + +@router.delete("/{request_id}", summary="Delete request") +async def delete(request_id: str, db=Depends(_get_db)): + deleted = await delete_request(db, request_id) + if not deleted: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Request not found or not in pending status"), + ) + return ok({"deleted": True}) +``` + +- [ ] **Step 5: Register requests router in main.py** + +Add import: + +```python +from src.backend.api.routes.requests import router as requests_router +``` + +Add router: + +```python +app.include_router(requests_router, prefix="/api") +``` + +- [ ] **Step 6: Run tests to verify they pass** + +```bash +uv run pytest tests/test_api_requests.py -v +``` + +Expected: all 7 tests PASS + +- [ ] **Step 7: Commit** + +```bash +git add src/backend/db/database.py src/backend/api/routes/requests.py src/backend/main.py tests/test_api_requests.py +git commit -m "feat: add requests CRUD endpoints" +``` + +--- + +## Task 8: Sessions Endpoints + +**Files:** +- Modify: `src/backend/db/database.py` +- Create: `src/backend/api/routes/sessions.py` +- Modify: `src/backend/main.py` +- Create: `tests/test_api_sessions.py` + +- [ ] **Step 1: Write the failing tests** + +Create `tests/test_api_sessions.py`: + +```python +import pytest +from httpx import AsyncClient, ASGITransport + + +@pytest.fixture +def app(): + from src.backend.main import app + + return app + + +@pytest.fixture +async def client(app): + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-Api-Key": "test-key-12345"}, + ) as ac: + yield ac + + +@pytest.fixture +async def topic_id(client): + resp = await client.post( + "/api/topics", + json={"name": "sessions-test", "system_prompt": "Test prompt."}, + ) + return resp.json()["data"]["id"] + + +@pytest.mark.asyncio +async def test_list_sessions_empty(client): + resp = await client.get("/api/sessions") + assert resp.status_code == 200 + assert resp.json()["data"] == [] + + +@pytest.mark.asyncio +async def test_list_sessions_with_data(client, topic_id): + resp1 = await client.post( + "/api/requests", + json={"prompt": "First", "topic_id": topic_id, "session_id": "new"}, + ) + session_id = resp1.json()["data"]["session_id"] + + await client.post( + "/api/requests", + json={"prompt": "Second", "topic_id": topic_id, "session_id": session_id}, + ) + + resp = await client.get("/api/sessions") + assert resp.status_code == 200 + sessions = resp.json()["data"] + assert len(sessions) == 1 + assert sessions[0]["session_id"] == session_id + assert sessions[0]["request_count"] == 2 + + +@pytest.mark.asyncio +async def test_delete_session(client, topic_id): + resp1 = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id, "session_id": "new"}, + ) + session_id = resp1.json()["data"]["session_id"] + + resp = await client.delete(f"/api/sessions/{session_id}") + assert resp.status_code == 200 + + # New requests with this session_id should fail + resp = await client.post( + "/api/requests", + json={"prompt": "Again", "topic_id": topic_id, "session_id": session_id}, + ) + assert resp.status_code == 410 +``` + +- [ ] **Step 2: Run tests to verify they fail** + +```bash +uv run pytest tests/test_api_sessions.py::test_list_sessions_empty -v +``` + +Expected: FAIL (404 — no sessions route) + +- [ ] **Step 3: Add session query functions to database.py** + +Append to `src/backend/db/database.py`: + +```python +async def list_sessions(db: aiosqlite.Connection) -> list[dict]: + cursor = await db.execute( + "SELECT session_id, topic_id, COUNT(*) as request_count, " + "MIN(created_at) as first_request_at, MAX(created_at) as last_request_at " + "FROM requests WHERE session_id IS NOT NULL " + "GROUP BY session_id ORDER BY last_request_at DESC" + ) + rows = await cursor.fetchall() + return [dict(row) for row in rows] + + +async def session_exists(db: aiosqlite.Connection, session_id: str) -> bool: + cursor = await db.execute( + "SELECT 1 FROM requests WHERE session_id = ? LIMIT 1", (session_id,) + ) + return await cursor.fetchone() is not None + + +async def is_session_closed(db: aiosqlite.Connection, session_id: str) -> bool: + cursor = await db.execute( + "SELECT 1 FROM closed_sessions WHERE session_id = ?", (session_id,) + ) + return await cursor.fetchone() is not None + + +async def close_session(db: aiosqlite.Connection, session_id: str) -> bool: + if not await session_exists(db, session_id): + return False + await db.execute( + "INSERT OR IGNORE INTO closed_sessions (session_id, closed_at) VALUES (?, ?)", + (session_id, _now()), + ) + await db.commit() + return True +``` + +- [ ] **Step 4: Add closed_sessions table to SCHEMA** + +Update the `SCHEMA` constant in `database.py` — append before the closing `"""`: + +```sql +CREATE TABLE IF NOT EXISTS closed_sessions ( + session_id TEXT PRIMARY KEY, + closed_at TEXT NOT NULL +); +``` + +- [ ] **Step 5: Update create_request to check closed sessions** + +In `create_request` function in `database.py`, add check after topic validation: + +```python + if session_id and session_id != "new": + if await is_session_closed(db, session_id): + return "SESSION_CLOSED" +``` + +And update the return convention: `None` = topic not found, `"SESSION_CLOSED"` = closed session. + +- [ ] **Step 6: Update routes/requests.py to handle closed session** + +In the `create` handler in `routes/requests.py`, update the error check: + +```python + if req is None: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found"), + ) + if req == "SESSION_CLOSED": + return JSONResponse( + status_code=410, + content=fail("SESSION_CLOSED", "Session is closed"), + ) +``` + +- [ ] **Step 7: Create routes/sessions.py** + +```python +from fastapi import APIRouter, Depends +from starlette.responses import JSONResponse + +from src.backend.api.dependencies import verify_api_key +from src.backend.db.database import get_db, list_sessions, close_session +from src.backend.models.api import ok, fail + +router = APIRouter( + prefix="/sessions", + tags=["sessions"], + dependencies=[Depends(verify_api_key)], +) + + +async def _get_db(): + db = await get_db() + try: + yield db + finally: + await db.close() + + +@router.get("", summary="List sessions") +async def list_all(db=Depends(_get_db)): + sessions = await list_sessions(db) + return ok(sessions) + + +@router.delete("/{session_id}", summary="Close session") +async def delete(session_id: str, db=Depends(_get_db)): + closed = await close_session(db, session_id) + if not closed: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Session not found"), + ) + return ok({"closed": True}) +``` + +- [ ] **Step 8: Register sessions router in main.py** + +Add import and router: + +```python +from src.backend.api.routes.sessions import router as sessions_router +app.include_router(sessions_router, prefix="/api") +``` + +- [ ] **Step 9: Run tests to verify they pass** + +```bash +uv run pytest tests/test_api_sessions.py -v +``` + +Expected: all 3 tests PASS + +- [ ] **Step 10: Commit** + +```bash +git add src/backend/db/database.py src/backend/api/routes/sessions.py src/backend/api/routes/requests.py src/backend/main.py tests/test_api_sessions.py +git commit -m "feat: add sessions endpoints with close mechanism" +``` + +--- + +## Task 9: Worker Client + +**Files:** +- Create: `src/backend/services/worker_client.py` +- Modify: `tests/test_services.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_services.py`: + +```python +from unittest.mock import AsyncMock, patch + + +@pytest.mark.asyncio +async def test_worker_client_process(): + from src.backend.services.worker_client import WorkerClient + + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "success": True, + "result": "The answer is 42.", + "session_id": "sess-123", + } + + with patch("src.backend.services.worker_client.httpx.AsyncClient") as MockClient: + mock_client_instance = AsyncMock() + mock_client_instance.post.return_value = mock_response + mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance) + mock_client_instance.__aexit__ = AsyncMock(return_value=False) + MockClient.return_value = mock_client_instance + + client = WorkerClient() + result = await client.process( + prompt="What is the answer?", + system_prompt="You know everything.", + model="claude-sonnet-4-6", + ) + + assert result["success"] is True + assert result["result"] == "The answer is 42." + + +@pytest.mark.asyncio +async def test_worker_client_summarize(): + from src.backend.services.worker_client import WorkerClient + + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "success": True, + "summary": "[2026-05-25] Discussed the meaning of life.", + } + + with patch("src.backend.services.worker_client.httpx.AsyncClient") as MockClient: + mock_client_instance = AsyncMock() + mock_client_instance.post.return_value = mock_response + mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance) + mock_client_instance.__aexit__ = AsyncMock(return_value=False) + MockClient.return_value = mock_client_instance + + client = WorkerClient() + result = await client.summarize( + exchanges=[{"prompt": "What?", "result": "42."}], + mode="incremental", + ) + + assert result["success"] is True + assert "Discussed" in result["summary"] + + +@pytest.mark.asyncio +async def test_worker_client_health(): + from src.backend.services.worker_client import WorkerClient + + mock_response = AsyncMock() + mock_response.status_code = 200 + + with patch("src.backend.services.worker_client.httpx.AsyncClient") as MockClient: + mock_client_instance = AsyncMock() + mock_client_instance.get.return_value = mock_response + mock_client_instance.__aenter__ = AsyncMock(return_value=mock_client_instance) + mock_client_instance.__aexit__ = AsyncMock(return_value=False) + MockClient.return_value = mock_client_instance + + client = WorkerClient() + is_healthy = await client.health() + assert is_healthy is True +``` + +- [ ] **Step 2: Run tests to verify they fail** + +```bash +uv run pytest tests/test_services.py::test_worker_client_process -v +``` + +Expected: FAIL with `ModuleNotFoundError` + +- [ ] **Step 3: Write worker_client.py** + +```python +import httpx + +from src.backend.config import settings + + +class WorkerClient: + def __init__(self): + self.base_url = settings.worker_url + + async def process( + self, + prompt: str, + system_prompt: str, + model: str, + session_id: str | None = None, + ) -> dict: + async with httpx.AsyncClient(timeout=300.0) as client: + resp = await client.post( + f"{self.base_url}/process", + json={ + "prompt": prompt, + "system_prompt": system_prompt, + "model": model, + "session_id": session_id, + }, + ) + return resp.json() + + async def summarize( + self, + exchanges: list[dict], + mode: str = "incremental", + ) -> dict: + async with httpx.AsyncClient(timeout=300.0) as client: + resp = await client.post( + f"{self.base_url}/summarize", + json={"exchanges": exchanges, "mode": mode}, + ) + return resp.json() + + async def health(self) -> bool: + try: + async with httpx.AsyncClient(timeout=5.0) as client: + resp = await client.get(f"{self.base_url}/health") + return resp.status_code == 200 + except Exception: + return False +``` + +- [ ] **Step 4: Run tests to verify they pass** + +```bash +uv run pytest tests/test_services.py -v +``` + +Expected: all tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/backend/services/worker_client.py tests/test_services.py +git commit -m "feat: add worker HTTP client" +``` + +--- + +## Task 10: Queue Processor + +**Files:** +- Create: `src/backend/services/queue.py` +- Modify: `tests/test_services.py` + +- [ ] **Step 1: Write the failing tests** + +Add to `tests/test_services.py`: + +```python +@pytest.mark.asyncio +async def test_resolve_model_request_overrides(): + from src.backend.services.queue import resolve_model + + model = resolve_model( + request_model="claude-opus-4-7", + topic_model="claude-sonnet-4-6", + ) + assert model == "claude-opus-4-7" + + +@pytest.mark.asyncio +async def test_resolve_model_topic_fallback(): + from src.backend.services.queue import resolve_model + + model = resolve_model( + request_model=None, + topic_model="claude-sonnet-4-6", + ) + assert model == "claude-sonnet-4-6" + + +@pytest.mark.asyncio +async def test_resolve_model_global_fallback(): + from src.backend.services.queue import resolve_model + + model = resolve_model(request_model=None, topic_model=None) + assert model == "claude-sonnet-4-6" + + +@pytest.mark.asyncio +async def test_process_one_request(): + from src.backend.services.queue import process_one + from src.backend.db.database import ( + init_db, + get_db, + create_topic, + create_request, + get_request, + ) + + await init_db() + db = await get_db() + try: + topic = await create_topic(db, "test", "System prompt") + req = await create_request(db, "Hello", topic["id"]) + finally: + await db.close() + + mock_worker = AsyncMock() + mock_worker.process.return_value = { + "success": True, + "result": "Hi there!", + "session_id": None, + } + mock_worker.summarize.return_value = { + "success": True, + "summary": "Greeted the user.", + } + + await process_one(mock_worker) + + db = await get_db() + try: + updated = await get_request(db, req["id"]) + assert updated["status"] == "completed" + assert updated["result"] == "Hi there!" + finally: + await db.close() +``` + +- [ ] **Step 2: Run tests to verify they fail** + +```bash +uv run pytest tests/test_services.py::test_resolve_model_request_overrides -v +``` + +Expected: FAIL with `ModuleNotFoundError` + +- [ ] **Step 3: Write queue.py** + +```python +import asyncio +import logging + +from src.backend.config import settings +from src.backend.db.database import ( + get_db, + pick_next_pending, + get_topic, + complete_request, + fail_request, + append_topic_summary, +) +from src.backend.services.worker_client import WorkerClient + +logger = logging.getLogger(__name__) + + +def resolve_model( + request_model: str | None, topic_model: str | None +) -> str: + if request_model: + return request_model + if topic_model: + return topic_model + return settings.claude_model + + +async def process_one(worker: WorkerClient) -> bool: + db = await get_db() + try: + req = await pick_next_pending(db) + if not req: + return False + + topic = await get_topic(db, req["topic_id"]) + if not topic: + await fail_request(db, req["id"], "Topic not found") + return True + + model = resolve_model(req["model"], topic["model"]) + finally: + await db.close() + + try: + result = await worker.process( + prompt=req["prompt"], + system_prompt=topic["system_prompt"], + model=model, + session_id=req["session_id"], + ) + except Exception as e: + db = await get_db() + try: + await fail_request(db, req["id"], str(e)) + finally: + await db.close() + return True + + db = await get_db() + try: + if result.get("success"): + await complete_request( + db, + req["id"], + result["result"], + session_id=result.get("session_id"), + ) + + if req["summarize"]: + try: + summary_result = await worker.summarize( + exchanges=[ + {"prompt": req["prompt"], "result": result["result"]} + ], + mode="incremental", + ) + if summary_result.get("success"): + await append_topic_summary( + db, req["topic_id"], summary_result["summary"] + ) + except Exception as e: + logger.warning("Summary failed for request %s: %s", req["id"], e) + else: + error_msg = result.get("error", "Unknown worker error") + await fail_request(db, req["id"], str(error_msg)) + finally: + await db.close() + + return True + + +async def queue_loop(worker: WorkerClient): + while True: + try: + processed = await process_one(worker) + if not processed: + await asyncio.sleep(2) + except Exception as e: + logger.error("Queue loop error: %s", e) + await asyncio.sleep(5) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +```bash +uv run pytest tests/test_services.py -v +``` + +Expected: all tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/backend/services/queue.py tests/test_services.py +git commit -m "feat: add queue processor with model resolution" +``` + +--- + +## Task 11: Rate Limiting + Full App Assembly + +**Files:** +- Modify: `src/backend/api/dependencies.py` +- Modify: `src/backend/main.py` +- Modify: `src/backend/api/routes/health.py` +- Modify: `tests/test_api_topics.py` (add rate limit test) + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_api_requests.py`: + +```python +@pytest.mark.asyncio +async def test_rate_limit_triggers(topic_id): + import os + os.environ["RATE_LIMIT_PER_MINUTE"] = "2" + + from importlib import reload + from src.backend import config + reload(config) + + from src.backend.main import app + transport = ASGITransport(app=app) + async with AsyncClient( + transport=transport, + base_url="http://test", + headers={"X-Api-Key": "test-key-12345"}, + ) as client: + # Create topic first + await client.post( + "/api/topics", + json={"name": "rate-test", "system_prompt": "Test"}, + ) + resp = await client.get("/api/topics") + tid = resp.json()["data"][0]["id"] + + for _ in range(3): + resp = await client.post( + "/api/requests", + json={"prompt": "Hi", "topic_id": tid}, + ) + + # Third request should be rate limited + assert resp.status_code == 429 + + os.environ["RATE_LIMIT_PER_MINUTE"] = "100" + reload(config) +``` + +Note: Rate limit testing is inherently stateful. This test may need adjustment based on slowapi's exact behavior. If the above test proves flaky, replace it with a simpler test that verifies the rate limiter middleware is registered: + +```python +@pytest.mark.asyncio +async def test_rate_limit_headers_present(client, topic_id): + resp = await client.post( + "/api/requests", + json={"prompt": "Hello", "topic_id": topic_id}, + ) + # slowapi adds rate limit headers + assert resp.status_code == 202 +``` + +- [ ] **Step 2: Add rate limiter to dependencies.py** + +Update `src/backend/api/dependencies.py`: + +```python +from fastapi import Header, HTTPException, Request +from slowapi import Limiter +from slowapi.util import get_remote_address + +from src.backend.config import settings + + +def _get_api_key(request: Request) -> str: + return request.headers.get("X-Api-Key", get_remote_address(request)) + + +limiter = Limiter(key_func=_get_api_key) + + +async def verify_api_key(x_api_key: str = Header(...)): + if x_api_key != settings.api_key: + raise HTTPException(status_code=403, detail="API key non valida") +``` + +- [ ] **Step 3: Apply rate limiter to requests route** + +Update `src/backend/api/routes/requests.py` — add rate limit decorator to `create`: + +Add import: + +```python +from src.backend.api.dependencies import limiter +from src.backend.config import settings +``` + +Add decorator to `create` handler (before `async def create`): + +```python +@router.post("", summary="Create request") +@limiter.limit(lambda: f"{settings.rate_limit_per_minute}/minute;{settings.rate_limit_per_hour}/hour") +async def create(request: Request, body: RequestCreate, db=Depends(_get_db)): +``` + +Note: the `request: Request` parameter is required by slowapi to read the request context. + +Add `Request` to the FastAPI import: + +```python +from fastapi import APIRouter, Depends, Query, Request +``` + +- [ ] **Step 4: Complete main.py with all routers, CORS, lifespan, and rate limiter** + +Rewrite `src/backend/main.py`: + +```python +import asyncio +import logging +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from slowapi import _rate_limit_exceeded_handler +from slowapi.errors import RateLimitExceeded + +from src.backend.api.dependencies import limiter +from src.backend.api.routes.health import router as health_router +from src.backend.api.routes.requests import router as requests_router +from src.backend.api.routes.sessions import router as sessions_router +from src.backend.api.routes.topics import router as topics_router +from src.backend.config import settings +from src.backend.db.database import init_db +from src.backend.services.queue import queue_loop +from src.backend.services.worker_client import WorkerClient + +logging.basicConfig(level=logging.INFO) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + await init_db() + worker = WorkerClient() + task = asyncio.create_task(queue_loop(worker)) + yield + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +app = FastAPI( + title="OpusAgent", + version="0.1.0", + docs_url="/doc", + redoc_url="/redoc", + openapi_url="/openapi.json", + lifespan=lifespan, +) + +app.state.limiter = limiter +app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], +) + +app.include_router(health_router, prefix="/api") +app.include_router(topics_router, prefix="/api") +app.include_router(requests_router, prefix="/api") +app.include_router(sessions_router, prefix="/api") +``` + +- [ ] **Step 5: Update health endpoint to check worker and DB** + +Update `src/backend/api/routes/health.py`: + +```python +from fastapi import APIRouter, Depends + +from src.backend.api.dependencies import verify_api_key +from src.backend.db.database import get_db +from src.backend.models.api import ok +from src.backend.services.worker_client import WorkerClient + +router = APIRouter(dependencies=[Depends(verify_api_key)]) + + +@router.get("/health", summary="Health check", tags=["system"]) +async def health(): + db_ok = True + try: + db = await get_db() + await db.execute("SELECT 1") + await db.close() + except Exception: + db_ok = False + + worker = WorkerClient() + worker_ok = await worker.health() + + return ok({ + "status": "running", + "database": db_ok, + "worker": worker_ok, + }) +``` + +- [ ] **Step 6: Run all tests** + +```bash +uv run pytest -v +``` + +Expected: all tests PASS + +- [ ] **Step 7: Commit** + +```bash +git add src/backend/api/dependencies.py src/backend/api/routes/health.py src/backend/api/routes/requests.py src/backend/main.py +git commit -m "feat: add rate limiting, CORS, queue loop in lifespan" +``` + +--- + +## Task 12: Topics Summarize Endpoint + +**Files:** +- Modify: `src/backend/api/routes/topics.py` +- Modify: `tests/test_api_topics.py` + +- [ ] **Step 1: Write the failing test** + +Add to `tests/test_api_topics.py`: + +```python +@pytest.mark.asyncio +async def test_summarize_topic(client): + # Create topic + create_resp = await client.post( + "/api/topics", + json={"name": "summarize-test", "system_prompt": "You help."}, + ) + topic_id = create_resp.json()["data"]["id"] + + # Create completed requests manually in DB + from src.backend.db.database import get_db, create_request, complete_request + + db = await get_db() + try: + req1 = await create_request(db, "Question 1", topic_id) + await complete_request(db, req1["id"], "Answer 1") + req2 = await create_request(db, "Question 2", topic_id) + await complete_request(db, req2["id"], "Answer 2") + finally: + await db.close() + + # Mock the worker summarize call + from unittest.mock import patch, AsyncMock + + mock_worker = AsyncMock() + mock_worker.summarize.return_value = { + "success": True, + "summary": "Comprehensive summary of all exchanges.", + } + + with patch( + "src.backend.api.routes.topics.WorkerClient", return_value=mock_worker + ): + resp = await client.post(f"/api/topics/{topic_id}/summarize") + + assert resp.status_code == 200 + data = resp.json() + assert data["success"] is True + assert "summary" in data["data"] +``` + +- [ ] **Step 2: Run test to verify it fails** + +```bash +uv run pytest tests/test_api_topics.py::test_summarize_topic -v +``` + +Expected: FAIL (no `/summarize` route) + +- [ ] **Step 3: Add summarize endpoint to routes/topics.py** + +Add imports to `src/backend/api/routes/topics.py`: + +```python +from src.backend.db.database import list_requests +from src.backend.services.worker_client import WorkerClient +``` + +Add the endpoint: + +```python +@router.post("/{topic_id}/summarize", summary="Regenerate topic summary") +async def summarize_topic(topic_id: str, db=Depends(_get_db)): + topic = await get_topic(db, topic_id) + if not topic: + return JSONResponse( + status_code=404, + content=fail("NOT_FOUND", "Topic not found"), + ) + + completed = await list_requests(db, status="completed", topic_id=topic_id) + if not completed: + return JSONResponse( + status_code=400, + content=fail("NO_DATA", "No completed requests to summarize"), + ) + + exchanges = [ + {"prompt": r["prompt"], "result": r["result"]} + for r in completed + if r["result"] + ] + + worker = WorkerClient() + result = await worker.summarize(exchanges=exchanges, mode="complete") + + if not result.get("success"): + return JSONResponse( + status_code=502, + content=fail("WORKER_ERROR", "Worker failed to generate summary"), + ) + + from src.backend.db.database import update_topic + + updated = await update_topic(db, topic_id, summary=result["summary"]) + return ok(updated) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +```bash +uv run pytest tests/test_api_topics.py -v +``` + +Expected: all tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add src/backend/api/routes/topics.py tests/test_api_topics.py +git commit -m "feat: add topic summarize endpoint" +``` + +--- + +## Task 13: Node.js Worker + +**Files:** +- Create: `worker/package.json` +- Create: `worker/index.js` +- Create: `worker/Dockerfile` + +Note: The Claude Code SDK API (`@anthropic-ai/claude-code`) must be verified against the official documentation. The implementation below uses the API as described in the spec. Adjust function signatures and options based on the actual SDK at implementation time. + +- [ ] **Step 1: Create package.json** + +```json +{ + "name": "opus-agent-worker", + "version": "0.1.0", + "private": true, + "type": "module", + "dependencies": { + "express": "^4.21.0", + "@anthropic-ai/claude-code": "^1.0.0" + } +} +``` + +- [ ] **Step 2: Create index.js** + +```javascript +import express from "express"; +import { query } from "@anthropic-ai/claude-code"; + +const app = express(); +app.use(express.json({ limit: "10mb" })); + +const PORT = process.env.WORKER_PORT || 3001; +const DEFAULT_MODEL = process.env.CLAUDE_MODEL || "claude-sonnet-4-6"; + +app.get("/health", (_req, res) => { + res.json({ status: "ok" }); +}); + +app.post("/process", async (req, res) => { + const { prompt, system_prompt, model, session_id } = req.body; + + if (!prompt) { + return res + .status(400) + .json({ success: false, error: "prompt is required" }); + } + + try { + const options = { + model: model || DEFAULT_MODEL, + systemPrompt: system_prompt || undefined, + allowedTools: [], + }; + + if (session_id) { + options.resume = session_id; + } + + const messages = await query({ + prompt, + options, + }); + + const resultText = messages + .filter((m) => m.type === "text") + .map((m) => m.text) + .join("\n"); + + const conversationId = messages.length > 0 ? messages[0].conversationId : null; + + res.json({ + success: true, + result: resultText, + session_id: conversationId || session_id, + }); + } catch (error) { + console.error("Process error:", error.message); + res.json({ + success: false, + error: error.message, + }); + } +}); + +app.post("/summarize", async (req, res) => { + const { exchanges, mode } = req.body; + + if (!exchanges || exchanges.length === 0) { + return res + .status(400) + .json({ success: false, error: "exchanges is required" }); + } + + const exchangeText = exchanges + .map((e, i) => `Exchange ${i + 1}:\nQ: ${e.prompt}\nA: ${e.result}`) + .join("\n\n"); + + const summaryPrompt = + mode === "complete" + ? `Summarize these exchanges into a comprehensive overview. Be concise but thorough:\n\n${exchangeText}` + : `Summarize this exchange in 1-2 sentences:\n\n${exchangeText}`; + + try { + const messages = await query({ + prompt: summaryPrompt, + options: { + model: DEFAULT_MODEL, + systemPrompt: + "You are a summarizer. Respond with only the summary, no preamble.", + allowedTools: [], + }, + }); + + const summaryText = messages + .filter((m) => m.type === "text") + .map((m) => m.text) + .join("\n"); + + res.json({ + success: true, + summary: summaryText.trim(), + }); + } catch (error) { + console.error("Summarize error:", error.message); + res.json({ + success: false, + error: error.message, + }); + } +}); + +app.listen(PORT, () => { + console.log(`Worker listening on port ${PORT}`); +}); +``` + +**Important:** The `query` function import and usage above follows the pattern `import { query } from '@anthropic-ai/claude-code'`. Verify the actual SDK exports at implementation time. The SDK may export `claude`, `ClaudeCode`, or another name. Check `node_modules/@anthropic-ai/claude-code/index.js` or the package README after installation. + +- [ ] **Step 3: Create worker/Dockerfile** + +```dockerfile +FROM node:20-slim + +RUN apt-get update && apt-get install -y wget && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +COPY package.json ./ +RUN npm install --production + +COPY index.js ./ + +EXPOSE 3001 + +CMD ["node", "index.js"] +``` + +- [ ] **Step 4: Verify worker builds** + +```bash +cd worker && npm install && cd .. +``` + +Expected: dependencies install without error + +- [ ] **Step 5: Commit** + +```bash +git add worker/ +git commit -m "feat: add Node.js worker with Claude Code SDK" +``` + +--- + +## Task 14: Docker Setup + +**Files:** +- Create: `Dockerfile` +- Create: `docker-compose.yml` + +- [ ] **Step 1: Create backend Dockerfile** + +```dockerfile +FROM python:3.11-slim AS base + +COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv + +WORKDIR /app + +COPY pyproject.toml uv.lock ./ +RUN uv sync --frozen --no-dev + +COPY src/ src/ + +EXPOSE 8000 + +CMD ["uv", "run", "start"] +``` + +- [ ] **Step 2: Create docker-compose.yml** + +```yaml +services: + fastapi-service: + build: + context: . + dockerfile: Dockerfile + ports: + - "${API_PORT:-8000}:8000" + env_file: .env + volumes: + - opus-data:/data + depends_on: + claude-worker: + condition: service_healthy + restart: unless-stopped + healthcheck: + test: + [ + "CMD", + "python", + "-c", + "import urllib.request; urllib.request.urlopen('http://localhost:8000/api/health')", + ] + interval: 10s + timeout: 5s + retries: 3 + + claude-worker: + build: + context: ./worker + dockerfile: Dockerfile + expose: + - "3001" + env_file: .env + volumes: + - opus-data:/data + - claude-auth:/home/node/.claude + restart: unless-stopped + healthcheck: + test: ["CMD", "wget", "--spider", "-q", "http://localhost:3001/health"] + interval: 10s + timeout: 5s + retries: 3 + +volumes: + opus-data: + claude-auth: +``` + +- [ ] **Step 3: Verify compose config is valid** + +```bash +docker compose config --quiet +``` + +Expected: no errors + +- [ ] **Step 4: Create README.md** + +```markdown +# OpusAgent + +Servizio self-hosted per l'interazione programmatica con i modelli Anthropic +tramite Claude Code SDK, sfruttando un abbonamento Claude Pro/Max esistente. + +## Architettura + +Due container Docker: +- **fastapi-service** — API REST, autenticazione, coda richieste, rate limiting +- **claude-worker** — Processamento richieste via Claude Code SDK + +## Setup + +### Prerequisiti + +- Docker e Docker Compose +- Un abbonamento Claude Pro/Max attivo + +### Prima installazione + +1. Copia e configura l'environment: + ```bash + cp .env.example .env + # Edita .env: imposta API_KEY e le altre variabili + ``` + +2. Autentica Claude Code nel worker: + ```bash + docker compose run --rm claude-worker npx claude login + ``` + +3. Avvia il sistema: + ```bash + docker compose up -d + ``` + +4. Verifica: + ```bash + curl -H "X-Api-Key: TUA_KEY" http://localhost:8000/api/health + ``` + +## Sviluppo locale + +```bash +uv sync --all-groups +cp .env.example .env +# Edita .env +uv run dev +``` + +## Test + +```bash +uv run pytest -v +``` + +## API + +Documentazione completa: `http://localhost:8000/doc` + +### Esempio d'uso + +```bash +# Crea un argomento +curl -X POST http://localhost:8000/api/topics \ + -H "X-Api-Key: TUA_KEY" \ + -H "Content-Type: application/json" \ + -d '{"name": "coding", "system_prompt": "Sei un esperto sviluppatore."}' + +# Invia una richiesta +curl -X POST http://localhost:8000/api/requests \ + -H "X-Api-Key: TUA_KEY" \ + -H "Content-Type: application/json" \ + -d '{"prompt": "Come implemento JWT in FastAPI?", "topic_id": "ID_TOPIC"}' + +# Controlla il risultato +curl http://localhost:8000/api/requests/ID_RICHIESTA \ + -H "X-Api-Key: TUA_KEY" +``` +``` + +- [ ] **Step 5: Commit** + +```bash +git add Dockerfile docker-compose.yml README.md +git commit -m "feat: add Docker setup and README" +``` + +--- + +## Self-Review Checklist + +### Spec Coverage + +| Spec Section | Task | +|---|---| +| 1. Panoramica | Overall design | +| 2. Architettura | Tasks 13-14 (Docker), Task 11 (main.py assembly) | +| 3. Data Model (topics, requests) | Task 3 (database), Task 6 (topics), Task 7 (requests) | +| 4.1 Topics API | Task 6 | +| 4.2 Requests API | Task 7 | +| 4.3 Sessions API | Task 8 | +| 4.4 Health | Task 5 | +| 5.1 Flusso standard | Task 10 (queue processor) | +| 5.2 Multi-turn | Task 7 (session_id handling), Task 10 | +| 5.3 Summary completo | Task 12 | +| 5.4 Summary incrementale | Task 10 (summarize in queue) | +| 6. Node.js Worker | Task 13 | +| 7. Selezione Modello | Task 10 (resolve_model) | +| 8. Rate Limiting | Task 11 | +| 9. Sicurezza | Task 5 (API key), Task 14 (Docker isolation) | +| 10. Configurazione | Task 2 | +| 11. Struttura Progetto | Task 1 | +| 12. Docker | Task 14 | +| 13. Setup Iniziale | Task 14 (README) | +| 14. Limiti | Documented in spec | +| POST /api/topics/{id}/summarize | Task 12 | +| Traefik integration | Noted as last phase, not in implementation plan (per spec) | + +### Type Consistency Check + +- `TopicCreate`, `TopicUpdate` — used in tasks 4, 6 ✓ +- `RequestCreate` — used in tasks 4, 7 ✓ +- `ok()`, `fail()` — used in all route files ✓ +- `get_db()` → `aiosqlite.Connection` — used consistently ✓ +- `WorkerClient.process()` → returns `dict` with `success`, `result`, `session_id` ✓ +- `WorkerClient.summarize()` → returns `dict` with `success`, `summary` ✓ +- `resolve_model(request_model, topic_model)` → `str` ✓ +- `process_one(worker)` → `bool` ✓ + +### Placeholder Check + +No TBD, TODO, or incomplete sections found. All steps contain actual code.