deerflow-code/skills-data/xdfx-split-controller/scripts/xdfx_flow.py

#!/usr/bin/env python3
"""Robust xdfx single-agent flow runner.

The skill text should stay thin; this script owns the slow, failure-prone
workflow details: HTTP calls, streamed answer parsing, JSON extraction, id
handoff, partial failure handling, and state persistence.
"""

from __future__ import annotations

import argparse
import ast
import copy
import json
import os
import re
import ssl
import sys
import time
import traceback
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


ENDPOINTS = {
    "aiuser": "/gpt/chat/chat_aiuser",
    "group": "/gpt/chat/chat_qianqian_group_chat",
    "spread": "/gpt/chat/chat_qianqian_spread_chat",
    "cuiruo": "/gpt/chat/chat_qianqian_cuiruo_chat",
    "strategy_model": "/gpt/chat/chat_qianqian_newboci_chat",
    "save_design": "/api/task/task-action-design",
    "save_design_account": "/api/task/task-action-design-account",
    "save_pathway": "/api/task/task-action-pathway-account",
    "save_relation": "/api/task/task-action-design-pathway-relation/saveRelation",
    "save_design_vuln": "/api/task/task-action-design-vulns",
    "save_pathway_vuln": "/api/task/task-action-pathway-account-vulns",
    "save_strategy": "/api/task/task-action-strategy",
    "save_channel": "/api/task/task-action-channel",
    "save_narrate": "/api/base/base-narrate",
    "save_message": "/api/task/task-action-message",
    "save_strategy_relation": "/api/task/task-action-design-strategy-relation",
}

ENDPOINT_LABELS = {
    "login": "登录获取访问 token",
    "aiuser": "调用候选账号/受众分析模型接口",
    "group": "调用受众目标结构化抽取接口",
    "spread": "调用关键传播节点结构化抽取接口",
    "cuiruo": "调用脆弱点分析模型接口",
    "strategy_model": "调用策略生成模型接口",
    "save_design": "保存受众目标",
    "save_design_account": "保存受众目标账号",
    "save_pathway": "保存关键传播节点账号",
    "save_relation": "保存受众与关键传播节点关系",
    "save_design_vuln": "保存受众脆弱点",
    "save_pathway_vuln": "保存关键传播节点脆弱点",
    "save_strategy": "保存策略主表",
    "save_channel": "保存策略平台与人设",
    "save_narrate": "保存策略信息主题",
    "save_message": "保存策略消息配置",
    "save_strategy_relation": "保存策略与受众/传播节点关系",
}

PLATFORM_MAP = {
    "facebook": 1,
    "fb": 1,
    "ptt": 13,
    "instagram": 2,
    "ig": 2,
    "x": 3,
    "twitter": 3,
    "youtube": 4,
    "yt": 4,
}

MODEL_RETRY_MARKERS = (
    "y模型没有响应",
    "模型没有响应",
    "模型无响应",
    "接口超时",
    "请求超时",
    "响应超时",
    "读取超时",
    "连接超时",
    "超时",
    "timeout",
    "timed out",
    "read timed out",
    "gateway timeout",
    "504 gateway",
)

DEFAULT_STRATEGY_PROMPT_AUDIENCE = (
    "一、叙事主题设计，根据议程设置理论，需优先选择与目标受众利益高度重合的议题。"
    "叙事框架构建采用系统叙事与认同叙事。二、行动人设设计需强调专业背书、地域关联和跨领域合作。"
    "三、传播策略需与目标受众所在平台、语言风格、文化禁忌和信息接受方式一致。"
    "请至少生成1个策略，字段使用：总体理由、发布人设1、账号类型1、社交平台1、信息标题1、"
    "信息形式1、数量1、信息发布时间段1，可继续输出2、3。只返回JSON。"
)

DEFAULT_STRATEGY_PROMPT_PATHWAY = (
    "一、叙事主题设计，根据议程设置理论，需优先选择与关键传播节点利益高度重合的议题。"
    "叙事框架构建采用系统叙事与认同叙事。二、行动人设设计需强调专业背书、地域关联和跨领域合作。"
    "三、传播节点突破策略需贴合该账号平台、粉丝结构、语言风格、文化禁忌和既有发文风格。"
    "请至少生成1个策略，字段使用：总体理由、发布人设1、账号类型1、社交平台1、信息标题1、"
    "信息形式1、数量1、信息发布时间段1，可继续输出2、3。只返回JSON。"
)


class FlowError(RuntimeError):
    pass


@dataclass
class FlowContext:
    api_base: str
    headers: dict[str, str]
    timeout: float
    retries: int
    dry_run: bool
    state_path: Path
    summary_path: Path | None
    max_audiences: int = 3
    max_pathways: int = 3
    max_vulns: int = 3
    max_strategies: int = 3
    sync_step_status: bool = True
    model_retries: int = 1
    verify_ssl: bool = False
    auto_login: bool = True
    login_username: str = "admin"
    login_password: str = "ch@user"
    login_path: str = "/api/v1/auth/login/username"
    login_attempt_count: int = 0


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def compact(value: Any, limit: int = 1400) -> str:
    try:
        if isinstance(value, str):
            text = value
        else:
            text = json.dumps(value, ensure_ascii=False)
    except Exception:
        text = repr(value)
    if len(text) <= limit:
        return text
    return text[:limit] + f"... <truncated {len(text) - limit} chars>"


def stable_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def as_int(value: Any) -> int | None:
    if value is None:
        return None
    if isinstance(value, bool):
        return None
    if isinstance(value, int):
        return value
    if isinstance(value, float) and value.is_integer():
        return int(value)
    text = str(value).strip()
    if re.fullmatch(r"-?\d+", text):
        return int(text)
    return None


def as_list(value: Any) -> list[Any]:
    if value is None:
        return []
    if isinstance(value, list):
        return value
    if isinstance(value, tuple):
        return list(value)
    return [value]


def deep_get(data: Any, *keys: str) -> Any:
    cur = data
    for key in keys:
        if not isinstance(cur, dict):
            return None
        cur = cur.get(key)
    return cur


def first_present(*values: Any) -> Any:
    for value in values:
        if value is None:
            continue
        if isinstance(value, str) and not value.strip():
            continue
        return value
    return None


def load_json_file(path: str | Path) -> dict[str, Any]:
    p = Path(path)
    if not p.exists():
        return {}
    return json.loads(p.read_text(encoding="utf-8"))


def save_json_file(path: str | Path, value: Any) -> None:
    p = Path(path)
    text = json.dumps(value, ensure_ascii=False, indent=2)
    try:
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(text, encoding="utf-8")
    except PermissionError:
        fallback = Path.cwd() / p.name
        fallback.write_text(text, encoding="utf-8")


def default_state() -> dict[str, Any]:
    return {
        "api_base": "",
        "taskId": "",
        "actionId": "",
        "mubiao": "",
        "direction": "U方向",
        "platform": "",
        "audienceRawText": "",
        "audienceCandidates": [],
        "topAudiences": [],
        "pathwayCandidates": [],
        "topPathways": [],
        "audienceVulns": {},
        "pathwayVulns": {},
        "audienceStrategies": [],
        "pathwayStrategies": [],
        "errors": [],
        "events": [],
        "startedAt": utc_now(),
        "updatedAt": utc_now(),
    }


def merge_config_into_state(state: dict[str, Any], config: dict[str, Any]) -> None:
    aliases = {
        "task_id": "taskId",
        "action_id": "actionId",
        "apiBase": "api_base",
        "target": "mubiao",
        "goal": "mubiao",
    }
    for key, value in config.items():
        dest = aliases.get(key, key)
        if value is None:
            continue
        if dest in {"taskId", "actionId"}:
            state[dest] = str(value)
        elif dest in state or dest in {
            "api_base",
            "headers",
            "auth_token",
            "authorization",
        }:
            state[dest] = value


def record_event(state: dict[str, Any], stage: str, message: str, data: Any | None = None) -> None:
    state.setdefault("events", []).append(
        {
            "time": utc_now(),
            "stage": stage,
            "message": message,
            "data": data,
        }
    )
    state["updatedAt"] = utc_now()


def feedback_step(stage: str) -> str:
    match = re.search(r"(step[1-6])", stage or "")
    return match.group(1) if match else ""


def request_keys(payload: Any) -> list[str]:
    if isinstance(payload, dict):
        return sorted(str(key) for key in payload.keys())
    return []


def endpoint_label(endpoint_key: str | None, method: str, url: str) -> str:
    if endpoint_key and endpoint_key in ENDPOINT_LABELS:
        return ENDPOINT_LABELS[endpoint_key]
    path = url.split("://", 1)[-1]
    path = "/" + path.split("/", 1)[1] if "/" in path else url
    if "/api/task/task-action/" in path and method == "GET":
        return "读取行动步骤状态"
    if "/api/task/task-action/" in path and method in {"PATCH", "PUT"}:
        return "更新行动步骤状态"
    if "setTopicKeywords" in path:
        return "同步策略关键词"
    return f"调用接口 {method}"


def emit_operation_feedback(
    state: dict[str, Any],
    stage: str,
    method: str,
    url: str,
    *,
    endpoint_key: str | None = None,
    payload: Any | None = None,
    attempt: int = 1,
    dry_run: bool = False,
) -> None:
    payload_obj = {
        "type": "OP_FEEDBACK",
        "step": feedback_step(stage),
        "stage": stage,
        "operation": endpoint_label(endpoint_key, method, url),
        "method": method,
        "url": url,
        "endpoint": endpoint_key or "",
        "attempt": attempt,
        "dryRun": dry_run,
        "requestKeys": request_keys(payload),
        "state": state.get("_statePath", ""),
        "summary": state.get("_summaryPath", ""),
        "time": utc_now(),
    }
    print("OP_FEEDBACK " + json.dumps(payload_obj, ensure_ascii=False), flush=True)


