demand-agent-new/examples/demand/run.py

"""demand 示例的最小可运行入口。"""
import asyncio
import copy
import importlib
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
from zoneinfo import ZoneInfo

# Clash Verge TUN 模式兼容：禁止 httpx/urllib 自动检测系统 HTTP 代理
os.environ.setdefault("no_proxy", "*")
# 该示例仅使用项目侧能力，禁用框架内置 skills
os.environ.setdefault("AGENT_DISABLE_BUILTIN_SKILLS", "1")
# 禁用内置工具自动注册，并开启严格工具白名单
os.environ.setdefault("AGENT_DISABLE_BUILTIN_TOOL_REGISTRATION", "1")
os.environ.setdefault("AGENT_STRICT_TOOL_SELECTION", "1")
# 禁用所有侧分支（压缩/反思）
os.environ.setdefault("AGENT_DISABLE_SIDE_BRANCHES", "1")

from dotenv import load_dotenv
from examples.demand.db_manager import (
    DEFAULT_DEMAND_PLATFORM,
    normalize_platform,
    query_category_level,
    query_latest_success_execution_id,
    query_video_ids_by_names,
)
from examples.demand.demand_agent_context import TopicBuildAgentContext

# 添加项目根目录到 Python 路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

load_dotenv()

from examples.demand.log_capture import build_log, log

CUSTOM_TOOL_MODULES = {
    # demand 示例：严格按工具名白名单加载对应模块
    "think_and_plan": "examples.demand.agent_tools",
    "get_category_tree": "examples.demand.demand_pattern_tools",
    "get_frequent_itemsets": "examples.demand.demand_pattern_tools",
    "get_itemset_detail": "examples.demand.demand_pattern_tools",
    "get_post_elements": "examples.demand.demand_pattern_tools",
    "search_elements": "examples.demand.demand_pattern_tools",
    "get_element_category_chain": "examples.demand.demand_pattern_tools",
    "get_category_detail": "examples.demand.demand_pattern_tools",
    "search_categories": "examples.demand.demand_pattern_tools",
    "get_category_elements": "examples.demand.demand_pattern_tools",
    "get_category_co_occurrences": "examples.demand.demand_pattern_tools",
    "get_element_co_occurrences": "examples.demand.demand_pattern_tools",
    "get_weight_score_topn": "examples.demand.weight_score_query_tools",
    "get_weight_score_by_name": "examples.demand.weight_score_query_tools",
    "create_demand_item": "examples.demand.demand_build_agent_tools",
    "create_demand_items": "examples.demand.demand_build_agent_tools",
    "write_execution_summary": "examples.demand.demand_build_agent_tools",
}

LOCAL_JSON_MODE = "local_json"
MYSQL_DEMAND_CONTENT_MODE = "mysql_demand_content"
LOCAL_ENTRYPOINT_ENV = "DEMAND_LOCAL_ENTRYPOINT"
LOCAL_ENTRYPOINT_NAME = "run_existing_execution_local"
MYSQL_ENTRYPOINT_ENV = "DEMAND_MYSQL_ENTRYPOINT"
MYSQL_ENTRYPOINT_NAMES = {"run_existing_execution_mysql", "run_hive_gap_mysql"}


def _resolve_run_platform(platform_type: Optional[str], demand_scope: Optional[dict[str, Any]]) -> str:
    scope = demand_scope if isinstance(demand_scope, dict) else {}
    raw_platform = platform_type or scope.get("platform") or scope.get("platform_type") or DEFAULT_DEMAND_PLATFORM
    return normalize_platform(raw_platform)


def _is_local_json_mode() -> bool:
    return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == LOCAL_JSON_MODE


def _is_mysql_demand_content_mode() -> bool:
    return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == MYSQL_DEMAND_CONTENT_MODE


def _raise_db_write_disabled(action: str) -> None:
    raise RuntimeError(
        f"{action} 已在 PG Pattern V2 MVP 中禁用；"
        "请通过 run_existing_execution_local 写本地 JSON，或通过 "
        "run_existing_execution_mysql 只写测试 MySQL demand_content"
    )


def _require_local_entrypoint() -> None:
    if not _is_local_json_mode():
        return
    entrypoint = os.getenv(LOCAL_ENTRYPOINT_ENV, "").strip()
    if entrypoint != LOCAL_ENTRYPOINT_NAME:
        raise RuntimeError(
            "DEMAND_OUTPUT_MODE=local_json 只能通过 "
            "examples.demand.run_existing_execution_local 入口运行"
        )


