"""Agent loop — 角色agent循环认领执行任务（v3.4.0）。

v3.4.0 新增：
- 所有agent具备文件读写 + git pull/push 能力
- agent 产出实际代码文件到项目仓库，而非仅文档
- PM agent 负责 clone 项目关联仓库、初始化工作区
- develop agent 产出可直接运行的源码文件
- 每个agent完成后自动 git commit + push

任务链：requirement → design → develop → test → deploy
"""

import asyncio
import json
import os
import re
import subprocess
import logging

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

ROLE_ALIASES = {
    'developer': 'develop', 'dev': 'develop', 'development': 'develop', 'coding': 'develop',
    'requirement': 'requirement', 'requirements': 'requirement', 'requirement_analysis': 'requirement',
    'designer': 'design', 'ui': 'design', 'ux': 'design',
    'testing': 'test', 'qa': 'test', 'tester': 'test',
    'deployment': 'deploy', 'release': 'deploy',
    'operation': 'ops', 'operations': 'ops', 'maintenance': 'ops', '运维': 'ops',
    'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm',
}

ROLE_CHAIN = {
    'requirement': 'design',
    'design': 'develop',
    'develop': 'test',
    'test': 'deploy',
    'deploy': None,
}

TASK_REVIEW = 'review'
TASK_APPROVED = 'approved'

# 允许的 shell 工作目录前缀（安全限制）
_ALLOWED_WORKDIRS = [
    os.path.expanduser('~/pipeline_ws'),
    '/d/pipeline/workspaces',
    '/tmp/pipeline_workspaces',
    '/tmp/pipeline_ws',
]


def _normalize_role(role):
    r = (role or '').strip().lower()
    return ROLE_ALIASES.get(r, r)


def _get_db():
    from sqlor.dbpools import DBPools
    db = DBPools()
    if not db.databases:
        from appPublic.jsonConfig import getConfig
        config = getConfig()
        if config.databases:
            db.databases = config.databases
    return db


def _resolve_workspace(workspace_dir):
    if not workspace_dir:
        workspace_dir = os.path.expanduser('~/pipeline_ws/default')
    try:
        parent = os.path.dirname(workspace_dir)
        if parent and not os.access(parent, os.W_OK):
            proj_name = os.path.basename(workspace_dir)
            workspace_dir = os.path.expanduser(f'~/pipeline_ws/{proj_name}')
    except Exception:
        workspace_dir = os.path.expanduser(f'~/pipeline_ws/{os.path.basename(workspace_dir) or "default"}')
    os.makedirs(workspace_dir, exist_ok=True)
    return workspace_dir


# ── Agent 工具函数 ──

def _is_safe_workdir(workdir):
    """检查目录是否在允许范围内。"""
    wd = os.path.abspath(workdir)
    for allowed in _ALLOWED_WORKDIRS:
        awd = os.path.abspath(os.path.expanduser(allowed))
        if wd.startswith(awd):
            return True
    return False


async def _run_shell(command, workdir, timeout=120):
    """安全执行 shell 命令。返回 {"rc": int, "stdout": str, "stderr": str}"""
    cwd = os.path.abspath(workdir) if workdir else os.path.expanduser('~/pipeline_ws')
    if not _is_safe_workdir(cwd):
        return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围"}
    if not os.path.isdir(cwd):
        return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}"}
    try:
        proc = await asyncio.create_subprocess_shell(
            command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
            cwd=cwd, executable='/bin/bash')
        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            return {"rc": -1, "stdout": "", "stderr": f"命令超时（{timeout}s）"}
        return {"rc": proc.returncode or 0,
                "stdout": stdout.decode('utf-8', 'replace')[-8000:],
                "stderr": stderr.decode('utf-8', 'replace')[-4000:]}
    except Exception as e:
        return {"rc": -1, "stdout": "", "stderr": str(e)[:500]}


async def _write_code_file(filepath, content):
    """写代码文件，自动创建父目录。"""
    try:
        d = os.path.dirname(filepath)
        if d:
            os.makedirs(d, exist_ok=True)
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(content or '')
        return True, filepath
    except Exception as e:
        return False, str(e)