def emit_error_feedback(error: dict[str, Any]) -> None:
    payload = {
        "type": "ERROR_FEEDBACK",
        "stage": error.get("stage"),
        "step": feedback_step(str(error.get("stage") or "")),
        "message": error.get("message"),
        "url": error.get("url"),
        "status": error.get("status"),
        "request": error.get("request"),
        "response": error.get("response"),
        "exceptionType": error.get("exceptionType"),
        "exception": error.get("exception"),
        "time": error.get("time"),
    }
    print("ERROR_FEEDBACK " + json.dumps(payload, ensure_ascii=False), flush=True)


def emit_json_parse_review(
    state: dict[str, Any],
    stage: str,
    endpoint_key: str,
    text: str,
    candidate_count: int,
) -> None:
    payload = {
        "type": "JSON_PARSE_REVIEW",
        "stage": stage,
        "step": feedback_step(stage),
        "endpoint": endpoint_key,
        "message": "模型输出经过多候选提取和修复后仍无法解析为可用 JSON，请执行智能体复核是否需要重新调用本步骤接口。",
        "candidateCount": candidate_count,
        "responseExcerpt": compact(text, 3000),
        "state": state.get("_statePath", ""),
        "summary": state.get("_summaryPath", ""),
        "time": utc_now(),
    }
    print("JSON_PARSE_REVIEW " + json.dumps(payload, ensure_ascii=False), flush=True)


def record_error(
    state: dict[str, Any],
    stage: str,
    message: str,
    *,
    url: str | None = None,
    status: int | None = None,
    request: Any | None = None,
    response: Any | None = None,
    exception: BaseException | None = None,
) -> None:
    item = {
        "time": utc_now(),
        "stage": stage,
        "message": message,
        "url": url,
        "status": status,
        "request": compact(request, 1200) if request is not None else None,
        "response": compact(response, 2000) if response is not None else None,
        "exceptionType": type(exception).__name__ if exception else None,
        "exception": str(exception) if exception else None,
    }
    state.setdefault("errors", []).append(item)
    state["updatedAt"] = utc_now()
    emit_error_feedback(item)


def stash_recovered_errors(state: dict[str, Any], error_count_before: int, stage: str, reason: str) -> None:
    errors = state.setdefault("errors", [])
    if len(errors) <= error_count_before:
        return
    recovered = errors[error_count_before:]
    del errors[error_count_before:]
    state.setdefault("recoveredErrors", []).extend(recovered)
    record_event(
        state,
        stage,
        "模型调用首次异常，准备重试；该异常如重试成功则不计入最终 errors",
        {"reason": reason, "recoveredErrorCount": len(recovered)},
    )


def require_context(state: dict[str, Any], *keys: str) -> None:
    missing = [key for key in keys if not str(state.get(key) or "").strip()]
    if missing:
        raise FlowError(f"缺少必要参数：{', '.join(missing)}")


def url_for(ctx: FlowContext, endpoint_key: str) -> str:
    return ctx.api_base.rstrip("/") + ENDPOINTS[endpoint_key]


def url_for_path(ctx: FlowContext, path: str) -> str:
    return ctx.api_base.rstrip("/") + "/" + path.lstrip("/")


def parse_headers(header_values: list[str], auth_token: str | None, authorization: str | None) -> dict[str, str]:
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if auth_token:
        headers["Authorization"] = auth_token if auth_token.lower().startswith("bearer ") else f"Bearer {auth_token}"
    if authorization:
        headers["Authorization"] = authorization
    for raw in header_values:
        if ":" not in raw:
            raise FlowError(f"header 格式错误，应为 Key: Value，实际：{raw}")
        key, value = raw.split(":", 1)
        headers[key.strip()] = value.strip()
    return headers


def has_authorization(headers: dict[str, str]) -> bool:
    return any(key.lower() == "authorization" and str(value).strip() for key, value in headers.items())


def ssl_context(ctx: FlowContext) -> ssl.SSLContext | None:
    if ctx.verify_ssl:
        return None
    return ssl._create_unverified_context()


def auth_header_value(token: str) -> str:
    token = token.strip()
    return token if token.lower().startswith("bearer ") else f"Bearer {token}"


def extract_access_token(data: Any) -> str | None:
    if isinstance(data, dict):
        for key in ("access_token", "accessToken", "token", "jwt", "id_token", "authorization", "Authorization"):
            value = data.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
        for value in data.values():
            nested = extract_access_token(value)
            if nested:
                return nested
    elif isinstance(data, list):
        for value in data:
            nested = extract_access_token(value)
            if nested:
                return nested
    return None


def redact_url(url: str) -> str:
    return re.sub(r"([?&](?:password|token|authToken|access_token)=)[^&]+", r"\1***", url, flags=re.IGNORECASE)


def build_login_url(ctx: FlowContext, path: str) -> str:
    base = path if re.match(r"^https?://", path, flags=re.IGNORECASE) else url_for_path(ctx, path)
    if ctx.login_password:
        separator = "&" if "?" in base else "?"
        base = f"{base}{separator}password={urllib.parse.quote(ctx.login_password, safe='')}"
    return base


def login_candidate_paths(ctx: FlowContext) -> list[str]:
    paths = [
        ctx.login_path or "/api/v1/auth/login/username",
        "/api/v1/auth/login/username",
    ]
    seen: set[str] = set()
    unique: list[str] = []
    for path in paths:
        normalized = str(path or "").strip()
        if not normalized or normalized in seen:
            continue
        seen.add(normalized)
        unique.append(normalized)
    return unique


def try_auto_login(ctx: FlowContext, state: dict[str, Any], stage: str, *, force: bool = False) -> bool:
    if not ctx.auto_login:
        return False
    if ctx.dry_run:
        record_event(state, stage, "dry-run 跳过自动登录", {"username": ctx.login_username})
        return False
    if not ctx.login_username:
        record_error(state, stage, "自动登录失败：缺少 login_username")
        return False
    if not force and ctx.login_attempt_count > 0:
        return has_authorization(ctx.headers)
    if force and ctx.login_attempt_count >= 2:
        return False

    ctx.login_attempt_count += 1
    payload = {"username": ctx.login_username}
    last_status: int | None = None
    last_response = ""
    last_exc: BaseException | None = None
    for attempt, path in enumerate(login_candidate_paths(ctx), start=1):
        url = build_login_url(ctx, path)
        display_url = redact_url(url)
        emit_operation_feedback(
            state,
            stage,
            "POST",
            display_url,
            endpoint_key="login",
            payload=payload,
            attempt=attempt,
        )
        try:
            data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
            headers = {
                "Content-Type": "application/json",
                "Accept": "application/json",
            }
            req = urllib.request.Request(url, data=data, headers=headers, method="POST")
            with urllib.request.urlopen(req, timeout=ctx.timeout, context=ssl_context(ctx)) as resp:
                raw = resp.read().decode("utf-8", errors="replace")
                parsed = parse_json_lenient(raw)
                token = extract_access_token(parsed)
                if not token:
                    last_status = getattr(resp, "status", None)
                    last_response = raw
                    record_event(
                        state,
                        stage,
                        "登录接口已响应但未找到 access_token",
                        {"url": display_url, "status": last_status, "response": compact(raw, 1000)},
                    )
                    continue
                ctx.headers["Authorization"] = auth_header_value(token)
                record_event(
                    state,
                    stage,
                    "自动登录成功，已写入 Authorization Bearer token",
                    {
                        "url": display_url,
                        "status": getattr(resp, "status", None),
                        "username": ctx.login_username,
                    },
                )
                return True
        except urllib.error.HTTPError as exc:
            last_status = exc.code
            last_response = exc.read().decode("utf-8", errors="replace") if exc.fp else ""
            last_exc = exc
            record_event(
                state,
                stage,
                "登录接口返回错误，准备尝试下一个候选登录路径",
                {"url": display_url, "status": exc.code, "response": compact(last_response, 1000)},
            )
        except Exception as exc:  # noqa: BLE001 - auto-login must report every failure.
            last_exc = exc
            last_response = traceback.format_exc()
            record_event(
                state,
                stage,
                "登录接口调用异常，准备尝试下一个候选登录路径",
                {"url": display_url, "exceptionType": type(exc).__name__, "exception": str(exc)},
            )

    record_error(
        state,
        stage,
        "自动登录失败，未能获取 access_token",
        url=redact_url(build_login_url(ctx, ctx.login_path or "/api/v1/auth/login/username")),
        status=last_status,
        request=payload,
        response=last_response,
        exception=last_exc,
    )
    return False


def http_request_json_url(
    ctx: FlowContext,
    state: dict[str, Any],
    method: str,
    url: str,
    payload: dict[str, Any] | None,
    stage: str,
    endpoint_key: str | None = None,
    allow_auto_login: bool = True,
) -> tuple[Any | None, str]:
    method = method.upper()
    if ctx.dry_run:
        emit_operation_feedback(state, stage, method, url, endpoint_key=endpoint_key, payload=payload, dry_run=True)
        record_event(state, stage, f"DRY RUN {method} {url}", {"payload": payload})
        return {"dryRun": True}, json.dumps({"dryRun": True})

    data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
    last_exc: BaseException | None = None
    for attempt in range(ctx.retries + 1):
        emit_operation_feedback(
            state,
            stage,
            method,
            url,
            endpoint_key=endpoint_key,
            payload=payload,
            attempt=attempt + 1,
        )
        try:
            req = urllib.request.Request(url, data=data, headers=ctx.headers, method=method)
            with urllib.request.urlopen(req, timeout=ctx.timeout, context=ssl_context(ctx)) as resp:
                raw = resp.read().decode("utf-8", errors="replace")
                parsed = parse_json_lenient(raw)
                record_event(state, stage, f"{method} {url} ok", {"status": getattr(resp, "status", None)})
                return parsed, raw
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace") if exc.fp else ""
            if exc.code == 401 and allow_auto_login and ctx.auto_login:
                refreshed = try_auto_login(ctx, state, f"{stage}/auto-login", force=True)
                if refreshed:
                    record_event(
                        state,
                        stage,
                        "接口返回 401，自动登录刷新 token 后重试原请求",
                        {"url": url, "status": exc.code, "response": compact(body, 1000)},
                    )
                    return http_request_json_url(
                        ctx,
                        state,
                        method,
                        url,
                        payload,
                        f"{stage}/retry-after-login",
                        endpoint_key=endpoint_key,
                        allow_auto_login=False,
                    )
            record_error(
                state,
                stage,
                "HTTP 接口返回错误",
                url=url,
                status=exc.code,
                request=payload,
                response=body,
                exception=exc,
            )
            if exc.code < 500 or attempt >= ctx.retries:
                return None, body
            last_exc = exc
        except Exception as exc:  # noqa: BLE001 - CLI runner must capture every failure.
            last_exc = exc
            if attempt >= ctx.retries:
                record_error(
                    state,
                    stage,
                    "HTTP 调用异常",
                    url=url,
                    request=payload,
                    response=traceback.format_exc(),
                    exception=exc,
                )
                return None, ""
        time.sleep(min(2**attempt, 5))

    if last_exc:
        record_error(state, stage, "HTTP 调用失败", url=url, request=payload, exception=last_exc)
    return None, ""