def _require_mysql_entrypoint() -> None:
    if not _is_mysql_demand_content_mode():
        return
    entrypoint = os.getenv(MYSQL_ENTRYPOINT_ENV, "").strip()
    if entrypoint not in MYSQL_ENTRYPOINT_NAMES:
        raise RuntimeError(
            "DEMAND_OUTPUT_MODE=mysql_demand_content 只能通过 "
            "examples.demand.run_existing_execution_mysql 或 "
            "examples.demand.run_hive_gap_mysql 入口运行"
        )


def _default_local_output_root(execution_id: int) -> Path:
    ts = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d_%H%M%S")
    return Path(__file__).parent / "test_output_data" / f"execution_{execution_id}_{ts}"


def _ensure_local_output_paths(execution_id: int) -> Path:
    root_value = os.getenv("DEMAND_LOCAL_OUTPUT_ROOT")
    root = Path(root_value) if root_value else _default_local_output_root(execution_id)
    os.environ["DEMAND_LOCAL_OUTPUT_ROOT"] = str(root)
    os.environ.setdefault("DEMAND_RESULT_BASE_DIR", str(root / "intermediate" / "result"))
    os.environ.setdefault("DEMAND_WEIGHT_DATA_DIR", str(root / "intermediate" / "data"))
    os.environ.setdefault("DEMAND_TRACE_STORE_PATH", str(root / ".trace"))
    os.environ.setdefault("DEMAND_OUTPUT_BASE_DIR", str(root / "output"))
    return root


def _get_result_base_dir() -> Path:
    redirected = os.getenv("DEMAND_RESULT_BASE_DIR")
    if redirected:
        return Path(redirected)
    if _is_local_json_mode():
        return _ensure_local_output_paths(0) / "result"
    return Path.cwd() / "result"


def _get_output_base_dir() -> Path:
    redirected = os.getenv("DEMAND_OUTPUT_BASE_DIR")
    if redirected:
        return Path(redirected)
    if _is_local_json_mode():
        return _ensure_local_output_paths(0) / "output"
    return Path(__file__).parent / "output"


def _get_trace_store_base_dir() -> Path:
    redirected = os.getenv("DEMAND_TRACE_STORE_PATH")
    if redirected:
        return Path(redirected)
    if _is_local_json_mode():
        return _ensure_local_output_paths(0) / ".trace"
    from examples.demand.config import TRACE_STORE_PATH

    return Path(TRACE_STORE_PATH)


def _get_weight_data_base_dir() -> Path:
    redirected = os.getenv("DEMAND_WEIGHT_DATA_DIR")
    if redirected:
        return Path(redirected)
    return Path(__file__).parent / "data"


def _get_local_output_root() -> Optional[Path]:
    root = os.getenv("DEMAND_LOCAL_OUTPUT_ROOT")
    return Path(root) if root else None


def _write_json_file(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)


def get_execution_id_by_merge_level2(cluster_name: str):
    """Return the latest successful PG Pattern V2 execution_id.

    PG Pattern V2 executions are global snapshots rather than the old MySQL
    merge_leve2-scoped execution rows. Local MVP callers should pass an explicit
    execution_id; this fallback only keeps non-local paths from using MySQL.
    """
    del cluster_name
    return query_latest_success_execution_id()


def resolve_model(prompt: Any, run_config: Any) -> str:
    for env_key in ("DEMAND_LLM_MODEL", "ARK_MODEL", "OPEN_ROUTER_MODEL", "OPENROUTER_MODEL"):
        model_from_env = os.getenv(env_key)
        if model_from_env:
            return model_from_env.strip()

    model_from_prompt = prompt.config.get("model")
    if model_from_prompt:
        return model_from_prompt
    return f"anthropic/{run_config.model}" if "/" not in run_config.model else run_config.model


def extract_assistant_text(message: Any) -> str:
    if message.role != "assistant":
        return ""
    content = message.content
    if isinstance(content, str):
        return content
    if isinstance(content, dict):
        text = content.get("text", "")
        # 即使本轮包含工具调用，也打印模型给出的文本，便于观察每一步输出
        if text:
            return text
    return ""


def register_selected_tools(tool_names: list[str]) -> None:
    for tool_name in tool_names:
        module_path = CUSTOM_TOOL_MODULES.get(tool_name)
        if not module_path:
            raise ValueError(f"未配置工具模块映射: {tool_name}")
        importlib.import_module(module_path)