async def _git_setup(workdir):
    """确保 git 用户已配置。"""
    r = await _run_shell('git config user.email', workdir, 5)
    if 'opencomputing' not in r.get('stdout', '') and '@' not in r.get('stdout', ''):
        await _run_shell('git config user.email "pipeline@opencomputing.cn"', workdir, 5)
        await _run_shell('git config user.name "Pipeline Agent"', workdir, 5)


async def _git_commit_push(workdir, commit_message, branch='main'):
    """git add → commit → push。"""
    await _git_setup(workdir)
    r1 = await _run_shell('git add -A', workdir, 10)
    r2 = await _run_shell(f'git diff --cached --stat', workdir, 10)
    if not r2.get('stdout', '').strip():
        return {"rc": 0, "message": "没有变更需要提交"}
    r3 = await _run_shell(f'git commit -m "{commit_message}"', workdir, 15)
    if r3['rc'] != 0:
        return {"rc": r3['rc'], "message": f"commit 失败: {r3['stderr'][:200]}"}
    r4 = await _run_shell(f'git push origin {branch}', workdir, 30)
    return {"rc": r4['rc'], "message": f"push {'成功' if r4['rc']==0 else '失败'}: {r4['stderr'][:200]}"}


async def _git_clone(repo_url, target_dir, branch='main'):
    """克隆仓库到目标目录。已存在则 pull。"""
    if os.path.isdir(os.path.join(target_dir, '.git')):
        r = await _run_shell(f'git checkout {branch} && git pull origin {branch}', target_dir, 30)
        return {"rc": r['rc'], "message": f"已存在，pull: {r['stdout'][:200]}"}
    parent = os.path.dirname(target_dir)
    os.makedirs(parent, exist_ok=True)
    r = await _run_shell(f'git clone -b {branch} {repo_url} {target_dir}', parent, 120)
    return {"rc": r['rc'], "message": f"clone: {r['stdout'][:200] if r['rc']==0 else r['stderr'][:200]}"}


# ── Prompts ──

AGENT_TOOLS = [
    {"name":"read_file","description":"读取工作空间中的文件","params":{"path":"相对路径"}},
    {"name":"write_file","description":"写入文件（自动创建父目录）","params":{"path":"相对路径","content":"文件内容"}},
    {"name":"list_files","description":"列出目录内容","params":{"path":"相对路径(可选,默认工作空间根)"}},
    {"name":"run_shell","description":"在工作空间中执行shell命令","params":{"command":"命令"}},
    {"name":"git_clone","description":"克隆git仓库到工作空间repos/下","params":{"repo_url":"仓库URL","repo_name":"仓库目录名(可选,默认从URL推断)","branch":"分支(可选,默认main)"}},
    {"name":"git_status","description":"查看git仓库状态","params":{"repo_dir":"仓库子目录(可选,默认repos下第一个)"}},
    {"name":"git_commit_push","description":"git add + commit + push","params":{"message":"提交信息","repo_dir":"仓库子目录(可选)"}},
    {"name":"ask_question","description":"向用户提问（缺少信息时使用）","params":{"question":"问题"}},
]

AGENT_SYSTEM_PROMPT = """你是软件开发产线中的「__ROLE__」角色Agent。

## 任务
__TITLE__
__QNA__

## 工作环境
工作空间：__WORKSPACE__
产出要求：__ROLE_SPECIFIC__

## 工具
你可以使用以下工具完成工作：
__TOOLS__

## 工作流
1. 先用 read_file/list_files 了解现有代码
2. 用 write_file 产出代码文件到 repos/ 下
3. 用 run_shell 验证（编译/测试）
4. 用 git_commit_push 提交到远端
5. 用 deliver 提交最终交付件

## 输出格式（每次只输出一个JSON对象）
调工具：
{"action":"tool_call","tool":"工具名","params":{}}

提交交付件：
{"action":"deliver","deliverable_type":"code_files","summary":"概述","result":"文档内容","files":[{"path":"repos/仓库/src/file.py","content":"代码"}],"git_commit_message":"feat: 描述"}

提问：
{"action":"ask","question":"问题"}

注意：每次只输出一个JSON！收到工具结果后再决定下一步。"""

