Update app/agent_system.py

app/agent_system.py

@@ -1,548 +1,410 @@
-import os
-import json
-import asyncio
-import datetime
-import traceback
-import subprocess
-import tempfile
-import base64
-import io
+"""
+PraisonChat Agent System — Real Execution Engine
+================================================
+Agents actually execute tools. No simulation.
+"""
+import os, json, asyncio, datetime, traceback
 from openai import AsyncOpenAI
 from typing import AsyncGenerator
 from docs_context import PRAISONAI_DOCS
+from tool_executor import (
+    execute_tool, run_builtin_tool, pip_install,
+    BUILTIN_TOOL_IMPLEMENTATIONS, PKG_DIR
+)
 
-LONGCAT_BASE_URL = "https://api.longcat.chat/openai/v1"
+LONGCAT_BASE = "https://api.longcat.chat/openai/v1"
 MODEL_MAP = {
-    "LongCat-Flash-Lite":         "LongCat-Flash-Lite",
-    "LongCat-Flash-Chat":         "LongCat-Flash-Chat",
-    "LongCat-Flash-Thinking-2601":"LongCat-Flash-Thinking-2601",
+    "LongCat-Flash-Lite":          "LongCat-Flash-Lite",
+    "LongCat-Flash-Chat":          "LongCat-Flash-Chat",
+    "LongCat-Flash-Thinking-2601": "LongCat-Flash-Thinking-2601",
 }
 DEFAULT_MODEL = "LongCat-Flash-Lite"
 
-# ── Built-in tools (always available to every agent) ────────────────────────
-
-def get_current_datetime() -> str:
-    now = datetime.datetime.now()
-    utc = datetime.datetime.utcnow()
-    return (f"Local: {now.strftime('%A, %B %d, %Y at %I:%M:%S %p')}\n"
-            f"UTC:   {utc.strftime('%Y-%m-%d %H:%M:%S')} UTC\n"
-            f"Unix:  {int(now.timestamp())}")
-
-def calculate_math(expression: str) -> str:
-    try:
-        safe_chars = set("0123456789+-*/.() %**^")
-        clean = expression.replace("^", "**")
-        if not all(c in safe_chars or c.isspace() for c in clean):
-            return "Error: unsafe characters in expression"
-        result = eval(clean, {"__builtins__": {}}, {})
-        return f"Result: {result}"
-    except Exception as e:
-        return f"Error: {e}"
-
-def run_python_code(code: str) -> str:
-    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
-        f.write(code)
-        tmp = f.name
-    try:
-        result = subprocess.run(
-            ['python3', tmp], capture_output=True, text=True, timeout=15
-        )
-        out = (result.stdout + result.stderr).strip()
-        return out or "(no output)"
-    except subprocess.TimeoutExpired:
-        return "Error: execution timed out after 15s"
-    except Exception as e:
-        return f"Error: {e}"
-    finally:
-        try:
-            os.unlink(tmp)
-        except Exception:
-            pass
-
-def create_voice_response(text: str) -> str:
-    try:
-        from gtts import gTTS
-        tts = gTTS(text=text, lang='en', slow=False)
-        buf = io.BytesIO()
-        tts.write_to_fp(buf)
-        buf.seek(0)
-        b64 = base64.b64encode(buf.read()).decode('utf-8')
-        return f"AUDIO_B64:{b64}"
-    except Exception as e:
-        return f"VOICE_FALLBACK:{text[:2000]}"
-
-def search_information(query: str) -> str:
-    # Simple stub - returns a helpful message since we don't have a search API key
-    # The agent can use its training knowledge to answer
-    return f"Searching for: {query}\n[Search tool: returning from internal knowledge - agent should answer from training data]"
-
-BUILTIN_TOOLS = {
-    "get_current_datetime": get_current_datetime,
-    "calculate_math":       calculate_math,
-    "run_python_code":      run_python_code,
-    "create_voice_response":create_voice_response,
-    "search_information":   search_information,
-}
-
-BUILTIN_TOOLS_DOC = """
-## Always-Available Built-in Tools
-These tools exist in every agent — no need to create them:
-
-- get_current_datetime() -> str
-  Returns the exact current date and time (local + UTC + unix timestamp).
-  USE THIS whenever user asks about date, time, day, etc.
-
-- calculate_math(expression: str) -> str
-  Evaluates math: "2 + 2", "100 * 3.14", "2**10", etc.
 
-- run_python_code(code: str) -> str
-  Executes Python code in a sandbox. Returns stdout + stderr.
-  Use for data processing, file operations, complex calculations.
+def now_str() -> str:
+    n = datetime.datetime.now()
+    return n.strftime("%A, %B %d, %Y at %I:%M:%S %p")
 
-- create_voice_response(text: str) -> str
-  Converts text to MP3 audio via gTTS. Returns AUDIO_B64:<base64>.
-  USE THIS when user explicitly asks for voice/audio/spoken response.
 
-- search_information(query: str) -> str
-  Queries knowledge base. Use for research tasks.
-
-CRITICAL RULES:
-1. If user asks "what time is it" / "what date" / "what day" -> use get_current_datetime
-2. If user asks for "voice" / "speak" / "audio" / "say it" -> use create_voice_response
-3. NEVER say "I cannot" for tasks these tools handle
-4. Always prefer tools over saying you lack capability
-"""
-
-def inject_datetime_context() -> str:
-    now = datetime.datetime.now()
-    return (f"[System context: Current datetime = "
-            f"{now.strftime('%A, %B %d, %Y %I:%M:%S %p')} local time | "
-            f"{datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC]\n")
-
-
-def build_orchestrator_system() -> str:
-    return f"""{inject_datetime_context()}
-You are the Main Orchestrator Agent for PraisonChat — a powerful AI that solves tasks by
-dynamically creating specialized sub-agents with custom-built Python tools.
+ORCHESTRATOR_SYSTEM = f"""You are the Main Orchestrator Agent for PraisonChat.
+Current datetime: {{DATETIME}}
+Python package dir: {PKG_DIR}
 
 {PRAISONAI_DOCS}
 