def http_get_json_path(ctx: FlowContext, state: dict[str, Any], path: str, stage: str) -> tuple[Any | None, str]:
    return http_request_json_url(ctx, state, "GET", url_for_path(ctx, path), None, stage)


def http_patch_json_path(ctx: FlowContext, state: dict[str, Any], path: str, payload: dict[str, Any], stage: str) -> tuple[Any | None, str]:
    return http_request_json_url(ctx, state, "PATCH", url_for_path(ctx, path), payload, stage)


def http_get_json_endpoint(ctx: FlowContext, state: dict[str, Any], endpoint_path: str, stage: str) -> tuple[Any | None, str]:
    return http_request_json_url(ctx, state, "GET", url_for_path(ctx, endpoint_path), None, stage)


def http_post_json(ctx: FlowContext, state: dict[str, Any], endpoint_key: str, payload: dict[str, Any], stage: str) -> tuple[Any | None, str]:
    url = url_for(ctx, endpoint_key)
    if ctx.dry_run:
        emit_operation_feedback(state, stage, "POST", url, endpoint_key=endpoint_key, payload=payload, dry_run=True)
        record_event(state, stage, f"DRY RUN POST {url}", {"payload": payload})
        if endpoint_key.startswith("save_"):
            synthetic_id = abs(hash(stage + stable_json(payload))) % 100000000
            return {"id": synthetic_id, "dryRun": True}, json.dumps({"id": synthetic_id, "dryRun": True})
        return {"dryRun": True}, json.dumps({"dryRun": True})
    return http_request_json_url(ctx, state, "POST", url, payload, stage, endpoint_key=endpoint_key)


def parse_json_lenient(text: str) -> Any | None:
    stripped = text.strip()
    if not stripped:
        return None
    try:
        return json.loads(stripped)
    except Exception:
        pass
    objects = parse_concatenated_json_objects(stripped)
    if objects:
        return objects
    return None


def parse_concatenated_json_objects(text: str) -> list[Any]:
    decoder = json.JSONDecoder()
    objects: list[Any] = []
    i = 0
    while i < len(text):
        while i < len(text) and text[i].isspace():
            i += 1
        if i >= len(text):
            break
        if text.startswith("data:", i):
            line_end = text.find("\n", i)
            if line_end == -1:
                line_end = len(text)
            line = text[i + 5 : line_end].strip()
            if line and line != "[DONE]":
                try:
                    objects.append(json.loads(line))
                except Exception:
                    pass
            i = line_end + 1
            continue
        if text[i] not in "{[":
            nxt = min([p for p in (text.find("{", i + 1), text.find("[", i + 1)) if p != -1], default=-1)
            if nxt == -1:
                break
            i = nxt
            continue
        try:
            obj, end = decoder.raw_decode(text, i)
            objects.append(obj)
            i = end
        except Exception:
            i += 1
    return objects


def model_text_from_response(parsed: Any, raw: str) -> tuple[str, list[Any]]:
    objects: list[Any]
    if isinstance(parsed, list):
        objects = parsed
    elif parsed is None:
        objects = parse_concatenated_json_objects(raw)
    else:
        objects = [parsed]

    parts: list[str] = []
    for obj in objects:
        collect_answer_parts(obj, parts)
    if parts:
        return "".join(parts), objects
    if isinstance(parsed, (dict, list)):
        return json.dumps(parsed, ensure_ascii=False), objects
    return raw, objects


def model_retry_reason(text: str, raw: str, parsed: Any) -> str | None:
    pieces: list[str] = []
    if text:
        pieces.append(text)
    if raw and raw != text:
        pieces.append(raw)
    if parsed is not None:
        pieces.append(compact(parsed, 2000))
    combined = "\n".join(pieces).strip()
    if not combined:
        return "模型接口空响应"
    lowered = combined.lower()
    for marker in MODEL_RETRY_MARKERS:
        if marker.lower() in lowered:
            return f"模型接口返回异常标记：{marker}"
    return None


def collect_answer_parts(obj: Any, parts: list[str]) -> None:
    if isinstance(obj, dict):
        if "answer" in obj and isinstance(obj.get("answer"), str):
            parts.append(obj["answer"])
        for value in obj.values():
            collect_answer_parts(value, parts)
    elif isinstance(obj, list):
        for item in obj:
            collect_answer_parts(item, parts)


def strip_think(text: str) -> str:
    text = text.replace("\ufeff", "").replace("\u200b", "")
    return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL | re.IGNORECASE).strip()


def extract_json_from_text(text: str) -> Any | None:
    candidates = extract_json_candidates_from_text(text)
    return select_json_candidate(candidates, endpoint_key="", stage="")


def extract_json_candidates_from_text(text: str) -> list[Any]:
    cleaned = strip_think(text)
    candidates: list[Any] = []
    seen: set[str] = set()

    def add_candidate(value: Any) -> None:
        if value is None:
            return
        try:
            key = stable_json(value)
        except Exception:
            key = repr(value)
        if key in seen:
            return
        seen.add(key)
        candidates.append(value)

    for block in re.findall(r"```(?:json|JSON)?\s*(.*?)```", cleaned, flags=re.DOTALL):
        add_candidate(try_parse_json_object(block))
        for obj in parse_concatenated_json_objects(block):
            add_candidate(obj)
        for segment in iter_balanced_segments(block, "{", "}"):
            add_candidate(try_parse_json_object(segment))
        for segment in iter_balanced_segments(block, "[", "]"):
            add_candidate(try_parse_json_object(segment))

    add_candidate(try_parse_json_object(cleaned))
    for obj in parse_concatenated_json_objects(cleaned):
        add_candidate(obj)
    for segment in iter_balanced_segments(cleaned, "{", "}"):
        add_candidate(try_parse_json_object(segment))
    for segment in iter_balanced_segments(cleaned, "[", "]"):
        add_candidate(try_parse_json_object(segment))
    return candidates


def select_json_candidate(candidates: list[Any], endpoint_key: str, stage: str) -> Any | None:
    if not candidates:
        return None
    scored = [(json_candidate_score(candidate, endpoint_key, stage), idx, candidate) for idx, candidate in enumerate(candidates)]
    scored.sort(key=lambda item: (item[0], item[1]), reverse=True)
    best_score, _, best = scored[0]
    if best_score <= 0:
        return candidates[-1]
    return best


def json_candidate_score(candidate: Any, endpoint_key: str, stage: str) -> int:
    keys = flatten_json_keys(candidate)
    text = stable_json(candidate)[:6000] if candidate is not None else ""
    score = 1 if isinstance(candidate, (dict, list)) else 0
    expected = expected_json_terms(endpoint_key, stage)
    for term in expected:
        if term in keys or term in text:
            score += 8
    if isinstance(candidate, dict):
        score += min(len(candidate), 20)
        if "items" in candidate and isinstance(candidate.get("items"), list):
            score += len(candidate["items"])
    elif isinstance(candidate, list):
        score += min(len(candidate), 20)
    if "answer" in keys and not any(term in keys or term in text for term in expected):
        score -= 5
    return score


def expected_json_terms(endpoint_key: str, stage: str) -> list[str]:
    haystack = f"{endpoint_key} {stage}"
    if "group" in haystack or "audience" in haystack:
        return ["群体", "受众", "targetAudience", "用户编号列表", "recReason", "理由"]
    if "spread" in haystack or "pathway" in haystack:
        return ["账号", "传播节点", "account", "recReason", "理由"]
    if "cuiruo" in haystack or "vuln" in haystack:
        return ["脆弱点", "vulnerability", "vuln", "vulnerabilities", "理由"]
    if "strategy" in haystack or "newboci" in haystack:
        return ["信息标题", "发布人设", "账号类型", "社交平台", "topic", "persona", "accountType"]
    return ["items", "data", "result"]


def flatten_json_keys(value: Any) -> set[str]:
    keys: set[str] = set()
    if isinstance(value, dict):
        for key, item in value.items():
            keys.add(str(key))
            keys.update(flatten_json_keys(item))
    elif isinstance(value, list):
        for item in value:
            keys.update(flatten_json_keys(item))
    return keys


def try_parse_json_object(text: str) -> Any | None:
    candidate = text.strip().strip("\u200b")
    if not candidate:
        return None
    normalized = normalize_json_text(candidate)
    repairs = [
        candidate,
        normalized,
        quote_unquoted_json_keys(normalized),
        re.sub(r",\s*([}\]])", r"\1", quote_unquoted_json_keys(normalized)),
    ]
    for item in repairs:
        try:
            return json.loads(item)
        except Exception:
            pass
        try:
            parsed = ast.literal_eval(item)
            if isinstance(parsed, (dict, list)):
                return parsed
        except Exception:
            pass
    return None


def normalize_json_text(text: str) -> str:
    normalized = (
        text.strip()
        .replace("\ufeff", "")
        .replace("\u200b", "")
        .replace("“", '"')
        .replace("”", '"')
        .replace("„", '"')
        .replace("＂", '"')
        .replace("‘", "'")
        .replace("’", "'")
        .replace("，", ",")
        .replace("：", ":")
    )
    normalized = re.sub(r",\s*([}\]])", r"\1", normalized)
    normalized = re.sub(r"\bNone\b", "null", normalized)
    normalized = re.sub(r"\bTrue\b", "true", normalized)
    normalized = re.sub(r"\bFalse\b", "false", normalized)
    return normalized