ROLE_SPECIFICS = {
    'requirement': """你是需求分析师。按 SDLC 仓库标准产出文档。

产出路径: docs/00-requirement/requirement-spec.md
同时创建（项目至少一个应用）:
  - apps/<应用名>.md: 应用描述、部署环境、端口
  - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖
  - 每个应用至少关联一个模块
内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""",

    'design': """你是系统设计师。按 SDLC 仓库标准产出文档。

产出路径: docs/01-design/ 目录下:
  - architecture.md: 系统架构、技术选型理由
  - database-design.md: ER图描述、表结构DDL
  - api-design.md: 接口列表(method/path/request/response)
  - ui-design.md: 页面结构、组件树(如适用)
产出后用 result 输出文档，files 列出所有文件路径。""",

    'develop': """你是开发工程师。源码写入模块独立仓库，不在项目仓库。

准备工作:
  1. read_file 读 docs/01-design/ 下的设计文档
  2. read_file 读 modules/<模块名>.md 获取模块仓库 URL
  3. git_clone 克隆模块仓库到工作空间
开发流程:
  4. write_file 写代码到模块仓库目录下
  5. run_shell 编译/运行验证
  6. git_commit_push 提交 "develop: <简述>"
  7. write_file 更新 modules/<模块名>.md 状态为已完成
  8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容
  9. deliver 交付，files 列出所有产出文件路径
PM审核: git_status 检查模块仓库有提交记录。""",

    'test': """你是测试工程师。按 SDLC 仓库标准产出文档。

产出路径: docs/03-test/ 目录下:
  - test-plan.md: 测试策略(单元/集成/端到端)
  - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果)
  - test-report.md: 执行结果、Bug清单、覆盖率
测试脚本放到 files。""",

    'deploy': """你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。

产出路径:
  - docs/04-deploy/deploy-guide.md: 环境要求、部署步骤、回滚方案
  - docs/04-deploy/release-notes.md: 发布说明
  - config/ 下: Dockerfile、nginx配置、docker-compose.yml 等
产出后用 result 输出部署说明，files 列出所有配置文件路径。""",
}

PM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。

## 项目
工作目录：__WORKSPACE__
关联仓库：__REPOS__

## 待审核任务
标题：__TITLE__
角色：__ROLE__
交付件摘要：__SUMMARY__
交付件内容(前6000字)：__DELIVERABLE__

## 仓库状态
__REPO_STATE__

## 决策
输出纯JSON：
1. 通过：{"status":"approved","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"详细描述"}
2. 驳回：{"status":"rejected","comment":"驳回原因","questions":"修改要求"}
3. 完成：{"status":"completed","comment":"项目总结"}

审核标准：
- requirement：需求是否清晰完整可量化
- design：方案合理、覆盖需求、技术可行
- develop：代码文件是否实际产出、目录结构是否合理、是否可编译运行
- test：测试覆盖充分、发现问题记录完整
- deploy：部署配置完整、可一键部署"""

PM_SYSTEM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。

## 项目信息
工作目录：__WORKSPACE__
关联仓库：__REPOS__

## 待审核
标题：__TITLE__
角色：__ROLE__

## 仓库状态
__REPO_STATE__

## 工具
- read_file(path) — 读工作空间文件
- list_files(path) — 列目录
- git_status() — 查看git状态
- run_shell(command) — 执行命令（编译/测试验证）

## 审核流程
1. 先用工具检查代码是否实际产出（git_status/list_files/read_file）
2. 如需编译验证，用 run_shell
3. 最后给出决策

## 输出格式（每次一个JSON）
查看文件：{"action":"tool_call","tool":"read_file","params":{"path":"相对路径"}}
查看git：{"action":"tool_call","tool":"git_status","params":{}}
批准：{"action":"review_approve","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"描述"}
驳回：{"action":"review_reject","comment":"原因","questions":"修改要求"}
完成：{"action":"review_complete","comment":"总结"}

审核标准：
- requirement：需求是否清晰完整可量化
- design：方案合理、覆盖需求、技术可行  
- develop：必须用 git_status/read_file 检查代码是否实际写入仓库
- test：测试覆盖充分、发现问题记录完整
- deploy：部署配置完整、可一键部署"""


# ── Agent 核心逻辑 ──

async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True):
    role = _normalize_role(role)
    where_role = "AND (role=${role}$ OR role='')" if match_role else ""
    recs = await sor.sqlExe(
        "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks "
        "WHERE tenant_id=${tid}$ AND state=${state}$ " + where_role + " "
        "AND NOT EXISTS (SELECT 1 FROM pipeline_task_steps s WHERE s.task_id=pipeline_tasks.id) "
        "ORDER BY created_at ASC LIMIT 1",
        {"tid": tenant_id, "state": state, "role": role})
    if not recs:
        return None
    task = recs[0]
    task_id = task.id
    from appPublic.uniqueID import getID
    claim_token = getID()
    await sor.sqlExe(
        "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ "
        "WHERE id=${tid}$ AND state=${state}$",
        {"cb": claim_token, "tid": task_id, "state": state})
    check = await sor.sqlExe(
        "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state='running' AND claimed_by=${cb}$",
        {"tid": task_id, "cb": claim_token})
    if not check:
        logger.info(f"claim lost race: task={task_id}")
        return None
    return task