-{BUILTIN_TOOLS_DOC}
-
-## Your Job
-When a user sends a task:
-1. Analyze what kind of work is needed
-2. Use built-in tools directly for simple things (datetime, math, voice, code)
-3. For complex tasks, design sub-agents each focused on one responsibility
-4. For each sub-agent, design exact Python tools they need
-5. Return a structured JSON execution plan
-
-## Response Format — valid JSON ONLY, no markdown fences:
+## Available Built-in Tools (always executable, no creation needed)
+- get_current_datetime  → real system time
+- search_web(query)     → real DuckDuckGo search results
+- fetch_webpage(url)    → real HTML content extraction
+- run_python_code(code) → real Python execution with output
+- create_voice(text)    → real MP3 audio via gTTS
+- calculate(expression) → real math evaluation
+
+## Your Task
+Analyze the user's request and respond with a JSON plan.
+RULES:
+1. For date/time → use builtin: get_current_datetime
+2. For web search → use builtin: search_web  
+3. For voice/audio → use builtin: create_voice
+4. For math → use builtin: calculate
+5. For code execution → use builtin: run_python_code
+6. For complex multi-step tasks → create sub_agents
+7. Sub-agent tools MUST have real working Python code using real libraries
+8. Specify pip packages needed — they will be ACTUALLY installed
+
+## Tool Implementation Rules (CRITICAL)
+- Use duckduckgo_search for searches — NOT requests to google.com
+- Use requests + BeautifulSoup for web scraping
+- Use real API calls for external services
+- Tool code must be complete, working Python
+- Add: import sys; sys.path.insert(0, '{PKG_DIR}') at top of tool code
+- Use # comments only, NEVER triple-quoted strings inside tool code
+
+## Response Format — JSON only, no markdown fences:
 {{
-  "task_analysis": "Clear explanation of what needs to be done",
-  "needs_sub_agents": true/false,
-  "builtin_tools_to_use": ["get_current_datetime", "calculate_math"],
+  "task_analysis": "what needs to be done",
+  "builtin_tools": [
+    {{"name": "search_web", "call_arg": "actual search query here"}}
+  ],
+  "needs_agents": false,
   "sub_agents": [
     {{
-      "name": "AgentName",
-      "role": "Specific professional role",
-      "goal": "What this agent achieves",
-      "backstory": "Brief agent background",
+      "name": "ResearchAgent",
+      "role": "Web Research Specialist",
+      "goal": "Find real information",
+      "task": "Search for X and summarize",
+      "expected_output": "Summary of findings",
       "tools": [
         {{
-          "name": "tool_function_name",
-          "description": "What this tool does",
-          "parameters": "param1: str, param2: int = 10",
-          "return_type": "str",
-          "implementation": "# Python code body (4-space indent, use # comments not triple quotes)\\n    result = do_something(param1)\\n    return str(result)"
+          "name": "search_news",
+          "description": "Search for latest news",
+          "packages": ["duckduckgo-search"],
+          "call_arg_key": "query",
+          "call_arg_value": "actual search terms",
+          "code": "import sys\\nsys.path.insert(0, '{PKG_DIR}')\\ndef search_news(query: str) -> str:\\n    from duckduckgo_search import DDGS\\n    results = []\\n    with DDGS() as ddgs:\\n        for r in ddgs.news(query, max_results=5):\\n            results.append(f\\"{{r['title']}}: {{r['body']}}\\" )\\n    return '\\\\n'.join(results)"
         }}
-      ],
-      "task_description": "Exact task for this agent to perform",
-      "expected_output": "What format/content to expect back"
+      ]
     }}
   ],
-  "execution_order": ["AgentName1", "AgentName2"],
-  "synthesis_instruction": "How to combine all agent results into the final answer",
+  "synthesis": "how to combine results",
   "output_format": "text"
 }}