def quote_unquoted_json_keys(text: str) -> str:
    return re.sub(r'([{\[,]\s*)([A-Za-z_\u4e00-\u9fff][\w\u4e00-\u9fff\-]*\d*)\s*:', r'\1"\2":', text)


def iter_balanced_segments(text: str, open_ch: str, close_ch: str, limit: int = 80) -> list[str]:
    segments: list[str] = []
    start = text.find(open_ch)
    while start != -1 and len(segments) < limit:
        segment = extract_balanced_from(text, start, open_ch, close_ch)
        if segment:
            segments.append(segment)
            start = text.find(open_ch, start + 1)
        else:
            start = text.find(open_ch, start + 1)
    return segments


def extract_balanced_from(text: str, start: int, open_ch: str, close_ch: str) -> str | None:
    depth = 0
    in_string = False
    quote_char = ""
    escape = False
    for idx in range(start, len(text)):
        ch = text[idx]
        if in_string:
            if escape:
                escape = False
            elif ch == "\\":
                escape = True
            elif ch == quote_char:
                in_string = False
            continue
        if ch in {'"', "'"}:
            in_string = True
            quote_char = ch
        elif ch == open_ch:
            depth += 1
        elif ch == close_ch:
            depth -= 1
            if depth == 0:
                return text[start : idx + 1]
    return None


def extract_first_balanced(text: str, open_ch: str, close_ch: str) -> str | None:
    segments = iter_balanced_segments(text, open_ch, close_ch, limit=1)
    return segments[0] if segments else None


def call_model_text(ctx: FlowContext, state: dict[str, Any], endpoint_key: str, payload: dict[str, Any], stage: str) -> tuple[str, list[Any], Any | None]:
    max_attempts = max(ctx.model_retries, 0) + 1
    last_text = ""
    last_objects: list[Any] = []
    last_parsed: Any | None = None
    last_reason = ""
    for attempt in range(max_attempts):
        attempt_stage = f"{stage}/model-call-attempt-{attempt + 1}"
        error_count_before = len(state.get("errors") or [])
        parsed, raw = http_post_json(ctx, state, endpoint_key, payload, attempt_stage)
        text, objects = model_text_from_response(parsed, raw)
        reason = model_retry_reason(text, raw, parsed)
        last_text, last_objects, last_parsed, last_reason = text, objects, parsed, reason or ""
        if reason is None:
            if attempt > 0:
                record_event(state, stage, f"模型接口第 {attempt + 1} 次调用恢复成功")
            return text, objects, parsed
        if attempt < max_attempts - 1:
            stash_recovered_errors(state, error_count_before, attempt_stage, reason)
            record_event(
                state,
                stage,
                f"模型接口第 {attempt + 1} 次调用无有效响应，自动重试一次",
                {"reason": reason, "endpoint": endpoint_key},
            )
            time.sleep(min(2**attempt, 3))
            continue
        record_error(
            state,
            stage,
            "模型接口重试后仍无有效响应",
            url=url_for(ctx, endpoint_key),
            request=payload,
            response={"reason": last_reason, "text": last_text, "parsed": last_parsed},
        )
    return last_text, last_objects, last_parsed


def call_model_json(ctx: FlowContext, state: dict[str, Any], endpoint_key: str, payload: dict[str, Any], stage: str) -> tuple[dict[str, Any], str, list[Any]]:
    last_text = ""
    last_objects: list[Any] = []
    last_candidate_count = 0
    current = copy.deepcopy(payload)
    for attempt in range(2):
        text, objects, parsed = call_model_text(ctx, state, endpoint_key, current, f"{stage}/model-json-attempt-{attempt + 1}")
        last_text = text
        last_objects = objects
        if isinstance(parsed, dict) and not any("answer" in obj for obj in as_list(parsed) if isinstance(obj, dict)):
            return parsed, text, objects
        candidates = extract_json_candidates_from_text(text)
        last_candidate_count = len(candidates)
        extracted = select_json_candidate(candidates, endpoint_key, stage)
        if isinstance(extracted, dict):
            return extracted, text, objects
        if isinstance(extracted, list):
            return {"items": extracted}, text, objects
        current = copy.deepcopy(payload)
        suffix = "\n\n重要：上一轮输出无法解析。请只返回一个合法 JSON 对象，不要解释，不要 Markdown，不要 think。"
        if isinstance(current.get("user_message"), str):
            current["user_message"] += suffix
        elif isinstance(current.get("usermessage"), str):
            current["usermessage"] += suffix
        else:
            current["user_message"] = suffix.strip()
    emit_json_parse_review(state, stage, endpoint_key, last_text, last_candidate_count)
    record_error(state, stage, "模型输出无法解析为 JSON", request=payload, response=last_text)
    return {}, last_text, last_objects


def is_account_like(obj: dict[str, Any]) -> bool:
    if not isinstance(obj, dict):
        return False
    has_text = any(k in obj for k in ("page_content", "content", "frofileInfo", "发文", "docs"))
    has_identity = any(k in obj for k in ("accountId", "accountName", "user_id", "name", "id", "screen_name")) or isinstance(obj.get("metadata"), dict)
    return has_text and has_identity


def collect_accounts(obj: Any, out: list[dict[str, Any]]) -> None:
    if isinstance(obj, dict):
        docs = obj.get("docs")
        if isinstance(docs, dict):
            collect_accounts(docs, out)
        if is_account_like(obj):
            out.append(obj)
        for value in obj.values():
            if value is docs:
                continue
            collect_accounts(value, out)
    elif isinstance(obj, list):
        for item in obj:
            collect_accounts(item, out)


def accounts_from_response(parsed: Any, objects: list[Any], text: str) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    collect_accounts(parsed, out)
    collect_accounts(objects, out)
    if not out:
        for candidate in extract_json_candidates_from_text(text):
            collect_accounts(candidate, out)
    return dedupe_accounts(out)


def dedupe_accounts(accounts: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen: set[str] = set()
    result: list[dict[str, Any]] = []
    for account in accounts:
        key = str(first_present(account_id(account), account_name(account), stable_json(account)[:200]))
        if key in seen:
            continue
        seen.add(key)
        result.append(account)
    return result


def profile(account: dict[str, Any]) -> dict[str, Any]:
    meta = account.get("metadata") if isinstance(account.get("metadata"), dict) else {}
    user_profile = meta.get("用户画像") if isinstance(meta.get("用户画像"), dict) else {}
    return {**meta, **user_profile}


def account_id(account: dict[str, Any]) -> str:
    prof = profile(account)
    value = first_present(account.get("accountId"), account.get("user_id"), deep_get(account, "metadata", "user_id"), prof.get("user_id"))
    if value is None:
        value = account.get("id")
    return str(value or "").strip()


def account_name(account: dict[str, Any]) -> str:
    prof = profile(account)
    value = first_present(account.get("accountName"), account.get("name"), account.get("user_name"), prof.get("user_name"), account.get("screen_name"), prof.get("screen_name"))
    return str(value or "").strip()


def account_platform_raw(account: dict[str, Any]) -> Any:
    prof = profile(account)
    return first_present(account.get("platformType"), account.get("accountPlat"), account.get("platform"), deep_get(account, "metadata", "平台"), prof.get("平台"))


def platform_to_number(value: Any) -> Any:
    number = as_int(value)
    if number is not None:
        return number
    text = str(value or "").strip().lower()
    return PLATFORM_MAP.get(text, value if value not in (None, "") else "")


def platform_label(value: Any) -> str:
    raw = account_platform_raw(value) if isinstance(value, dict) else value
    if raw is None:
        return ""
    number = as_int(raw)
    if number == 1:
        return "facebook"
    if number == 13:
        return "ptt"
    text = str(raw).strip()
    return {"fb": "facebook"}.get(text.lower(), text)


def fans_count(account: dict[str, Any]) -> int:
    prof = profile(account)
    for value in (
        account.get("fansCount"),
        account.get("followerCount"),
        account.get("followers_count"),
        account.get("fans_count"),
        deep_get(account, "metadata", "fans_count"),
        prof.get("粉丝数"),
        prof.get("fans_count"),
    ):
        number = as_int(value)
        if number is not None:
            return number
    return 0


def account_issue(account: dict[str, Any]) -> str:
    value = first_present(account.get("issue"), deep_get(account, "metadata", "issue"))
    if isinstance(value, list):
        return ";".join(str(v) for v in value)
    return str(value or "")


def profile_info(account: dict[str, Any]) -> str:
    value = account.get("frofileInfo")
    if isinstance(value, str) and value.strip():
        return value
    return json.dumps(account, ensure_ascii=False)


def parse_index_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(v).strip() for v in value if str(v).strip()]
    text = str(value)
    return [part.strip().strip("<>").strip() for part in re.split(r"[,，;；、\s]+", text) if part.strip()]


def resolve_account_ref(ref: Any, accounts: list[dict[str, Any]]) -> dict[str, Any] | None:
    if not accounts:
        return None
    text = str(ref).strip().strip("<>").strip()
    number = as_int(text)
    if number is not None:
        for account in accounts:
            if any(as_int(account.get(key)) == number for key in ("引用序号", "索引", "index")):
                return account
        if 0 <= number < len(accounts):
            return accounts[number]
        if 1 <= number <= len(accounts):
            return accounts[number - 1]
    lower = text.lower()
    for account in accounts:
        names = [account_name(account), account_id(account), str(account.get("screen_name") or ""), str(deep_get(account, "metadata", "screen_name") or "")]
        if any(name and (lower in name.lower() or name.lower() in lower) for name in names):
            return account
    return None