async def _get_workspace_dir(sor, project_id):
    recs = await sor.sqlExe("SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", {"pid": project_id})
    ws = getattr(recs[0], 'workspace_dir', '') if recs else ''
    return _resolve_workspace(ws)


async def _build_qna_section(sor, task_id):
    from .questions import get_task_qna
    qna = await get_task_qna(task_id)
    if not qna:
        return ""
    lines = ["历史问答："]
    for item in qna:
        src = "客户" if item.get('answer_source') == "customer" else "主agent"
        lines.append(f"问：{item.get('question','')}")
        lines.append(f"答({src})：{item.get('answer','')}")
    return "\n".join(lines)


def _parse_result(raw):
    raw = (raw or "").strip()
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[1].rsplit("```", 1)[0]
    try:
        d = json.loads(raw)
        if isinstance(d, dict):
            return d
    except (json.JSONDecodeError, ValueError):
        pass
    return {"status": "done", "result": raw}


async def _get_deliverable_content(sor, task_id):
    recs = await sor.sqlExe(
        "SELECT content, deliverable_type FROM pipeline_deliverables "
        "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1", {"tid": task_id})
    if recs:
        return getattr(recs[0], 'content', '') or '', getattr(recs[0], 'deliverable_type', '') or ''
    return '', ''


async def _get_project_repos(sor, project_id):
    """获取项目关联的所有仓库。"""
    recs = await sor.sqlExe(
        "SELECT id, repo_name, repo_url, default_branch, local_path FROM sd_project_repos "
        "WHERE project_id=${pid}$", {"pid": project_id})
    return [{'id': getattr(r, 'id', ''), 'name': getattr(r, 'repo_name', ''),
             'url': getattr(r, 'repo_url', ''), 'branch': getattr(r, 'default_branch', 'main'),
             'local_path': getattr(r, 'local_path', '') or ''} for r in (recs or [])]


async def _setup_repos(sor, workspace_dir, project_id):
    """PM：clone 所有项目关联仓库到 workspace/repos/。"""
    repos = await _get_project_repos(sor, project_id)
    results = []
    repos_dir = os.path.join(workspace_dir, 'repos')
    os.makedirs(repos_dir, exist_ok=True)
    for repo in repos:
        target = os.path.join(repos_dir, repo['name'])
        r = await _git_clone(repo['url'], target, repo['branch'])
        results.append({'repo': repo['name'], **r})
    return results


async def _get_repo_state(workspace_dir):
    """获取仓库当前状态（供PM审核时查看）。"""
    repos_dir = os.path.join(workspace_dir, 'repos')
    if not os.path.isdir(repos_dir):
        return "暂无仓库"
    lines = []
    for name in sorted(os.listdir(repos_dir)):
        rp = os.path.join(repos_dir, name)
        if os.path.isdir(rp) and os.path.isdir(os.path.join(rp, '.git')):
            r = await _run_shell('git log --oneline -3', rp, 5)
            lines.append(f"\n[{name}]")
            lines.append(r.get('stdout', '') or '(空仓库)')
    return "\n".join(lines) if lines else "仓库无提交记录"


async def _get_next_role(current_role):
    return ROLE_CHAIN.get(_normalize_role(current_role))


