fix: Python 3.11 compat - debug_agent f-string + reasoning_agent import

- Fixed debug_agent.py: Removed backslash from f-string expressions (Python 3.11 doesn't allow backslash in f-string expressions before 3.12)
- Fixed reasoning_agent.py: Changed incorrect import 'from core.agent import BaseAgent' to 'from .base_agent import BaseAgent'
- Fixed reasoning_agent.py: Updated __init__ and run() to match BaseAgent interface
- These fixes were blocking HF Space deployment (RUNTIME_ERROR -> RUNNING)

backend/agents/debug_agent.py

@@ -44,12 +44,12 @@ class DebugAgent(BaseAgent):
         messages = [
             {"role": "system", "content": DEBUG_SYSTEM},
             {"role": "user", "content": (
-                f"Debug and fix this issue (attempt {attempt}):\n\n"
-                f"{task}\n\n"
-                f"Provide:\n"
-                f"1. Root cause analysis\n"
-                f"2. Exact fix (code/config)\n"
-                f"3. Prevention strategy"
+                "Debug and fix this issue (attempt " + str(attempt) + "):\n\n"
+                + task + "\n\n"
+                "Provide:\n"
+                "1. Root cause analysis\n"
+                "2. Exact fix (code/config)\n"
+                "3. Prevention strategy"
             )},
         ]
 
@@ -71,14 +71,19 @@ class DebugAgent(BaseAgent):
 
     async def analyze_error(self, error_output: str, source_code: str = "", task_id: str = "", session_id: str = "") -> Dict:
         """Deep error analysis with structured output."""
+        if source_code:
+            source_section = "Source Code:\n```\n" + source_code[:1000] + "\n```"
+        else:
+            source_section = ""
+
         messages = [
             {"role": "system", "content": DEBUG_SYSTEM},
             {"role": "user", "content": (
-                f"Analyze this error and provide structured diagnosis:\n\n"
-                f"Error:\n{error_output[:2000]}\n\n"
-                f"{'Source Code:\\n```\\n' + source_code[:1000] + '\\n```' if source_code else ''}\n\n"
-                f"Respond with JSON:\n"
-                f'{{"error_type": "...", "root_cause": "...", "fix": "...", "prevention": "...", "severity": "low|medium|high|critical"}}'
+                "Analyze this error and provide structured diagnosis:\n\n"
+                "Error:\n" + error_output[:2000] + "\n\n"
+                + source_section + "\n\n"
+                "Respond with JSON:\n"
+                '{"error_type": "...", "root_cause": "...", "fix": "...", "prevention": "...", "severity": "low|medium|high|critical"}'
             )},
         ]
         raw = await self.llm(messages, task_id=task_id, session_id=session_id, temperature=0.1, max_tokens=1000)
@@ -111,10 +116,10 @@ class DebugAgent(BaseAgent):
             messages = [
                 {"role": "system", "content": DEBUG_SYSTEM},
                 {"role": "user", "content": (
-                    f"Fix attempt {attempt}/{max_retries}:\n\n"
-                    f"Error: {current_error}\n\n"
-                    f"Code:\n```\n{current_code[:3000]}\n```\n\n"
-                    f"Return ONLY the fixed code."
+                    "Fix attempt " + str(attempt) + "/" + str(max_retries) + ":\n\n"
+                    "Error: " + current_error + "\n\n"
+                    "Code:\n```\n" + current_code[:3000] + "\n```\n\n"
+                    "Return ONLY the fixed code."
                 )},
             ]
 
@@ -143,7 +148,7 @@ class DebugAgent(BaseAgent):
             compile(code, "<string>", "exec")
             return {"valid": True, "error": ""}
         except SyntaxError as e:
-            return {"valid": False, "error": f"SyntaxError: {e}"}
+            return {"valid": False, "error": "SyntaxError: " + str(e)}
         except Exception as e:
             return {"valid": False, "error": str(e)}
 

backend/agents/reasoning_agent.py

@@ -1,24 +1,34 @@
 """
-🚀 GOD MODE+ v3 - Reasoning Agent
-Specialized agent for complex reasoning tasks using DeepSeek R1, Qwen QwQ, o1-mini
-Version: 3.0.0
+ReasoningAgent v7 — Complex reasoning, analysis, and multi-step problem solving
+GOD AGENT OS — Using DeepSeek R1, Qwen QwQ, o1-mini style reasoning
 """
 
 import asyncio
 import json
-from typing import Dict, Any, Optional
+from typing import Dict, Any, List, Optional
 
 import structlog
 
-from core.agent import BaseAgent
+from .base_agent import BaseAgent
 
 log = structlog.get_logger()
 
+REASONING_SYSTEM = """You are an elite autonomous reasoning and analysis agent.
+You excel at:
+- Multi-step reasoning with chain-of-thought
+- Complex problem decomposition and analysis
+- Mathematical and logical reasoning
+- Strategic planning and decision making
+- Root cause analysis
+
+Always think step by step, show your reasoning, and provide confident conclusions.
+"""
+
 
 class ReasoningAgent(BaseAgent):
     """
     Specialized agent for complex reasoning, analysis, and problem-solving tasks.
-    
+
     Capabilities:
     - Multi-step reasoning with chain-of-thought
     - Complex problem decomposition
@@ -27,259 +37,127 @@ class ReasoningAgent(BaseAgent):
     - Strategic planning
     """
 
-    def __init__(self, ws_manager, ai_router):
-        """Initialize Reasoning Agent."""
-        super().__init__(
-            name="ReasoningAgent",
-            color="🟦",
-            description="Complex reasoning and analysis",
-            ws_manager=ws_manager,
-            ai_router=ai_router,
-        )
-        self.reasoning_depth = 3  # Number of reasoning steps
+    def __init__(self, ws_manager=None, ai_router=None):
+        super().__init__("ReasoningAgent", ws_manager, ai_router)
+        self.reasoning_depth = 3
         self.max_reasoning_tokens = 16000
 
-    async def process(self, task: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Process reasoning task with multi-step reasoning.
-        """
-        user_message = task.get("content", "")
-        session_id = task.get("session_id", "")
-        context = task.get("context", {})
+    async def run(self, task: str, context: Dict = {}, **kwargs) -> str:
+        """Execute reasoning task."""
+        session_id = kwargs.get("session_id", "")
+        task_id = kwargs.get("task_id", "")
 
-        log.info("🧠 Reasoning Agent activated", message=user_message[:100])
+        await self.emit(task_id, "agent_start", {
+            "agent": "ReasoningAgent",
+            "task": task[:80],
+        }, session_id)
 
         try:
             # Step 1: Analyze the problem
-            analysis = await self._analyze_problem(user_message, context)
-            await self._broadcast(session_id, {
-                "type": "reasoning_step",
-                "step": "analysis",
-                "data": analysis,
-            })
-
-            # Step 2: Break down into sub-problems
-            sub_problems = await self._decompose_problem(user_message, analysis)
-            await self._broadcast(session_id, {
-                "type": "reasoning_step",
-                "step": "decomposition",
-                "data": sub_problems,
-            })
-
-            # Step 3: Solve each sub-problem
-            solutions = []
-            for i, sub_problem in enumerate(sub_problems):
-                solution = await self._solve_sub_problem(sub_problem, context)
-                solutions.append(solution)
-                await self._broadcast(session_id, {
-                    "type": "reasoning_step",
-                    "step": f"solution_{i+1}",
-                    "data": solution,
-                })
-
-            # Step 4: Synthesize final answer
-            final_answer = await self._synthesize_answer(
-                user_message,
-                analysis,
-                sub_problems,
-                solutions
-            )
-
-            await self._broadcast(session_id, {
-                "type": "reasoning_complete",
-                "answer": final_answer,
+            analysis = await self._analyze_problem(task, context, task_id, session_id)
+
+            # Step 2: Decompose if complex
+            if len(task.split()) > 30:
+                sub_problems = await self._decompose_problem(task, analysis, task_id, session_id)
+                solutions = []
+                for sub in sub_problems[:3]:
+                    sol = await self._solve_sub_problem(sub, context, task_id, session_id)
+                    solutions.append(sol)
+                result = await self._synthesize_answer(task, analysis, sub_problems, solutions, task_id, session_id)
+            else:
+                result = analysis
+
+            await self.emit(task_id, "reasoning_complete", {
+                "agent": "ReasoningAgent",
                 "reasoning_depth": self.reasoning_depth,
-            })
+            }, session_id)
 
-            return {
-                "success": True,
-                "agent": self.name,
-                "answer": final_answer,
-                "reasoning_steps": {
-                    "analysis": analysis,
-                    "sub_problems": sub_problems,
-                    "solutions": solutions,
-                },
-            }
+            return result
 
         except Exception as e:
-            log.error("❌ Reasoning Agent failed", error=str(e))
-            return {
-                "success": False,
-                "agent": self.name,
-                "error": str(e),
-            }
-
-    async def _analyze_problem(self, problem: str, context: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Analyze the problem using reasoning model.
-        """
-        prompt = f"""Analyze this problem and identify:
-1. Core problem statement
-2. Key constraints
-3. Required information
-4. Potential approaches
-
-Problem: {problem}
-
-Provide structured analysis."""
-
-        response = await self.ai_router.route(
-            prompt,
-            context={"task_type": "reasoning"},
-            optimize_for="quality"
-        )
-
-        return {
-            "problem_type": self._classify_problem(problem),
-            "complexity": self._estimate_complexity(problem),
-            "analysis": response.get("response", ""),
-        }
-
-    async def _decompose_problem(
-        self,
-        problem: str,
-        analysis: Dict[str, Any]
-    ) -> list:
-        """
-        Break down complex problem into manageable sub-problems.
-        """
-        prompt = f"""Based on this analysis, break down the problem into 3-5 specific sub-problems:
-
-Problem: {problem}
-Analysis: {json.dumps(analysis, indent=2)}
-
-List each sub-problem clearly and explain the dependencies."""
-
-        response = await self.ai_router.route(
-            prompt,
-            context={"task_type": "reasoning"},
-            optimize_for="quality"
+            log.error("ReasoningAgent failed", error=str(e))
+            return "Reasoning agent encountered an error: " + str(e)
+
+    async def _analyze_problem(self, problem: str, context: Dict, task_id: str, session_id: str) -> str:
+        """Analyze the problem using reasoning model."""
+        messages = [
+            {"role": "system", "content": REASONING_SYSTEM},
+            {"role": "user", "content": (
+                "Analyze this problem step by step:\n\n"
+                + problem + "\n\n"
+                "Provide:\n"
+                "1. Problem type and complexity\n"
+                "2. Key constraints and requirements\n"
+                "3. Step-by-step solution\n"
+                "4. Final answer/recommendation"
+            )},
+        ]
+        return await self.llm(
+            messages,
+            task_id=task_id,
+            session_id=session_id,
+            temperature=0.2,
+            max_tokens=self.max_reasoning_tokens,
         )
 
-        # Parse sub-problems from response
-        sub_problems = self._parse_sub_problems(response.get("response", ""))
-        return sub_problems
-
-    async def _solve_sub_problem(
-        self,
-        sub_problem: str,
-        context: Dict[str, Any]
-    ) -> Dict[str, Any]:
-        """
-        Solve individual sub-problem.
-        """
-        prompt = f"""Solve this sub-problem step by step:
-
-{sub_problem}
-
-Provide:
-1. Step-by-step solution
-2. Key insights
-3. Confidence level (0-100)"""
-
-        response = await self.ai_router.route(
-            prompt,
-            context={"task_type": "reasoning"},
-            optimize_for="quality"
-        )
+    async def _decompose_problem(self, problem: str, analysis: str, task_id: str, session_id: str) -> List[str]:
+        """Break down complex problem into sub-problems."""
+        messages = [
+            {"role": "system", "content": REASONING_SYSTEM},
+            {"role": "user", "content": (
+                "Break this complex problem into 3 specific sub-problems:\n\n"
+                + problem + "\n\n"
+                "Initial analysis: " + analysis[:500] + "\n\n"
+                "List each sub-problem on a new line starting with '1.', '2.', '3.'"
+            )},
+        ]
+        raw = await self.llm(messages, task_id=task_id, session_id=session_id, temperature=0.3, max_tokens=2000)
+        # Parse numbered sub-problems
+        lines = raw.split("\n")
+        sub_problems = []
+        for line in lines:
+            line = line.strip()
+            if line and any(line.startswith(str(i) + ".") for i in range(1, 10)):
+                sub_problems.append(line)
+        return sub_problems if sub_problems else [problem]
 
-        return {
-            "sub_problem": sub_problem,
-            "solution": response.get("response", ""),
-            "model_used": response.get("model", "unknown"),
-        }
+    async def _solve_sub_problem(self, sub_problem: str, context: Dict, task_id: str, session_id: str) -> str:
+        """Solve individual sub-problem."""
+        messages = [
+            {"role": "system", "content": REASONING_SYSTEM},
+            {"role": "user", "content": "Solve this specific problem:\n\n" + sub_problem},
+        ]
+        return await self.llm(messages, task_id=task_id, session_id=session_id, temperature=0.2, max_tokens=4096)
 
     async def _synthesize_answer(
         self,
-        original_problem: str,
-        analysis: Dict[str, Any],
-        sub_problems: list,
-        solutions: list
+        original: str,
+        analysis: str,
+        sub_problems: List[str],
+        solutions: List[str],
+        task_id: str,
+        session_id: str,
     ) -> str:
-        """
-        Synthesize final answer from all reasoning steps.
-        """
-        synthesis_prompt = f"""Based on the analysis and solutions, provide a comprehensive answer:
-
-Original Problem: {original_problem}
-
-Analysis: {json.dumps(analysis, indent=2)}
-
-Solutions:
-{json.dumps(solutions, indent=2)}
-
-Provide a clear, well-reasoned final answer that:
-1. Directly addresses the original problem
-2. Integrates insights from all sub-problems
-3. Explains the reasoning clearly
-4. Suggests any follow-up actions if needed"""
-
-        response = await self.ai_router.route(
-            synthesis_prompt,
-            context={"task_type": "reasoning"},
-            optimize_for="quality"
+        """Synthesize final answer from all reasoning steps."""
+        solutions_text = "\n".join(
+            "Sub-problem " + str(i + 1) + ": " + sol[:300]
+            for i, sol in enumerate(solutions)
         )
-
-        return response.get("response", "Unable to synthesize answer")
-
-    def _classify_problem(self, problem: str) -> str:
-        """Classify problem type."""
-        problem_lower = problem.lower()
-        
-        if any(word in problem_lower for word in ["math", "calculate", "equation"]):
-            return "mathematical"
-        elif any(word in problem_lower for word in ["logic", "reason", "why"]):
-            return "logical"
-        elif any(word in problem_lower for word in ["plan", "strategy", "approach"]):
-            return "strategic"
-        elif any(word in problem_lower for word in ["analyze", "compare", "evaluate"]):
-            return "analytical"
-        else:
-            return "general"
-
-    def _estimate_complexity(self, problem: str) -> str:
-        """Estimate problem complexity."""
-        word_count = len(problem.split())
-        
-        if word_count < 20:
-            return "simple"
-        elif word_count < 100:
-            return "moderate"
-        else:
-            return "complex"
-
-    def _parse_sub_problems(self, response: str) -> list:
-        """Parse sub-problems from model response."""
-        # Simple parsing - can be enhanced
-        lines = response.split("\n")
-        sub_problems = []
-        
-        for line in lines:
-            line = line.strip()
-            if line and any(line.startswith(f"{i}.") for i in range(1, 10)):
-                sub_problems.append(line)
-        
-        return sub_problems if sub_problems else [response]
-
-    async def _broadcast(self, session_id: str, data: Dict[str, Any]):
-        """Broadcast reasoning progress to client."""
-        if self.ws_manager:
-            await self.ws_manager.broadcast(
-                room=f"chat:{session_id}",
-                message={
-                    "type": "agent_message",
-                    "agent": self.name,
-                    "color": self.color,
-                    **data,
-                }
-            )
+        messages = [
+            {"role": "system", "content": REASONING_SYSTEM},
+            {"role": "user", "content": (
+                "Original problem: " + original + "\n\n"
+                "Analysis: " + analysis[:800] + "\n\n"
+                "Solutions to sub-problems:\n" + solutions_text + "\n\n"
+                "Provide a comprehensive final answer that integrates all insights."
+            )},
+        ]
+        return await self.llm(messages, task_id=task_id, session_id=session_id, temperature=0.3, max_tokens=8192)
 
     def get_status(self) -> Dict[str, Any]:
         """Get agent status."""
         return {
             "name": self.name,
-            "color": self.color,
             "status": "ready",
             "capabilities": [
                 "Multi-step reasoning",
@@ -289,7 +167,6 @@ Provide a clear, well-reasoned final answer that:
                 "Strategic planning",
             ],
             "reasoning_depth": self.reasoning_depth,
-            "max_reasoning_tokens": self.max_reasoning_tokens,
         }