diff --git a/src/backend/db/database.py b/src/backend/db/database.py index 96f7d40..e12e3f5 100644 --- a/src/backend/db/database.py +++ b/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, @@ -156,7 +157,7 @@ async def create_request( now = _now() if session_id == "new": - session_id = _new_id() + session_id = None await db.execute( "INSERT INTO requests (id, session_id, topic_id, prompt, model, summarize, status, created_at) " @@ -246,19 +247,14 @@ async def complete_request( db: aiosqlite.Connection, request_id: str, result: str, - session_id: str | None = None, + sdk_conversation_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.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() @@ -297,6 +293,17 @@ async def list_sessions(db: aiosqlite.Connection) -> list[dict]: 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,) diff --git a/src/backend/services/queue.py b/src/backend/services/queue.py index a9b1d0b..e8bd7b5 100644 --- a/src/backend/services/queue.py +++ b/src/backend/services/queue.py @@ -9,6 +9,7 @@ from src.backend.db.database import ( complete_request, fail_request, append_topic_summary, + get_session_sdk_id, ) from src.backend.services.worker_client import WorkerClient @@ -38,6 +39,11 @@ async def process_one(worker: WorkerClient) -> bool: 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() @@ -46,7 +52,7 @@ async def process_one(worker: WorkerClient) -> bool: prompt=req["prompt"], system_prompt=topic["system_prompt"], model=model, - session_id=req["session_id"], + session_id=resume_id, ) except Exception as e: db = await get_db() @@ -63,7 +69,7 @@ async def process_one(worker: WorkerClient) -> bool: db, req["id"], result["result"], - session_id=result.get("session_id"), + sdk_conversation_id=result.get("session_id"), ) if req["summarize"]: diff --git a/worker/index.js b/worker/index.js index e201dbe..072cddc 100644 --- a/worker/index.js +++ b/worker/index.js @@ -33,19 +33,26 @@ app.post("/process", async (req, res) => { const conv = query({ prompt, options }); const messages = []; - for await (const msg of conv) { - messages.push(msg); + 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 && resultMsg.is_error) { + 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, + result: resultMsg.result || "", + session_id: resultMsg.session_id || session_id, }); } catch (error) { console.error("Process error:", error.message); @@ -85,8 +92,12 @@ app.post("/summarize", async (req, res) => { }, }); const msgs = []; - for await (const msg of conv) { - msgs.push(msg); + 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");