docs: add API guide for AI agents

Complete reference for interacting with OpusAgent: auth, async flow,
multi-turn sessions, model resolution, summarization, and error handling.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

CLAUDE.md

@@ -0,0 +1,125 @@
+# OpusAgent
+
+Self-hosted REST API proxy to Claude models via Claude Code SDK, leveraging a Claude Pro/Max subscription.
+
+## Architecture
+
+Two Docker containers behind Traefik reverse proxy:
+
+```
+Client → Traefik (HTTPS) → FastAPI (Python, :8000) ↔ Node.js Worker (:3001, internal)
+                                    ↓                         ↓
+                              SQLite (WAL)           Claude Code SDK
+```
+
+- **fastapi-service**: API REST, auth, rate limiting, async queue processing
+- **claude-worker**: Express server wrapping `@anthropic-ai/claude-code` SDK
+
+## Key Paths
+
+| Area | Path |
+|------|------|
+| FastAPI entrypoint | `src/backend/main.py` |
+| Config (Pydantic Settings) | `src/backend/config.py` |
+| Database layer (aiosqlite) | `src/backend/db/database.py` |
+| API models + envelope | `src/backend/models/api.py` |
+| Routes | `src/backend/api/routes/{health,topics,requests,sessions}.py` |
+| Auth + rate limiter | `src/backend/api/dependencies.py` |
+| Worker HTTP client | `src/backend/services/worker_client.py` |
+| Queue processor | `src/backend/services/queue.py` |
+| Node.js worker | `worker/index.js` |
+| Tests | `tests/` |
+| Design spec (Italian) | `docs/superpowers/specs/` |
+| Implementation plan | `docs/superpowers/plans/` |
+
+## Tech Stack
+
+- **Python 3.11+**: FastAPI, uvicorn, aiosqlite, httpx, slowapi, pydantic-settings
+- **Node.js 20**: Express, @anthropic-ai/claude-code SDK (v1.0.128+, async iterable API)
+- **Package manager**: uv (Python), npm (Node.js)
+- **Database**: SQLite with WAL mode, foreign keys enabled
+- **Infra**: Docker Compose, Traefik v3 (TLS via Let's Encrypt)
+
+## Database Schema
+
+Three tables in SQLite:
+
+- **topics**: id (UUID), name (unique), system_prompt, summary, model (nullable), timestamps
+- **requests**: id (UUID), session_id, sdk_conversation_id, topic_id (FK), prompt, result, error, model, summarize (bool), status (pending/processing/completed/failed), timestamps
+- **closed_sessions**: session_id (PK), closed_at
+
+## API Endpoints
+
+All require `X-Api-Key` header. All responses use envelope: `{success, data, error}`.
+
+- `GET /api/health` — health check (DB + worker)
+- `GET /api/models` — list available Claude models (aliases + default)
+- `POST|GET /api/topics`, `GET|PUT|DELETE /api/topics/{id}` — topic CRUD
+- `POST /api/topics/{id}/summarize` — regenerate complete topic summary
+- `POST /api/requests` (202) — submit prompt (async queue), `GET|DELETE /api/requests[/{id}]`
+- `GET /api/sessions`, `DELETE /api/sessions/{id}` — session management
+
+## Processing Flow
+
+1. `POST /api/requests` → 202 Accepted (queued)
+2. Background loop picks oldest pending, calls worker `POST /process`
+3. Worker runs Claude Code SDK (`query()` async iterable) with optional session resume
+4. Result stored in DB; SDK conversation ID saved in `sdk_conversation_id`; optional incremental summary appended to topic
+5. Client polls `GET /api/requests/{id}` for result
+
+Model resolution: request-level > topic-level > `CLAUDE_MODEL` env var.
+
+### Multi-turn sessions
+
+- `session_id: "new"` → stored as NULL; after completion, set to SDK's conversation ID
+- Custom `session_id` → preserved as-is; SDK conversation ID stored separately in `sdk_conversation_id`
+- Resume logic: queue looks up `sdk_conversation_id` from previous completed requests in the same session; only passes it to worker if found (avoids resume on non-existent SDK sessions)
+
+## Commands
+
+```bash
+# Development
+uv sync --all-groups
+uv run dev                          # uvicorn with --reload
+
+# Production
+docker compose up -d                # both services
+docker compose exec claude-worker npx claude login  # first-time auth (must run on active container, NOT run --rm)
+
+# Tests
+uv run pytest -v
+
+# Logs
+docker compose logs -f fastapi-service
+docker compose logs -f claude-worker
+```
+
+## Environment Variables (.env)
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `API_KEY` | (required) | Auth key for all endpoints |
+| `API_HOST` | 0.0.0.0 | Server bind address |
+| `API_PORT` | 8000 | Server port |
+| `WORKER_URL` | http://claude-worker:3001 | Internal worker address |
+| `DB_PATH` | /data/opus-agent.db | SQLite database path |
+| `CLAUDE_MODEL` | claude-sonnet-4-6 | Default Claude model |
+| `RATE_LIMIT_PER_MINUTE` | 10 | Rate limit per API key |
+| `RATE_LIMIT_PER_HOUR` | 100 | Rate limit per API key |
+
+## Design Decisions
+
+- **Sequential processing**: one request at a time (FIFO), no concurrency — simplifies Claude Code SDK session handling
+- **Async Python**: FastAPI + aiosqlite for non-blocking I/O while queue runs in background
+- **SQLite + WAL**: sufficient for single-writer pattern, no external DB needed
+- **Envelope pattern**: consistent `{success, data, error}` on all responses
+- **Session = conversation**: session_id maps to Claude Code SDK `resume` option
+- **Worker isolation**: not exposed to host, only reachable from fastapi-service via Docker internal network
+- **Claude auth volume**: credentials stored at `/root/.claude` (not `/home/node/.claude`) because the worker process runs as root inside the container
+- **SDK async iteration**: `query()` returns an async iterable (not a promise); iterate with `for await...of` and extract the message with `type === "result"`
+
+## Deployment
+
+Domain: `opus-agent.tielogic.xyz` → Traefik → fastapi-service:8000
+
+Traefik config: `/opt/docker/traefik/dynamic.yml` (file provider, auto-reload).

Dockerfile

@@ -0,0 +1,14 @@
+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", "uvicorn", "src.backend.main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -0,0 +1,112 @@
+# 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. Avvia il sistema:
+   ```bash
+   docker compose up -d
+   ```
+
+3. Autentica Claude Code nel worker (nel container attivo, non con `run --rm`):
+   ```bash
+   docker compose exec claude-worker npx claude login
+   ```
+   Le credenziali vengono salvate nel volume Docker `claude-auth` (mount: `/root/.claude`)
+   e persistono tra riavvii e rebuild. Si perdono solo con `docker compose down -v`.
+
+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 interattiva:
+- Swagger UI: `https://opus-agent.tielogic.xyz/docs`
+- ReDoc: `https://opus-agent.tielogic.xyz/redoc`
+
+### Modelli disponibili
+
+```bash
+curl http://localhost:8000/api/models -H "X-Api-Key: TUA_KEY"
+```
+
+Ritorna i modelli disponibili dall'SDK (`default`, `opus`, `sonnet`, `opusplan`) e il modello
+di default configurato. Gli alias possono essere usati nel campo `model` di topic e richieste.
+
+### 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"
+```
+
+### Sessioni multi-turn
+
+```bash
+# Primo messaggio: session_id "new" o un ID personalizzato
+curl -X POST http://localhost:8000/api/requests \
+  -H "X-Api-Key: TUA_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"prompt": "Mi chiamo Marco.", "topic_id": "ID_TOPIC", "session_id": "new"}'
+
+# Il risultato contiene il session_id (generato dall'SDK)
+# Usalo per i messaggi successivi nella stessa conversazione
+curl -X POST http://localhost:8000/api/requests \
+  -H "X-Api-Key: TUA_KEY" \
+  -H "Content-Type: application/json" \
+  -d '{"prompt": "Come mi chiamo?", "topic_id": "ID_TOPIC", "session_id": "SESSION_ID_DAL_RISULTATO"}'
+```
+
+Il campo `sdk_conversation_id` nella risposta contiene l'ID interno della conversazione
+Claude Code SDK, usato automaticamente per il resume nei turni successivi.

docker-compose.yml

@@ -0,0 +1,46 @@
+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:/root/.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:

docs/agent-guide.md

@@ -0,0 +1,264 @@
+# OpusAgent - Guida per Agenti AI
+
+Questo documento spiega come interagire con l'API REST di OpusAgent.
+OpusAgent e' un proxy che inoltra richieste ai modelli Claude tramite Claude Code SDK.
+
+## Informazioni base
+
+- **Base URL**: `https://opus-agent.tielogic.xyz`
+- **Autenticazione**: header `X-Api-Key` obbligatorio su ogni richiesta
+- **Formato risposte**: JSON con envelope `{"success": bool, "data": any, "error": {"code": str, "message": str} | null}`
+- **Elaborazione asincrona**: le richieste vengono accodate e processate sequenzialmente (FIFO)
+
+## Flusso operativo
+
+### 1. Crea un topic
+
+Un topic definisce il contesto della conversazione (system prompt) e opzionalmente un modello di default.
+
+```
+POST /api/topics
+Content-Type: application/json
+X-Api-Key: TUA_KEY
+
+{
+  "name": "analisi-dati",
+  "system_prompt": "Sei un esperto analista dati. Rispondi in italiano.",
+  "model": "sonnet"  // opzionale, default dal server
+}
+```
+
+Risposta (201):
+```json
+{
+  "success": true,
+  "data": {
+    "id": "uuid-del-topic",
+    "name": "analisi-dati",
+    "system_prompt": "...",
+    "summary": "",
+    "model": "sonnet",
+    "created_at": "...",
+    "updated_at": "..."
+  }
+}
+```
+
+### 2. Invia una richiesta
+
+Le richieste vengono accodate e processate in background. La risposta immediata e' un 202 Accepted con l'ID della richiesta.
+
+```
+POST /api/requests
+Content-Type: application/json
+X-Api-Key: TUA_KEY
+
+{
+  "topic_id": "uuid-del-topic",
+  "prompt": "Analizza i trend di vendita Q1 2026",
+  "model": "opus",       // opzionale, sovrascrive il modello del topic
+  "session_id": "new",   // opzionale, per conversazioni multi-turn
+  "summarize": true       // opzionale, genera riassunto incrementale nel topic
+}
+```
+
+Risposta (202):
+```json
+{
+  "success": true,
+  "data": {
+    "id": "uuid-della-richiesta",
+    "session_id": null,
+    "status": "pending",
+    "created_at": "..."
+  }
+}
+```
+
+### 3. Attendi il risultato (polling)
+
+Interroga periodicamente lo stato della richiesta fino a `status: "completed"` o `status: "failed"`.
+
+```
+GET /api/requests/{id}
+X-Api-Key: TUA_KEY
+```
+
+Risposta quando completata:
+```json
+{
+  "success": true,
+  "data": {
+    "id": "uuid-della-richiesta",
+    "session_id": "sdk-conversation-id",
+    "topic_id": "uuid-del-topic",
+    "prompt": "Analizza i trend di vendita Q1 2026",
+    "model": "opus",
+    "summarize": true,
+    "status": "completed",
+    "result": "Risposta di Claude...",
+    "error": null,
+    "created_at": "...",
+    "completed_at": "...",
+    "sdk_conversation_id": "sdk-uuid"
+  }
+}
+```
+
+**Intervallo di polling consigliato**: 2-3 secondi. Il tempo medio di risposta e' 3-5 secondi per richieste semplici.
+
+**Stati possibili**: `pending` -> `processing` -> `completed` | `failed`
+
+## Conversazioni multi-turn
+
+Per mantenere il contesto tra messaggi successivi, usa le sessioni.
+
+### Prima richiesta
+
+Invia con `"session_id": "new"` oppure un ID personalizzato a tua scelta.
+
+```json
+{
+  "topic_id": "uuid",
+  "prompt": "Mi chiamo Marco. Lavorero' su un progetto Python.",
+  "session_id": "new"
+}
+```
+
+Il `session_id` nella risposta 202 sara' `null`. Dopo il completamento, il campo `session_id` conterra' l'ID della conversazione generato dall'SDK.
+
+### Richieste successive
+
+Usa il `session_id` ottenuto dal risultato della prima richiesta completata.
+
+```json
+{
+  "topic_id": "uuid",
+  "prompt": "Come mi chiamo? Su cosa lavoriamo?",
+  "session_id": "session-id-dal-risultato-precedente"
+}
+```
+
+Claude ricordera' tutto il contesto della conversazione.
+
+### Session ID personalizzato
+
+Puoi passare un tuo ID (es. `"progetto-alpha-chat-1"`) invece di `"new"`. Il sistema lo mantiene come identificativo della sessione e gestisce internamente la mappatura con la conversazione SDK.
+
+### Chiudere una sessione
+
+```
+DELETE /api/sessions/{session_id}
+X-Api-Key: TUA_KEY
+```
+
+Dopo la chiusura, nuove richieste con lo stesso `session_id` verranno rifiutate con status 410 Gone.
+
+## Risoluzione del modello
+
+Il modello usato segue questa priorita' (dal piu' alto al piu' basso):
+
+1. **Campo `model` nella richiesta** - sovrascrive tutto
+2. **Campo `model` nel topic** - default per il topic
+3. **Variabile d'ambiente `CLAUDE_MODEL`** - default globale del server
+
+### Valori accettati per il campo model
+
+Alias corti (consigliati):
+- `default` - Opus o Sonnet in base ai limiti d'uso
+- `opus` - Opus 4.1, per task complessi
+- `sonnet` - Sonnet 4, per uso quotidiano
+- `opusplan` - Opus in plan mode, Sonnet altrimenti
+
+Model ID completi:
+- `claude-opus-4-7`
+- `claude-sonnet-4-6`
+- `claude-haiku-4-5-20251001`
+
+Per ottenere la lista aggiornata dei modelli disponibili:
+
+```
+GET /api/models
+X-Api-Key: TUA_KEY
+```
+
+## Riassunti automatici
+
+### Riassunto incrementale
+
+Aggiungi `"summarize": true` alla richiesta. Dopo il completamento, una sintesi di 1-2 frasi viene appesa al campo `summary` del topic con la data.
+
+### Riassunto completo
+
+Rigenera il riassunto del topic analizzando tutte le richieste completate:
+
+```
+POST /api/topics/{id}/summarize
+X-Api-Key: TUA_KEY
+```
+
+Il summary viene sovrascritto con un'overview completa di tutte le interazioni.
+
+## Riferimento completo degli endpoint
+
+| Metodo | Endpoint | Descrizione |
+|--------|----------|-------------|
+| GET | `/api/health` | Stato del sistema (DB, worker) |
+| GET | `/api/models` | Modelli Claude disponibili |
+| POST | `/api/topics` | Crea topic |
+| GET | `/api/topics` | Lista topic |
+| GET | `/api/topics/{id}` | Dettaglio topic |
+| PUT | `/api/topics/{id}` | Aggiorna topic |
+| DELETE | `/api/topics/{id}` | Elimina topic (fallisce se ha richieste pendenti) |
+| POST | `/api/topics/{id}/summarize` | Rigenera riassunto completo |
+| POST | `/api/requests` | Invia richiesta (202 Accepted) |
+| GET | `/api/requests` | Lista richieste (filtri: `?status=`, `?topic_id=`, `?session_id=`) |
+| GET | `/api/requests/{id}` | Stato e risultato richiesta |
+| DELETE | `/api/requests/{id}` | Elimina richiesta pendente |
+| GET | `/api/sessions` | Lista sessioni attive |
+| DELETE | `/api/sessions/{id}` | Chiudi sessione |
+
+## Gestione errori
+
+Le risposte di errore seguono lo stesso formato envelope:
+
+```json
+{
+  "success": false,
+  "data": null,
+  "error": {
+    "code": "NOT_FOUND",
+    "message": "Topic not found"
+  }
+}
+```
+
+Codici HTTP comuni:
+- **202**: richiesta accettata e accodata
+- **400**: parametri mancanti o invalidi
+- **403**: API key mancante o errata
+- **404**: risorsa non trovata
+- **409**: conflitto (es. nome topic duplicato)
+- **410**: sessione chiusa
+- **429**: rate limit superato
+
+## Rate limiting
+
+Le richieste `POST /api/requests` sono limitate per API key:
+- **Per minuto**: configurabile (default 10)
+- **Per ora**: configurabile (default 100)
+
+Le altre operazioni (lettura, topic CRUD) non hanno limiti.
+
+## Esempio completo di workflow
+
+```
+1. GET  /api/models                     -> scegli il modello
+2. POST /api/topics                     -> crea il contesto
+3. POST /api/requests (session_id=new)  -> primo messaggio
+4. GET  /api/requests/{id}              -> polling fino a completed
+5. POST /api/requests (session_id=...)  -> messaggi successivi
+6. GET  /api/requests/{id}              -> polling risultato
+7. POST /api/topics/{id}/summarize      -> riassunto finale
+8. DELETE /api/sessions/{id}            -> chiudi sessione
+```

src/backend/api/dependencies.py

@@ -0,0 +1,17 @@
+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 | None = Header(default=None)):
+    if x_api_key is None or x_api_key != settings.api_key:
+        raise HTTPException(status_code=403, detail="API key non valida")

