feat: add requests CRUD endpoints

src/backend/api/routes/requests.py

@@ -0,0 +1,90 @@
+from fastapi import APIRouter, Depends, Query
+from starlette.responses import JSONResponse
+
+from src.backend.api.dependencies import verify_api_key
+from src.backend.db.database import (
+    get_db,
+    create_request,
+    get_request,
+    list_requests,
+    delete_request,
+)
+from src.backend.models.api import RequestCreate, ok, fail
+
+router = APIRouter(
+    prefix="/requests",
+    tags=["requests"],
+    dependencies=[Depends(verify_api_key)],
+)
+
+
+async def _get_db():
+    db = await get_db()
+    try:
+        yield db
+    finally:
+        await db.close()
+
+
+@router.post("", summary="Create request")
+async def create(body: RequestCreate, db=Depends(_get_db)):
+    req = await create_request(
+        db,
+        prompt=body.prompt,
+        topic_id=body.topic_id,
+        session_id=body.session_id,
+        model=body.model,
+        summarize=body.summarize,
+    )
+    if req is None:
+        return JSONResponse(
+            status_code=404,
+            content=fail("NOT_FOUND", "Topic not found"),
+        )
+    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/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
+);
 """
 
 
@@ -129,3 +134,189 @@ async def delete_topic(db: aiosqlite.Connection, topic_id: str) -> bool:
     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

@@ -4,6 +4,8 @@ from fastapi import FastAPI
 
 from src.backend.api.routes.health import router as health_router
 from src.backend.api.routes.topics import router as topics_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.db.database import init_db
 
 
@@ -17,3 +19,5 @@ app = FastAPI(title="OpusAgent", version="0.1.0", lifespan=lifespan)
 
 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")

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