async def _create_next_task(sor, project_id, task, next_role, pm_comment=''):
    from appPublic.uniqueID import getID
    title = getattr(task, 'title', '') or ''
    params_str = getattr(task, 'params', '{}') or '{}'
    try:
        params = json.loads(params_str) if isinstance(params_str, str) else params_str
    except (json.JSONDecodeError, TypeError):
        params = {}
    new_title = f"{title}（{next_role}阶段）"
    new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')),
                  'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment}
    new_task_id = getID()
    await sor.C('pipeline_tasks', {
        'id': new_task_id, 'tenant_id': project_id, 'pipeline_id': 'role_task',
        'owner_id': 'pm', 'title': new_title,
        'params': json.dumps(new_params, ensure_ascii=False),
        'role': next_role, 'state': 'submitted', 'claimed_by': None,
    })
    logger.info(f"_create_next_task: {next_role} task={new_task_id}")
    return new_task_id, new_title


# ── Agent 工具执行 ──

def _parse_agent_action(raw):
    raw = (raw or "").strip()
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip()

    # deepseek 原生 function calling 输出 XML：<tool_calls><invoke name="..."><parameter name="..." string="true">...</parameter></invoke></tool_calls>
    m = re.search(r'<invoke\s+name="([^"]+)"[^>]*>(.*?)</invoke>', raw, re.DOTALL)
    if m:
        tool = m.group(1).strip()
        body = m.group(2)
        params = {}
        for pm in re.finditer(r'<parameter\s+name="([^"]+)"[^>]*>(.*?)</parameter>', body, re.DOTALL):
            val = pm.group(2).strip()
            try:
                val = json.loads(val)
            except (json.JSONDecodeError, ValueError):
                pass
            params[pm.group(1)] = val
        if tool in ('deliver', 'deliver_result', 'submit', 'finish'):
            return {"action": "deliver", **params}
        if tool in ('ask', 'ask_question', 'ask_user'):
            return {"action": "ask", "question": params.get("question", "")}
        return {"action": "tool_call", "tool": tool, "params": params}

    try:
        d = json.loads(raw)
        if isinstance(d, dict) and 'action' in d:
            return d
    except (json.JSONDecodeError, ValueError):
        pass
    return {"action": "deliver", "result": raw}


def _resolve_repo_target(workspace_dir, repo_dir):
    """解析仓库目录：空→repos/下第一个git仓库；'repos/xxx'→直接；'xxx'→repos/xxx。"""
    repos_dir = os.path.join(workspace_dir, 'repos')
    if repo_dir:
        if repo_dir.startswith('repos/') or repo_dir.startswith('repos\\'):
            return os.path.join(workspace_dir, repo_dir)
        return os.path.join(repos_dir, repo_dir)
    if os.path.isdir(repos_dir):
        dirs = [d for d in os.listdir(repos_dir)
                if os.path.isdir(os.path.join(repos_dir, d, '.git'))]
        if dirs:
            return os.path.join(repos_dir, dirs[0])
    return workspace_dir