+output_format: text | voice | code | markdown
+"""
 
-output_format options: "text", "voice", "code", "table", "json"
 
-## Rules
-- Simple questions (time, math, quick facts) = no sub-agents, use builtin_tools_to_use
-- Tool implementations: use # comments ONLY, never triple-quoted strings inside code
-- Max 4 sub-agents per task
-- Tool code must be valid Python, no imports not in stdlib
-- If voice requested: set output_format to "voice" AND use create_voice_response tool
-"""
+def get_orchestrator_system():
+    return ORCHESTRATOR_SYSTEM.replace("{DATETIME}", now_str())
 
 
-def build_tool_function(spec: dict):
-    name = spec.get("name", "unnamed_tool")
-    params = spec.get("parameters", "input: str")
-    rtype = spec.get("return_type", "str")
-    impl = spec.get("implementation", "return str(input)")
-
-    lines = impl.strip().splitlines()
-    indented = "\n".join("    " + l if l.strip() else "" for l in lines)
-
-    src = f"def {name}({params}) -> {rtype}:\n{indented}\n"
-    ns = {}
-    try:
-        exec(src, ns)
-        fn = ns[name]
-        fn.__doc__ = spec.get("description", "")
-        return fn, None
-    except Exception as e:
-        def fallback(**kwargs) -> str:
-            return f"[Tool '{name}' build error: {e}]"
-        fallback.__name__ = name
-        return fallback, str(e)
-
-
-def call_builtin_tool(name: str, agent_task: str) -> str:
-    fn = BUILTIN_TOOLS.get(name)
-    if not fn:
-        return f"Unknown built-in tool: {name}"
-    try:
-        if name == "get_current_datetime":
-            return fn()
-        elif name == "calculate_math":
-            return fn(agent_task)
-        elif name == "run_python_code":
-            return fn(agent_task)
-        elif name == "create_voice_response":
-            return fn(agent_task)
-        elif name == "search_information":
-            return fn(agent_task)
-        else:
-            return fn()
-    except Exception as e:
-        return f"Tool error: {e}"
+def get_agent_system(spec: dict, context: str) -> str:
+    tools_desc = "\n".join(
+        f"- {t['name']}: {t.get('description','')}" for t in spec.get("tools", [])
+    )
+    return f"""You are {spec['name']}, a specialized AI agent.
+Current datetime: {now_str()}
+Role: {spec['role']}
+Goal: {spec['goal']}
+
+Tools available (already executed with real results):
+{tools_desc}
+
+Context from previous agents:
+{context or 'You are the first agent.'}
+
+Synthesize the real tool results into a clear, accurate response.
+Expected output: {spec.get('expected_output', 'Detailed results')}
+IMPORTANT: Use the actual tool results provided — do NOT make up or estimate data."""
 
 
 class AgentOrchestrator:
     def __init__(self):
         self._clients: dict = {}
 
-    def get_client(self, api_key: str, model: str = DEFAULT_MODEL) -> AsyncOpenAI:
-        key = f"{api_key}:{model}"
-        if key not in self._clients:
-            self._clients[key] = AsyncOpenAI(
-                api_key=api_key,
-                base_url=LONGCAT_BASE_URL,
-            )
-        return self._clients[key]
+    def client(self, api_key: str) -> AsyncOpenAI:
+        if api_key not in self._clients:
+            self._clients[api_key] = AsyncOpenAI(api_key=api_key, base_url=LONGCAT_BASE)
+        return self._clients[api_key]
 