def _enabled_tools_for_run(configured_tools: list[str]) -> list[str]:
    tools = configured_tools.copy()
    if _is_mysql_demand_content_mode():
        allowed = {
            "think_and_plan",
            "get_category_tree",
            "get_frequent_itemsets",
            "get_itemset_detail",
            "get_post_elements",
            "search_elements",
            "get_element_category_chain",
            "get_category_detail",
            "search_categories",
            "get_category_elements",
            "get_category_co_occurrences",
            "get_element_co_occurrences",
            "get_weight_score_topn",
            "get_weight_score_by_name",
            "create_demand_item",
            "create_demand_items",
        }
        return [tool_name for tool_name in tools if tool_name in allowed]
    if _is_local_json_mode():
        # 本地批量输出只需要 DemandItem JSON。长摘要会显著放大最后一轮上下文，
        # 且不参与下游 CFA 验证契约。
        tools = [tool_name for tool_name in tools if tool_name != "write_execution_summary"]
    return tools


def _join_element_names_to_name(element_names: object) -> str:
    """把 tool 入参 element_names 转成 demand_content.name（逗号分隔）。"""
    if element_names is None:
        return ""
    if isinstance(element_names, list):
        parts = [str(x).strip() for x in element_names if x is not None and str(x).strip()]
        return ",".join(parts)
    # 兼容异常数据：比如历史版本可能传了字符串
    return str(element_names).strip()


def _safe_truncate(s: object, max_len: int) -> str:
    if s is None:
        return ""
    s_str = str(s)
    if max_len and len(s_str) > max_len:
        return s_str[:max_len]
    return s_str


def _load_name_score_map(execution_id: int) -> dict:
    """读取 data/{execution_id} 下所有 JSON 的「名字->score」（同名取最高分）。

    兼容两类数据结构：
    - `*_元素.json`：字段 `name` 表示名字
    - `*_分类.json`：字段 `category` 表示名字
    """
    data_dir = _get_weight_data_base_dir() / str(execution_id)
    if not data_dir.exists():
        return {}

    score_map = {}
    for json_path in data_dir.glob("*.json"):
        try:
            with open(json_path, "r", encoding="utf-8") as f:
                payload = json.load(f)
        except Exception:
            continue

        if not isinstance(payload, list):
            continue

        for item in payload:
            if not isinstance(item, dict):
                continue
            # 元素数据以 name 为主；分类数据以 category 为主。
            name = item.get("name")
            if not isinstance(name, str) or not name:
                name = item.get("category")
            score = item.get("score")
            if isinstance(name, str) and isinstance(score, (int, float)):
                prev = score_map.get(name)
                score_f = float(score)
                if prev is None or score_f > prev:
                    score_map[name] = score_f
    return score_map


def _ensure_pg_weight_score_files(execution_id: int) -> None:
    """Build local PG weight files when local JSON mode needs them."""
    if not (_is_local_json_mode() or _is_mysql_demand_content_mode()):
        return
    from examples.demand.pg_weight_score_builder import build_pg_weight_score_files

    base_dir = _get_weight_data_base_dir()
    expected_dir = base_dir / str(execution_id)
    required_files = [
        expected_dir / f"{dimension}_{level}.json"
        for dimension in ("实质", "形式", "意图")
        for level in ("元素", "分类")
    ]
    if all(path.exists() for path in required_files):
        return
    counts = build_pg_weight_score_files(execution_id=execution_id, base_dir=base_dir)
    log(f"[weight][pg] 生成 PG 权重 JSON，execution_id={execution_id}, files={counts}")


def _avg_score_for_joined_name(name: str, score_map: dict) -> float:
    """按逗号拆分 name，分别取分后求平均。"""
    parts = [part.strip() for part in str(name).split(",") if part and part.strip()]
    if not parts:
        return 0.0
    return sum(float(score_map.get(part, 0.0)) for part in parts) / len(parts)


def _resolve_video_ids_by_name_and_execution_id(name: str, execution_id: int) -> list[str]:
    """按 name(逗号分隔) 与 execution_id 解析去重后的 post_id 列表。"""
    name_parts = [part.strip() for part in str(name).split(",") if part and part.strip()]
    if not name_parts:
        return []
    return query_video_ids_by_names(execution_id=execution_id, names=name_parts)


def _demand_items_output_path(execution_id: int) -> Path:
    return _get_result_base_dir() / str(execution_id) / f"execution_id_{execution_id}_demand_items.json"


def _resolve_demand_items_path(execution_id: int) -> Optional[Path]:
    configured_path = _demand_items_output_path(execution_id)
    if configured_path.exists() or _is_local_json_mode():
        return configured_path if configured_path.exists() else None

    fallbacks = [
        Path.cwd() / "result" / str(execution_id) / f"execution_id_{execution_id}_demand_items.json",
        Path(__file__).parent / "result" / str(execution_id) / f"execution_id_{execution_id}_demand_items.json",
    ]
    for path in fallbacks:
        if path.exists():
            return path
    return None


