Adriano Dal Pastro
9b68f772f6
- Add sdk_conversation_id column to track SDK conversation separately from user session_id - Queue processor looks up SDK conversation ID for resume instead of passing user session_id - session_id "new" stores NULL initially, gets set from SDK on completion - Custom session_ids are preserved (not overwritten by SDK) - Worker catches SDK process exit error after messages are collected Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
from src.backend.config import settings
|
|
from src.backend.db.database import (
|
|
get_db,
|
|
pick_next_pending,
|
|
get_topic,
|
|
complete_request,
|
|
fail_request,
|
|
append_topic_summary,
|
|
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)
|