"""pipeline_service - 通用产线执行引擎模块。

把 Hermes Agent 验证过的业务流程固化为可重复、可并发的产线业务环境。
支持多租户隔离、DAG 步骤调度、可插拔步骤处理器、artifact 版本管理。
支持人工交互步骤（human_task/approval_gate）：人机协作产线。

任何宿主应用都可以通过 load_pipeline_service() 加载本模块。
"""

import json
import asyncio
import os
from ahserver.serverenv import ServerEnv
from appPublic.uniqueID import getID
from appPublic.log import debug

from .state import (
    TASK_SUBMITTED, TASK_RUNNING, TASK_COMPLETED, TASK_FAILED, TASK_PAUSED, TASK_CANCELLED, TASK_WAITING,
    build_step_graph, get_cascade_rerun_steps, get_rerun_from_next,
)
from .storage import (
    create_task, init_task_steps, get_task, get_task_steps,
    get_artifact, get_all_artifacts, list_tasks,
    update_task_state, update_task_version,
    get_pipeline_steps, reset_steps, save_artifact,
    get_human_task,
)
from .executor import start_task, resume_task, stop_task, is_running
from .handler import register_handler, list_handlers, register_default_handler
from .step_registry import (
    register_step_type, get_step_type, list_step_types,
    unregister_step_type, load_builtin_types,
)
from .human import human_complete, approval_approve, approval_reject, human_list
from .questions import (
    agent_ask, answer_question, forward_question,
    get_question, list_questions, get_task_qna,
)
from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_loop

MODULE_NAME = "pipeline_service"
MODULE_VERSION = "3.2.0"


async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None):
    """提交新产线任务。

    Args:
        tenant_id: 租户ID（由宿主应用提供，可以是 org_id、user_id 等）
        pipeline_id: 产线定义ID（来自 pipelines 表）
        owner_id: 提交人ID
        title: 任务标题
        params: 提交参数（dict）

    Returns:
        JSON string with status, task_id
    """
    result = {"success": False}
    try:
        if not tenant_id or not pipeline_id:
            result["message"] = "缺少 tenant_id 或 pipeline_id"
            return json.dumps(result, ensure_ascii=False)

        params = params or {}
        task_id = await create_task(tenant_id, pipeline_id, owner_id, title, params)

        # Read step definitions from pipeline_steps table
        step_records = await get_pipeline_steps(pipeline_id)
        if not step_records:
            result["message"] = f"产线 {pipeline_id} 没有步骤定义"
            return json.dumps(result, ensure_ascii=False)

        # Create step execution records
        await init_task_steps(task_id, step_records)

        # Start execution
        await start_task(task_id)

        result["success"] = True
        result["task_id"] = task_id
        result["message"] = "任务已提交并开始执行"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False)


async def pipeline_role_submit(tenant_id, pipeline_id, owner_id, title, params=None, role=""):
    """提交角色任务（主agent → 角色agent 模式，v3.2.0）。

    与 pipeline_submit 的区别：
    - 带 role 字段，由角色agent循环（role_agent_loop）认领执行
    - 不创建步骤记录、不启动 DAG executor

    Args:
        role: 目标角色（requirement/design/dev/test/ops 等）。必填。

    Returns:
        JSON string with success, task_id
    """
    result = {"success": False}
    try:
        if not tenant_id:
            result["message"] = "缺少 tenant_id"
            return json.dumps(result, ensure_ascii=False)
        if not role:
            result["message"] = "缺少 role（角色任务必须指定目标角色）"
            return json.dumps(result, ensure_ascii=False)

        params = params or {}
        task_id = await create_task(tenant_id, pipeline_id or "", owner_id, title, params, role=role)

        result["success"] = True
        result["task_id"] = task_id
        result["role"] = role
        result["message"] = f"角色任务已提交（role={role}），等待角色agent认领执行"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False)