def _load_demand_items(execution_id: int, log_prefix: str) -> list[dict]:
    demand_items_path = _resolve_demand_items_path(execution_id)
    if not demand_items_path:
        expected_path = _demand_items_output_path(execution_id)
        log(f"[{log_prefix}] 未找到需求 JSON：{expected_path}，跳过写入")
        return []

    try:
        with open(demand_items_path, "r", encoding="utf-8") as f:
            loaded = json.load(f)
    except Exception as e:
        log(f"[{log_prefix}] 读取需求 JSON 失败：{demand_items_path}，error={e}")
        return []

    items = loaded["items"] if isinstance(loaded, dict) and isinstance(loaded.get("items"), list) else loaded
    if not isinstance(items, list):
        log(f"[{log_prefix}] 需求 JSON 非数组，跳过写入：type={type(items)}")
        return []
    return [item for item in items if isinstance(item, dict)]


def _normalize_evidence_refs(value: object) -> dict:
    if isinstance(value, dict):
        return value
    if isinstance(value, str) and value.strip():
        try:
            loaded = json.loads(value)
            if isinstance(loaded, dict):
                return loaded
        except json.JSONDecodeError:
            pass
    return {}


def _build_demand_content_rows(
        execution_id: int,
        merge_level2: str,
        *,
        trace_id: Optional[str] = None,
        task_id: Optional[int] = None,
        serialize_ext_data: bool = True,
        require_evidence: bool = False,
        demand_scope: Optional[dict[str, Any]] = None,
) -> tuple[list[dict], list[dict], list[dict]]:
    from examples.demand.evidence_pack_builder import build_evidence_pack

    items = _load_demand_items(execution_id=execution_id, log_prefix="result")
    if not items:
        return [], [], []

    dt_value = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d")
    score_map = _load_name_score_map(execution_id)
    rows: list[dict] = []
    normalized_items: list[dict] = []
    rejected_items: list[dict] = []
    for index, di in enumerate(items, start=1):
        normalized_item = dict(di)
        evidence_refs = _normalize_evidence_refs(di.get("evidence_refs"))
        normalized_item["evidence_refs"] = evidence_refs
        normalized_items.append(normalized_item)

        name = _join_element_names_to_name(di.get("element_names"))
        if not name:
            rejected_items.append(
                {
                    "item_index": index,
                    "reject_reason": "missing element_names/name",
                    "demand_item": normalized_item,
                }
            )
            continue

        score = _avg_score_for_joined_name(name, score_map)
        reason = di.get("reason")
        desc_value = di.get("desc")
        item_type = di.get("type")
        type_str = str(item_type).strip() if item_type is not None else ""
        if type_str == "分类":
            category_level = query_category_level(execution_id=execution_id, name=name)
            if category_level:
                type_str = type_str + f"L{category_level}"

        video_ids = _resolve_video_ids_by_name_and_execution_id(name=name, execution_id=execution_id)
        ext_data: dict[str, Any] = {
            "reason": reason,
            "desc": desc_value,
            "type": type_str,
            "video_ids": video_ids,
        }
        if trace_id:
            ext_data["trace_id"] = trace_id
        if task_id is not None:
            ext_data["demand_task_id"] = task_id

        content_id = len(rows) + 1 if not serialize_ext_data else None
        if require_evidence or evidence_refs:
            evidence_result = build_evidence_pack(
                execution_id=int(execution_id),
                demand_item=normalized_item,
                trace_id=trace_id or "",
                demand_task_id=task_id,
                demand_content_id=content_id,
                demand_scope=demand_scope or TopicBuildAgentContext.get_metadata("demand_scope", {}),
                merge_leve2=merge_level2,
                platform=TopicBuildAgentContext.get_metadata("platform"),
            )
            if not evidence_result.get("success"):
                if require_evidence:
                    rejected_items.append(
                        {
                            "item_index": index,
                            "reject_reason": evidence_result.get("reject_reason") or "evidence validation failed",
                            "demand_item": normalized_item,
                        }
                    )
                    continue
                log(
                    "[result] evidence 校验未通过，生产兼容路径保留旧输出："
                    f"item_index={index}, reason={evidence_result.get('reject_reason')}"
                )
            else:
                evidence_pack = evidence_result["evidence_pack"]
                ext_data["evidence_pack"] = evidence_pack
                if evidence_pack.get("video_ids"):
                    ext_data["video_ids"] = evidence_pack["video_ids"]

        row = {
            "merge_leve2": _safe_truncate(merge_level2, 32),
            "name": _safe_truncate(name, 64),
            "reason": reason,
            "suggestion": desc_value,
            "score": float(score),
            "ext_data": json.dumps(ext_data, ensure_ascii=False) if serialize_ext_data else ext_data,
            "dt": dt_value,
        }
        if not serialize_ext_data:
            row["id"] = content_id
        rows.append(row)

    if not rows:
        log("[result] 生成行为空，跳过写入")
    if rejected_items:
        log(f"[result] evidence 校验拒绝 {len(rejected_items)} 条 DemandItem")
    return rows, normalized_items, rejected_items