async def _exec_agent_tool(tool, params, workspace_dir):
    p = params or {}
    try:
        if tool == 'read_file':
            path = p.get('path', '')
            if not path: return 'FAIL: 需要文件路径'
            full = os.path.join(workspace_dir, path)
            if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围'
            if not os.path.isfile(full): return f'FAIL: 文件不存在 {path}'
            with open(full, encoding='utf-8') as f:
                return f.read()[:8000]
        elif tool == 'write_file':
            path = p.get('path', '')
            content = p.get('content', '')
            if not path: return 'FAIL: 需要文件路径'
            full = os.path.join(workspace_dir, path)
            if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围'
            os.makedirs(os.path.dirname(full), exist_ok=True)
            with open(full, 'w', encoding='utf-8') as f:
                f.write(content)
            return f'OK: 已写入 {path} ({len(content)} 字符)'
        elif tool == 'list_files':
            path = p.get('path', '') or '.'
            full = os.path.join(workspace_dir, path)
            if not _is_safe_workdir(full): return 'FAIL: 路径不在允许范围'
            if not os.path.isdir(full): return f'FAIL: 目录不存在 {path}'
            items = os.listdir(full)[:50]
            lines = []
            for name in sorted(items):
                fp = os.path.join(full, name)
                t = 'DIR' if os.path.isdir(fp) else 'FILE'
                size = os.path.getsize(fp) if os.path.isfile(fp) else 0
                lines.append(f"[{t}] {name} ({size}B)")
            return '\n'.join(lines) if lines else '(空目录)'
        elif tool == 'run_shell':
            cmd = p.get('command', '')
            if not cmd: return 'FAIL: 需要命令'
            r = await _run_shell(cmd, workspace_dir, timeout=120)
            return f"rc={r['rc']}\nSTDOUT:\n{r['stdout'][:2000]}\nSTDERR:\n{r['stderr'][:1000]}"
        elif tool == 'git_clone':
            url = p.get('repo_url', '')
            if not url: return 'FAIL: 需要仓库URL'
            name = p.get('repo_name', '') or url.rstrip('/').split('/')[-1].replace('.git', '')
            target = os.path.join(workspace_dir, 'repos', name)
            r = await _git_clone(url, target, p.get('branch', 'main'))
            return f"rc={r['rc']} {r['message']}"
        elif tool == 'git_status':
            target = _resolve_repo_target(workspace_dir, p.get('repo_dir', ''))
            r = await _run_shell('git status --short', target, 10)
            r2 = await _run_shell('git log --oneline -3', target, 10)
            return f"Status:\n{r['stdout'][:1000] or '(clean)'}\nRecent:\n{r2['stdout'][:500]}"
        elif tool == 'git_commit_push':
            msg = p.get('message', '') or 'agent update'
            target = _resolve_repo_target(workspace_dir, p.get('repo_dir', ''))
            r = await _git_commit_push(target, msg)
            return f"rc={r['rc']} {r['message']}"
        return f'未实现: {tool}'
    except Exception as e:
        return f'ERROR: {str(e)[:300]}'


# ── 角色 Agent ──

