feat(talkbot): reprocess cited messages and raise pipeline timeouts

containers/talkbot/server.py

@@ -24,6 +24,7 @@ NC_USER = os.environ["NC_USER"]
 NC_PASSWORD = os.environ["NC_PASSWORD"]
 
 WHISPER_URL = os.environ.get("WHISPER_URL", "http://whisper:8080").rstrip("/")
+WHISPER_TIMEOUT = int(os.environ.get("WHISPER_TIMEOUT") or "300")
 LLAMACPP_URL = os.environ.get("LLAMACPP_URL", "http://llamacpp:7000").rstrip("/")
 LLAMACPP_MODEL = os.environ.get("LLAMACPP_MODEL", "").strip()
 LLAMACPP_API_KEY = os.environ.get("LLAMACPP_API_KEY", "").strip()
@@ -328,6 +329,28 @@ def extract_file_param(parameters) -> Optional[dict]:
     return None
 
 
+def is_bot_mentioned(parameters) -> bool:
+    if isinstance(parameters, dict):
+        values = list(parameters.values())
+    elif isinstance(parameters, list):
+        values = parameters
+    else:
+        values = []
+
+    for value in values:
+        if not isinstance(value, dict) or value.get("type") != "user":
+            continue
+
+        mentioned_id = str(value.get("id", "") or "")
+        if "/" in mentioned_id:
+            mentioned_id = mentioned_id.split("/", 1)[-1]
+
+        if mentioned_id == MANAGER_USER:
+            return True
+
+    return False
+
+
 def is_audio_file(file_param: dict) -> bool:
     mimetype = str(file_param.get("mimetype", "") or "").lower()
     if mimetype.startswith("audio/"):
@@ -370,6 +393,80 @@ def extract_reply_to_message_id(obj: dict) -> Optional[int]:
         return None
 
 
+def fetch_chat_message(token: str, message_id: int) -> Optional[dict]:
+    response = httpx.get(
+        f"{NC_URL}/ocs/v2.php/apps/spreed/api/v1/chat/{token}",
+        headers={
+            "OCS-APIRequest": "true",
+            "Accept": "application/json",
+        },
+        auth=(NC_USER, NC_PASSWORD),
+        params={
+            "lookIntoFuture": 0,
+            "limit": 1,
+            "lastKnownMessageId": message_id,
+            "includeLastKnown": 1,
+        },
+        timeout=30,
+    )
+
+    if response.status_code == 304:
+        return None
+
+    response.raise_for_status()
+
+    messages = response.json().get("ocs", {}).get("data", [])
+
+    for message in messages:
+        if isinstance(message, dict) and int(message.get("id", -1)) == message_id:
+            return message
+
+    return None
+
+
+def build_reprocess_job(token: str, message_id: int) -> Optional[dict]:
+    """Build a job that reprocesses an already-posted message.
+
+    Used when someone replies to a message mentioning the bot instead of
+    sending a new message: we fetch the cited message from Nextcloud Talk
+    and re-run the pipeline on it exactly as if it had just arrived, be it
+    a voice message or a text message.
+    """
+    message = fetch_chat_message(token, message_id)
+
+    if not message:
+        return None
+
+    parameters = message.get("messageParameters", {})
+
+    file_param = extract_file_param(parameters)
+    if file_param and not is_audio_file(file_param):
+        file_param = None
+
+    message_text = "" if file_param else str(message.get("message") or "").strip()
+
+    if not file_param and not message_text:
+        return None
+
+    sender_user_id = actor_user_id({"id": message.get("actorId", "")})
+    sender_display_name = (
+        message.get("actorDisplayName")
+        or sender_user_id
+        or "User"
+    )
+
+    return {
+        "token": token,
+        "message_id": int(message_id),
+        "sender_user_id": sender_user_id,
+        "sender_display_name": sender_display_name,
+        "file_param": file_param,
+        "message_text": message_text,
+        "preserve_original": False,
+        "reply_to_message_id": None,
+    }
+
+
 def get_conversation_participants(token: str) -> list[dict]:
     response = httpx.get(
         f"{NC_URL}/ocs/v2.php/apps/spreed/api/v4/room/{token}/participants",
@@ -610,7 +707,7 @@ def transcribe(audio_bytes: bytes) -> tuple[str, str]:
         data={
             "response_format": "json",
         },
-        timeout=300,
+        timeout=WHISPER_TIMEOUT,
     )
 
     if response.is_error:
@@ -1435,6 +1532,7 @@ def health():
     return {
         "status": "ok",
         "manager": MANAGER_USER,
+        "whisper_timeout": WHISPER_TIMEOUT,
         "llamacpp_timeout": LLAMACPP_TIMEOUT,
         "llamacpp_thinking": LLAMACPP_ENABLE_THINKING,
         "qwen_tts_timeout": QWEN_TTS_TIMEOUT,
@@ -1551,6 +1649,49 @@ async def webhook(
         parameters,
     )
 
+    if (
+        not file_param
+        and reply_to_message_id
+        and is_bot_mentioned(parameters)
+    ):
+        reprocess_job = build_reprocess_job(
+            token,
+            reply_to_message_id,
+        )
+
+        if reprocess_job:
+            enqueue_event(reprocess_job)
+
+            log.info(
+                "Queued reprocessing of message %s (requested by %s citing "
+                "the bot in message %s)",
+                reply_to_message_id,
+                sender_user_id,
+                message_id,
+            )
+
+            return JSONResponse(
+                {
+                    "status": "queued",
+                    "type": "voice-reprocess" if reprocess_job["file_param"] else "text-reprocess",
+                    "reprocessed_message_id": reply_to_message_id,
+                }
+            )
+
+        log.info(
+            "Message %s mentions the bot but cited message %s has no "
+            "content to reprocess",
+            message_id,
+            reply_to_message_id,
+        )
+
+        return JSONResponse(
+            {
+                "status": "ignored",
+                "reason": "cited message has no content to reprocess",
+            }
+        )
+
     preserve_original = False
 
     if file_param and not is_audio_file(file_param):

containers/talkbot/talkbot.container

@@ -31,19 +31,27 @@ Environment=NC_BOT_SECRET=changeme
 Environment=NC_USER=changeme
 Environment=NC_PASSWORD=changeme
 
-# --- Backend services (adjust host:port to match your setup/network) ---
+# Whisper
 Environment=WHISPER_URL=http://whisper:8080
+Environment=WHISPER_TIMEOUT=1800
+
+# LLAMAcpp
 Environment=LLAMACPP_URL=http://llamacpp:8090
 #Environment=LLAMACPP_MODEL=
 #Environment=LLAMACPP_API_KEY=
 Environment=LLAMACPP_MAX_TOKENS=8192
 Environment=LLAMACPP_ENABLE_THINKING=false
-#Environment=LOG_WEBHOOK_PAYLOAD=false
+Environment=LLAMACPP_TIMEOUT=1800
+
+# QWEN
 Environment=QWEN_TTS_URL=http://qwen-tts:8000
+Environment=QWEN_TTS_TIMEOUT=5400
+Environment=QWEN_TTS_SPEAKER=Ryan
+
 
 # --- Behaviour ---
+#Environment=LOG_WEBHOOK_PAYLOAD=false
 Environment=DEFAULT_TARGET_LANGUAGE=it
-Environment=QWEN_TTS_SPEAKER=Ryan
 
 [Service]
 Restart=on-failure