def extract_audiences(data: dict[str, Any]) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    for i in range(1, 21):
        group = first_present(data.get(f"群体{i}"), data.get(f"受众{i}"), data.get(f"targetAudience{i}"))
        reason = first_present(data.get(f"理由{i}"), data.get(f"推荐理由{i}"), data.get(f"recReason{i}"))
        users = first_present(data.get(f"用户编号列表{i}"), data.get(f"用户列表{i}"), data.get(f"账号编号列表{i}"))
        if group:
            items.append(
                {
                    "targetAudience": str(group).strip(),
                    "recReason": str(reason or "").strip(),
                    "userIndexes": parse_index_list(users),
                    "rank": i,
                }
            )
    if not items and isinstance(data.get("items"), list):
        for i, item in enumerate(data["items"], start=1):
            if isinstance(item, dict):
                group = first_present(item.get("群体"), item.get("受众"), item.get("targetAudience"), item.get("name"))
                if group:
                    items.append(
                        {
                            "targetAudience": str(group).strip(),
                            "recReason": str(first_present(item.get("理由"), item.get("recReason"), item.get("reason")) or "").strip(),
                            "userIndexes": parse_index_list(first_present(item.get("用户编号列表"), item.get("users"), item.get("accounts"))),
                            "rank": i,
                        }
                    )
    return items


def audience_score(item: dict[str, Any]) -> tuple[int, int]:
    text = (item.get("targetAudience", "") + " " + item.get("recReason", "")).lower()
    keywords = ["重点", "核心", "高度相关", "关键", "军事", "安全", "两岸", "美国", "对台", "影响力", "传播力", "现实派"]
    score = sum(3 for kw in keywords if kw.lower() in text)
    score += min(len(item.get("userIndexes") or []), 5)
    score += max(0, 6 - int(item.get("rank") or 99))
    return score, -int(item.get("rank") or 99)


def select_top_audiences(items: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]:
    unique: dict[str, dict[str, Any]] = {}
    for item in items:
        key = item.get("targetAudience", "")
        if key and key not in unique:
            unique[key] = item
    return sorted(unique.values(), key=audience_score, reverse=True)[:limit]


def extract_pathway_refs(data: dict[str, Any]) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    for i in range(1, 21):
        ref = first_present(data.get(f"账号{i}"), data.get(f"传播节点{i}"), data.get(f"account{i}"))
        reason = first_present(data.get(f"理由{i}"), data.get(f"推荐理由{i}"), data.get(f"recReason{i}"))
        if ref is not None:
            items.append({"ref": ref, "recReason": str(reason or "").strip(), "rank": i})
    return items


