feat: add Docker setup and README

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", "start"]

README.md

@@ -0,0 +1,79 @@
+# 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"
+```

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:/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:

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,28 @@
+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,
+    })

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

@@ -33,6 +33,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 +64,259 @@ 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 = _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()
+
+
+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

src/backend/main.py

@@ -0,0 +1,57 @@
+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.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")

src/backend/services/queue.py

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

src/backend/services/worker_client.py

@@ -0,0 +1,47 @@
+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

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()
+    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

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):
+    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
+
+    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()

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,109 @@
+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}`);
+});

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"
+  }
+}