async def pipeline_list(tenant_id, pipeline_id=None, limit=100):
    """查询租户的任务列表。"""
    result = {"success": False}
    try:
        tasks = await list_tasks(tenant_id, pipeline_id, limit)
        result["success"] = True
        result["tasks"] = tasks
        result["total"] = len(tasks)
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def pipeline_detail(tenant_id, task_id):
    """获取任务详情 + 步骤状态树。"""
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        steps = await get_task_steps(task_id)

        # Enrich steps with human task info for interactive steps
        for step in steps:
            if step.get('state') == 'waiting':
                ht = await get_human_task(task_id, step['step_name'])
                if ht:
                    step['human_task'] = ht

        task["steps"] = steps
        task["is_running"] = is_running(task_id)

        result["success"] = True
        result["task"] = task
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def pipeline_node(tenant_id, task_id, step_name, version=None):
    """获取某节点某版本的 input/output artifact。"""
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        v = version or task.get("current_version", 1)
        if isinstance(v, str):
            v = int(v)

        input_data = await get_artifact(task_id, v, step_name, "input")
        output_data = await get_artifact(task_id, v, step_name, "output")

        result["success"] = True
        result["step_name"] = step_name
        result["version"] = v
        result["input"] = input_data
        result["output"] = output_data
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def pipeline_modify(tenant_id, task_id, updates, rerun_from="node"):
    """修改节点 artifact 并触发级联重跑。

    Args:
        updates: {step_name: {content: {...}}, ...}
        rerun_from: "node" = 从该节点重跑, "next" = 从下游节点重跑
    """
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        if is_running(task_id):
            result["message"] = "任务正在执行中，请先暂停"
            return json.dumps(result, ensure_ascii=False)

        pipeline_id = task.get("pipeline_id", task.get("Pipeline_id", ""))
        current_version = task.get("current_version", task.get("current_Version", 1))
        if isinstance(current_version, str):
            current_version = int(current_version)

        # Load step graph
        step_records = await get_pipeline_steps(pipeline_id)
        step_graph = build_step_graph(step_records)

        # Calculate affected steps
        all_rerun = set()
        for step_name in updates:
            if step_name not in step_graph:
                result["message"] = f"未知步骤: {step_name}"
                return json.dumps(result, ensure_ascii=False)
            if rerun_from == "node":
                affected = get_cascade_rerun_steps(step_graph, step_name)
            else:
                affected = get_rerun_from_next(step_graph, step_name)
            all_rerun.update(affected)

        # Create new version
        new_version = current_version + 1
        await update_task_version(task_id, new_version)

        # Save modified artifacts
        for step_name, step_update in updates.items():
            content = step_update.get("content", step_update)
            io_type = "input" if rerun_from == "node" else "output"
            await save_artifact(task_id, new_version, step_name, io_type, content)

        # Reset affected steps
        all_rerun_list = sorted(all_rerun, key=lambda s: step_graph.get(s, {}).get("order", 999))
        await reset_steps(task_id, all_rerun_list)

        # Update task state and resume
        await update_task_state(task_id, TASK_RUNNING)
        await resume_task(task_id)

        result["success"] = True
        result["new_version"] = new_version
        result["rerun_steps"] = all_rerun_list
        result["message"] = f"创建 v{new_version}，重跑 {len(all_rerun_list)} 个步骤"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def pipeline_pause(tenant_id, task_id):
    """暂停任务。"""
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        await stop_task(task_id)
        await update_task_state(task_id, TASK_PAUSED)

        result["success"] = True
        result["message"] = "任务已暂停"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False)


async def pipeline_resume(tenant_id, task_id):
    """恢复任务。"""
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        await update_task_state(task_id, TASK_RUNNING)
        await resume_task(task_id)

        result["success"] = True
        result["message"] = "任务已恢复"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False)