def pathway_score(item: dict[str, Any]) -> tuple[int, int, int]:
    account = item.get("account") or {}
    text = (item.get("recReason", "") + " " + account_name(account) + " " + account_issue(account)).lower()
    keywords = ["核心", "关键", "传播", "影响力", "粉丝", "贴合", "军事", "安全", "两岸", "美国", "专业"]
    score = sum(3 for kw in keywords if kw.lower() in text)
    score += min(fans_count(account) // 1000, 50)
    score += max(0, 8 - int(item.get("rank") or 99))
    return score, fans_count(account), -int(item.get("rank") or 99)


def merge_pathway_candidates(candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
    merged: dict[str, dict[str, Any]] = {}
    for item in candidates:
        account = item.get("account") or {}
        key = account_id(account) or account_name(account) or stable_json(account)[:200]
        if not key:
            continue
        existing = merged.get(key)
        if existing is None:
            item["designIds"] = list(dict.fromkeys(item.get("designIds") or []))
            item["relatedAudiences"] = list(dict.fromkeys(item.get("relatedAudiences") or []))
            merged[key] = item
        else:
            existing["designIds"] = list(dict.fromkeys((existing.get("designIds") or []) + (item.get("designIds") or [])))
            existing["relatedAudiences"] = list(dict.fromkeys((existing.get("relatedAudiences") or []) + (item.get("relatedAudiences") or [])))
            if pathway_score(item) > pathway_score(existing):
                keep_ids = existing["designIds"]
                keep_audiences = existing["relatedAudiences"]
                merged[key] = item
                merged[key]["designIds"] = keep_ids
                merged[key]["relatedAudiences"] = keep_audiences
    return list(merged.values())


def extract_vulns(data: dict[str, Any], limit: int) -> list[dict[str, str]]:
    vulns: list[dict[str, str]] = []
    for i in range(1, 21):
        vuln = first_present(data.get(f"脆弱点{i}"), data.get(f"vulnerability{i}"), data.get(f"vuln{i}"))
        reason = first_present(data.get(f"理由{i}"), data.get(f"reason{i}"))
        if vuln:
            text = str(vuln).strip()
            if text and text not in [v["vulnerabilities"] for v in vulns]:
                vulns.append({"vulnerabilities": text, "reason": str(reason or "").strip(), "rank": str(i)})
        if len(vulns) >= limit:
            break
    return vulns


def extract_strategies(data: dict[str, Any], limit: int) -> list[dict[str, Any]]:
    strategies: list[dict[str, Any]] = []
    for i in range(1, 21):
        title = first_present(data.get(f"信息标题{i}"), data.get(f"主题{i}"), data.get(f"topic{i}"))
        persona = first_present(data.get(f"发布人设{i}"), data.get(f"人设{i}"), data.get(f"persona{i}"), "专业议题分析者")
        account_type = first_present(data.get(f"账号类型{i}"), data.get(f"accountType{i}"))
        social_platform = first_present(data.get(f"社交平台{i}"), data.get(f"平台{i}"), data.get(f"platform{i}"))
        format_value = first_present(data.get(f"信息形式{i}"), data.get(f"format{i}"), data.get(f"modelType{i}"))
        language = first_present(data.get(f"语种{i}"), data.get(f"language{i}"))
        count = first_present(data.get(f"数量{i}"), data.get(f"信息数量{i}"), data.get(f"respondingCount{i}"))
        persona_size = first_present(data.get(f"人设数量{i}"), data.get(f"personaSize{i}"))
        if title:
            strategies.append(
                {
                    "persona": str(persona or "专业议题分析者").strip(),
                    "accountType": str(account_type or "").strip(),
                    "platform": str(social_platform or "").strip(),
                    "title": str(title).strip(),
                    "format": format_value,
                    "language": language,
                    "respondingCount": count,
                    "personaSize": persona_size,
                    "rank": i,
                }
            )
        if len(strategies) >= limit:
            break
    if not strategies:
        title = first_present(data.get("信息标题"), data.get("主题"), data.get("topic"))
        if title:
            strategies.append(
                {
                    "persona": str(first_present(data.get("发布人设"), data.get("人设"), "专业议题分析者")).strip(),
                    "accountType": str(first_present(data.get("账号类型"), data.get("accountType"), "")).strip(),
                    "platform": str(first_present(data.get("社交平台"), data.get("平台"), "")).strip(),
                    "title": str(title).strip(),
                    "format": first_present(data.get("信息形式"), data.get("format"), data.get("modelType")),
                    "language": first_present(data.get("语种"), data.get("language")),
                    "respondingCount": first_present(data.get("数量"), data.get("信息数量"), data.get("respondingCount")),
                    "personaSize": first_present(data.get("人设数量"), data.get("personaSize")),
                    "rank": 1,
                }
            )
    return strategies


def response_id(response: Any) -> int | None:
    if isinstance(response, dict):
        number = as_int(response.get("id"))
        if number is not None:
            return number
        data = response.get("data")
        if isinstance(data, dict):
            return response_id(data)
    return None


def save_with_id(ctx: FlowContext, state: dict[str, Any], endpoint_key: str, payload: dict[str, Any], stage: str) -> tuple[int | None, Any | None]:
    response, _ = http_post_json(ctx, state, endpoint_key, payload, stage)
    rid = response_id(response)
    if rid is None:
        record_error(state, stage, "保存接口未返回 id，视为失败", request=payload, response=response)
    return rid, response


DEFAULT_STEP_FLAGS = {
    "step-analysis-three": 1,
    "step-info-search-ceh": 1,
    "step-analysis-audience": 0,
    "step-analysis-keypoint": 0,
    "step-analysis-weakpoint": 0,
    "step-analysis-strategy": 0,
    "step-analysis-strategy-gantt": 0,
    "step-effect-analysis": 0,
}


def parse_step_flags(raw: Any) -> dict[str, Any]:
    if isinstance(raw, dict):
        return dict(raw)
    if isinstance(raw, str) and raw.strip():
        try:
            parsed = json.loads(raw)
            if isinstance(parsed, dict):
                return parsed
        except Exception:
            return {}
    return {}


def mark_step_status(ctx: FlowContext, state: dict[str, Any], links: list[str], stage: str) -> None:
    if not ctx.sync_step_status:
        record_event(state, stage, "跳过 task-action.step 状态同步")
        return
    action_id = as_int(state.get("actionId")) or state.get("actionId")
    if not action_id:
        record_error(state, stage, "缺少 actionId，无法同步步骤状态")
        return

    step_obj = {**DEFAULT_STEP_FLAGS, **parse_step_flags(state.get("actionStep"))}
    if not ctx.dry_run:
        action, _ = http_get_json_path(ctx, state, f"/api/task/task-action/{action_id}", stage + "/get-action-step")
        if isinstance(action, dict):
            step_obj = {**DEFAULT_STEP_FLAGS, **parse_step_flags(action.get("step"))}
        elif action is None:
            record_error(state, stage, "读取 task-action 失败，仍尝试用本地步骤状态写回")

    for link in links:
        step_obj[link] = 1
    state["actionStep"] = step_obj
    payload = {"step": json.dumps(step_obj, ensure_ascii=False)}
    http_patch_json_path(ctx, state, f"/api/task/task-action/{action_id}", payload, stage + "/update-action-step")


def chat_number(value: Any, default: int = 10) -> int:
    number = as_int(value)
    if number is not None and number > 0:
        return number
    match = re.search(r"\d+", str(value or ""))
    return int(match.group(0)) if match else default


def strategy_platform(value: Any) -> Any:
    text = str(value or "").strip().lower()
    if not text:
        return 1
    if "fb" in text or "facebook" in text:
        return 1
    if text in {"x", "twitter"} or "twitter" in text:
        return 3
    if "instagram" in text or text in {"ig", "in"}:
        return 2
    if "reddit" in text or text.startswith("re"):
        return 14
    if "threads" in text or text.startswith("th"):
        return 15
    return platform_to_number(value) or 1


def strategy_model_type(value: Any) -> str:
    # Frontend defaults to text model type "1" when no modal type is selected.
    raw = first_present(value, "1")
    if isinstance(raw, list):
        return ",".join(str(v) for v in raw if str(v).strip()) or "1"
    return str(raw or "1")


def strategy_language(value: Any) -> int:
    # Frontend languageTypeChange falls back to 3.
    return as_int(value) or 3


def save_strategy_details(
    ctx: FlowContext,
    state: dict[str, Any],
    strategy_id: int,
    item: dict[str, Any],
    strategy: dict[str, Any],
    topic: str,
    stage: str,
    *,
    audience: bool,
) -> None:
    platform = strategy_platform(first_present(strategy.get("platform"), state.get("platform"), platform_label(item.get("account") or {})))
    persona = strategy.get("persona") or "专业议题分析者"
    persona_size = chat_number(strategy.get("personaSize"), 3)
    responding_count = chat_number(first_present(strategy.get("respondingCount"), strategy.get("数量")), 10)
    title = str(strategy.get("title") or "").strip()

    channel_id, _ = save_with_id(
        ctx,
        state,
        "save_channel",
        {
            "taskId": as_int(state["taskId"]) or state["taskId"],
            "actionId": as_int(state["actionId"]) or state["actionId"],
            "strategyId": strategy_id,
            "platform": platform,
            "personaJson": json.dumps([{"persona": persona, "size": persona_size}], ensure_ascii=False),
        },
        stage + "/save-channel",
    )
    narrate_id, _ = save_with_id(ctx, state, "save_narrate", {"narrate": title}, stage + "/save-narrate")
    if channel_id is not None and narrate_id is not None:
        save_with_id(
            ctx,
            state,
            "save_message",
            {
                "taskId": as_int(state["taskId"]) or state["taskId"],
                "actionId": as_int(state["actionId"]) or state["actionId"],
                "strategyId": strategy_id,
                "channelId": channel_id,
                "modelType": strategy_model_type(first_present(strategy.get("format"), strategy.get("信息形式"), strategy.get("modelType"))),
                "language": strategy_language(first_present(strategy.get("language"), strategy.get("语种"))),
                "respondingCount": responding_count,
                "messageCount": responding_count,
                "respondingTopic": narrate_id,
            },
            stage + "/save-message",
        )

    design_ids = [item.get("designId")] if audience else item.get("designIds") or []
    pathway_ids = [] if audience else [item.get("pathwayAccountId")]
    for design_id in [x for x in design_ids if x]:
        save_with_id(ctx, state, "save_strategy_relation", {"designId": design_id, "strategyId": strategy_id}, stage + "/save-strategy-relation")
    for pathway_id in [x for x in pathway_ids if x]:
        save_with_id(ctx, state, "save_strategy_relation", {"pathwayId": pathway_id, "strategyId": strategy_id}, stage + "/save-strategy-relation")

    if not ctx.dry_run:
        http_get_json_endpoint(
            ctx,
            state,
            f"/api/task/task-action-strategy/setTopicKeywords/{as_int(state['actionId']) or state['actionId']}",
            stage + "/set-topic-keywords",
        )


def object_for_audience(audience: dict[str, Any]) -> dict[str, Any]:
    return {
        "targetAudience": audience.get("targetAudience"),
        "recReason": audience.get("recReason"),
        "accounts": audience.get("accounts") or [],
    }


def run_step1(ctx: FlowContext, state: dict[str, Any]) -> None:
    require_context(state, "taskId", "actionId", "mubiao")
    stage = "step1-audience"
    if state.get("topAudiences") and not ctx.dry_run:
        record_event(state, stage, "topAudiences 已存在，跳过。如需重跑请清空 state 中该字段。")
        return
    payload = {
        "usermessage": "你是中国对台湾方向的认知域专家，你要围绕驱动目标传播行动中的叙事主题，请详细分析你在各社交平台的受众目标群体，可以从账号画像和账号近期发文等方面考虑。可以从受众目标的兴趣爱好、关注议题、职业定位、政治立场等方面划分。",
        "otherinfo": {"方向": state.get("direction") or "U方向"},
        "history": [],
        "stream": True,
        "model_name": "OpenAI3",
        "temperature": 0.7,
        "max_tokens": 20000,
        "prompt_name": f"任务背景和目的：{state['mubiao']}",
    }
    text, objects, parsed = call_model_text(ctx, state, "aiuser", payload, stage + "/analyze")
    state["audienceRawText"] = text
    accounts = accounts_from_response(parsed, objects, text)
    state["audienceAccounts"] = accounts

    group_payload = {
        "mubiao": state["mubiao"],
        "user_message": "分析其中关键信息。请基于上一轮受众分析结果，抽取受众目标群体，字段必须是：群体1、用户编号列表1、理由1、群体2、用户编号列表2、理由2 ...，最多输出5个，只返回JSON。",
        "zhanghao": accounts,
        "history": [],
        "stream": True,
        "model_name": "OpenAI3",
        "temperature": 0.7,
        "max_tokens": 20000,
        "prompt_name": state["mubiao"],
    }
    data, _, _ = call_model_json(ctx, state, "group", group_payload, stage + "/extract")
    candidates = extract_audiences(data)
    state["audienceCandidates"] = candidates
    top = select_top_audiences(candidates, ctx.max_audiences)
    saved: list[dict[str, Any]] = []
    for item in top:
        design_payload = {
            "taskId": as_int(state["taskId"]) or state["taskId"],
            "actionId": as_int(state["actionId"]) or state["actionId"],
            "parentTargetAudience": "大模型生成",
            "targetAudience": item["targetAudience"],
            "recReason": item.get("recReason", ""),
        }
        design_id, resp = save_with_id(ctx, state, "save_design", design_payload, stage + "/save-design")
        if design_id is None:
            continue
        item = {**item, "designId": design_id, "saveResponse": resp, "accounts": [], "savedAccountIds": []}
        for ref in item.get("userIndexes") or []:
            account = resolve_account_ref(ref, accounts)
            if account is None:
                record_error(state, stage + "/save-design-account", f"用户编号 {ref} 未匹配到候选账号")
                continue
            aid = account_id(account)
            name = account_name(account)
            if not aid or not name:
                record_error(state, stage + "/save-design-account", "账号缺少 accountId 或 accountName，跳过", request=account)
                continue
            account_payload = {
                "taskId": as_int(state["taskId"]) or state["taskId"],
                "actionId": as_int(state["actionId"]) or state["actionId"],
                "accountId": aid,
                "accountName": name,
                "accountPlat": platform_to_number(account_platform_raw(account)),
                "accountSource": first_present(account.get("dataSource"), account.get("accountSource"), "target"),
                "frofileInfo": profile_info(account),
                "designId": design_id,
            }
            saved_account_id, _ = save_with_id(ctx, state, "save_design_account", account_payload, stage + "/save-design-account")
            item["accounts"].append(account)
            if saved_account_id is not None:
                item["savedAccountIds"].append(saved_account_id)
        saved.append(item)
    state["topAudiences"] = saved
    if saved:
        mark_step_status(ctx, state, ["step-analysis-audience", "step-analysis-keypoint"], stage + "/step-status")
    record_event(state, stage, f"完成，保存受众 {len(saved)} 个")


def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
    require_context(state, "taskId", "actionId", "mubiao")
    stage = "step2-pathway"
    audiences = [a for a in state.get("topAudiences") or [] if a.get("designId")]
    if not audiences:
        record_error(state, stage, "没有已入库受众，无法分析传播节点")
        return
    if state.get("topPathways") and not ctx.dry_run:
        record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
        return
    all_candidates: list[dict[str, Any]] = []
    for audience in audiences:
        payload = {
            "usermessage": "你是中国对台湾方向的认知域专家，你要围绕驱动目标传播行动中的叙事主题，请分析哪些社交账号是你本次任务的核心传播节点，通过撬动该账号可实现扩大传播影响力的目的。可以从账号粉丝量、发文内容互动量、所在领域的影响力、账号画像与你传播议题的贴合度等角度分析。",
            "otherinfo": {
                "方向": state.get("direction") or "U方向",
                "任务": "传播",
                "平台": state.get("platform") or "fb",
                "受众": {"名称": audience.get("targetAudience"), "理由": audience.get("recReason")},
            },
            "history": [],
            "stream": True,
            "model_name": "OpenAI3",
            "temperature": 0.7,
            "max_tokens": 20000,
            "prompt_name": f"任务背景和目的：{state['mubiao']}",
        }
        text, objects, parsed = call_model_text(ctx, state, "aiuser", payload, stage + "/analyze")
        accounts = accounts_from_response(parsed, objects, text)
        spread_payload = {
            "mubiao": state["mubiao"],
            "user_message": "分析其中关键信息。请从候选账号中抽取核心传播节点，字段必须是：账号1、理由1、账号2、理由2 ...，最多输出5个。账号字段优先返回候选数组的引用编号或索引，只返回JSON。",
            "zhanghao": accounts,
            "history": [],
            "stream": True,
            "model_name": "liyou",
            "temperature": 0.7,
            "max_tokens": 1000,
            "prompt_name": f"{audience.get('targetAudience')}:{audience.get('recReason')};",
        }
        data, _, _ = call_model_json(ctx, state, "spread", spread_payload, stage + "/extract")
        refs = extract_pathway_refs(data)
        for ref_item in refs:
            account = resolve_account_ref(ref_item.get("ref"), accounts)
            if account is None:
                record_error(state, stage, f"传播节点引用 {ref_item.get('ref')} 未匹配账号", response=data)
                continue
            all_candidates.append(
                {
                    "account": account,
                    "accountId": account_id(account),
                    "accountName": account_name(account),
                    "recReason": ref_item.get("recReason", ""),
                    "rank": ref_item.get("rank"),
                    "designIds": [audience["designId"]],
                    "relatedAudiences": [audience.get("targetAudience", "")],
                }
            )
    merged = merge_pathway_candidates(all_candidates)
    state["pathwayCandidates"] = merged
    top = sorted(merged, key=pathway_score, reverse=True)[: ctx.max_pathways]
    saved: list[dict[str, Any]] = []
    for item in top:
        account = item["account"]
        aid = account_id(account)
        name = account_name(account)
        if not aid or not name:
            record_error(state, stage + "/save-pathway", "传播节点账号缺少 id 或名称，跳过", request=account)
            continue
        pathway_payload = {
            "taskId": as_int(state["taskId"]) or state["taskId"],
            "actionId": as_int(state["actionId"]) or state["actionId"],
            "accountName": name,
            "accountId": aid,
            "accountSource": first_present(account.get("dataSource"), account.get("accountSource"), "target"),
            "fansCount": fans_count(account),
            "platform": platform_to_number(account_platform_raw(account)),
            "recReason": item.get("recReason", ""),
            "issue": account_issue(account),
            "frofileInfo": profile_info(account),
        }
        pathway_id, resp = save_with_id(ctx, state, "save_pathway", pathway_payload, stage + "/save-pathway")
        if pathway_id is None:
            continue
        item = {
            **item,
            "pathwayAccountId": pathway_id,
            "pathwayId": pathway_id,
            "platform": platform_label(account),
            "saveResponse": resp,
        }
        relation_payload = {
            "actionId": as_int(state["actionId"]) or state["actionId"],
            "pathwayId": pathway_id,
            "designIds": item.get("designIds") or [],
        }
        http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
        saved.append(item)
    state["topPathways"] = saved
    if saved:
        mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")


def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:
    run_vuln_step(ctx, state, audience=True)


def run_step4(ctx: FlowContext, state: dict[str, Any]) -> None:
    run_vuln_step(ctx, state, audience=False)


def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> None:
    require_context(state, "taskId", "actionId", "mubiao")
    stage = "step3-audience-vulns" if audience else "step4-pathway-vulns"
    items = state.get("topAudiences" if audience else "topPathways") or []
    output_key = "audienceVulns" if audience else "pathwayVulns"
    state.setdefault(output_key, {})
    for item in items:
        owner_id = item.get("designId") if audience else item.get("pathwayAccountId")
        if not owner_id:
            continue
        if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run:
            continue
        zhanghao = item.get("accounts") or [object_for_audience(item)] if audience else [item.get("account") or item]
        prompt = (
            "你是中国对台湾方向的认知战专家，你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证，分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3，只返回JSON。"
            if audience
            else "你是中国对台湾方向的认知战专家，你要传播行动要求的叙事主题，经分析你发现该账号是你本次传播任务中最重要的传播节点，请基于该账号画像、发文历史、社会分工定位、面临的利益诉求等角度，分析该账号有哪些脆弱点可以成为你撬动目标的关键要素，例如该账号比较脆弱的利益或内心诉求等。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3，只返回JSON。"
        )
        payload = {
            "user_message": prompt,
            "mubiao": state["mubiao"],
            "zhanghao": zhanghao,
            "history": [],
            "stream": True,
            "model_name": "OpenAI3",
            "temperature": 0.7,
            "max_tokens": 1000,
            "prompt_name": item.get("targetAudience") if audience else "大V",
        }
        data, _, _ = call_model_json(ctx, state, "cuiruo", payload, stage + "/extract")
        vulns = extract_vulns(data, ctx.max_vulns)
        saved: list[dict[str, Any]] = []
        for vuln in vulns:
            if audience:
                save_payload = {
                    "taskId": as_int(state["taskId"]) or state["taskId"],
                    "actionId": as_int(state["actionId"]) or state["actionId"],
                    "actionDesignId": owner_id,
                    "vulnerabilities": vuln["vulnerabilities"],
                    "recScore": 100,
                }
                endpoint = "save_design_vuln"
            else:
                save_payload = {
                    "taskId": as_int(state["taskId"]) or state["taskId"],
                    "actionId": as_int(state["actionId"]) or state["actionId"],
                    "pathwayAccountId": owner_id,
                    "vulnerabilities": vuln["vulnerabilities"],
                    "recScore": 100,
                }
                endpoint = "save_pathway_vuln"
            vuln_id, resp = save_with_id(ctx, state, endpoint, save_payload, stage + "/save")
            if vuln_id is not None:
                saved.append({**vuln, "id": vuln_id, "saveResponse": resp})
        state[output_key][str(owner_id)] = saved
    if any(state.get(output_key, {}).values()):
        mark_step_status(ctx, state, ["step-analysis-weakpoint"], stage + "/step-status")
    record_event(state, stage, "完成")


def extract_goal(mubiao: str) -> str:
    match = re.search(r"实现[“\"](.+?)[”\"]作战目的", mubiao)
    if match:
        return match.group(1)
    match = re.search(r"实现(.+?)(?:作战目的|目的)", mubiao)
    if match:
        return match.group(1).strip("：: ，,。")
    return mubiao


def run_step5(ctx: FlowContext, state: dict[str, Any]) -> None:
    run_strategy_step(ctx, state, audience=True)


def run_step6(ctx: FlowContext, state: dict[str, Any]) -> None:
    run_strategy_step(ctx, state, audience=False)


def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) -> None:
    require_context(state, "taskId", "actionId", "mubiao")
    stage = "step5-audience-strategy" if audience else "step6-pathway-strategy"
    items = state.get("topAudiences" if audience else "topPathways") or []
    vulns_by_owner = state.get("audienceVulns" if audience else "pathwayVulns") or {}
    output_key = "audienceStrategies" if audience else "pathwayStrategies"
    state.setdefault(output_key, [])
    existing_topics = {s.get("topic") for s in state[output_key] if isinstance(s, dict)}
    for item in items:
        owner_id = item.get("designId") if audience else item.get("pathwayAccountId")
        vulns = vulns_by_owner.get(str(owner_id)) or []
        vuln_texts = [v.get("vulnerabilities") for v in vulns if v.get("vulnerabilities")]
        if not vuln_texts:
            record_error(state, stage, "没有可用脆弱点，跳过策略生成", request=item)
            continue
        target_names = [item.get("targetAudience")] if audience else (item.get("relatedAudiences") or [item.get("accountName")])
        pingtai = state.get("platform") or (platform_label(item.get("account") or {}) if not audience else "")
        payload = {
            "user_message": DEFAULT_STRATEGY_PROMPT_AUDIENCE if audience else DEFAULT_STRATEGY_PROMPT_PATHWAY,
            "cuiruodian": vuln_texts,
            "pingtai": pingtai or "facebook",
            "shouzhong": [name for name in target_names if name],
            "mudi": extract_goal(state["mubiao"]),
            "heiliao": [],
            "history": [],
            "stream": True,
            "model_name": "OpenAI3",
            "temperature": 0.7,
            "max_tokens": 1000,
            "prompt_name": "default",
        }
        data, _, _ = call_model_json(ctx, state, "strategy_model", payload, stage + "/extract")
        strategies = extract_strategies(data, ctx.max_strategies)
        for strategy in strategies:
            target = (target_names[0] if target_names else "") or (item.get("accountName") if not audience else "")
            account_type = strategy.get("accountType") or ("目标受众账号" if audience else "关键传播节点账号")
            topic = f"利用\"{strategy.get('persona') or '专业议题分析者'}\"人设,传播\"{strategy.get('title')}\"主题"
            if not strategy.get("title"):
                record_error(state, stage, "策略缺少信息标题，跳过", request=strategy, response=data)
                continue
            if topic in existing_topics:
                continue
            save_payload = {
                "taskId": as_int(state["taskId"]) or state["taskId"],
                "actionId": as_int(state["actionId"]) or state["actionId"],
                "targetAudience": target,
                "accountType": account_type,
                "topic": topic,
                "strategyTitle": "",
            }
            strategy_id, resp = save_with_id(ctx, state, "save_strategy", save_payload, stage + "/save")
            if strategy_id is not None:
                save_strategy_details(ctx, state, strategy_id, item, strategy, topic, stage, audience=audience)
                existing_topics.add(topic)
                state[output_key].append({**strategy, "id": strategy_id, "targetAudience": target, "accountType": account_type, "topic": topic, "saveResponse": resp})
    if state[output_key]:
        mark_step_status(ctx, state, ["step-analysis-strategy", "step-analysis-strategy-gantt"], stage + "/step-status")
    record_event(state, stage, "完成")


STEP_FUNCS = {
    "step1": run_step1,
    "step2": run_step2,
    "step3": run_step3,
    "step4": run_step4,
    "step5": run_step5,
    "step6": run_step6,
}

STEP_LABELS = {
    "step1": "受众目标分析与入库",
    "step2": "关键传播节点分析与入库",
    "step3": "受众脆弱点分析与入库",
    "step4": "传播节点脆弱点分析与入库",
    "step5": "受众策略分析与入库",
    "step6": "传播节点策略分析与入库",
}


def state_counts(state: dict[str, Any]) -> dict[str, int]:
    return {
        "audiences": len(state.get("topAudiences") or []),
        "pathways": len(state.get("topPathways") or []),
        "audienceVulns": sum(len(v) for v in (state.get("audienceVulns") or {}).values()),
        "pathwayVulns": sum(len(v) for v in (state.get("pathwayVulns") or {}).values()),
        "audienceStrategies": len(state.get("audienceStrategies") or []),
        "pathwayStrategies": len(state.get("pathwayStrategies") or []),
    }


def emit_step_feedback(
    state: dict[str, Any],
    step: str,
    before_error_count: int,
    *,
    status: str | None = None,
    note: str = "",
) -> None:
    total_errors = len(state.get("errors") or [])
    new_errors = max(0, total_errors - before_error_count)
    final_status = status or ("partial" if new_errors else "ok")
    payload = {
        "type": "STEP_FEEDBACK",
        "step": step,
        "label": STEP_LABELS.get(step, step),
        "status": final_status,
        "note": note,
        "newErrors": new_errors,
        "totalErrors": total_errors,
        "counts": state_counts(state),
        "state": state.get("_statePath", ""),
        "summary": state.get("_summaryPath", ""),
        "updatedAt": state.get("updatedAt"),
    }
    print("STEP_FEEDBACK " + json.dumps(payload, ensure_ascii=False), flush=True)


def finish_step(
    ctx: FlowContext,
    state: dict[str, Any],
    step: str,
    before_error_count: int,
    *,
    status: str | None = None,
    note: str = "",
) -> None:
    state["updatedAt"] = utc_now()
    save_json_file(ctx.state_path, state)
    if ctx.summary_path:
        write_summary(ctx.summary_path, state)
    emit_step_feedback(state, step, before_error_count, status=status, note=note)


def run_step_with_feedback(ctx: FlowContext, state: dict[str, Any], step: str) -> None:
    before_errors = len(state.get("errors") or [])
    STEP_FUNCS[step](ctx, state)
    finish_step(ctx, state, step, before_errors)


def skip_step_with_feedback(ctx: FlowContext, state: dict[str, Any], step: str, reason: str) -> None:
    before_errors = len(state.get("errors") or [])
    record_event(state, step, f"跳过：{reason}")
    finish_step(ctx, state, step, before_errors, status="skipped", note=reason)


def run_full(ctx: FlowContext, state: dict[str, Any]) -> None:
    run_step_with_feedback(ctx, state, "step1")
    run_step_with_feedback(ctx, state, "step2")
    run_step_with_feedback(ctx, state, "step3")
    if state.get("topPathways"):
        run_step_with_feedback(ctx, state, "step4")
    else:
        skip_step_with_feedback(ctx, state, "step4", "没有已入库关键传播节点")
    run_step_with_feedback(ctx, state, "step5")
    if state.get("topPathways"):
        run_step_with_feedback(ctx, state, "step6")
    else:
        skip_step_with_feedback(ctx, state, "step6", "没有已入库关键传播节点")


def write_summary(path: Path, state: dict[str, Any]) -> None:
    audience_vuln_count = sum(len(v) for v in (state.get("audienceVulns") or {}).values())
    pathway_vuln_count = sum(len(v) for v in (state.get("pathwayVulns") or {}).values())
    lines = [
        "# xdfx 分析入库执行摘要",
        "",
        f"- 更新时间：{state.get('updatedAt')}",
        f"- taskId：{state.get('taskId')}",
        f"- actionId：{state.get('actionId')}",
        f"- 受众入库：{len(state.get('topAudiences') or [])}",
        f"- 传播节点入库：{len(state.get('topPathways') or [])}",
        f"- 受众脆弱点入库：{audience_vuln_count}",
        f"- 传播节点脆弱点入库：{pathway_vuln_count}",
        f"- 受众策略入库：{len(state.get('audienceStrategies') or [])}",
        f"- 传播节点策略入库：{len(state.get('pathwayStrategies') or [])}",
        f"- 错误数：{len(state.get('errors') or [])}",
        "",
        "## 受众",
    ]
    for item in state.get("topAudiences") or []:
        lines.append(f"- {item.get('targetAudience')} (designId={item.get('designId')})")
    lines += ["", "## 传播节点"]
    for item in state.get("topPathways") or []:
        lines.append(f"- {item.get('accountName')} (pathwayAccountId={item.get('pathwayAccountId')})")
    if state.get("errors"):
        lines += ["", "## 失败明细"]
        for err in state["errors"]:
            lines.append(f"- [{err.get('stage')}] {err.get('message')} status={err.get('status')} url={err.get('url')}")
            if err.get("response"):
                lines.append(f"  - response: {err.get('response')}")
    text = "\n".join(lines) + "\n"
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(text, encoding="utf-8")
    except PermissionError:
        fallback = Path.cwd() / path.name
        fallback.write_text(text, encoding="utf-8")


def build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Run xdfx analysis/storage flow.")
    parser.add_argument("--mode", choices=["full", *STEP_FUNCS.keys()], default="full")
    parser.add_argument("--input", help="JSON config file. Keys: api_base, taskId, actionId, mubiao, direction, platform, headers.")
    parser.add_argument("--state", default="xdfx-flow-state.json", help="State JSON path for resume and step handoff.")
    parser.add_argument("--summary", default="xdfx-flow-summary.md", help="Markdown summary output path.")
    parser.add_argument("--api-base")
    parser.add_argument("--task-id")
    parser.add_argument("--action-id")
    parser.add_argument("--mubiao")
    parser.add_argument("--direction", default=None)
    parser.add_argument("--platform", default=None)
    parser.add_argument("--auth-token", default=os.getenv("XDFX_AUTH_TOKEN"))
    parser.add_argument("--authorization", default=os.getenv("AUTHORIZATION"))
    parser.add_argument("--header", action="append", default=[], help="Extra HTTP header, format 'Key: Value'.")
    parser.add_argument("--login-username", default=os.getenv("XDFX_LOGIN_USERNAME", "admin"), help="Username for automatic login when Authorization is missing or stale.")
    parser.add_argument("--login-password", default=os.getenv("XDFX_LOGIN_PASSWORD", "ch@user"), help="Password query parameter for automatic username login.")
    parser.add_argument("--login-path", default=os.getenv("XDFX_LOGIN_PATH", "/api/v1/auth/login/username"), help="Login endpoint path or absolute URL.")
    parser.add_argument("--no-auto-login", action="store_true", help="Disable automatic username login and 401 token refresh.")
    parser.add_argument("--verify-ssl", action="store_true", help="Verify HTTPS certificates. Default is disabled for internal/UAT certificates.")
    parser.add_argument("--timeout", type=float, default=90)
    parser.add_argument("--retries", type=int, default=2)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--no-step-status", action="store_true", help="Do not update task/task-action.step after completed steps.")
    parser.add_argument("--model-retries", type=int, default=1, help="Retry model/stream endpoints when the first response is empty, timeout-like, or says the model did not respond.")
    parser.add_argument("--max-audiences", type=int, default=3)
    parser.add_argument("--max-pathways", type=int, default=3)
    parser.add_argument("--max-vulns", type=int, default=3)
    parser.add_argument("--max-strategies", type=int, default=3)
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_arg_parser()
    args = parser.parse_args(argv)
    state_path = Path(args.state)
    state = default_state()
    if state_path.exists():
        loaded = load_json_file(state_path)
        if isinstance(loaded, dict):
            state.update(loaded)
    if args.input:
        config = load_json_file(args.input)
        if not isinstance(config, dict):
            raise FlowError("--input 必须是 JSON 对象")
        merge_config_into_state(state, config)
        input_headers = config.get("headers")
        if isinstance(input_headers, dict):
            args.header.extend(f"{k}: {v}" for k, v in input_headers.items())
        args.auth_token = args.auth_token or config.get("auth_token")
        args.authorization = args.authorization or config.get("authorization")
        if config.get("login_username") is not None:
            args.login_username = str(config.get("login_username") or "")
        if config.get("login_password") is not None:
            args.login_password = str(config.get("login_password") or "")
        if config.get("login_path") is not None:
            args.login_path = str(config.get("login_path") or "")
        if config.get("auto_login") is False:
            args.no_auto_login = True
        if config.get("verify_ssl") is not None:
            args.verify_ssl = bool(config.get("verify_ssl"))

    cli_config = {
        "api_base": args.api_base,
        "taskId": args.task_id,
        "actionId": args.action_id,
        "mubiao": args.mubiao,
        "direction": args.direction,
        "platform": args.platform,
    }
    merge_config_into_state(state, {k: v for k, v in cli_config.items() if v is not None})
    api_base = state.get("api_base") or "https://ch1.b.uat.4.cn"
    headers = parse_headers(args.header, args.auth_token, args.authorization)
    ctx = FlowContext(
        api_base=api_base,
        headers=headers,
        timeout=args.timeout,
        retries=max(args.retries, 0),
        dry_run=args.dry_run,
        state_path=state_path,
        summary_path=Path(args.summary) if args.summary else None,
        max_audiences=max(args.max_audiences, 1),
        max_pathways=max(args.max_pathways, 1),
        max_vulns=max(args.max_vulns, 1),
        max_strategies=max(args.max_strategies, 1),
        sync_step_status=not args.no_step_status,
        model_retries=max(args.model_retries, 0),
        verify_ssl=bool(args.verify_ssl),
        auto_login=not args.no_auto_login,
        login_username=args.login_username,
        login_password=args.login_password,
        login_path=args.login_path,
    )
    state["api_base"] = api_base
    state["_statePath"] = str(state_path)
    state["_summaryPath"] = str(ctx.summary_path) if ctx.summary_path else ""
    state["_autoLogin"] = ctx.auto_login
    state["_verifySsl"] = ctx.verify_ssl
    record_event(state, "runner", f"start mode={args.mode} dry_run={ctx.dry_run}")
    try:
        require_context(state, "taskId", "actionId", "mubiao")
        if ctx.auto_login and not has_authorization(ctx.headers):
            try_auto_login(ctx, state, "runner/auto-login")
        if args.mode == "full":
            run_full(ctx, state)
        else:
            run_step_with_feedback(ctx, state, args.mode)
    except FlowError as exc:
        record_error(state, "runner", str(exc), exception=exc)
        save_json_file(state_path, state)
        if ctx.summary_path:
            write_summary(ctx.summary_path, state)
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    except Exception as exc:  # noqa: BLE001
        record_error(state, "runner", "未捕获异常", response=traceback.format_exc(), exception=exc)
        save_json_file(state_path, state)
        if ctx.summary_path:
            write_summary(ctx.summary_path, state)
        print(traceback.format_exc(), file=sys.stderr)
        return 1

    state["updatedAt"] = utc_now()
    save_json_file(state_path, state)
    if ctx.summary_path:
        write_summary(ctx.summary_path, state)
    print(json.dumps(
        {
            "ok": True,
            "mode": args.mode,
            "state": str(state_path),
            "summary": str(ctx.summary_path) if ctx.summary_path else "",
            "audiences": len(state.get("topAudiences") or []),
            "pathways": len(state.get("topPathways") or []),
            "audienceStrategies": len(state.get("audienceStrategies") or []),
            "pathwayStrategies": len(state.get("pathwayStrategies") or []),
            "errors": len(state.get("errors") or []),
        },
        ensure_ascii=False,
    ))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())