feat(talkbot): split voice/text queues and trim verbose log output

containers/talkbot/server.py

@@ -38,6 +38,7 @@ LOG_WEBHOOK_PAYLOAD = (
     os.environ.get("LOG_WEBHOOK_PAYLOAD", "false").lower()
     in {"1", "true", "yes", "on"}
 )
+LOG_TEXT_PREVIEW_CHARS = int(os.environ.get("LOG_TEXT_PREVIEW_CHARS") or "120")
 QWEN_TTS_URL = os.environ.get("QWEN_TTS_URL", "http://qwen-tts:8000").rstrip("/")
 QWEN_TTS_TIMEOUT = int(os.environ.get("QWEN_TTS_TIMEOUT") or "1800")
 
@@ -85,9 +86,14 @@ MANAGER_USER = NC_USER
 
 app = FastAPI(title="Nextcloud Talk voice translator")
 
-# Events are processed one at a time by a single background worker: the
-# pipeline (whisper -> llama.cpp -> qwen-tts) is heavy and the host has
-# limited resources, so we serialize instead of processing concurrently.
+# Events are processed by two background workers, one per job type:
+# - "voice" jobs run the heavy whisper -> llama.cpp -> qwen-tts pipeline
+#   and are processed one at a time, since whisper/qwen-tts are the
+#   resource-heavy steps and the host has limited GPU/CPU for them.
+# - "text" jobs only call llama.cpp for translation, which is fast and
+#   uses its own service, so they are processed on a separate worker and
+#   never wait behind a long-running voice job.
+# Each worker still processes its own queue serially.
 # The webhook handler only validates/parses the payload (no blocking I/O)
 # and enqueues the job, so Nextcloud always gets an immediate response
 # and never times out waiting for a translation to finish.
@@ -99,6 +105,10 @@ app = FastAPI(title="Nextcloud Talk voice translator")
 new_job_signal = threading.Event()
 
 
+def job_type_of(job: dict) -> str:
+    return "voice" if job.get("file_param") else "text"
+
+
 def init_queue_db() -> None:
     os.makedirs(os.path.dirname(QUEUE_DB_PATH), exist_ok=True)
 
@@ -127,16 +137,31 @@ def init_queue_db() -> None:
         if "next_attempt_at" not in existing_columns:
             conn.execute("ALTER TABLE queue ADD COLUMN next_attempt_at TEXT")
 
+        if "job_type" not in existing_columns:
+            conn.execute("ALTER TABLE queue ADD COLUMN job_type TEXT")
+
+            # Backfill rows left over from before job_type existed so they
+            # are picked up by the right worker instead of neither.
+            for row_id, payload in conn.execute(
+                "SELECT id, payload FROM queue WHERE job_type IS NULL"
+            ):
+                job_type = job_type_of(json.loads(payload))
+                conn.execute(
+                    "UPDATE queue SET job_type = ? WHERE id = ?",
+                    (job_type, row_id),
+                )
+
         conn.commit()
 
 
 def enqueue_event(job: dict) -> None:
     with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn:
         conn.execute(
-            "INSERT INTO queue (payload, created_at) VALUES (?, ?)",
+            "INSERT INTO queue (payload, created_at, job_type) VALUES (?, ?, ?)",
             (
                 json.dumps(job),
                 datetime.now(timezone.utc).isoformat(),
+                job_type_of(job),
             ),
         )
         conn.commit()
@@ -144,17 +169,18 @@ def enqueue_event(job: dict) -> None:
     new_job_signal.set()
 
 
-def dequeue_next_event() -> Optional[tuple[int, dict, int]]:
+def dequeue_next_event(job_type: str) -> Optional[tuple[int, dict, int]]:
     now = datetime.now(timezone.utc).isoformat()
 
     with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn:
         row = conn.execute(
             """
             SELECT id, payload, attempts FROM queue
-            WHERE next_attempt_at IS NULL OR next_attempt_at <= ?
+            WHERE job_type = ?
+              AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
             ORDER BY id LIMIT 1
             """,
-            (now,),
+            (job_type, now),
         ).fetchone()
 
     if not row:
@@ -245,6 +271,13 @@ def language_short_code(language: str) -> str:
     return LANGUAGE_CODES.get(code, code.upper()[:8])
 
 
+def truncate_for_log(text: str) -> str:
+    text = text or ""
+    if len(text) <= LOG_TEXT_PREVIEW_CHARS:
+        return text
+    return text[:LOG_TEXT_PREVIEW_CHARS] + "…"
+
+
 def sanitize_filename_part(value: str) -> str:
     value = (value or "").strip()
     value = re.sub(r"[^\w.-]+", "_", value, flags=re.UNICODE)
@@ -1170,7 +1203,7 @@ def process_voice_message(
     log.info(
         "Transcribed (%s): %s",
         detected_language,
-        text,
+        truncate_for_log(text),
     )
 
     if is_blank_transcription(text):
@@ -1232,7 +1265,7 @@ def process_voice_message(
         log.info(
             "Translated (%s): %s",
             recipient_language_name,
-            translated_text,
+            truncate_for_log(translated_text),
         )
 
         translations.append(
@@ -1416,7 +1449,7 @@ def process_event(job: dict) -> None:
                 "Translated text for %s (%s): %s",
                 recipient,
                 language_name(target_language),
-                translated,
+                truncate_for_log(translated),
             )
 
             if translated.strip().lower() == message_text.lower():
@@ -1472,9 +1505,9 @@ def notify_processing_failure(job: dict) -> None:
         )
 
 
-def event_worker() -> None:
+def event_worker(job_type: str) -> None:
     while True:
-        job_row = dequeue_next_event()
+        job_row = dequeue_next_event(job_type)
 
         if job_row is None:
             new_job_signal.wait(timeout=5)
@@ -1488,7 +1521,8 @@ def event_worker() -> None:
             process_event(job)
         except Exception as exc:
             log.exception(
-                "Error processing message %s (attempt %s/%s): %s",
+                "Error processing %s message %s (attempt %s/%s): %s",
+                job_type,
                 job.get("message_id"),
                 attempts,
                 MAX_ATTEMPTS,
@@ -1500,7 +1534,8 @@ def event_worker() -> None:
                 continue
 
             log.error(
-                "Giving up on message %s after %s attempts",
+                "Giving up on %s message %s after %s attempts",
+                job_type,
                 job.get("message_id"),
                 attempts,
             )
@@ -1522,7 +1557,15 @@ def start_event_worker() -> None:
 
     threading.Thread(
         target=event_worker,
-        name="talkbot-event-worker",
+        args=("voice",),
+        name="talkbot-voice-worker",
+        daemon=True,
+    ).start()
+
+    threading.Thread(
+        target=event_worker,
+        args=("text",),
+        name="talkbot-text-worker",
         daemon=True,
     ).start()