fix: router - remove response_format for Qwen thinking model compatibility

qwen/qwen3.6-27b rejects response_format={'type':'json_object'} with
a 400 json_validate_failed error, causing both primary and fallback
router calls to fail and always defaulting to PATH_A.

Fixes:
- Remove response_format from groq_call in classify_query()
- Add _extract_json_from_text() to handle:
* <think>...</think> blocks that Qwen prepends to output
* Regex fallback extraction if json.loads fails
- Increase router max_tokens 80 -> 500 to allow full thinking output
- Add _VALID_CLASSIFICATIONS guard to catch garbage model outputs
- Config: qwen/qwen3.6-27b set as route model (user preference)

services/agent_loop.py

@@ -48,7 +48,7 @@ from services.tools import REAL_ESTATE_TOOLS, COMMERCIAL_TOOLS
 
 logger = setup_logger(__name__)
 
-MAX_TOOL_ROUNDS = 4   # Maximum tool-call rounds before forcing synthesis
+MAX_TOOL_ROUNDS = 4  # Maximum tool-call rounds before forcing synthesis
 
 # ─── STATIC RESPONSES ─────────────────────────────────────────────────────────
 
@@ -69,41 +69,96 @@ _REPLY_FALLBACK = (
 
 # ─── ROUTER ───────────────────────────────────────────────────────────────────
 
+
+# Valid classification labels - used to validate router output
+_VALID_CLASSIFICATIONS = frozenset({
+    "OUT_OF_SCOPE", "PATH_A", "PATH_B", "BOTH", "GREETING", "COMMERCIAL_HANDOFF"
+})
+
+
+def _extract_json_from_text(text: str) -> dict:
+    """
+    Extract a JSON object from raw model output.
+
+    Handles two patterns:
+      1. Thinking models (qwen): output wrapped in <think>...</think> before JSON.
+      2. Standard: clean JSON object in the content.
+
+    Falls back to regex extraction if json.loads fails on the full text.
+    """
+    if not text:
+        return {}
+
+    # Strip <think>...</think> block if present (Qwen thinking models)
+    clean = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
+
+    # Try direct parse
+    try:
+        return json.loads(clean)
+    except json.JSONDecodeError:
+        pass
+
+    # Regex: find first {...} block
+    match = re.search(r"\{[^{}]*\}", clean, re.DOTALL)
+    if match:
+        try:
+            return json.loads(match.group(0))
+        except json.JSONDecodeError:
+            pass
+
+    return {}
+
+
 async def classify_query(user_query: str) -> str:
     """
-    Use a small, fast model to classify the query into one of 6 route tags.
-    Fix #4: Previously used a 20B reasoning model (484ms). Now uses the
-    lightweight GROQ_ROUTE_MODEL (default: llama-3.1-8b-instant, ~120ms).
+    Classify the user query into one of 6 route tags using a fast router model.
+
+    NOTE: response_format={"type": "json_object"} is intentionally NOT used.
+    Qwen thinking models (qwen/qwen3.6-27b) reject JSON mode with a 400
+    json_validate_failed error. We ask for JSON via the prompt instead and
+    extract it with _extract_json_from_text() which handles <think> blocks
+    and falls back to regex extraction.
     """
     messages = [
         {"role": "system", "content": ROUTER_PROMPT},
         {"role": "user", "content": user_query},
     ]
-    # Try primary route model, fall back to secondary
     for model in [GROQ_ROUTE_MODEL, GROQ_FALLBACK_ROUTE_MODEL]:
         try:
             res = await groq_call(
                 model=model,
                 messages=messages,
-                max_tokens=80,
+                max_tokens=500,   # Allow thinking models to finish <think> block
                 temperature=0.0,
-                response_format={"type": "json_object"},
+                # No response_format - unsupported by Qwen thinking models
             )
             raw = res.choices[0].message.content or "{}"
-            data = json.loads(raw)
-            classification = data.get("classification", "PATH_A")
-            logger.info(f"[router] {model} -> {classification}: {data.get('reason', '')}")
+            data = _extract_json_from_text(raw)
+            classification = data.get("classification", "")
+
+            if classification not in _VALID_CLASSIFICATIONS:
+                logger.warning(
+                    f"[router] {model} returned invalid classification '{classification}', defaulting PATH_A"
+                )
+                classification = "PATH_A"
+
+            logger.info(
+                f"[router] {model} -> {classification}: {data.get('reason', '')}"
+            )
             return classification
+
         except Exception as exc:
             if is_structural_error(exc):
                 logger.warning(f"[router] structural error on {model}: {exc}")
             else:
                 logger.warning(f"[router] {model} failed, trying fallback: {exc}")
-    return "PATH_A"   # Safe default
+
+    return "PATH_A"  # Safe default
 
 
 # ─── AGENTIC TOOL LOOP ────────────────────────────────────────────────────────
 
+
 async def run_agent_loop(
     messages: list[dict],
     active_tools: list | None,
@@ -144,15 +199,19 @@ async def run_agent_loop(
                     # or schema mismatch — recover by executing the tool directly.
                     fn, fn_args = parse_failed_generation(failed_gen)
                     if fn and isinstance(fn_args, dict):
-                        logger.info(f"[agent_loop] recovering tool={fn} from failed_generation")
+                        logger.info(
+                            f"[agent_loop] recovering tool={fn} from failed_generation"
+                        )
                         tool_result = await execute_tool_by_name(fn, fn_args)
                         tool_results.append({"tool": fn, "args": fn_args})
-                        messages.append({
-                            "tool_call_id": f"recov_{int(time.time())}",
-                            "role": "tool",
-                            "name": fn,
-                            "content": compress_tool_output(fn, tool_result),
-                        })
+                        messages.append(
+                            {
+                                "tool_call_id": f"recov_{int(time.time())}",
+                                "role": "tool",
+                                "name": fn,
+                                "content": compress_tool_output(fn, tool_result),
+                            }
+                        )
                         # After injecting result, break to top of outer loop to retry
                         break
                 logger.warning(f"[agent_loop] {model} error round={round_num}: {exc}")
@@ -171,18 +230,20 @@ async def run_agent_loop(
         if finish_reason == "tool_calls":
             messages.append(normalize_assistant_message(response_message))  # Fix #5
             for tc in response_message.tool_calls:
-                fn   = tc.function.name
+                fn = tc.function.name
                 args = parse_tool_args(tc.function.arguments)
                 logger.info(f"[agent_loop] executing tool={fn} args={args}")
 
                 result = await execute_tool_by_name(fn, args)
                 tool_results.append({"tool": fn, "args": args})
-                messages.append({
-                    "tool_call_id": tc.id,
-                    "role": "tool",
-                    "name": fn,
-                    "content": compress_tool_output(fn, result),
-                })
+                messages.append(
+                    {
+                        "tool_call_id": tc.id,
+                        "role": "tool",
+                        "name": fn,
+                        "content": compress_tool_output(fn, result),
+                    }
+                )
             # Continue loop — model may want more tools or synthesis next round
 
         # ── Direct text reply — done ──────────────────────────────────────────
@@ -193,13 +254,15 @@ async def run_agent_loop(
     logger.warning("[agent_loop] max rounds reached, forcing synthesis")
     for model in [GROQ_SYNTHESIS_MODEL, GROQ_FALLBACK_SYNTHESIS_MODEL]:
         try:
-            synth_messages = list(messages) + [{
-                "role": "system",
-                "content": (
-                    "Synthesize the real estate data gathered above into a clear, "
-                    "concise Markdown reply for the user. Do NOT call any more tools."
-                ),
-            }]
+            synth_messages = list(messages) + [
+                {
+                    "role": "system",
+                    "content": (
+                        "Synthesize the real estate data gathered above into a clear, "
+                        "concise Markdown reply for the user. Do NOT call any more tools."
+                    ),
+                }
+            ]
             final_res = await groq_call(
                 model=model,
                 messages=synth_messages,
@@ -215,6 +278,7 @@ async def run_agent_loop(
             fn, fn_args = parse_failed_generation(failed_gen)
             if fn == "generate_data_export" and isinstance(fn_args, dict):
                 from services.appwrite_service import upload_document_to_appwrite
+
                 content = fn_args.get("content", "")
                 upload_res = await upload_document_to_appwrite(content, "md")
                 dl_url = upload_res.get("download_url", "")
@@ -228,6 +292,7 @@ async def run_agent_loop(
 
 # ─── STREAMING AGENT LOOP (for SSE) ───────────────────────────────────────────
 
+
 async def run_agent_loop_streaming(
     messages: list[dict],
     active_tools: list | None,
@@ -269,12 +334,14 @@ async def run_agent_loop_streaming(
                     if fn and isinstance(fn_args, dict):
                         tool_result = await execute_tool_by_name(fn, fn_args)
                         tool_results.append({"tool": fn, "args": fn_args})
-                        messages.append({
-                            "tool_call_id": f"recov_{int(time.time())}",
-                            "role": "tool",
-                            "name": fn,
-                            "content": compress_tool_output(fn, tool_result),
-                        })
+                        messages.append(
+                            {
+                                "tool_call_id": f"recov_{int(time.time())}",
+                                "role": "tool",
+                                "name": fn,
+                                "content": compress_tool_output(fn, tool_result),
+                            }
+                        )
                         yield {"type": "tool_call", "tool": fn, "args": fn_args}
                         break
                 logger.warning(f"[stream_loop] {model} error round={round_num}: {exc}")
@@ -286,17 +353,19 @@ async def run_agent_loop_streaming(
         if has_tool_calls:
             messages.append(normalize_assistant_message(response_message))
             for tc in response_message.tool_calls:
-                fn   = tc.function.name
+                fn = tc.function.name
                 args = parse_tool_args(tc.function.arguments)
                 yield {"type": "tool_call", "tool": fn, "args": args}
                 result = await execute_tool_by_name(fn, args)
                 tool_results.append({"tool": fn, "args": args})
-                messages.append({
-                    "tool_call_id": tc.id,
-                    "role": "tool",
-                    "name": fn,
-                    "content": compress_tool_output(fn, result),
-                })
+                messages.append(
+                    {
+                        "tool_call_id": tc.id,
+                        "role": "tool",
+                        "name": fn,
+                        "content": compress_tool_output(fn, result),
+                    }
+                )
         else:
             # Model returned a direct text reply — stream it token by token
             if response_message.content:
@@ -308,10 +377,12 @@ async def run_agent_loop_streaming(
     # Forced synthesis with streaming
     for model in [GROQ_SYNTHESIS_MODEL, GROQ_FALLBACK_SYNTHESIS_MODEL]:
         try:
-            synth_messages = list(messages) + [{
-                "role": "system",
-                "content": "Synthesize the real estate data gathered above into a clear Markdown reply. Do NOT call any more tools.",
-            }]
+            synth_messages = list(messages) + [
+                {
+                    "role": "system",
+                    "content": "Synthesize the real estate data gathered above into a clear Markdown reply. Do NOT call any more tools.",
+                }
+            ]
 
             def _stream_call():
                 return groq_client.chat.completions.create(
@@ -323,13 +394,17 @@ async def run_agent_loop_streaming(
                 )
 
             from services.groq_client import groq_client as _gc
-            stream = await loop.run_in_executor(None, lambda: _gc.chat.completions.create(
-                model=model,
-                messages=synth_messages,
-                temperature=0.2,
-                max_tokens=1000,
-                stream=True,
-            ))
+
+            stream = await loop.run_in_executor(
+                None,
+                lambda: _gc.chat.completions.create(
+                    model=model,
+                    messages=synth_messages,
+                    temperature=0.2,
+                    max_tokens=1000,
+                    stream=True,
+                ),
+            )
             for chunk in stream:
                 delta = chunk.choices[0].delta.content or ""
                 if delta:
@@ -358,6 +433,7 @@ def _split_tokens(text: str, chunk_size: int = 4):
 
 # ─── CLARIFICATION PARSER ─────────────────────────────────────────────────────
 
+
 def extract_clarification_options(reply: str) -> tuple[str, list[str]]:
     """Strip embedded JSON clarification options from the reply text."""
     suggested_actions: list[str] = []
@@ -373,12 +449,13 @@ def extract_clarification_options(reply: str) -> tuple[str, list[str]]:
             suggested_actions = data.get("clarification_options", [])
             reply = reply.replace(json_match.group(0), "").strip()
         except json.JSONDecodeError:
-            pass
+            logger.error(f"Failed to parse clarification options from reply: {reply}")
     return reply, suggested_actions
 
 
 # ─── PUBLIC ENTRY POINT ───────────────────────────────────────────────────────
 
+
 @observe(name="process-chat")
 async def process_chat_message(
     user_query: str,
@@ -397,13 +474,21 @@ async def process_chat_message(
 
         if classification == "OUT_OF_SCOPE":
             get_client().update_current_span(output=_REPLY_OUT_OF_SCOPE)
-            return {"reply": _REPLY_OUT_OF_SCOPE, "path_used": "OUT_OF_SCOPE",
-                    "tools_called": [], "suggested_actions": []}
+            return {
+                "reply": _REPLY_OUT_OF_SCOPE,
+                "path_used": "OUT_OF_SCOPE",
+                "tools_called": [],
+                "suggested_actions": [],
+            }
 
         if classification == "GREETING":
             get_client().update_current_span(output=_REPLY_GREETING)
-            return {"reply": _REPLY_GREETING, "path_used": "GREETING",
-                    "tools_called": [], "suggested_actions": []}
+            return {
+                "reply": _REPLY_GREETING,
+                "path_used": "GREETING",
+                "tools_called": [],
+                "suggested_actions": [],
+            }
 
         # STEP 2: RAG retrieval for knowledge-base paths
         rag_chunks: list[str] = []
@@ -413,17 +498,20 @@ async def process_chat_message(
         # STEP 3: Build message context
         system_prompt = build_system_prompt(classification)  # Fix #3
         messages: list[dict] = [{"role": "system", "content": system_prompt}]
-        messages.extend(get_context_window(session_id))       # Fix #6 (trimmed history)
+        messages.extend(get_context_window(session_id))  # Fix #6 (trimmed history)
 
         user_msg = f"User Context Filters: {json.dumps(session_context)}\nUser Query: {user_query}"
         messages.append({"role": "user", "content": user_msg})
         append_message(session_id, {"role": "user", "content": user_msg})
 
         if rag_chunks:
-            messages.append({
-                "role": "system",
-                "content": "Retrieved Methodology Context:\n" + "\n---\n".join(rag_chunks),
-            })
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Retrieved Methodology Context:\n"
+                    + "\n---\n".join(rag_chunks),
+                }
+            )
 
         # STEP 4: Select tools
         active_tools = None
@@ -436,7 +524,11 @@ async def process_chat_message(
         tool_results: list[dict] = []
         final_reply = await run_agent_loop(messages, active_tools, tool_results)
 
-        if not final_reply or final_reply.strip().startswith('{"name":') or final_reply.strip().startswith("<function="):
+        if (
+            not final_reply
+            or final_reply.strip().startswith('{"name":')
+            or final_reply.strip().startswith("<function=")
+        ):
             final_reply = _REPLY_FALLBACK
 
         # STEP 6: Store assistant reply in session
@@ -465,22 +557,31 @@ async def stream_chat_message(
     SSE streaming handler.
     Yields: {"type": "status"|"tool_call"|"token"|"done", ...}
     """
-    with propagate_attributes(session_id=session_id, tags=["real-estate-chat", "stream"]):
-
+    with propagate_attributes(
+        session_id=session_id, tags=["real-estate-chat", "stream"]
+    ):
         # STEP 1: Route
         classification = await classify_query(user_query)
         yield {"type": "status", "classification": classification}
 
         if classification == "OUT_OF_SCOPE":
             yield {"type": "token", "token": _REPLY_OUT_OF_SCOPE}
-            yield {"type": "done", "tools_called": [], "suggested_actions": [],
-                   "path_used": "OUT_OF_SCOPE"}
+            yield {
+                "type": "done",
+                "tools_called": [],
+                "suggested_actions": [],
+                "path_used": "OUT_OF_SCOPE",
+            }
             return
 
         if classification == "GREETING":
             yield {"type": "token", "token": _REPLY_GREETING}
-            yield {"type": "done", "tools_called": [], "suggested_actions": [],
-                   "path_used": "GREETING"}
+            yield {
+                "type": "done",
+                "tools_called": [],
+                "suggested_actions": [],
+                "path_used": "GREETING",
+            }
             return
 
         # STEP 2: RAG
@@ -496,10 +597,13 @@ async def stream_chat_message(
         messages.append({"role": "user", "content": user_msg})
         append_message(session_id, {"role": "user", "content": user_msg})
         if rag_chunks:
-            messages.append({
-                "role": "system",
-                "content": "Retrieved Methodology Context:\n" + "\n---\n".join(rag_chunks),
-            })
+            messages.append(
+                {
+                    "role": "system",
+                    "content": "Retrieved Methodology Context:\n"
+                    + "\n---\n".join(rag_chunks),
+                }
+            )
 
         # STEP 4: Tools
         active_tools = None
@@ -512,7 +616,9 @@ async def stream_chat_message(
         tool_results: list[dict] = []
         full_reply_parts: list[str] = []
 
-        async for event in run_agent_loop_streaming(messages, active_tools, tool_results):
+        async for event in run_agent_loop_streaming(
+            messages, active_tools, tool_results
+        ):
             if event["type"] == "token":
                 full_reply_parts.append(event["token"])
             yield event

services/prompts.py

@@ -23,7 +23,10 @@ Analyze the user query and classify it into EXACTLY ONE of six classifications:
 6. "COMMERCIAL_HANDOFF": Query asks about getting started, hiring Joule Dynamics, custom builds, custom dashboards, pricing for software, or requests tracking for their own specific portfolio outside the demo scope.
 
 Respond ONLY with valid JSON matching this schema:
-{"classification": "OUT_OF_SCOPE" | "PATH_A" | "PATH_B" | "BOTH" | "GREETING" | "COMMERCIAL_HANDOFF", "reason": "1-sentence justification"}
+{
+    "classification": "OUT_OF_SCOPE" | "PATH_A" | "PATH_B" | "BOTH" | "GREETING" | "COMMERCIAL_HANDOFF", 
+    "reason": "1-sentence justification"
+}
 """
 
 # ─── SYNTHESIS PROMPT — CORE (always included) ────────────────────────────────
@@ -103,6 +106,7 @@ When a user asks about custom builds, deploying this system for their business,
 
 # ─── COMPOSER ─────────────────────────────────────────────────────────────────
 
+
 def build_system_prompt(classification: str) -> str:
     """
     Compose the system prompt from core + relevant extension blocks only.