async def pipeline_cancel(tenant_id, task_id):
    """取消任务。"""
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        await stop_task(task_id)
        await update_task_state(task_id, TASK_CANCELLED)

        result["success"] = True
        result["message"] = "任务已取消"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def pipeline_restart(tenant_id, task_id):
    """重启已完成/失败/取消的任务 — 重置所有步骤为pending，新版本继续。"""
    result = {"success": False}
    try:
        task = await get_task(tenant_id, task_id)
        if not task:
            result["message"] = "任务不存在"
            return json.dumps(result, ensure_ascii=False)

        if is_running(task_id):
            result["message"] = "任务正在执行中，请先暂停"
            return json.dumps(result, ensure_ascii=False)

        # Stop any lingering executor
        await stop_task(task_id)

        # Get step names to reset
        steps = await get_task_steps(task_id)
        step_names = [s['step_name'] for s in steps]

        # New version
        current_version = task.get("current_version", task.get("current_Version", 1))
        if isinstance(current_version, str):
            current_version = int(current_version)
        new_version = current_version + 1
        await update_task_version(task_id, new_version)

        # Reset all steps to pending
        await reset_steps(task_id, step_names)

        # Restart execution
        await update_task_state(task_id, TASK_RUNNING)
        await start_task(task_id)

        result["success"] = True
        result["new_version"] = new_version
        result["message"] = f"任务已重新启动 v{new_version}"
    except Exception as e:
        result["message"] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


def pipeline_handlers():
    """查看已注册的步骤处理器。"""
    return json.dumps(list_handlers(), ensure_ascii=False)


def pipeline_step_types():
    """查看所有注册的步骤类型（含元数据）。"""
    return json.dumps(list_step_types(), ensure_ascii=False)


def pipeline_register_step_type(step_type, metadata):
    """注册步骤类型（可装卸）。"""
    register_step_type(step_type, metadata)
    return json.dumps({"success": True, "step_type": step_type}, ensure_ascii=False)


def pipeline_unregister_step_type(step_type):
    """卸载步骤类型。"""
    unregister_step_type(step_type)
    return json.dumps({"success": True, "step_type": step_type}, ensure_ascii=False)


# ── Git / Shell 操作（v3.2.1: 主agent devops 能力）──

_SHELL_BASE_DIR = '/d/pipeline/workspaces'

async def shell_exec(command: str, workdir: str = None, timeout: int = 60):
    """安全外壳执行。限制在 workspace 目录下，超时自动终止。
    
    Returns: {"rc": exit_code, "stdout": "...", "stderr": "..."}
    """
    cwd = workdir or _SHELL_BASE_DIR
    cwd = os.path.abspath(cwd)
    if not cwd.startswith(os.path.abspath(_SHELL_BASE_DIR)):
        return {"rc": -1, "stdout": "", "stderr": f"安全限制：工作目录必须在 {_SHELL_BASE_DIR} 下"}
    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 skill_import_git(repo_url: str, skills_dir: str = 'skills'):
    """克隆 git 仓库，扫描指定目录下所有子目录（每个=一个企业技能）。
    
    Returns: {"success": bool, "repo":..., "target":..., "skills": [{"name","path","description"}]}
    """
    from appPublic.uniqueID import getID
    repo_name = repo_url.rstrip('/').split('/')[-1].replace('.git', '') or f"repo_{getID()[:8]}"
    target = os.path.join(_SHELL_BASE_DIR, repo_name)
    
    if not os.path.isdir(target):
        r = await shell_exec(f'git clone {repo_url} {target}', workdir=_SHELL_BASE_DIR)
        if r['rc'] != 0:
            return {"success": False, "error": f"clone 失败: {r['stderr'][:500]}"}
    
    full_skills = os.path.join(target, skills_dir)
    if not os.path.isdir(full_skills):
        entries = ', '.join(os.listdir(target)[:10]) if os.path.isdir(target) else '(dir not found)'
        return {"success": False, "error": f"skills目录不存在: {full_skills}，仓库内容: {entries}"}
    
    found = []
    for entry in sorted(os.listdir(full_skills)):
        entry_path = os.path.join(full_skills, entry)
        if os.path.isdir(entry_path):
            desc = ''
            skill_file = os.path.join(entry_path, 'SKILL.md')
            if os.path.isfile(skill_file):
                try:
                    with open(skill_file) as f:
                        content = f.read(500)
                    for line in content.split('\n'):
                        if line.startswith('description:'):
                            desc = line.split(':', 1)[1].strip()
                            break
                except:
                    pass
            found.append({"name": entry, "path": entry_path, "description": desc})
    
    return {"success": True, "repo": repo_url, "target": target,
            "skills_dir": skills_dir, "skills": found}