def _create_demand_task(
        execution_id: int,
        name: Optional[str] = None,
        platform: Optional[str] = None,
) -> Optional[int]:
    """Disabled production demand_task writer retained for legacy imports."""
    del execution_id, name, platform
    _raise_db_write_disabled("生产 _create_demand_task")


def _finish_demand_task(task_id: Optional[int], status: int, task_log: str) -> None:
    """Disabled production demand_task updater retained for legacy imports."""
    if not task_id:
        return
    del status, task_log
    _raise_db_write_disabled("生产 _finish_demand_task")


def _start_local_demand_task(
        *,
        execution_id: int,
        task_id: Optional[int],
        name: Optional[str],
        platform: Optional[str],
) -> int:
    from examples.demand.local_output_sink import LocalOutputSink

    local_task_id = int(task_id or os.getenv("DEMAND_LOCAL_TASK_ID", "1"))
    os.environ["DEMAND_LOCAL_TASK_ID"] = str(local_task_id)
    root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
    task_payload = {
        "id": local_task_id,
        "execution_id": execution_id,
        "name": name,
        "platform": platform,
        "status": 0,
        "mode": LOCAL_JSON_MODE,
        "is_simulated_id": True,
        "log": "",
        "started_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
    }
    LocalOutputSink(root).write_all(
        run_manifest={
            "output_mode": LOCAL_JSON_MODE,
            "execution_id": execution_id,
            "merge_leve2": name,
            "platform_type": platform,
            "task_id": local_task_id,
            "status": 0,
            "initialized": True,
            "created_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
        },
        demand_task=task_payload,
    )
    log(f"[task][local_json] 写入本地 demand_task，task_id={local_task_id}, execution_id={execution_id}")
    return local_task_id


def _finish_local_demand_task(task_id: Optional[int], status: int, task_log: str) -> None:
    if not task_id:
        return
    root = _get_local_output_root()
    if not root:
        return
    task_path = root / "demand_task.json"
    payload: dict[str, Any] = {}
    if task_path.exists():
        try:
            with open(task_path, "r", encoding="utf-8") as f:
                loaded = json.load(f)
            if isinstance(loaded, dict):
                payload = loaded
        except Exception:
            payload = {}
    payload.update(
        {
            "id": task_id,
            "status": int(status),
            "mode": LOCAL_JSON_MODE,
            "is_simulated_id": True,
            "log": task_log or "",
            "finished_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
        }
    )
    _write_json_file(task_path, payload)
    log(f"[task][local_json] 更新本地 demand_task，task_id={task_id}, status={status}")


def write_demand_items_to_mysql(
        execution_id: int,
        merge_level2: str,
        *,
        trace_id: Optional[str] = None,
        task_id: Optional[int] = None,
        demand_scope: Optional[dict[str, Any]] = None,
) -> int:
    """Write DB-validated demand_content rows to the test MySQL sink only."""
    if not _is_mysql_demand_content_mode():
        _raise_db_write_disabled("生产 write_demand_items_to_mysql")

    from examples.demand.mysql_demand_content_sink import write_demand_content_rows

    rows, demand_items, rejected_items = _build_demand_content_rows(
        execution_id=execution_id,
        merge_level2=merge_level2,
        trace_id=trace_id,
        task_id=task_id,
        serialize_ext_data=False,
        require_evidence=True,
        demand_scope=demand_scope,
    )
    run_label = os.getenv("DEMAND_RUN_LABEL", "").strip()
    result = write_demand_content_rows(rows, run_label=run_label)
    log(
        "[mysql_demand_content] 写入测试 MySQL demand_content 完成，"
        f"demand_items={len(demand_items)}, inserted={result.inserted_count}, "
        f"rejected={len(rejected_items)}, skipped={result.skipped_count}"
    )
    if rejected_items:
        log(
            "[mysql_demand_content] rejected_items="
            + json.dumps(rejected_items, ensure_ascii=False, default=str)
        )
    if result.inserted_ids:
        log(f"[mysql_demand_content] inserted_ids={result.inserted_ids}")
    return result.inserted_count