-    async def plan_task(self, client, user_message: str, history: list, model: str) -> dict:
-        messages = [{"role": "system", "content": build_orchestrator_system()}]
+    async def plan(self, client, msg: str, history: list, model: str) -> dict:
+        messages = [{"role": "system", "content": get_orchestrator_system()}]
         for m in history[-6:]:
-            messages.append({"role": m["role"], "content": str(m.get("content", ""))[:2000]})
-        messages.append({
-            "role": "user",
-            "content": f"Plan execution for: {user_message}"
-        })
+            messages.append({"role": m["role"], "content": str(m.get("content",""))[:1500]})
+        messages.append({"role": "user", "content": f"Plan execution for: {msg}"})
 
         resp = await client.chat.completions.create(
-            model=model,
-            messages=messages,
-            max_tokens=6000,
-            temperature=0.1,
+            model=model, messages=messages, max_tokens=5000, temperature=0.1
         )
         raw = resp.choices[0].message.content.strip()
-        # Strip possible markdown fences
         if "```" in raw:
-            parts = raw.split("```")
-            for p in parts:
-                p2 = p.strip()
-                if p2.startswith("json"):
-                    p2 = p2[4:].strip()
-                if p2.startswith("{"):
-                    raw = p2
-                    break
+            for part in raw.split("```"):
+                p = part.strip().lstrip("json").strip()
+                if p.startswith("{"):
+                    raw = p; break
         try:
             return json.loads(raw)
         except Exception:
-            return {
-                "task_analysis": "Direct response",
-                "needs_sub_agents": False,
-                "builtin_tools_to_use": [],
-                "sub_agents": [],
-                "execution_order": [],
-                "synthesis_instruction": "Respond directly",
-                "output_format": "text",
-            }
-
-    async def run_sub_agent(self, client, spec: dict, context: str, model: str) -> dict:
-        tools_built = []
-        tool_errors = []
-        tool_descriptions = "\n".join(
-            f"- {t['name']}: {t.get('description','')}" for t in spec.get("tools", [])
-        )
-        # Build custom tools
-        for t in spec.get("tools", []):
-            fn, err = build_tool_function(t)
-            if err:
-                tool_errors.append(f"{t['name']}: {err}")
-            tools_built.append({"name": t["name"], "fn": fn, "desc": t.get("description", ""), "error": err})
-
-        system = f"""{inject_datetime_context()}
-You are {spec['name']}, a specialized AI agent.
-Role: {spec['role']}
-Goal: {spec['goal']}
-Backstory: {spec.get('backstory', '')}
-
-Built-in tools always available:
-{BUILTIN_TOOLS_DOC}
-
-Custom tools for this task:
-{tool_descriptions if tool_descriptions else 'None — use built-in tools and your knowledge'}
-
-Context from previous agents:
-{context if context else 'You are the first agent.'}
-
-Execute your task. Show reasoning and tool usage step by step.
-Expected output: {spec.get('expected_output', 'Detailed results')}"""
+            return {"task_analysis":"Direct","builtin_tools":[],"needs_agents":False,
+                    "sub_agents":[],"synthesis":"","output_format":"text"}
 
+    async def synthesize_agent(self, client, spec: dict, tool_results: dict,
+                                context: str, model: str) -> str:
+        tool_results_text = "\n\n".join(
+            f"[Tool: {k}]\n{v}" for k, v in tool_results.items()
+        )
+        system = get_agent_system(spec, context)
         resp = await client.chat.completions.create(
             model=model,
             messages=[
                 {"role": "system", "content": system},
-                {"role": "user", "content": spec["task_description"]},
+                {"role": "user", "content": f"Task: {spec['task']}\n\nReal tool results:\n{tool_results_text}"},
             ],
-            max_tokens=12000,
-            temperature=0.7,
-        )
-        result = resp.choices[0].message.content
-        return {
-            "result": result,
-            "tools_built": [{"name": t["name"], "desc": t["desc"], "error": t.get("error")} for t in tools_built],
-            "tool_errors": tool_errors,
-        }
-
-    async def synthesize(self, client, user_message: str, agent_results: dict,
-                         synthesis_instruction: str, output_format: str, model: str) -> AsyncGenerator:
-        combined = "\n\n".join(
-            f"=== {name} ===\n{r['result']}" for name, r in agent_results.items()
+            max_tokens=8000, temperature=0.7,
         )