src/backend/api/routes/health.py

@@ -0,0 +1,43 @@
+from fastapi import APIRouter, Depends
+
+from src.backend.api.dependencies import verify_api_key
+from src.backend.config import settings
+from src.backend.db.database import get_db
+from src.backend.models.api import ok, fail
+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,
+    })
+
+
+@router.get("/models", summary="List available models", tags=["system"])
+async def list_models():
+    worker = WorkerClient()
+    result = await worker.models()
+
+    if not result.get("success"):
+        return fail("WORKER_ERROR", result.get("error", "Failed to fetch models"))
+
+    return ok({
+        "default_model": settings.claude_model,
+        "models": result["models"],
+    })

src/backend/api/routes/requests.py

@@ -0,0 +1,92 @@
+from fastapi import APIRouter, Depends, Query, Request as FastAPIRequest
+from starlette.responses import JSONResponse
+
+from src.backend.api.dependencies import verify_api_key, limiter
+from src.backend.config import settings
+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")
+@limiter.limit(lambda: f"{settings.rate_limit_per_minute}/minute;{settings.rate_limit_per_hour}/hour")
+async def create(request: FastAPIRequest, 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"),
+        )
+    if req == "SESSION_CLOSED":
+        return JSONResponse(
+            status_code=410,
+            content=fail("SESSION_CLOSED", "Session is closed"),
+        )
+    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})