def write_demand_items_to_local_json(
        execution_id: int,
        merge_level2: str,
        *,
        trace_id: Optional[str] = None,
        task_id: Optional[int] = None,
        demand_scope: Optional[dict[str, Any]] = None,
) -> int:
    """把普通需求池输出镜像到本地 JSON，不写 demand_content / Hive。"""
    from examples.demand.local_output_sink import LocalOutputSink

    root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
    rows, demand_items, rejected_items = _build_demand_content_rows(
        execution_id=execution_id,
        merge_level2=merge_level2,
        trace_id=trace_id,
        task_id=task_id,
        serialize_ext_data=False,
        require_evidence=True,
        demand_scope=demand_scope,
    )

    sink = LocalOutputSink(root)
    sink.write_demand_items(demand_items)
    sink.write_rejected_demand_items(rejected_items)
    sink.write_demand_content(rows)
    sink.write_dwd_multi_demand_pool_di_from_demand_content(rows)
    sink.write_feature_point_data([])
    log(
        "[local_json] 写入本地 demand_content/dwd_multi_demand_pool_di 完成，"
        f"demand_content_rows={len(rows)}, rejected={len(rejected_items)}, root={root}"
    )
    return len(rows)


def write_global_tree_demand_items_to_hive(execution_id: int, merge_level2: str) -> int:
    """Disabled production Hive writer retained for legacy imports."""
    del execution_id, merge_level2
    _raise_db_write_disabled("生产 write_global_tree_demand_items_to_hive")


def write_global_tree_demand_items_to_local_json(
        execution_id: int,
        merge_level2: str,
        *,
        trace_id: Optional[str] = None,
        task_id: Optional[int] = None,
) -> int:
    """把全局树 feature_point_data 输出镜像到本地 JSON，不写 Hive。"""
    from examples.demand.local_output_sink import LocalOutputSink

    root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
    items = _load_demand_items(execution_id=execution_id, log_prefix="local_json")
    normalized_items = []
    names: list[str] = []
    for di in items:
        normalized = dict(di)
        normalized["evidence_refs"] = _normalize_evidence_refs(di.get("evidence_refs"))
        normalized_items.append(normalized)
        name = _join_element_names_to_name(di.get("element_names"))
        if name:
            names.append(name)

    sink = LocalOutputSink(root)
    sink.write_demand_items(normalized_items)
    sink.write_rejected_demand_items([])
    sink.write_demand_content([])
    sink.write_dwd_multi_demand_pool_di([])
    sink.write_feature_point_data_from_names(names)
    log(
        "[local_json] 写入本地 feature_point_data 完成，"
        f"rows={len(names)}, merge_leve2={merge_level2}, root={root}"
    )
    return len(names)


def _write_local_run_manifest(
        *,
        execution_id: int,
        merge_level2: str,
        platform_type: Optional[str],
        count: int,
        task_id: Optional[int],
        trace_id: Optional[str],
        final_text: str,
        task_status: int,
        total_tokens: int,
        total_cost: float,
        result_write_count: int,
        log_file_path: Path,
        result_write_error: Optional[str] = None,
) -> None:
    if not _is_local_json_mode():
        return
    root = _get_local_output_root() or _ensure_local_output_paths(execution_id)
    manifest = {
        "output_mode": LOCAL_JSON_MODE,
        "execution_id": execution_id,
        "merge_leve2": merge_level2,
        "platform_type": platform_type,
        "count": count,
        "task_id": task_id,
        "trace_id": trace_id,
        "status": task_status,
        "total_tokens": total_tokens,
        "total_cost": total_cost,
        "result_write_count": result_write_count,
        "result_write_error": result_write_error,
        "final_text_length": len(final_text or ""),
        "paths": {
            "root": str(root),
            "result": str(_get_result_base_dir()),
            "trace": str(_get_trace_store_base_dir()),
            "output": str(_get_output_base_dir()),
            "demand_items_raw": str(_demand_items_output_path(execution_id)),
            "log_file": str(log_file_path),
        },
        "created_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(),
    }
    _write_json_file(root / "run_manifest.json", manifest)