def load_pipeline_service():
    """注册所有函数到 ServerEnv。任何宿主应用调用此函数即可使用产线引擎。"""
    env = ServerEnv()

    # Task lifecycle
    env.pipeline_submit = pipeline_submit
    env.pipeline_role_submit = pipeline_role_submit
    env.pipeline_list = pipeline_list
    env.pipeline_detail = pipeline_detail
    env.pipeline_node = pipeline_node
    env.pipeline_modify = pipeline_modify
    env.pipeline_pause = pipeline_pause
    env.pipeline_resume = pipeline_resume
    env.pipeline_cancel = pipeline_cancel
    env.pipeline_restart = pipeline_restart

    # Handler management
    env.pipeline_register_handler = register_handler
    env.pipeline_handlers = pipeline_handlers

    # Step type registry (pluggable)
    env.pipeline_step_types = pipeline_step_types
    env.pipeline_register_step_type = pipeline_register_step_type
    env.pipeline_unregister_step_type = pipeline_unregister_step_type

    # Human task operations
    env.human_task_complete = human_complete
    env.approval_approve = approval_approve
    env.approval_reject = approval_reject
    env.human_task_list = human_list

    # Register default handler
    register_default_handler()

    # Load built-in interactive step types
    load_builtin_types()

    # Register intent classifier + LLM bridge (shared across all pipelines)
    from .intent_classifier import intent_classify
    from .llm_bridge import llm_call
    env.intent_classify = intent_classify
    env.pipeline_llm_call = llm_call

    # Role-agent loop (v3.2.0: 主agent + 角色agent 认领执行)
    env.role_agent_run = role_agent_run
    env.role_agent_loop = role_agent_loop
    env.agent_loop = agent_loop          # 向后兼容
    env.run_agent_loop = run_agent_loop  # 向后兼容

    # Question loop (角色agent ↔ 主agent ↔ 客户)
    env.agent_ask = agent_ask
    env.question_answer = answer_question
    env.question_forward = forward_question
    env.question_detail = get_question
    env.question_list = list_questions
    env.question_qna = get_task_qna

    # DevOps: git/shell operations (v3.2.1)
    env.shell_exec = shell_exec
    env.skill_import_git = skill_import_git

    # Background poller: auto-dispatch submitted role_tasks to role agents

    async def _role_poller(app):
        """启动后台轮询器，spawn 后立即返回（不阻塞 aiohttp 启动）。"""
        from sqlor.dbpools import DBPools
        poll_db = DBPools()

        async def _poll_loop():
            while True:
                try:
                    async with poll_db.sqlorContext("pipeline") as sor:
                        recs = await sor.sqlExe(
                            "SELECT id, tenant_id, role FROM pipeline_tasks "
                            "WHERE state='submitted' AND pipeline_id='role_task' "
                            "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 5",
                            {})
                        for rec in (recs or []):
                            tid = getattr(rec, 'id', '')
                            pid = getattr(rec, 'tenant_id', '')
                            r = getattr(rec, 'role', '')
                            if tid and pid and r:
                                debug(f"agent_poller dispatching task={tid} role={r}")
                                asyncio.ensure_future(
                                    role_agent_loop(pid, r, agent_id=f"poller-{r}"))
                except Exception as e:
                    debug(f"agent_poller error: {e}")
                await asyncio.sleep(10)

        asyncio.create_task(_poll_loop())
        debug("agent poller started")

    from ahserver.configuredServer import add_startup
    add_startup(_role_poller)

    debug(f"[{MODULE_NAME}] v{MODULE_VERSION} loaded — pipeline engine with role-agent + question loop")
    return True