async def role_agent_run(project_id, role, agent_id=None, model_name=None):
    role = _normalize_role(role)
    if role == 'pm':
        return {"status": "idle", "message": "PM请使用 pm_review_run"}

    db = _get_db()
    async with db.sqlorContext("pipeline") as sor:
        task = await _claim_task(sor, project_id, role)
        if not task:
            return {"status": "idle", "message": "没有待办任务"}

        task_id = task.id
        title = getattr(task, "title", "") or ""
        params_str = getattr(task, "params", "{}") or "{}"
        workspace_dir = await _get_workspace_dir(sor, project_id)

        try:
            await sor.sqlExe("COMMIT", {})
        except Exception:
            pass

        repos_dir = os.path.join(workspace_dir, 'repos')
        os.makedirs(repos_dir, exist_ok=True)

        # 确保项目关联仓库已 clone（幂等：已存在则 pull）——否则源码写不进 git 仓库
        try:
            clone_results = await _setup_repos(sor, workspace_dir, project_id)
            logger.info(f"role_agent_run setup_repos: {clone_results}")
        except Exception as e:
            logger.warning(f"role_agent_run setup_repos failed: {e}")

        role_specific = ROLE_SPECIFICS.get(role, ROLE_SPECIFICS.get('develop', ''))
        qna_section = await _build_qna_section(sor, task_id)
        tools_text = json.dumps(AGENT_TOOLS, ensure_ascii=False)

        system = (AGENT_SYSTEM_PROMPT
                  .replace('__ROLE__', role)
                  .replace('__TITLE__', title)
                  .replace('__QNA__', qna_section)
                  .replace('__WORKSPACE__', workspace_dir)
                  .replace('__ROLE_SPECIFIC__', role_specific)
                  .replace('__TOOLS__', tools_text))

        msgs = [{"role": "system", "content": system}]
        msgs.append({"role": "user", "content": f"执行任务：{title}\n参数：{params_str}"})

        from .llm_bridge import llm_call_msgs

        deliverable = None
        ask_question = None

        # ── Tool Loop ──
        for turn in range(15):
            try:
                raw = await llm_call_msgs(msgs, model=model_name, temperature=0.4)
            except Exception as e:
                await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id})
                logger.error(f"role_agent_run llm failed: task={task_id} err={e}")
                return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}

            act = _parse_agent_action(raw)

            if act.get('action') == 'deliver':
                deliverable = act
                break
            elif act.get('action') == 'ask':
                ask_question = act.get('question', '')
                break
            elif act.get('action') == 'tool_call':
                tool = act.get('tool', '')
                params = act.get('params', {})
                result = await _exec_agent_tool(tool, params, workspace_dir)
                msgs.append({"role": "assistant", "content": raw})
                msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"})
                logger.info(f"role_agent tool_call: {tool} -> {result[:100]}")
            else:
                deliverable = {"result": raw}
                break

        if ask_question:
            from .questions import agent_ask
            qid = await agent_ask(project_id, task_id, role, ask_question,
                                  context={"title": title})
            return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": ask_question}

        if not deliverable:
            deliverable = {"result": "Agent未产出交付件"}

        # ── 处理产出 ──
        from appPublic.uniqueID import getID
        did = getID()
        result_text = deliverable.get("result") or ""
        deliverable_type = deliverable.get("deliverable_type") or role
        summary = deliverable.get("summary", "")

        # 写入代码文件
        files_written = []
        code_files = deliverable.get("files") or []
        if isinstance(code_files, list):
            for f in code_files:
                if isinstance(f, dict) and f.get("path") and f.get("content"):
                    abs_path = os.path.join(workspace_dir, f["path"])
                    ok, msg = await _write_code_file(abs_path, f["content"])
                    if ok:
                        files_written.append(abs_path)
                        logger.info(f"code file written: {abs_path}")
                    else:
                        logger.error(f"code file failed: {abs_path} err={msg}")

        # 写交付件文档
        file_path = ''
        if result_text:
            deliverable_dir = os.path.join(workspace_dir, 'deliverables', role)
            os.makedirs(deliverable_dir, exist_ok=True)
            safe_type = (deliverable_type or 'deliverable').replace('/', '_')
            file_path = os.path.join(deliverable_dir, f"{task_id}_{safe_type}.md")
            try:
                with open(file_path, 'w', encoding='utf-8') as f:
                    f.write(result_text or '')
            except Exception as e:
                logger.error(f"deliverable write failed: {file_path} err={e}")
                file_path = ''

        await sor.C("pipeline_deliverables", {
            "id": did, "project_id": project_id, "task_id": task_id,
            "deliverable_type": deliverable_type, "title": title,
            "content": result_text, "file_path": file_path,
            "quality_score": 80, "review_status": "pending",
            "created_by": agent_id or role,
        })

        # Git commit + push
        commit_msg = deliverable.get("git_commit_message") or f"{role}: {title[:80]}"
        git_result = {"rc": 0, "message": "无git操作"}
        if files_written:
            for repo_name in os.listdir(repos_dir):
                repo_path = os.path.join(repos_dir, repo_name)
                if os.path.isdir(os.path.join(repo_path, '.git')):
                    git_result = await _git_commit_push(repo_path, commit_msg)

        await sor.sqlExe(
            "UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$",
            {"st": TASK_REVIEW, "tid": task_id})
        logger.info(f"role_agent_run completed: task={task_id} role={role} "
                     f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)}")
        return {"status": "completed", "task_id": task_id, "deliverable_id": did,
                "files_written": len(files_written), "git_result": git_result}


# ── PM 审核 ──