src/backend/api/routes/sessions.py

@@ -0,0 +1,37 @@
+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})

src/backend/api/routes/topics.py

@@ -0,0 +1,130 @@
+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,
+    list_requests,
+)
+from src.backend.models.api import TopicCreate, TopicUpdate, ok, fail
+from src.backend.services.worker_client import WorkerClient
+
+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})
+
+
+@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"),
+        )
+
+    updated = await update_topic(db, topic_id, summary=result["summary"])
+    return ok(updated)

src/backend/db/database.py

@@ -19,6 +19,7 @@ CREATE TABLE IF NOT EXISTS topics (
 CREATE TABLE IF NOT EXISTS requests (
     id TEXT PRIMARY KEY,
     session_id TEXT,
+    sdk_conversation_id TEXT,
     topic_id TEXT NOT NULL REFERENCES topics(id),
     prompt TEXT NOT NULL,
     model TEXT,
@@ -33,6 +34,11 @@ CREATE TABLE IF NOT EXISTS requests (
 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);
+
+CREATE TABLE IF NOT EXISTS closed_sessions (
+    session_id TEXT PRIMARY KEY,
+    closed_at TEXT NOT NULL
+);
 """
 
 
@@ -59,3 +65,265 @@ async def init_db():
         await db.commit()
     finally:
         await db.close()
+
+
+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
+
+
+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 | None:
+    topic = await get_topic(db, topic_id)
+    if not topic:
+        return None
+
+    if session_id and session_id != "new":
+        if await is_session_closed(db, session_id):
+            return "SESSION_CLOSED"
+
+    request_id = _new_id()
+    now = _now()
+
+    if session_id == "new":
+        session_id = None
+
+    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,
+    sdk_conversation_id: str | None = None,
+) -> None:
+    now = _now()
+    await db.execute(
+        "UPDATE requests SET status = 'completed', result = ?, completed_at = ?, "
+        "sdk_conversation_id = ?, session_id = COALESCE(session_id, ?) WHERE id = ?",
+        (result, now, sdk_conversation_id, sdk_conversation_id, 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()
+
+
+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 get_session_sdk_id(db: aiosqlite.Connection, session_id: str) -> str | None:
+    cursor = await db.execute(
+        "SELECT sdk_conversation_id FROM requests "
+        "WHERE session_id = ? AND status = 'completed' AND sdk_conversation_id IS NOT NULL "
+        "ORDER BY completed_at DESC LIMIT 1",
+        (session_id,),
+    )
+    row = await cursor.fetchone()
+    return row[0] if row else None
+
+
+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

src/backend/main.py

@@ -0,0 +1,74 @@
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from slowapi import _rate_limit_exceeded_handler
+from slowapi.errors import RateLimitExceeded
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+
+from src.backend.api.dependencies import limiter
+from src.backend.config import settings
+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.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="/docs",
+    redoc_url="/redoc",
+    openapi_url="/openapi.json",
+    lifespan=lifespan,
+)
+
+PROTECTED_DOCS_PATHS = {"/docs", "/redoc"}
+
+
+class DocsAuthMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        if request.url.path in PROTECTED_DOCS_PATHS:
+            key = request.headers.get("X-Api-Key") or request.query_params.get("api_key")
+            if key != settings.api_key:
+                return JSONResponse(status_code=403, content={"detail": "API key required"})
+        return await call_next(request)
+
+
+app.add_middleware(DocsAuthMiddleware)
+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")

src/backend/services/queue.py

@@ -0,0 +1,106 @@
+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,
+    get_session_sdk_id,
+)
+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"])
+
+        # Only resume if there's a previous completed request with an SDK conversation ID
+        resume_id = None
+        if req["session_id"]:
+            resume_id = await get_session_sdk_id(db, req["session_id"])
+    finally:
+        await db.close()
+
+    try:
+        result = await worker.process(
+            prompt=req["prompt"],
+            system_prompt=topic["system_prompt"],
+            model=model,
+            session_id=resume_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"],
+                sdk_conversation_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)

src/backend/services/worker_client.py

@@ -0,0 +1,52 @@
+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 models(self) -> dict:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            resp = await client.get(f"{self.base_url}/models")
+            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

tests/conftest.py

@@ -23,6 +23,7 @@ async def setup_db():
     yield
     db = await get_db()
     try:
+        await db.execute("DELETE FROM closed_sessions")
         await db.execute("DELETE FROM requests")
         await db.execute("DELETE FROM topics")
         await db.commit()

tests/test_api_requests.py

@@ -0,0 +1,114 @@
+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()
+    # session_id is None at creation; gets set from SDK after completion
+    assert data["data"]["session_id"] is None
+
+
+@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

tests/test_api_sessions.py

@@ -0,0 +1,74 @@
+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):
+    # Use a custom session_id (not "new") so it's set immediately
+    session_id = "test-session-abc"
+    await client.post(
+        "/api/requests",
+        json={"prompt": "First", "topic_id": topic_id, "session_id": 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):
+    session_id = "test-session-delete"
+    await client.post(
+        "/api/requests",
+        json={"prompt": "Hello", "topic_id": topic_id, "session_id": session_id},
+    )
+
+    resp = await client.delete(f"/api/sessions/{session_id}")
+    assert resp.status_code == 200
+
+    resp = await client.post(
+        "/api/requests",
+        json={"prompt": "Again", "topic_id": topic_id, "session_id": session_id},
+    )
+    assert resp.status_code == 410

tests/test_api_topics.py

@@ -0,0 +1,205 @@
+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
+
+
+@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
+
+
+@pytest.mark.asyncio
+async def test_summarize_topic(client):
+    create_resp = await client.post(
+        "/api/topics",
+        json={"name": "summarize-test", "system_prompt": "You help."},
+    )
+    topic_id = create_resp.json()["data"]["id"]
+
+    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()
+
+    from unittest.mock import patch, AsyncMock, MagicMock
+
+    mock_worker = MagicMock()
+    mock_response = MagicMock()
+    mock_response.status_code = 200
+    mock_response.json.return_value = {
+        "success": True,
+        "summary": "Comprehensive summary of all exchanges.",
+    }
+
+    mock_client = AsyncMock()
+    mock_client.post.return_value = mock_response
+    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+    mock_client.__aexit__ = AsyncMock(return_value=False)
+
+    with patch("src.backend.services.worker_client.httpx.AsyncClient", return_value=mock_client):
+        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"]
+
+
+@pytest.mark.asyncio
+async def test_summarize_topic_no_requests(client):
+    create_resp = await client.post(
+        "/api/topics",
+        json={"name": "empty-topic", "system_prompt": "Empty."},
+    )
+    topic_id = create_resp.json()["data"]["id"]
+
+    resp = await client.post(f"/api/topics/{topic_id}/summarize")
+    assert resp.status_code == 400

tests/test_services.py

@@ -1,6 +1,7 @@
 from src.backend.config import settings
 
 import pytest
+from unittest.mock import AsyncMock, MagicMock, patch
 
 
 def test_settings_loads_api_key():
@@ -109,3 +110,151 @@ def test_envelope_fail():
     result = fail("NOT_FOUND", "Resource not found")
     assert result["success"] is False
     assert result["error"]["code"] == "NOT_FOUND"
+
+
+@pytest.mark.asyncio
+async def test_worker_client_process():
+    from src.backend.services.worker_client import WorkerClient
+
+    mock_response = MagicMock()
+    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 = MagicMock()
+    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
+
+
+@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-queue", "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()

uv.lock

@@ -226,6 +226,12 @@ dev = [
     { name = "pytest-asyncio" },
 ]
 
+[package.dev-dependencies]
+dev = [
+    { name = "pytest" },
+    { name = "pytest-asyncio" },
+]
+
 [package.metadata]
 requires-dist = [
     { name = "aiosqlite", specifier = ">=0.20" },
@@ -241,6 +247,12 @@ requires-dist = [
 ]
 provides-extras = ["dev"]
 
+[package.metadata.requires-dev]
+dev = [
+    { name = "pytest", specifier = ">=9.0.3" },
+    { name = "pytest-asyncio", specifier = ">=1.3.0" },
+]
+
 [[package]]
 name = "packaging"
 version = "26.2"

worker/Dockerfile

@@ -0,0 +1,14 @@
+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"]

worker/index.js

@@ -0,0 +1,133 @@
+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.get("/models", async (_req, res) => {
+  try {
+    const conv = query({
+      prompt: ".",
+      options: { model: "claude-sonnet-4-6", allowedTools: [] },
+    });
+    const models = await conv.supportedModels();
+    await conv.return();
+    res.json({ success: true, models });
+  } catch (error) {
+    res.json({ success: false, error: error.message });
+  }
+});
+
+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 conv = query({ prompt, options });
+    const messages = [];
+    try {
+      for await (const msg of conv) {
+        messages.push(msg);
+      }
+    } catch (_) {
+      // SDK throws after process exits; messages already collected
+    }
+
+    const resultMsg = messages.find((m) => m.type === "result");
+    if (!resultMsg) {
+      return res.json({ success: false, error: "No result received from Claude" });
+    }
+    if (resultMsg.is_error) {
+      return res.json({ success: false, error: resultMsg.result });
+    }
+
+    res.json({
+      success: true,
+      result: resultMsg.result || "",
+      session_id: resultMsg.session_id || 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 conv = query({
+      prompt: summaryPrompt,
+      options: {
+        model: DEFAULT_MODEL,
+        systemPrompt:
+          "You are a summarizer. Respond with only the summary, no preamble.",
+        allowedTools: [],
+      },
+    });
+    const msgs = [];
+    try {
+      for await (const msg of conv) {
+        msgs.push(msg);
+      }
+    } catch (_) {
+      // SDK throws after process exits; messages already collected
+    }
+
+    const resultMsg = msgs.find((m) => m.type === "result");
+    res.json({
+      success: true,
+      summary: (resultMsg?.result || "").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}`);
+});

worker/package.json

@@ -0,0 +1,10 @@
+{
+  "name": "opus-agent-worker",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "dependencies": {
+    "express": "^4.21.0",
+    "@anthropic-ai/claude-code": "^1.0.0"
+  }
+}