+        return resp.choices[0].message.content
+
+    async def final_synthesis(self, client, user_msg: str, agent_results: dict,
+                               builtin_results: dict, synthesis_note: str,
+                               output_format: str, model: str) -> AsyncGenerator:
+        all_results = ""
+        if builtin_results:
+            all_results += "## Real Tool Results\n"
+            all_results += "\n\n".join(f"[{k}]:\n{v}" for k,v in builtin_results.items())
+        if agent_results:
+            all_results += "\n\n## Agent Results\n"
+            all_results += "\n\n".join(f"[{k}]:\n{v}" for k,v in agent_results.items())
+
         voice_note = ""
         if output_format == "voice":
-            voice_note = "\nIMPORTANT: The user wants a voice response. End your message with: [VOICE_RESPONSE: <the exact text to speak>]"
+            voice_note = "\nEnd with: [VOICE: <the text to speak aloud>]"
 
-        system = f"""{inject_datetime_context()}
-You are the Main Orchestrator. Synthesize results from sub-agents into a final response.
-Synthesis instruction: {synthesis_instruction}
-Output format: {output_format}
+        system = f"""You are PraisonChat, synthesizing REAL tool results into a final response.
+Current datetime: {now_str()}
+{synthesis_note}
 {voice_note}
 
-Sub-agent results:
-{combined}
-
-Provide a comprehensive, well-structured markdown response."""
+CRITICAL: Use ONLY the actual data from tool results. Do NOT estimate or fabricate data.
+If a search returned real results, quote and cite them.
+Format response in clear markdown."""
 
         stream = await client.chat.completions.create(
             model=model,
             messages=[
                 {"role": "system", "content": system},
-                {"role": "user", "content": user_message},
+                {"role": "user", "content": f"User asked: {user_msg}\n\nActual results to use:\n{all_results}"},
             ],
-            max_tokens=16000,
-            temperature=0.7,
-            stream=True,
+            max_tokens=12000, temperature=0.7, stream=True
         )
         async for chunk in stream:
             c = chunk.choices[0].delta.content
