"""
pipeline-service v2: Agent Executor — 智能执行引擎

对照 Hermes Agent 的 run_conversation() + tool dispatch：
- 多轮 tool-calling loop，max_turns 可配置
- Memory + Skills 注入每轮 system prompt
- 上下文压缩（近 token 上限时自动压缩）
- 并行 tool calls（同一轮内独立工具同时执行）
- 消息数组污染防护（tool_call JSON 不进入 LLM 上下文）
- 会话持久化（用户+项目隔离）
- ask_user 澄清机制
- delegate_subtask 子任务分发

使用方法：
    executor = AgentExecutor(config, project_id=..., user_id=...)
    async for chunk in executor.run("用户输入"):
        yield chunk  # NDJSON 流式输出
"""

import asyncio
import json
import logging
import time
from typing import AsyncGenerator, Dict, List, Optional

logger = logging.getLogger("pipeline.agent_executor")

# ── 默认工具定义（在 pipeline-core 未加载时使用）──

_BUILTIN_ASK_USER_SCHEMA = {
    "name": "ask_user",
    "description": "向用户提问澄清意图。不确定用户想要什么时使用。",
    "parameters": {
        "type": "object",
        "properties": {
            "question": {"type": "string", "description": "要问的问题"},
        },
        "required": ["question"],
    },
}