async def run_once(
        execution_id,
        merge_level2,
        count: int = 30,
        task_id: Optional[int] = None,
        platform_type: Optional[str] = None,
        demand_scope: Optional[dict[str, Any]] = None,
) -> dict:
    from agent.core.runner import AgentRunner
    from agent.llm import create_openrouter_llm_call
    from agent.llm.prompts import SimplePrompt
    from agent.trace import FileSystemTraceStore, Message, Trace
    from agent.utils import setup_logging
    from examples.demand.config import DEBUG, ENABLED_TOOLS, LOG_FILE, LOG_LEVEL, RUN_CONFIG

    task_log_text = ""
    task_status = 0
    result_write_count = 0
    result_write_error: Optional[str] = None

    if _is_local_json_mode():
        _ensure_local_output_paths(execution_id)

    if _is_local_json_mode() or _is_mysql_demand_content_mode():
        _ensure_pg_weight_score_files(int(execution_id))

    platform_type = _resolve_run_platform(platform_type, demand_scope)
    TopicBuildAgentContext.set_execution_id(execution_id)
    TopicBuildAgentContext.set_metadata("result_base_dir", str(_get_result_base_dir()))
    TopicBuildAgentContext.set_metadata("merge_leve2", merge_level2)
    TopicBuildAgentContext.set_metadata("platform", platform_type)
    resolved_demand_scope = dict(demand_scope or {})
    resolved_demand_scope.setdefault("scope_source", "manual_cli")
    resolved_demand_scope.setdefault("merge_leve2", merge_level2)
    resolved_demand_scope["platform"] = _resolve_run_platform(platform_type, resolved_demand_scope)
    resolved_demand_scope.setdefault("pattern_execution_id", int(execution_id))
    TopicBuildAgentContext.set_metadata("demand_scope", resolved_demand_scope)

    base_dir = Path(__file__).parent
    output_dir = _get_output_base_dir()
    output_dir.mkdir(parents=True, exist_ok=True)

    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
    enabled_tools = _enabled_tools_for_run(ENABLED_TOOLS)
    register_selected_tools(enabled_tools)

    prompt_file = "demand_mysql.md" if _is_mysql_demand_content_mode() else "demand.md"
    prompt = SimplePrompt(base_dir / prompt_file)

    run_config = copy.deepcopy(RUN_CONFIG)
    model = resolve_model(prompt, run_config)
    run_config.model = model
    run_config.temperature = float(prompt.config.get("temperature", run_config.temperature))
    run_config.max_iterations = int(prompt.config.get("max_iterations", run_config.max_iterations))
    run_config.tools = enabled_tools
    # 禁用反思/总结经验相关流程（避免进入 reflection 侧分支）
    run_config.enable_research_flow = False
    run_config.goal_compression = "none"
    run_config.force_side_branch = None
    run_config.knowledge.enable_extraction = False
    run_config.knowledge.enable_completion_extraction = False
    run_config.knowledge.enable_injection = False
    run_config.trace_id = None

    initial_messages = prompt.build_messages(merge_level2=merge_level2, count=count)

    trace_store_path = _get_trace_store_base_dir()
    store = FileSystemTraceStore(base_path=str(trace_store_path))
    runner = AgentRunner(
        trace_store=store,
        llm_call=create_openrouter_llm_call(model=model),
        skills_dir=None,
        debug=DEBUG,
    )

    final_text = ""
    total_tokens = 0
    total_cost = 0.0
    trace_id: Optional[str] = None
    has_completed_trace_cost = False
    log_file_path = output_dir / f"{execution_id}" / f"run_log_{execution_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
    log_file_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        with build_log(execution_id) as log_buffer:
            async for item in runner.run(messages=initial_messages, config=run_config):
                if isinstance(item, Trace):
                    trace_id = getattr(item, "trace_id", None) or trace_id
                    if getattr(item, "status", None) == "completed":
                        total_tokens = int(getattr(item, "total_tokens", 0) or 0)
                        total_cost = float(getattr(item, "total_cost", 0.0) or 0.0)
                        has_completed_trace_cost = True
                    continue
                elif isinstance(item, Message):
                    text = extract_assistant_text(item)
                    if text:
                        final_text = text
                        log(f"[assistant] {text}")
                    if not has_completed_trace_cost:
                        total_tokens += int(getattr(item, "total_tokens", 0) or 0)
                        total_cost += float(getattr(item, "cost", 0.0) or 0.0)

            if final_text:
                output_file = output_dir / f"{execution_id}" / "result.txt"
                output_file.parent.mkdir(parents=True, exist_ok=True)
                with open(output_file, "w", encoding="utf-8") as f:
                    f.write(final_text)

            log(f"[cost] total_tokens={total_tokens}, total_cost=${total_cost:.6f}")

            # agent 执行完成后：全局树写 Hive，其他写 MySQL
            try:
                if str(merge_level2).strip() == "全局树":
                    if _is_local_json_mode():
                        result_write_count = write_global_tree_demand_items_to_local_json(
                            execution_id=execution_id,
                            merge_level2=merge_level2,
                            trace_id=trace_id,
                            task_id=task_id,
                        )
                    else:
                        result_write_count = write_global_tree_demand_items_to_hive(
                            execution_id=execution_id,
                            merge_level2=merge_level2,
                        )
                else:
                    # element_names -> name（逗号分隔）；reason -> demand_content.reason；desc -> demand_content.suggestion；dt -> demand_content.dt
                    if _is_local_json_mode():
                        result_write_count = write_demand_items_to_local_json(
                            execution_id=execution_id,
                            merge_level2=merge_level2,
                            trace_id=trace_id,
                            task_id=task_id,
                            demand_scope=resolved_demand_scope,
                        )
                    else:
                        result_write_count = write_demand_items_to_mysql(
                            execution_id=execution_id,
                            merge_level2=merge_level2,
                            trace_id=trace_id,
                            task_id=task_id,
                            demand_scope=resolved_demand_scope,
                        )
            except Exception as e:
                result_write_error = str(e)
                log(f"[result-write] 写入结果异常：{e}")
                if _is_local_json_mode():
                    task_status = 2

            task_log_text = log_buffer.getvalue()
            if task_status != 2:
                task_status = 1
    except Exception as e:
        if not task_log_text:
            # 如果异常发生在 build_log 内部，尽量回收已产生的日志
            try:
                existing = locals().get("log_buffer")
                if existing is not None:
                    task_log_text = existing.getvalue()  # type: ignore[attr-defined]
            except Exception:
                pass
        if not task_log_text:
            task_log_text = f"[run] 执行异常: {e}"
        task_status = 2
        raise
    finally:
        if task_log_text:
            try:
                with open(log_file_path, "w", encoding="utf-8") as f:
                    f.write(task_log_text)
            except Exception:
                # 兜底：即使写文件失败，也要确保 MySQL 状态被更新
                pass
        if _is_local_json_mode():
            _finish_local_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
        elif not _is_mysql_demand_content_mode():
            _finish_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
        _write_local_run_manifest(
            execution_id=execution_id,
            merge_level2=merge_level2,
            platform_type=platform_type,
            count=int(count),
            task_id=task_id,
            trace_id=trace_id,
            final_text=final_text,
            task_status=task_status,
            total_tokens=total_tokens,
            total_cost=total_cost,
            result_write_count=result_write_count,
            log_file_path=log_file_path,
            result_write_error=result_write_error,
        )

    return {
        "final_text": final_text,
        "trace_id": trace_id,
        "output_dir": str(output_dir / f"{execution_id}"),
        "trace_store_path": str(trace_store_path),
        "result_base_dir": str(_get_result_base_dir()),
        "result_write_count": result_write_count,
    }