-            if c:
-                yield c
-
-    async def direct_response(self, client, user_message: str, history: list,
-                               builtin_tools: list, output_format: str, model: str) -> AsyncGenerator:
-        # Execute builtin tools first
-        tool_results = {}
-        for tool_name in (builtin_tools or []):
-            if tool_name in BUILTIN_TOOLS:
-                tool_results[tool_name] = call_builtin_tool(tool_name, user_message)
-
-        tool_context = ""
-        if tool_results:
-            tool_context = "\n\nTool results:\n" + "\n".join(
-                f"[{k}]: {v}" for k, v in tool_results.items()
-            )
-
-        voice_note = ""
-        if output_format == "voice":
-            voice_note = "\nThe user wants a voice response. End your reply with: [VOICE_RESPONSE: <text to speak>]"
-
-        messages = [{
-            "role": "system",
-            "content": (
-                f"{inject_datetime_context()}"
-                "You are PraisonChat, a powerful AI assistant. "
-                "Respond helpfully using markdown. "
-                "You have real-time access to date/time, code execution, and voice tools. "
-                "NEVER say you cannot check the time or date — you have it above."
-                f"{tool_context}{voice_note}"
+            if c: yield c
+
+    async def direct_stream(self, client, user_msg: str, history: list,
+                             builtin_results: dict, output_format: str, model: str) -> AsyncGenerator:
+        tool_ctx = ""
+        if builtin_results:
+            tool_ctx = "\n\nReal tool results:\n" + "\n\n".join(
+                f"[{k}]: {v}" for k,v in builtin_results.items()
             )
-        }]
+        voice_note = "\nEnd reply with: [VOICE: <text to speak>]" if output_format=="voice" else ""
+
+        messages = [{"role":"system","content":(
+            f"You are PraisonChat, a powerful AI assistant.\n"
+            f"Current datetime: {now_str()}\n"
+            f"Use the real tool results below to answer accurately.{tool_ctx}{voice_note}\n"
+            f"NEVER say you cannot check the time/date — it is shown above.\n"
+            f"Format in clear markdown."
+        )}]
         for m in history[-10:]:
-            messages.append({"role": m["role"], "content": str(m.get("content",""))[:3000]})
-        messages.append({"role": "user", "content": user_message})
+            messages.append({"role":m["role"],"content":str(m.get("content",""))[:2500]})
+        messages.append({"role":"user","content":user_msg})
 
         stream = await client.chat.completions.create(
-            model=model,
-            messages=messages,
-            max_tokens=16000,
-            temperature=0.7,
-            stream=True,
+            model=model, messages=messages, max_tokens=12000, temperature=0.7, stream=True
         )
         async for chunk in stream:
             c = chunk.choices[0].delta.content
-            if c:
-                yield c
+            if c: yield c
 
-    async def stream_response(self, user_message: str, history: list,
+    # ── Main streaming entry point ──────────────────────────────────────────
+    async def stream_response(self, user_msg: str, history: list,
                                api_key: str, model: str = DEFAULT_MODEL) -> AsyncGenerator:
-        def emit(data: dict) -> str:
-            return json.dumps(data)
+        def emit(d: dict) -> str:
+            return json.dumps(d)
 
         model = MODEL_MAP.get(model, DEFAULT_MODEL)
-        client = self.get_client(api_key, model)
+        cl = self.client(api_key)
 
         try:
-            # Step 1: Plan
-            yield emit({"type": "thinking", "text": "Analyzing your request..."})
+            # 1. Plan
+            yield emit({"type":"thinking","text":"Analyzing your request…"})
             await asyncio.sleep(0)
 
-            plan = await self.plan_task(client, user_message, history, model)
-
-            yield emit({"type": "thinking", "text": plan.get("task_analysis", "Planning...")})
+            plan = await self.plan(cl, user_msg, history, model)
+            yield emit({"type":"thinking","text": plan.get("task_analysis","Planning…")})
             await asyncio.sleep(0)
 
-            sub_agents = plan.get("sub_agents", [])
-            needs_sub = plan.get("needs_sub_agents", bool(sub_agents))
-            builtin_tools = plan.get("builtin_tools_to_use", [])
-            output_format = plan.get("output_format", "text")
+            builtin_results: dict = {}
+            agent_results: dict = {}
+
+            # 2. Execute built-in tools REALLY
+            for bt in plan.get("builtin_tools", []):
+                tool_name = bt.get("name","")
+                call_arg  = bt.get("call_arg", user_msg)
+                if tool_name not in BUILTIN_TOOL_IMPLEMENTATIONS:
+                    continue
+
+                yield emit({"type":"tool_call","tool":tool_name,"arg":call_arg[:80],"builtin":True})
+                await asyncio.sleep(0)
+
+                # Actual execution in thread pool
+                spec = BUILTIN_TOOL_IMPLEMENTATIONS[tool_name]
+                args_dict = {}
+                if spec["args"]:
+                    first_key = list(spec["args"].keys())[0]
+                    args_dict = {first_key: call_arg}
+
+                loop = asyncio.get_event_loop()
+                result = await loop.run_in_executor(None,
+                    lambda s=spec, a=args_dict: execute_tool(
+                        s["code"], tool_name, a, s["packages"], timeout=45
+                    )
+                )
+
+                if result["ok"]:
+                    builtin_results[tool_name] = result["result"]
+                    preview = result["result"][:200]
+                    yield emit({"type":"tool_result","tool":tool_name,"result":preview,"ok":True})
+
+                    # Handle audio immediately
+                    if tool_name == "create_voice" and result["result"].startswith("AUDIO_B64:"):
+                        yield emit({"type":"audio_response","audio_b64":result["result"][10:]})
+                else:
+                    err = result.get("error","unknown error")
+                    builtin_results[tool_name] = f"Error: {err}"
+                    yield emit({"type":"tool_result","tool":tool_name,"result":err,"ok":False})
 
-            # Emit builtin tool calls
-            for bt in builtin_tools:
-                if bt in BUILTIN_TOOLS:
-                    yield emit({"type": "tool_call", "tool": bt, "builtin": True})
-                    await asyncio.sleep(0)
-                    result = call_builtin_tool(bt, user_message)
-                    is_audio = result.startswith("AUDIO_B64:")
-                    preview = "[audio data]" if is_audio else result[:200]
-                    yield emit({"type": "tool_result", "tool": bt, "result": preview, "is_audio": is_audio,
-                                "audio_b64": result[10:] if is_audio else None})
-                    await asyncio.sleep(0)
+                await asyncio.sleep(0)
 
-            # Step 2: Sub-agents or direct
-            if needs_sub and sub_agents:
-                yield emit({"type": "step", "text": f"Spawning {len(sub_agents)} sub-agent(s)..."})
+            # 3. Sub-agents with real tool execution
+            if plan.get("needs_agents") and plan.get("sub_agents"):
+                sub_agents = plan["sub_agents"]
+                yield emit({"type":"step","text":f"Spawning {len(sub_agents)} real agent(s)…"})
 
                 for spec in sub_agents:
-                    tool_names = [t["name"] for t in spec.get("tools", [])]
-                    yield emit({
-                        "type": "agent_created",
-                        "name": spec["name"],
-                        "role": spec["role"],
-                        "goal": spec.get("goal", ""),
-                        "tools": tool_names,
-                        "tool_specs": spec.get("tools", []),
-                    })
+                    tool_names = [t["name"] for t in spec.get("tools",[])]
+                    yield emit({"type":"agent_created","name":spec["name"],
+                                "role":spec["role"],"goal":spec.get("goal",""),
+                                "tools":tool_names,"tool_specs":spec.get("tools",[])})
                     await asyncio.sleep(0.05)
 
                 context_so_far = ""
-                agent_results = {}
-                order = plan.get("execution_order", [s["name"] for s in sub_agents])
-
-                for agent_name in order:
-                    spec = next((s for s in sub_agents if s["name"] == agent_name), None)
-                    if not spec:
-                        continue
-
-                    yield emit({"type": "agent_working", "name": agent_name,
-                                "task": spec["task_description"][:120]})
+                for spec in sub_agents:
+                    yield emit({"type":"agent_working","name":spec["name"],"task":spec["task"][:100]})
                     await asyncio.sleep(0)
 
-                    # Emit tool builds
-                    for t in spec.get("tools", []):
-                        yield emit({"type": "tool_building", "agent": agent_name,
-                                    "tool": t["name"], "description": t.get("description", "")})
-                        await asyncio.sleep(0.05)
-
-                    try:
-                        r = await self.run_sub_agent(client, spec, context_so_far, model)
-                        agent_results[agent_name] = r
+                    agent_tool_results = {}
+
+                    for tool in spec.get("tools", []):
+                        t_name = tool["name"]
+                        t_code = tool.get("code","")
+                        t_pkgs = tool.get("packages",[])
+                        t_arg_key = tool.get("call_arg_key","input")
+                        t_arg_val = tool.get("call_arg_value", user_msg)
+
+                        yield emit({"type":"tool_building","agent":spec["name"],
+                                    "tool":t_name,"packages":t_pkgs,
+                                    "description":tool.get("description","")})
+                        await asyncio.sleep(0)
+
+                        # Install packages if needed
+                        if t_pkgs:
+                            loop = asyncio.get_event_loop()
+                            ok, msg = await loop.run_in_executor(None, pip_install, t_pkgs)
+                            yield emit({"type":"pkg_install","packages":t_pkgs,"ok":ok,"msg":msg})
+
+                        # Actually execute the tool
+                        call_args = {t_arg_key: t_arg_val}
+                        loop = asyncio.get_event_loop()
+                        result = await loop.run_in_executor(None,
+                            lambda c=t_code, n=t_name, a=call_args, p=t_pkgs:
+                                execute_tool(c, n, a, p, timeout=45)
+                        )
+
+                        if result["ok"]:
+                            agent_tool_results[t_name] = result["result"]
+                            preview = result["result"][:250]
+                            yield emit({"type":"tool_ready","agent":spec["name"],"tool":t_name,
+                                        "result":preview,"ok":True})
+
+                            # Voice from agent tool
+                            if result["result"].startswith("AUDIO_B64:"):
+                                yield emit({"type":"audio_response","audio_b64":result["result"][10:]})
+                        else:
+                            err = result.get("error","unknown")
+                            agent_tool_results[t_name] = f"Error: {err}"
+                            yield emit({"type":"tool_ready","agent":spec["name"],"tool":t_name,
+                                        "result":err,"ok":False,"error":err})
 
-                        # Emit tool results
-                        for tb in r.get("tools_built", []):
-                            yield emit({"type": "tool_ready", "agent": agent_name,
-                                        "tool": tb["name"], "error": tb.get("error")})
+                        await asyncio.sleep(0)
 
-                        context_so_far += f"\n\n{agent_name}: {r['result'][:600]}"
-                        preview = r["result"][:300] + ("..." if len(r["result"]) > 300 else "")
-                        yield emit({"type": "agent_done", "name": agent_name, "preview": preview})
+                    # Agent synthesizes its real results
+                    try:
+                        agent_summary = await self.synthesize_agent(
+                            cl, spec, agent_tool_results, context_so_far, model
+                        )
+                        agent_results[spec["name"]] = agent_summary
+                        context_so_far += f"\n\n{spec['name']}: {agent_summary[:500]}"
+                        yield emit({"type":"agent_done","name":spec["name"],
+                                    "preview":agent_summary[:300]})
                     except Exception as e:
-                        yield emit({"type": "agent_error", "name": agent_name, "error": str(e)[:200]})
-                        agent_results[agent_name] = {"result": f"Error: {e}", "tools_built": [], "tool_errors": [str(e)]}
+                        agent_results[spec["name"]] = f"Agent error: {e}"
+                        yield emit({"type":"agent_error","name":spec["name"],"error":str(e)[:150]})
 
-                yield emit({"type": "step", "text": "Synthesizing final response..."})
-                yield emit({"type": "response_start", "output_format": output_format})
+                # Final synthesis
+                yield emit({"type":"step","text":"Synthesizing real results into final answer…"})
 
-                full_text = ""
-                async for token in self.synthesize(client, user_message, agent_results,
-                                                    plan.get("synthesis_instruction", ""),
-                                                    output_format, model):
-                    full_text += token
-                    yield emit({"type": "token", "content": token})
-
-                # Handle voice in synthesized response
-                if output_format == "voice" and "[VOICE_RESPONSE:" in full_text:
-                    try:
-                        voice_text = full_text.split("[VOICE_RESPONSE:")[1].rsplit("]", 1)[0].strip()
-                        audio_result = create_voice_response(voice_text)
-                        if audio_result.startswith("AUDIO_B64:"):
-                            yield emit({"type": "audio_response", "audio_b64": audio_result[10:],
-                                        "text": voice_text})
-                        else:
-                            yield emit({"type": "voice_fallback", "text": voice_text})
-                    except Exception:
-                        pass
+            # 4. Stream final response
+            yield emit({"type":"response_start","output_format":plan.get("output_format","text")})
+            full_text = ""
 
+            if plan.get("needs_agents") and agent_results:
+                async for token in self.final_synthesis(
+                    cl, user_msg, agent_results, builtin_results,
+                    plan.get("synthesis",""), plan.get("output_format","text"), model
+                ):
+                    full_text += token
+                    yield emit({"type":"token","content":token})
             else:
-                # Direct response
-                if builtin_tools:
-                    yield emit({"type": "step", "text": f"Using built-in tools: {', '.join(builtin_tools)}"})
-                else:
-                    yield emit({"type": "step", "text": "Generating response..."})
-                yield emit({"type": "response_start", "output_format": output_format})
-
-                full_text = ""
-                async for token in self.direct_response(client, user_message, history,
-                                                         builtin_tools, output_format, model):
+                async for token in self.direct_stream(
+                    cl, user_msg, history, builtin_results,
+                    plan.get("output_format","text"), model
+                ):
                     full_text += token
-                    yield emit({"type": "token", "content": token})
-
-                # Handle voice in direct response
-                if output_format == "voice" or "[VOICE_RESPONSE:" in full_text:
-                    try:
-                        if "[VOICE_RESPONSE:" in full_text:
-                            voice_text = full_text.split("[VOICE_RESPONSE:")[1].rsplit("]", 1)[0].strip()
-                        else:
-                            voice_text = full_text[:1000]
-                        audio_result = create_voice_response(voice_text)
-                        if audio_result.startswith("AUDIO_B64:"):
-                            yield emit({"type": "audio_response", "audio_b64": audio_result[10:],
-                                        "text": voice_text})
-                        else:
-                            yield emit({"type": "voice_fallback", "text": voice_text})
-                    except Exception:
-                        pass
-
-            yield emit({"type": "done"})
+                    yield emit({"type":"token","content":token})
+
+            # Extract and deliver voice if requested
+            if "[VOICE:" in full_text:
+                try:
+                    voice_text = full_text.split("[VOICE:")[1].rsplit("]",1)[0].strip()
+                    loop = asyncio.get_event_loop()
+                    spec = BUILTIN_TOOL_IMPLEMENTATIONS["create_voice"]
+                    vresult = await loop.run_in_executor(None,
+                        lambda: execute_tool(spec["code"],"create_voice",
+                                             {"text":voice_text},spec["packages"],45)
+                    )
+                    if vresult["ok"] and vresult["result"].startswith("AUDIO_B64:"):
+                        yield emit({"type":"audio_response","audio_b64":vresult["result"][10:],
+                                    "text":voice_text})
+                    else:
+                        yield emit({"type":"voice_fallback","text":voice_text})
+                except Exception:
+                    pass
+
+            yield emit({"type":"done"})
 
         except Exception as e:
-            yield emit({"type": "error", "message": str(e), "detail": traceback.format_exc()[:800]})
+            yield emit({"type":"error","message":str(e),"detail":traceback.format_exc()[-600:]})
 
 
 orchestrator = AgentOrchestrator()