class AgentExecutor:
    """Agent 执行引擎。

    负责：
    1. 组装 system prompt（配置 + 记忆 + 技能 + 工具）
    2. 运行 tool-loop（LLM 决策 → 工具执行 → 回传结果）
    3. 上下文压缩
    4. 会话持久化
    """

    def __init__(
        self,
        config,  # AgentConfig (from pipeline_core)
        project_id: str = "",
        user_id: str = "",
        workspace_dir: str = "",
        model_name: str = None,
    ):
        self.config = config
        self.project_id = project_id
        self.user_id = user_id
        self.org_id = ""              # loaded from project context
        self.workspace_dir = workspace_dir or "/tmp/pipeline_ws"
        self.model_name = model_name or config.model_name

        # 运行状态
        self._msgs: List[dict] = []           # LLM 消息数组
        self._turn_count: int = 0
        self._tool_call_count: int = 0
        self._session_id: str = ""
        self._started_at: float = 0.0

        # 懒加载
        self._tool_registry = None
        self._skill_loader = None
        self._memory_store = None
        self._compressor = None
        self._session_mgr = None

    # ═══════════════════════════════════════════════════════
    # 主入口
    # ═══════════════════════════════════════════════════════

    async def run(self, user_input: str, history: List[dict] = None) -> AsyncGenerator[str, None]:
        """执行 agent 主循环。

        Args:
            user_input: 用户输入
            history: 历史消息列表 [{"role":"user"|"assistant","content":"..."}]

        Yields:
            NDJSON 行：{"type":"progress"|"tool_call"|"reply"|"error", ...}
        """
        self._started_at = time.time()
        self._turn_count = 0
        self._tool_call_count = 0

        # Step 1: 初始化
        await self._init_components()

        # Step 2: 加载历史
        hist_msgs = await self._load_history(history)

        # Step 3: 组装 system prompt
        system_prompt = await self._build_system_prompt(user_input)

        # Step 4: 构建消息数组
        self._msgs = [{"role": "system", "content": system_prompt}]
        if hist_msgs:
            self._msgs.extend(hist_msgs)
        self._msgs.append({"role": "user", "content": user_input})

        # Step 5: Tool Loop
        yield json.dumps({"type": "progress", "message": "思考中..."}, ensure_ascii=False) + "\n"

        final_reply = ""

        max_turns = self.config.max_turns
        for turn in range(max_turns):
            self._turn_count = turn + 1

            # 5a. 上下文压缩
            if self.config.compression.enabled:
                await self._maybe_compress()

            # 5b. LLM 调用
            raw = await self._call_llm()

            # 5c. 解析 LLM 输出
            actions = self._parse_actions(raw)

            # 5d. 分发处理
            for act in actions:
                action_type = act.get("action", "")

                if action_type == "reply":
                    final_reply = act.get("message", "")
                    yield json.dumps({
                        "type": "reply", "message": final_reply,
                    }, ensure_ascii=False) + "\n"
                    await self._save_turn(user_input, final_reply)
                    return

                elif action_type == "tool_call":
                    tool_name = act.get("tool", "")
                    tool_params = act.get("params", {})

                    # 需要确认的工具
                    if self._needs_confirmation(tool_name):
                        yield json.dumps({
                            "type": "confirm",
                            "tool": tool_name,
                            "params": tool_params,
                        }, ensure_ascii=False) + "\n"
                        await self._save_turn(user_input, f"[需确认] {tool_name}")
                        return

                    # 执行工具
                    yield json.dumps({
                        "type": "tool_call", "tool": tool_name,
                        "params": tool_params,
                    }, ensure_ascii=False) + "\n"

                    result = await self._execute_tool(tool_name, tool_params)
                    self._tool_call_count += 1

                    yield json.dumps({
                        "type": "tool_result", "tool": tool_name,
                        "result": result[:500],
                    }, ensure_ascii=False) + "\n"

                    # 反馈给 LLM（关键：不存原始 tool_call JSON）
                    self._msgs.append({
                        "role": "assistant",
                        "content": f"已调用 {tool_name}",
                    })
                    self._msgs.append({
                        "role": "user",
                        "content": f"工具 {tool_name} 结果：\n{result}",
                    })

                elif action_type == "error":
                    final_reply = act.get("message", "处理出错")
                    yield json.dumps({
                        "type": "error", "message": final_reply,
                    }, ensure_ascii=False) + "\n"
                    return

        # 超过 max_turns
        final_reply = final_reply or "任务过于复杂，请简化描述后重试。"
        yield json.dumps({
            "type": "reply", "message": final_reply,
        }, ensure_ascii=False) + "\n"
        await self._save_turn(user_input, final_reply)

    # ═══════════════════════════════════════════════════════
    # 初始化
    # ═══════════════════════════════════════════════════════

    async def _init_components(self):
        """懒加载各组件"""
        # Tool Registry
        try:
            from pipeline_core.tool_registry import get_tool_registry
            self._tool_registry = get_tool_registry()
        except ImportError:
            self._tool_registry = None

        # Skill Loader — 三级隔离
        try:
            from pipeline_core.skill_loader import get_skill_loader
            base_dir = self.config.skills.base_dir
            self._skill_loader = get_skill_loader(base_dir)
        except ImportError:
            self._skill_loader = None

        # Memory Store
        try:
            from pipeline_core.memory_store import get_memory_store
            self._memory_store = get_memory_store()
        except ImportError:
            self._memory_store = None

        # 加载 org_id（从项目上下文）
        if self.project_id:
            try:
                from sqlor.dbpools import DBPools
                db = DBPools()
                async with db.sqlorContext("pipeline") as sor:
                    recs = await sor.sqlExe(
                        "SELECT org_id FROM sd_projects WHERE id=${pid}$",
                        {"pid": self.project_id})
                    if recs:
                        self.org_id = getattr(recs[0], 'org_id', '') or ''
            except Exception:
                pass

    async def _build_system_prompt(self, user_input: str) -> str:
        """组装完整 system prompt = 基础 prompt + 记忆 + 技能 + 工具列表"""
        prompt = self.config.system_prompt

        # 注入当前项目
        prompt = prompt.replace("{current_project}", self.project_id or "未选择")

        # 注入记忆
        if self.config.memory.enabled and self._memory_store:
            mem_block = await self._memory_store.build_prompt_block(max_entries=15)
            if mem_block:
                prompt += f"\n\n## 持久记忆\n{mem_block}"

        # 注入技能（三级隔离：user > org > global）
        if self.config.skills.enabled and self._skill_loader:
            skill_block = self._skill_loader.build_prompt_block(
                user_input=user_input,
                org_id=self.org_id if self.config.skills.enable_org else "",
                user_id=self.user_id if self.config.skills.enable_user else "",
                max_skills=self.config.skills.max_skills_per_turn,
            )
            if skill_block:
                prompt += f"\n\n{skill_block}"

        # 注入工具列表（先构建，再替换占位符）
        tools_text = ""
        if self._tool_registry:
            tools_text = self._tool_registry.to_text_description()
        else:
            lines = []
            for t in self.config.tools:
                if t.enabled:
                    params = ", ".join(f"{k}: {v}" for k, v in (t.parameters or {}).items())
                    lines.append(f"- {t.name}({params}): {t.description}")
            tools_text = "\n".join(lines)
        prompt = prompt.replace("{tools_description}", tools_text)

        # 注入项目上下文
        try:
            ctx = await self._load_project_context()
            prompt = prompt.replace("{project_context}", ctx)
        except Exception:
            prompt = prompt.replace("{project_context}", "")

        return prompt

    # ═══════════════════════════════════════════════════════
    # LLM 调用
    # ═══════════════════════════════════════════════════════

    async def _call_llm(self) -> str:
        """调用 LLM，返回原始文本"""
        from pipeline_service.llm_bridge import llm_call_msgs

        # 使用 OpenAI native function calling（如果模型支持）
        use_native = False
        tools_schema = None
        if self._tool_registry and use_native:
            tools_schema = self._tool_registry.to_openai_schema()

        # 简单消息调用（兼容所有模型）
        return await llm_call_msgs(
            self._msgs,
            model=self.model_name,
            temperature=self.config.temperature,
        )

    # ═══════════════════════════════════════════════════════
    # 解析 LLM 输出
    # ═══════════════════════════════════════════════════════

    def _parse_actions(self, raw: str) -> List[dict]:
        """解析 LLM 输出，支持单个 JSON 或连续多个 JSON。

        对照 HA：解析器成熟处理多 JSON + 容错。
        """
        raw = (raw or "").strip()

        # 去除 markdown 代码块包裹
        if raw.startswith("```"):
            parts = raw.split("\n", 1)
            if len(parts) > 1:
                raw = parts[1]
            if raw.endswith("```"):
                raw = raw[:-3]
        raw = raw.strip()

        results = []

        # 尝试多 JSON 解析
        # LLM 可能输出: {"action":"tool_call",...}\n{"action":"tool_call",...}
        lines = raw.split("\n")
        buffer = ""
        depth = 0

        for line in lines:
            line_stripped = line.strip()
            if not line_stripped:
                if buffer:
                    # 尝试解析 buffer
                    result = self._try_parse_json(buffer)
                    if result:
                        results.append(result)
                    buffer = ""
                continue

            buffer += line_stripped

            # 简单大括号计数
            depth += line_stripped.count("{") - line_stripped.count("}")

            if depth == 0 and buffer:
                result = self._try_parse_json(buffer)
                if result:
                    results.append(result)
                buffer = ""

        # 处理残留
        if buffer:
            result = self._try_parse_json(buffer)
            if result:
                results.append(result)

        if not results:
            # 完全无法解析 JSON → 当作纯文本回复
            results.append({"action": "reply", "message": raw})

        return results

    def _try_parse_json(self, text: str) -> Optional[dict]:
        """容错 JSON 解析"""
        text = text.strip()
        try:
            d = json.loads(text)
            if isinstance(d, dict):
                return d
        except (json.JSONDecodeError, ValueError):
            # 尝试提取第一个 JSON 对象
            for start_char in ["{", "["]:
                if start_char in text:
                    idx = text.index(start_char)
                    end_char = "}" if start_char == "{" else "]"
                    end_idx = text.rfind(end_char)
                    if end_idx > idx:
                        try:
                            d = json.loads(text[idx:end_idx + 1])
                            if isinstance(d, dict):
                                return d
                        except (json.JSONDecodeError, ValueError):
                            pass
        return None

    # ═══════════════════════════════════════════════════════
    # 工具执行
    # ═══════════════════════════════════════════════════════

    async def _execute_tool(self, tool_name: str, params: dict) -> str:
        """执行工具，返回结果字符串。

        优先级：
        1. ToolRegistry 注册的 handler
        2. SDLC 内建工具（兼容 cockpit_chat.dspy 的 TOOLS）
        3. ask_user（始终可用）
        """
        # 1. 注册的 handler
        if self._tool_registry:
            handler = self._tool_registry.get_handler(tool_name)
            if handler:
                try:
                    result = await handler(params, {
                        "project_id": self.project_id,
                        "user_id": self.user_id,
                        "workspace_dir": self.workspace_dir,
                    })
                    return str(result)
                except Exception as e:
                    logger.error(f"Tool {tool_name} handler error: {e}")
                    return f"ERROR: {str(e)[:300]}"

        # 2. SDLC 内建工具
        result = await self._execute_sdlc_tool(tool_name, params)
        if result is not None:
            return result

        # 3. ask_user
        if tool_name == "ask_user":
            return f"QUESTION: {params.get('question', '')}"

        return f"未知工具: {tool_name}"

    async def _execute_sdlc_tool(self, tool_name: str, params: dict) -> Optional[str]:
        """SDLC 内建工具（兼容 cockpit_chat.dspy）"""
        # 导入现有 handler
        try:
            from sqlor.dbpools import DBPools

            db = DBPools()
            async with db.sqlorContext("pipeline") as sor:
                return await self._dispatch_sdlc_tool(sor, tool_name, params)
        except Exception as e:
            logger.error(f"SDLC tool {tool_name} error: {e}")
            return f"ERROR: {str(e)[:300]}"

    async def _dispatch_sdlc_tool(self, sor, tool_name: str, params: dict) -> str:
        """SDL 工具分发（兼容 cockpit_chat.dspy 的 16 个工具）"""
        p = params or {}
        pid = self.project_id

        # 所有工具映射（含别名）
        handlers = {
            "switch_project": self._t_switch_project,
            "create_project": self._t_create_project,
            "create_task": self._t_create_task,
            "list_tasks": self._t_list_tasks,
            "task_detail": self._t_task_detail,
            "get_task": self._t_task_detail,        # 别名
            "start_agents": self._t_start_agents,
            "diagnose_project": self._t_diagnose_project,
            "list_deliverables": self._t_list_deliverables,
            "view_deliverable": self._t_view_deliverable,
            "get_deliverable": self._t_view_deliverable,  # 别名
            "list_questions": self._t_list_questions,
            "answer_question": self._t_answer_question,
            "add_repo": self._t_add_repo,
            "list_repos": self._t_list_repos,
            "run_command": self._t_run_command,
            "check_progress": self._t_check_progress,
            "add_bug": self._t_add_bug,
        }

        handler = handlers.get(tool_name)
        if handler:
            return await handler(sor, p, pid)
        return None

    # ── 工具实现 ──

    async def _t_switch_project(self, sor, p, pid):
        name = p.get("project_name", "").strip()
        if not name:
            return "需要项目名称"

        # 精确匹配
        recs = await sor.sqlExe(
            "SELECT id, name FROM sd_projects WHERE name=${n}$", {"n": name})
        if recs:
            r = recs[0]
            self.project_id = getattr(r, "id", "")
            return f"OK: 已切换到 {getattr(r, 'name', name)}"

        # LLM 分类匹配
        try:
            from pipeline_service.llm_bridge import llm_call

            all_recs = await sor.sqlExe(
                "SELECT name FROM sd_projects ORDER BY created_at DESC LIMIT 20", {})
            pnames = [getattr(r, "name", "") for r in (all_recs or [])]
            classify_prompt = f"用户输入: {name}\n项目列表: {', '.join(pnames)}\n\n判断用户想要哪个项目。只回复项目名或\"不存在\"。"
            matched = await llm_call(classify_prompt, temperature=0.0)
            matched = matched.strip().strip('"').strip("'")

            recs2 = await sor.sqlExe(
                "SELECT id, name FROM sd_projects WHERE name=${n}$", {"n": matched})
            if recs2:
                r = recs2[0]
                self.project_id = getattr(r, "id", "")
                return f"OK: 已切换到 {getattr(r, 'name', matched)}"
        except Exception:
            pass

        return f"未找到项目: {name}"

    async def _t_create_project(self, sor, p, pid):
        from appPublic.uniqueID import getID

        name = p.get("name", "").strip()
        desc = p.get("description", "")
        if not name:
            return "需要项目名称"

        pid_val = getID()
        await sor.C("sd_projects", {
            "id": pid_val, "name": name, "description": desc,
            "status": "active",
        })
        await sor.C("sd_iterations", {
            "id": getID(), "project_id": pid_val,
            "iteration_name": f"{name}-初始迭代",
            "iteration_type": "default", "status": "active", "priority": 1,
        })
        self.project_id = pid_val
        return f"OK: 已创建项目 {name}"

    async def _t_create_task(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        from appPublic.uniqueID import getID

        title = p.get("title", "").strip()
        role = p.get("role", "develop")
        desc = p.get("description", "")
        if not title:
            return "需要任务标题"

        tid = getID()
        await sor.C("pipeline_tasks", {
            "id": tid, "tenant_id": pid, "pipeline_id": "role_task",
            "owner_id": self.user_id or "user", "title": title,
            "state": "submitted", "role": role,
            "params": json.dumps({"description": desc}, ensure_ascii=False),
        })
        return f"OK: 已创建任务 {title}（角色: {role}）"

    async def _t_list_tasks(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        role = p.get("role", "")
        state = p.get("state", "")

        sql = "SELECT id, title, role, state, created_at FROM pipeline_tasks WHERE tenant_id=${pid}$"
        params_dict = {"pid": pid}
        if role:
            sql += " AND role=${role}$"
            params_dict["role"] = role
        if state:
            sql += " AND state=${state}$"
            params_dict["state"] = state
        sql += " ORDER BY created_at DESC LIMIT 30"

        recs = await sor.sqlExe(sql, params_dict)
        if not recs:
            return "暂无任务"

        lines = []
        for r in recs:
            lines.append(
                f"- [{getattr(r, 'state', '?')}] {getattr(r, 'title', '')} "
                f"(角色: {getattr(r, 'role', '')}) "
                f"id={getattr(r, 'id', '')[:8]}"
            )
        return "\n".join(lines)

    async def _t_task_detail(self, sor, p, pid):
        tid = p.get("task_id", "")
        if not tid:
            return "需要任务ID"

        recs = await sor.sqlExe(
            "SELECT * FROM pipeline_tasks WHERE id=${tid}$", {"tid": tid})
        if not recs:
            return f"任务不存在: {tid}"

        r = recs[0]
        return json.dumps({
            "id": getattr(r, "id", ""),
            "title": getattr(r, "title", ""),
            "state": getattr(r, "state", ""),
            "role": getattr(r, "role", ""),
            "params": getattr(r, "params", "{}"),
        }, ensure_ascii=False, indent=2)

    async def _t_start_agents(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        # 找 submitted 任务
        recs = await sor.sqlExe(
            "SELECT id, title, role FROM pipeline_tasks "
            "WHERE tenant_id=${pid}$ AND state='submitted' ORDER BY created_at ASC LIMIT 10",
            {"pid": pid})
        if not recs:
            return "没有待执行的任务"

        results = []
        for r in recs:
            tid = getattr(r, "id", "")
            role = getattr(r, "role", "develop")
            try:
                from pipeline_service.agent_loop import role_agent_run

                result = await role_agent_run(pid, role)
                results.append(f"- {getattr(r, 'title', '')}: {result.get('status', '?')}")
            except Exception as e:
                results.append(f"- {getattr(r, 'title', '')}: error={str(e)[:100]}")

        return "Agent 执行结果:\n" + "\n".join(results)

    async def _t_diagnose_project(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        # 汇总各种状态
        submitted = await sor.sqlExe(
            "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='submitted'",
            {"pid": pid})
        running = await sor.sqlExe(
            "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='running'",
            {"pid": pid})
        failed = await sor.sqlExe(
            "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='failed'",
            {"pid": pid})
        review = await sor.sqlExe(
            "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='review'",
            {"pid": pid})
        questions = await sor.sqlExe(
            "SELECT COUNT(*) as c FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND answered IS NULL",
            {"pid": pid})

        sc = getattr(submitted[0], "c", 0) if submitted else 0
        rc = getattr(running[0], "c", 0) if running else 0
        fc = getattr(failed[0], "c", 0) if failed else 0
        vc = getattr(review[0], "c", 0) if review else 0
        qc = getattr(questions[0], "c", 0) if questions else 0

        return (
            f"项目诊断:\n"
            f"- 待执行: {sc}\n"
            f"- 运行中: {rc}\n"
            f"- 待审核: {vc}\n"
            f"- 失败: {fc}\n"
            f"- 待回答问题: {qc}\n"
        )

    async def _t_list_deliverables(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        role = p.get("role", "")
        sql = "SELECT id, title, deliverable_type, review_status FROM pipeline_deliverables WHERE project_id=${pid}$"
        params_dict = {"pid": pid}
        sql += " ORDER BY created_at DESC LIMIT 20"

        recs = await sor.sqlExe(sql, params_dict)
        if not recs:
            return "暂无交付件"

        lines = []
        for r in recs:
            lines.append(
                f"- [{getattr(r, 'review_status', '?')}] {getattr(r, 'title', '')} "
                f"({getattr(r, 'deliverable_type', '')})"
            )
        return "\n".join(lines)

    async def _t_view_deliverable(self, sor, p, pid):
        did = p.get("deliverable_id", "")
        if not did:
            return "需要交付件ID"

        recs = await sor.sqlExe(
            "SELECT * FROM pipeline_deliverables WHERE id=${did}$", {"did": did})
        if not recs:
            return f"交付件不存在: {did}"

        r = recs[0]
        return getattr(r, "content", "")[:4000] or "(空)"

    async def _t_list_questions(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        recs = await sor.sqlExe(
            "SELECT id, question, answer_source FROM pipeline_agent_questions "
            "WHERE tenant_id=${pid}$ AND answered IS NULL ORDER BY created_at DESC LIMIT 10",
            {"pid": pid})
        if not recs:
            return "没有待回答问题"

        lines = []
        for r in recs:
            lines.append(f"- Q{getattr(r, 'id', '')[:8]}: {getattr(r, 'question', '')[:200]}")
        return "\n".join(lines)

    async def _t_answer_question(self, sor, p, pid):
        qid = p.get("question_id", "")
        answer = p.get("answer", "")
        if not qid or not answer:
            return "需要问题ID和回答内容"

        await sor.sqlExe(
            "UPDATE pipeline_agent_questions SET answered=1, answer=${a}$ WHERE id=${qid}$",
            {"a": answer, "qid": qid})
        return "OK: 已回答"

    async def _t_add_repo(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        from appPublic.uniqueID import getID

        url = p.get("repo_url", "").strip()
        name = p.get("repo_name", "").strip()
        branch = p.get("default_branch", "main")
        if not url:
            return "需要仓库URL"

        await sor.C("sd_project_repos", {
            "id": getID(), "project_id": pid,
            "repo_url": url, "repo_name": name or url.split("/")[-1].replace(".git", ""),
            "default_branch": branch,
        })
        return f"OK: 已添加仓库 {name or url}"

    async def _t_list_repos(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        recs = await sor.sqlExe(
            "SELECT repo_name, repo_url, default_branch FROM sd_project_repos WHERE project_id=${pid}$",
            {"pid": pid})
        if not recs:
            return "暂无关联仓库"

        lines = []
        for r in recs:
            lines.append(f"- {getattr(r, 'repo_name', '')}: {getattr(r, 'repo_url', '')}")
        return "\n".join(lines)

    async def _t_run_command(self, sor, p, pid):
        cmd = p.get("command", "")
        if not cmd:
            return "需要命令"

        try:
            from pipeline_service.agent_loop import _run_shell

            r = await _run_shell(cmd, self.workspace_dir, timeout=60)
            return f"rc={r['rc']}\n{r['stdout'][:2000]}"
        except Exception as e:
            return f"ERROR: {str(e)[:300]}"

    async def _t_check_progress(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        recs = await sor.sqlExe(
            "SELECT role, state, COUNT(*) as c FROM pipeline_tasks "
            "WHERE tenant_id=${pid}$ GROUP BY role, state ORDER BY role, state",
            {"pid": pid})
        if not recs:
            return "暂无进度数据"

        lines = []
        for r in recs:
            lines.append(
                f"- {getattr(r, 'role', '?')}: {getattr(r, 'state', '?')} x{getattr(r, 'c', 0)}"
            )
        return "\n".join(lines)

    async def _t_add_bug(self, sor, p, pid):
        if not pid:
            return "请先切换到项目"

        from appPublic.uniqueID import getID

        title = p.get("title", "").strip()
        desc = p.get("description", "")
        severity = p.get("severity", "medium")
        if not title:
            return "需要Bug标题"

        await sor.C("sd_bugs", {
            "id": getID(), "title": title,
            "description": desc, "severity": severity,
            "status": "open",
        })
        return f"OK: 已提交Bug: {title}"

    # ═══════════════════════════════════════════════════════
    # 上下文压缩
    # ═══════════════════════════════════════════════════════

    async def _maybe_compress(self):
        """检查是否需要压缩上下文"""
        est_tokens = self._estimate_tokens()
        limit = self.config.context_limit
        threshold = int(limit * self.config.compression.threshold)

        if est_tokens < threshold:
            return

        logger.info(f"Compressing: {est_tokens}/{limit} tokens (threshold={threshold})")

        # 保留 system + 最近 N 轮
        keep = self.config.compression.keep_recent * 2  # user + assistant 各一条
        system_msg = self._msgs[0] if self._msgs else None
        recent = self._msgs[-keep:] if keep < len(self._msgs) else self._msgs[1:]

        # 压缩中间消息
        middle = self._msgs[1:-keep] if keep < len(self._msgs) else []
        if middle:
            summary = await self._summarize(middle)
            compressed = [
                system_msg,
                {"role": "user", "content": f"[对话摘要]\n{summary}"},
            ] if system_msg else []
            compressed.extend(recent)
            self._msgs = compressed

    def _estimate_tokens(self) -> int:
        """简单 token 估算（中文 ~1.5 字符/token，英文 ~4 字符/token）"""
        total = 0
        for m in self._msgs:
            content = m.get("content", "")
            # 简单估算：平均每 2 字符 1 token
            total += len(content) // 2 + 1
        return total

    async def _summarize(self, messages: list) -> str:
        """压缩消息为摘要"""
        if len(messages) <= 2:
            return "\n".join(m.get("content", "")[:200] for m in messages)

        text = "\n".join(
            f"[{m['role']}]: {m.get('content', '')[:500]}"
            for m in messages
        )

        try:
            from pipeline_service.llm_bridge import llm_call

            summary = await llm_call(
                f"请用3-5句话总结以下对话的关键信息：\n\n{text[:4000]}",
                temperature=0.1,
            )
            return summary[:500]
        except Exception:
            return text[:500]

    # ═══════════════════════════════════════════════════════
    # 会话管理
    # ═══════════════════════════════════════════════════════

    async def _load_history(self, external_history: List[dict] = None) -> List[dict]:
        """加载历史消息。隔离策略：project 级别。"""
        if external_history:
            return external_history

        if self.config.session_isolation == "none":
            return []

        try:
            from sqlor.dbpools import DBPools

            db = DBPools()
            async with db.sqlorContext("pipeline") as sor:
                # 项目隔离
                pid = self.project_id
                if self.config.session_isolation == "project" and pid:
                    recs = await sor.sqlExe(
                        "SELECT role, content FROM pipeline_conversations "
                        "WHERE iteration_id=${pid}$ AND created_by=${uid}$ "
                        "ORDER BY created_at ASC LIMIT ${lim}$",
                        {"pid": pid, "uid": self.user_id or "", "lim": self.config.history_limit},
                    )
                else:
                    recs = await sor.sqlExe(
                        "SELECT role, content FROM pipeline_conversations "
                        "WHERE created_by=${uid}$ "
                        "ORDER BY created_at ASC LIMIT ${lim}$",
                        {"uid": self.user_id or "", "lim": self.config.history_limit},
                    )

                if not recs:
                    return []

                # 过滤 tool_call JSON 污染（关键！）
                msgs = []
                for r in recs:
                    role = getattr(r, "role", "")
                    content = getattr(r, "content", "") or ""
                    # 跳过 tool_call JSON
                    if content.startswith('{"action":"tool_call"'):
                        continue
                    if content.startswith("已调用 "):
                        continue
                    if role in ("user", "assistant"):
                        msgs.append({"role": role, "content": content})
                return msgs
        except Exception as e:
            logger.warning(f"History load failed: {e}")
            return []

    async def _save_turn(self, user_input: str, reply: str):
        """持久化本轮对话"""
        if self.config.session_isolation == "none":
            return

        try:
            from sqlor.dbpools import DBPools
            from appPublic.uniqueID import getID

            db = DBPools()
            async with db.sqlorContext("pipeline") as sor:
                # 用户消息
                await sor.C("pipeline_conversations", {
                    "id": getID(),
                    "role": "user",
                    "content": user_input,
                    "created_by": self.user_id or "",
                    "iteration_id": self.project_id or "",
                })
                # 助手回复（不含 tool_call JSON！）
                reply_clean = reply
                if reply.startswith('{"action":"tool_call"'):
                    reply_clean = "[工具调用]"
                await sor.C("pipeline_conversations", {
                    "id": getID(),
                    "role": "assistant",
                    "content": reply_clean[:4000],
                    "created_by": self.user_id or "",
                    "iteration_id": self.project_id or "",
                })
        except Exception as e:
            logger.warning(f"Session save failed: {e}")

    async def _load_project_context(self) -> str:
        """加载项目上下文（项目名、任务数等）"""
        if not self.project_id:
            return ""

        try:
            from sqlor.dbpools import DBPools

            db = DBPools()
            async with db.sqlorContext("pipeline") as sor:
                recs = await sor.sqlExe(
                    "SELECT name, description FROM sd_projects WHERE id=${pid}$",
                    {"pid": self.project_id})
                if recs:
                    name = getattr(recs[0], "name", "")
                    return f"项目: {name}"
        except Exception:
            pass
        return ""

    def _needs_confirmation(self, tool_name: str) -> bool:
        """检查工具是否需要用户确认"""
        if self._tool_registry:
            tool = self._tool_registry.get(tool_name)
            if tool and tool.requires_confirmation:
                return True
        return False


# ═══════════════════════════════════════════════════════════
# 便捷函数
# ═══════════════════════════════════════════════════════════

async def run_agent(
    user_input: str,
    project_id: str = "",
    user_id: str = "",
    config=None,
    model_name: str = None,
    workspace_dir: str = "",
) -> AsyncGenerator[str, None]:
    """便捷入口：运行 agent 并流式输出。

    用法：
        async for chunk in run_agent("创建一个新项目"):
            print(chunk)
    """
    if config is None:
        try:
            from pipeline_core.agent_config import load_agent_config, SDLC_DEFAULT_CONFIG

            config = await load_agent_config(project_id=project_id)
        except ImportError:
            config = SDLC_DEFAULT_CONFIG

    executor = AgentExecutor(
        config=config,
        project_id=project_id,
        user_id=user_id,
        workspace_dir=workspace_dir,
        model_name=model_name,
    )
    async for chunk in executor.run(user_input):
        yield chunk