async def main(
        cluster_name: str,
        platform_type: str,
        count,
        execution_id: Optional[int] = None,
        task_id: Optional[int] = None,
        demand_scope: Optional[dict[str, Any]] = None,
) -> dict:
    if not (_is_local_json_mode() or _is_mysql_demand_content_mode()):
        _raise_db_write_disabled("非受控运行入口")

    if _is_local_json_mode():
        _require_local_entrypoint()
        if execution_id is None:
            raise ValueError("local_json 模式必须传入已有 execution_id，不能触发 prepare 写库流程")
        _ensure_local_output_paths(execution_id)

    if _is_mysql_demand_content_mode():
        _require_mysql_entrypoint()
        if execution_id is None:
            raise ValueError("mysql_demand_content 模式必须传入已有 execution_id，不能触发 prepare 写库流程")
        if str(cluster_name).strip() == "全局树":
            raise ValueError("mysql_demand_content 模式只写普通 demand_content，不支持全局树 Hive 输出")

    platform_type = _resolve_run_platform(platform_type, demand_scope)
    if execution_id is None:
        execution_id = get_execution_id_by_merge_level2(cluster_name)
    if not execution_id:
        return {"execution_id": None, "trace_id": None, "final_text": ""}

    if _is_local_json_mode():
        task_id = _start_local_demand_task(
            execution_id=int(execution_id),
            task_id=task_id,
            name=cluster_name,
            platform=platform_type,
        )

    run_result = await run_once(
        execution_id,
        cluster_name,
        count=count,
        task_id=task_id,
        platform_type=platform_type,
        demand_scope=demand_scope,
    )
    result = {
        "execution_id": execution_id,
        "trace_id": run_result.get("trace_id"),
        "final_text": run_result.get("final_text", ""),
        "output_dir": run_result.get("output_dir"),
        "trace_store_path": run_result.get("trace_store_path"),
        "result_base_dir": run_result.get("result_base_dir"),
        "result_write_count": run_result.get("result_write_count", 0),
    }
    if _is_local_json_mode():
        root = _get_local_output_root()
        result["local_output_root"] = str(root) if root else None
    return result


if __name__ == "__main__":
    asyncio.run(main('全局树', 'piaoquan', 50))
    # write_demand_items_to_mysql(execution_id=8, merge_level2='贪污腐败')