async def pm_review_run(project_id, agent_id=None, model_name=None):
    db = _get_db()
    async with db.sqlorContext("pipeline") as sor:
        task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False)
        if not task:
            return {"status": "idle", "message": "没有待审核任务"}

        task_id = task.id
        title = getattr(task, "title", "") or ""
        task_role = _normalize_role(getattr(task, "role", "") or "")
        workspace_dir = await _get_workspace_dir(sor, project_id)

        try:
            await sor.sqlExe("COMMIT", {})
        except Exception:
            pass

        repos = await _get_project_repos(sor, project_id)
        if repos:
            await _setup_repos(sor, workspace_dir, project_id)

        deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id)
        if not deliverable_content:
            await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id})
            return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"}

        repo_state = await _get_repo_state(workspace_dir)
        repos_str = ", ".join([r['name'] for r in repos]) if repos else "无"
        content_preview = deliverable_content[:6000]

        pm_system = PM_SYSTEM_PROMPT.replace('__TITLE__', title)\
            .replace('__ROLE__', task_role)\
            .replace('__WORKSPACE__', workspace_dir)\
            .replace('__REPOS__', repos_str)\
            .replace('__REPO_STATE__', repo_state)

        msgs = [{"role": "system", "content": pm_system}]
        msgs.append({"role": "user", "content": f"请审核以下交付件（类型:{deliverable_type}）：\n\n{content_preview}"})

        from .llm_bridge import llm_call_msgs

        decision = None
        for turn in range(5):
            try:
                raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3)
            except Exception as e:
                await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id})
                return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}

            act = _parse_agent_action(raw)

            if act.get('action') == 'review_approve':
                decision = {'status': 'approved', 'comment': act.get('comment', ''), 'next_title': act.get('next_task_title', ''), 'next_desc': act.get('next_task_description', '')}
                break
            elif act.get('action') == 'review_reject':
                decision = {'status': 'rejected', 'comment': act.get('comment', ''), 'questions': act.get('questions', '')}
                break
            elif act.get('action') == 'review_complete':
                decision = {'status': 'completed', 'comment': act.get('comment', '')}
                break
            elif act.get('action') == 'tool_call':
                tool = act.get('tool', '')
                params = act.get('params', {})
                result = await _exec_agent_tool(tool, params, workspace_dir)
                msgs.append({"role": "assistant", "content": raw})
                msgs.append({"role": "user", "content": f"工具 {tool} 结果：\n{result}"})
            else:
                decision = {'status': 'completed', 'comment': raw[:200]}
                break

        if not decision:
            decision = {'status': 'rejected', 'comment': '审核超时'}

        status = decision['status']
        comment = decision.get('comment', '')

        if status == 'approved':
            from appPublic.uniqueID import getID
            pm_did = getID()
            await sor.C("pipeline_deliverables", {
                "id": pm_did, "project_id": project_id, "task_id": task_id,
                "deliverable_type": "pm_review", "title": f"PM审核：{title}",
                "content": json.dumps(decision, ensure_ascii=False),
                "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_review.md"),
                "quality_score": 100, "review_status": "approved", "created_by": agent_id or "pm",
            })
            await sor.sqlExe("UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ WHERE task_id=${tid}$", {"cm": comment, "tid": task_id})
            await sor.sqlExe("UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", {"st": TASK_APPROVED, "tid": task_id})
            next_role = await _get_next_role(task_role)
            if next_role:
                next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment)
                return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment}
            else:
                return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}

        elif status == 'rejected':
            from .questions import agent_ask
            rejection_q = decision.get("questions") or comment or "交付件不满足要求"
            await agent_ask(project_id, task_id, "pm", rejection_q, context={"pm_comment": comment, "deliverable_type": deliverable_type})
            await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${tid}$", {"tid": task_id})
            return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q}

        else:
            await sor.sqlExe("UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", {"tid": task_id})
            return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}


async def role_agent_loop(project_id, role, agent_id=None, model_name=None, max_iterations=10):
    results = []
    for _ in range(max_iterations):
        r = await role_agent_run(project_id, role, agent_id, model_name)
        results.append(r)
        if r["status"] in ("idle", "need_info", "failed"):
            break
        await asyncio.sleep(1)
    return results


async def pm_review_loop(project_id, agent_id=None, model_name=None, max_iterations=20):
    results = []
    for _ in range(max_iterations):
        r = await pm_review_run(project_id, agent_id, model_name)
        results.append(r)
        if r["status"] in ("idle", "failed"):
            break
        await asyncio.sleep(1)
    return results


# 向后兼容
async def run_agent_loop(project_id, role_name, model_name=None):
    role = _normalize_role(role_name or "develop")
    if role == 'pm':
        return await pm_review_run(project_id, model_name=model_name)
    return await role_agent_run(project_id, role_name or "develop", model_name=model_name)


async def agent_loop(project_id, role_name, model_name=None, max_iterations=10):
    role = _normalize_role(role_name or "develop")
    if role == 'pm':
        return await pm_review_loop(project_id, model_name=model_name, max_iterations=max_iterations)
    return await role_agent_loop(project_id, role_name or "develop", model_name=model_name, max_iterations=max_iterations)