feat: add user-assisted xdfx skill flow

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

@@ -28,6 +28,10 @@ from typing import Any
 
 import xdfx_auth
 
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+if hasattr(sys.stderr, "reconfigure"):
+    sys.stderr.reconfigure(encoding="utf-8", errors="replace")
 
 # Keep every backend path in one table so endpoint drift is easy to audit.
 ENDPOINTS = {
@@ -76,6 +80,7 @@ LOG_PREFIXES = {
     "operation": "操作反馈",
     "error": "错误反馈",
     "json_review": "JSON解析复核",
+    "assist": "用户协助",
     "save": "入库反馈",
     "step": "步骤反馈",
     "done": "执行完成",
@@ -85,6 +90,7 @@ STATUS_LABELS = {
     "ok": "成功",
     "partial": "部分完成",
     "skipped": "已跳过",
+    "waiting_user": "等待用户选择",
 }
 
 STEP_NUMBER_LABELS = {
@@ -226,6 +232,14 @@ class FlowContext:
     max_strategies: int = 3
     downstream_audiences: int = 2
     downstream_pathways: int = 2
+    require_user_assist: bool = True
+    select_audience_refs: list[str] | None = None
+    select_pathway_refs: list[str] | None = None
+    select_audience_vuln_refs: list[str] | None = None
+    select_pathway_vuln_refs: list[str] | None = None
+    select_strategy_audience_refs: list[str] | None = None
+    select_strategy_pathway_refs: list[str] | None = None
+    strategy_target: str = ""
     sync_step_status: bool = True
     model_retries: int = 1
     verify_ssl: bool = False
@@ -284,6 +298,129 @@ def safe_compact(value: Any, limit: int = 1400) -> str:
     return compact(redact_sensitive(value), limit)
 
 
+def split_selection_refs(raw: Any) -> list[str]:
+    """Parse user choices like '1,3' or ['1', '3'] into stable strings."""
+    if raw is None:
+        return []
+    if isinstance(raw, (list, tuple, set)):
+        parts: list[str] = []
+        for item in raw:
+            parts.extend(split_selection_refs(item))
+        return parts
+    text = str(raw).strip()
+    if not text:
+        return []
+    if text.lower() in {"all", "全部"}:
+        return ["all"]
+    return [part.strip() for part in re.split(r"[,，;；\s]+", text) if part.strip()]
+
+
+def normalized_ref(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def has_selection(refs: list[str] | None) -> bool:
+    return bool([ref for ref in refs or [] if str(ref).strip()])
+
+
+def clear_pending_assist(state: dict[str, Any]) -> None:
+    state.pop("_pendingUserAssist", None)
+
+
+def has_pending_assist(state: dict[str, Any]) -> bool:
+    return isinstance(state.get("_pendingUserAssist"), dict)
+
+
+def user_assist_command_hint(state: dict[str, Any], mode: str, flag_name: str, example: str) -> str:
+    command = f"python scripts/xdfx_flow.py --mode {mode} --state {state.get('_statePath') or 'xdfx-flow-state.json'}"
+    if state.get("_summaryPath"):
+        command += f" --summary {state.get('_summaryPath')}"
+    return f"{command} --{flag_name} {example}"
+
+
+def emit_user_assist(
+    state: dict[str, Any],
+    stage: str,
+    *,
+    selection_key: str,
+    title: str,
+    prompt: str,
+    options: list[dict[str, Any]],
+    command_hint: str,
+    allow_multiple: bool = True,
+    max_count: int | None = None,
+    exclusive_note: str = "",
+) -> None:
+    payload = {
+        "日志类型": "用户协助",
+        "阶段": stage_label(stage),
+        "步骤": feedback_step_label(stage),
+        "选择类型": selection_key,
+        "标题": title,
+        "提示": prompt,
+        "是否允许多选": yes_no(allow_multiple),
+        "最多选择数量": max_count,
+        "互斥规则": exclusive_note,
+        "选项": options,
+        "继续命令示例": command_hint,
+        "状态文件": state.get("_statePath", ""),
+        "摘要文件": state.get("_summaryPath", ""),
+        "时间": utc_now(),
+    }
+    state["_pendingUserAssist"] = {
+        "selectionKey": selection_key,
+        "stage": stage,
+        "title": title,
+        "commandHint": command_hint,
+        "optionCount": len(options),
+        "createdAt": payload["时间"],
+    }
+    print(LOG_PREFIXES["assist"] + " " + json.dumps(payload, ensure_ascii=False), flush=True)
+    record_event(state, stage, "等待用户协助选择", {"selectionKey": selection_key, "optionCount": len(options)})
+
+
+def ref_matches(ref: str, ordinal: int, aliases: list[Any]) -> bool:
+    normalized = normalized_ref(ref)
+    if normalized == "all":
+        return True
+    if normalized == str(ordinal):
+        return True
+    return any(normalized and normalized == normalized_ref(alias) for alias in aliases if alias is not None)
+
+
+def select_by_refs(
+    items: list[Any],
+    refs: list[str],
+    alias_fn,
+    *,
+    max_count: int | None = None,
+) -> tuple[list[Any], list[str]]:
+    if not refs:
+        selected = list(items)
+    elif any(normalized_ref(ref) in {"all", "全部"} for ref in refs):
+        selected = list(items)
+    else:
+        selected = []
+        missing: list[str] = []
+        for ref in refs:
+            found = None
+            for index, item in enumerate(items, start=1):
+                if ref_matches(ref, index, alias_fn(item, index)):
+                    found = item
+                    break
+            if found is None:
+                missing.append(ref)
+                continue
+            if found not in selected:
+                selected.append(found)
+        if max_count is not None:
+            return selected[:max_count], missing
+        return selected, missing
+    if max_count is not None:
+        selected = selected[:max_count]
+    return selected, []
+
+
 def stable_json(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
 
@@ -336,7 +473,7 @@ 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"))
+    return json.loads(p.read_text(encoding="utf-8-sig"))
 
 
 def save_json_file(path: str | Path, value: Any) -> None:
@@ -369,6 +506,8 @@ def default_state() -> dict[str, Any]:
         "pathwayCandidates": [],
         "topPathways": [],
         "audiencesWithPathways": [],
+        "audienceVulnCandidates": {},
+        "pathwayVulnCandidates": {},
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2182,6 +2321,121 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     return selected
 
 
+def audience_option(item: dict[str, Any], index: int) -> dict[str, Any]:
+    return {
+        "序号": index,
+        "受众目标": item.get("targetAudience"),
+        "推荐理由": compact(item.get("recReason", ""), 300),
+        "账号引用": item.get("userIndexes") or [],
+    }
+
+
+def audience_aliases(item: dict[str, Any], index: int) -> list[Any]:
+    return [index, item.get("targetAudience"), item.get("rank")]
+
+
+def pathway_option(item: dict[str, Any], index: int) -> dict[str, Any]:
+    return {
+        "序号": index,
+        "账号名称": item.get("accountName") or account_name(item.get("account") or {}),
+        "账号ID": item.get("accountId") or account_id(item.get("account") or {}),
+        "关联受众": item.get("relatedAudiences") or [],
+        "推荐理由": compact(item.get("recReason", ""), 300),
+    }
+
+
+def pathway_aliases(item: dict[str, Any], index: int) -> list[Any]:
+    account = item.get("account") or item
+    return [
+        index,
+        item.get("pathwayAccountId"),
+        item.get("pathwayId"),
+        item.get("accountId"),
+        item.get("accountName"),
+        account_id(account),
+        account_name(account),
+    ]
+
+
+def vuln_candidate_entries(
+    items: list[dict[str, Any]],
+    candidate_store: dict[str, Any],
+    *,
+    audience: bool,
+) -> list[dict[str, Any]]:
+    entries: list[dict[str, Any]] = []
+    seq = 1
+    for item in items:
+        owner_id = item.get("designId") if audience else item.get("pathwayAccountId")
+        owner_name = item.get("targetAudience") if audience else item.get("accountName")
+        for index, vuln in enumerate(candidate_store.get(str(owner_id)) or [], start=1):
+            if not isinstance(vuln, dict):
+                continue
+            entries.append(
+                {
+                    "seq": seq,
+                    "ownerId": owner_id,
+                    "ownerName": owner_name,
+                    "localIndex": index,
+                    "vuln": vuln,
+                }
+            )
+            seq += 1
+    return entries
+
+
+def vuln_option(entry: dict[str, Any]) -> dict[str, Any]:
+    vuln = entry.get("vuln") or {}
+    return {
+        "序号": entry.get("seq"),
+        "对象ID": entry.get("ownerId"),
+        "对象名称": entry.get("ownerName"),
+        "本对象内序号": entry.get("localIndex"),
+        "脆弱点": vuln.get("vulnerabilities"),
+        "理由": compact(vuln.get("reason", ""), 260),
+    }
+
+
+def vuln_aliases(entry: dict[str, Any], index: int) -> list[Any]:
+    vuln = entry.get("vuln") or {}
+    owner_id = entry.get("ownerId")
+    return [
+        index,
+        entry.get("seq"),
+        f"{owner_id}:{entry.get('localIndex')}",
+        vuln.get("vulnerabilities"),
+    ]
+
+
+def strategy_target_option(item: dict[str, Any], index: int, *, audience: bool) -> dict[str, Any]:
+    if audience:
+        return {
+            "序号": index,
+            "类型": "受众",
+            "对象ID": item.get("designId"),
+            "对象名称": item.get("targetAudience"),
+            "已入库脆弱点数量": len((item.get("_vulns") or [])),
+            "关联传播节点": item.get("downstreamPathways") or [],
+        }
+    return {
+        "序号": index,
+        "类型": "关键传播节点",
+        "对象ID": item.get("pathwayAccountId"),
+        "对象名称": item.get("accountName"),
+        "已入库脆弱点数量": len((item.get("_vulns") or [])),
+        "关联受众": item.get("relatedAudiences") or [],
+    }
+
+
+def strategy_target_aliases(item: dict[str, Any], index: int, *, audience: bool) -> list[Any]:
+    return [
+        index,
+        item.get("designId") if audience else item.get("pathwayAccountId"),
+        item.get("targetAudience") if audience else item.get("accountName"),
+        item.get("accountId") if not audience else None,
+    ]
+
+
 def run_step1(ctx: FlowContext, state: dict[str, Any]) -> None:
     """Analyze audiences, save top 3, and attach any referenced accounts."""
     require_context(state, "taskId", "actionId", "mubiao")
@@ -2189,39 +2443,65 @@ def run_step1(ctx: FlowContext, state: dict[str, Any]) -> None:
     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
-    # The first model call may return raw text plus embedded account objects.
-    accounts = accounts_from_response(parsed, objects, text)
-    state["audienceAccounts"] = accounts
+    candidates = [item for item in state.get("audienceCandidates") or [] if isinstance(item, dict)]
+    accounts = [item for item in state.get("audienceAccounts") or [] if isinstance(item, dict)]
+    if not candidates:
+        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
+        # The first model call may return raw text plus embedded account objects.
+        accounts = accounts_from_response(parsed, objects, text)
+        state["audienceAccounts"] = accounts
 
-    # The second model call converts the long audience analysis into numbered
-    # groups that can be saved as task-action-design rows.
-    group_payload = {
-        "mubiao": state["mubiao"],
-        "user_message": "分析其中关键信息。请基于上一轮受众分析结果，抽取受众目标群体。字段必须是：群体1、用户编号列表1、理由1、群体2、用户编号列表2、理由2 ...，最多输出5个，只返回JSON。用户编号列表中的编号/id必须对应本次入参 zhanghao 数组里的账号索引值；例如返回 0 或 <<0>> 表示 zhanghao[0]，返回 1 或 <<1>> 表示 zhanghao[1]。不要生成不存在于 zhanghao 的编号。",
-        "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)
+        # The second model call converts the long audience analysis into numbered
+        # groups that can be saved as task-action-design rows.
+        group_payload = {
+            "mubiao": state["mubiao"],
+            "user_message": "分析其中关键信息。请基于上一轮受众分析结果，抽取受众目标群体。字段必须是：群体1、用户编号列表1、理由1、群体2、用户编号列表2、理由2 ...，最多输出5个，只返回JSON。用户编号列表中的编号/id必须对应本次入参 zhanghao 数组里的账号索引值；例如返回 0 或 <<0>> 表示 zhanghao[0]，返回 1 或 <<1>> 表示 zhanghao[1]。不要生成不存在于 zhanghao 的编号。",
+            "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
+    if not candidates:
+        record_error(state, stage, "没有生成可供选择的受众候选")
+        return
+    if ctx.require_user_assist and not has_selection(ctx.select_audience_refs):
+        emit_user_assist(
+            state,
+            stage,
+            selection_key="audience",
+            title="请选择要入库并继续分析的受众",
+            prompt=f"可多选，最多建议选择 {ctx.max_audiences} 个。回复序号即可，例如：1,2,3。",
+            options=[audience_option(item, index) for index, item in enumerate(candidates, start=1)],
+            command_hint=user_assist_command_hint(state, "step1", "select-audiences", "1,2,3"),
+            allow_multiple=True,
+            max_count=ctx.max_audiences,
+        )
+        return
+    selected_refs = ctx.select_audience_refs or []
+    top, missing = select_by_refs(candidates, selected_refs, audience_aliases, max_count=ctx.max_audiences)
+    clear_pending_assist(state)
+    if missing:
+        record_error(state, stage, f"用户选择的受众未匹配：{', '.join(missing)}")
+    if not top:
+        record_error(state, stage, "没有选中任何受众，无法继续")
+        return
     saved: list[dict[str, Any]] = []
     for item in top:
         # Save the audience first; account rows and downstream analyses depend
@@ -2282,60 +2562,85 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
-    all_candidates: list[dict[str, Any]] = []
-    for audience in audiences:
-        # Analyze pathway candidates in the context of each saved audience.
-        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)
-        # Ask the model to point back to candidate accounts, then resolve those
-        # references to concrete account objects.
-        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]
+    ranked = [item for item in state.get("pathwayCandidates") or [] if isinstance(item, dict)]
+    if not ranked:
+        all_candidates: list[dict[str, Any]] = []
+        for audience in audiences:
+            # Analyze pathway candidates in the context of each saved audience.
+            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)
+            # Ask the model to point back to candidate accounts, then resolve those
+            # references to concrete account objects.
+            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)
+        ranked = sorted(merged, key=pathway_score, reverse=True)
+        state["pathwayCandidates"] = ranked
+    if not ranked:
+        record_error(state, stage, "没有生成可供选择的关键传播节点候选")
+        return
+    if ctx.require_user_assist and not has_selection(ctx.select_pathway_refs):
+        emit_user_assist(
+            state,
+            stage,
+            selection_key="pathway",
+            title="请选择要入库并继续分析的关键传播节点",
+            prompt=f"可多选，最多建议选择 {ctx.max_pathways} 个。回复序号、账号名称或账号ID均可，例如：1,2,3。",
+            options=[pathway_option(item, index) for index, item in enumerate(ranked, start=1)],
+            command_hint=user_assist_command_hint(state, "step2", "select-pathways", "1,2,3"),
+            allow_multiple=True,
+            max_count=ctx.max_pathways,
+        )
+        return
+    top, missing = select_by_refs(ranked, ctx.select_pathway_refs or [], pathway_aliases, max_count=ctx.max_pathways)
+    clear_pending_assist(state)
+    if missing:
+        record_error(state, stage, f"用户选择的关键传播节点未匹配：{', '.join(missing)}")
+    if not top:
+        record_error(state, stage, "没有选中任何关键传播节点，无法继续")
+        return
     saved: list[dict[str, Any]] = []
     for item in top:
         # Save the account row before writing the design-pathway relation.
@@ -2395,7 +2700,10 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) ->
     stage = "step3-audience-vulns" if audience else "step4-pathway-vulns"
     items = downstream_items(ctx, state, stage, audience=audience)
     output_key = "audienceVulns" if audience else "pathwayVulns"
+    candidate_key = "audienceVulnCandidates" if audience else "pathwayVulnCandidates"
+    select_refs = ctx.select_audience_vuln_refs if audience else ctx.select_pathway_vuln_refs
     state.setdefault(output_key, {})
+    candidate_store = state.setdefault(candidate_key, {})
     for item in items:
         # owner_id is designId for audiences and pathwayAccountId for accounts.
         owner_id = item.get("designId") if audience else item.get("pathwayAccountId")
@@ -2403,6 +2711,8 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) ->
             continue
         if str(owner_id) in state.get(output_key, {}) and state[output_key][str(owner_id)] and not ctx.dry_run:
             continue
+        if str(owner_id) in candidate_store and candidate_store[str(owner_id)]:
+            continue
         zhanghao = selected_accounts_for_item(item, audience=audience)
         prompt = (
             "你是中国对台湾方向的认知战专家，你要传播行动要求的叙事主题。请基于该受众目标的画像、关注议题、社会分工定位、利益诉求与上一轮账号佐证，分析该受众目标有哪些脆弱点可以成为后续策略设计的关键要素。请输出理由、脆弱点1、理由1、脆弱点2、理由2、脆弱点3、理由3，只返回JSON。"
@@ -2422,8 +2732,43 @@ def run_vuln_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool) ->
         }
         data, _, _ = call_model_json(ctx, state, "cuiruo", payload, stage + "/extract")
         vulns = extract_vulns(data, ctx.max_vulns)
+        candidate_store[str(owner_id)] = vulns
+    entries = vuln_candidate_entries(items, candidate_store, audience=audience)
+    if not entries:
+        record_error(state, stage, "没有生成可供选择的脆弱点候选")
+        return
+    if ctx.require_user_assist and not has_selection(select_refs):
+        flag = "select-audience-vulns" if audience else "select-pathway-vulns"
+        emit_user_assist(
+            state,
+            stage,
+            selection_key="audience_vuln" if audience else "pathway_vuln",
+            title="请选择要入库的受众脆弱点" if audience else "请选择要入库的传播节点脆弱点",
+            prompt="可多选。回复全局序号即可，例如：1,2,4；也可用 对象ID:本对象内序号。",
+            options=[vuln_option(entry) for entry in entries],
+            command_hint=user_assist_command_hint(state, "step3" if audience else "step4", flag, "1,2,4"),
+            allow_multiple=True,
+        )
+        return
+    selected_entries, missing = select_by_refs(entries, select_refs or [], vuln_aliases)
+    clear_pending_assist(state)
+    if missing:
+        record_error(state, stage, f"用户选择的脆弱点未匹配：{', '.join(missing)}")
+    selected_by_owner: dict[str, list[dict[str, Any]]] = {}
+    for entry in selected_entries:
+        owner_key = str(entry.get("ownerId"))
+        selected_by_owner.setdefault(owner_key, []).append(entry.get("vuln") or {})
+    if not selected_by_owner:
+        record_error(state, stage, "没有选中任何脆弱点，无法继续")
+        return
+    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
         saved: list[dict[str, Any]] = []
-        for vuln in vulns:
+        for vuln in selected_by_owner.get(str(owner_id), []):
             # The two weak-point tables have nearly identical payloads but
             # different foreign-key names.
             if audience:
@@ -2475,8 +2820,53 @@ def run_strategy_step(ctx: FlowContext, state: dict[str, Any], *, audience: bool
     """Generate strategies from saved weak points and persist frontend child rows."""
     require_context(state, "taskId", "actionId", "mubiao")
     stage = "step5-audience-strategy" if audience else "step6-pathway-strategy"
+    if audience and (has_selection(ctx.select_strategy_pathway_refs) or ctx.strategy_target == "pathway"):
+        clear_pending_assist(state)
+        record_error(state, stage, "策略生成目标类型冲突：Step5 只能选择受众，不能同时选择关键传播节点")
+        return
+    if not audience and (has_selection(ctx.select_strategy_audience_refs) or ctx.strategy_target == "audience"):
+        clear_pending_assist(state)
+        record_error(state, stage, "策略生成目标类型冲突：Step6 只能选择关键传播节点，不能同时选择受众")
+        return
     items = downstream_items(ctx, state, stage, audience=audience)
     vulns_by_owner = state.get("audienceVulns" if audience else "pathwayVulns") or {}
+    selectable_items: list[dict[str, Any]] = []
+    for item in items:
+        owner_id = item.get("designId") if audience else item.get("pathwayAccountId")
+        vulns = vulns_by_owner.get(str(owner_id)) or []
+        if vulns:
+            selectable_items.append({**item, "_vulns": vulns})
+    select_refs = ctx.select_strategy_audience_refs if audience else ctx.select_strategy_pathway_refs
+    if selectable_items and ctx.require_user_assist and not has_selection(select_refs):
+        flag = "select-strategy-audiences" if audience else "select-strategy-pathways"
+        emit_user_assist(
+            state,
+            stage,
+            selection_key="strategy_audience" if audience else "strategy_pathway",
+            title="请选择要生成策略的受众" if audience else "请选择要生成策略的关键传播节点",
+            prompt=(
+                "可多选，但本步骤只能选择受众，不能混选关键传播节点。回复序号即可，例如：1,2。"
+                if audience
+                else "可多选，但本步骤只能选择关键传播节点，不能混选受众。回复序号即可，例如：1,2。"
+            ),
+            options=[strategy_target_option(item, index, audience=audience) for index, item in enumerate(selectable_items, start=1)],
+            command_hint=user_assist_command_hint(state, "step5" if audience else "step6", flag, "1,2"),
+            allow_multiple=True,
+            exclusive_note="受众策略和关键传播节点策略不能在同一步命令中交叉选择。",
+        )
+        return
+    selected_items, missing = select_by_refs(
+        selectable_items,
+        select_refs or [],
+        lambda item, index: strategy_target_aliases(item, index, audience=audience),
+    )
+    clear_pending_assist(state)
+    if missing:
+        record_error(state, stage, f"用户选择的策略生成对象未匹配：{', '.join(missing)}")
+    items = selected_items
+    if not items:
+        record_error(state, stage, "没有选中任何具备脆弱点的策略生成对象，无法继续")
+        return
     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)}
@@ -2583,14 +2973,15 @@ def emit_step_feedback(
     # equally transparent to the user.
     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")
+    final_status = status or ("waiting_user" if has_pending_assist(state) else ("partial" if new_errors else "ok"))
+    final_note = note or (state.get("_pendingUserAssist", {}).get("title", "") if has_pending_assist(state) else "")
     payload = {
         "日志类型": "步骤反馈",
         "步骤": STEP_NUMBER_LABELS.get(step, step),
         "步骤名称": STEP_LABELS.get(step, step),
         "状态": STATUS_LABELS.get(final_status, final_status),
         "执行状态": status_label(final_status),
-        "备注": note,
+        "备注": final_note,
         "本步骤新增错误数": new_errors,
         "累计错误数": total_errors,
         "累计入库数量": state_counts(state),
@@ -2638,15 +3029,27 @@ def run_full(ctx: FlowContext, state: dict[str, Any]) -> None:
     for dry-runs and unattended local troubleshooting.
     """
     run_step_with_feedback(ctx, state, "step1")
+    if has_pending_assist(state):
+        return
     run_step_with_feedback(ctx, state, "step2")
+    if has_pending_assist(state):
+        return
     run_step_with_feedback(ctx, state, "step3")
+    if has_pending_assist(state):
+        return
     if state.get("topPathways"):
         run_step_with_feedback(ctx, state, "step4")
+        if has_pending_assist(state):
+            return
     else:
         skip_step_with_feedback(ctx, state, "step4", "没有已入库关键传播节点")
     run_step_with_feedback(ctx, state, "step5")
+    if has_pending_assist(state):
+        return
     if state.get("topPathways"):
         run_step_with_feedback(ctx, state, "step6")
+        if has_pending_assist(state):
+            return
     else:
         skip_step_with_feedback(ctx, state, "step6", "没有已入库关键传播节点")
 
@@ -2722,6 +3125,14 @@ def build_arg_parser() -> argparse.ArgumentParser:
     parser.add_argument("--max-strategies", type=int, default=3)
     parser.add_argument("--downstream-audiences", type=int, default=2, help="继续进入脆弱点/策略步骤的已入库受众数量，默认 2。")
     parser.add_argument("--downstream-pathways", type=int, default=2, help="继续进入脆弱点/策略步骤的已入库传播节点数量，默认 2。")
+    parser.add_argument("--no-user-assist", action="store_true", help="关闭每步用户协助选择，使用脚本默认排序自动继续。")
+    parser.add_argument("--select-audiences", default="", help="Step1 用户选择的受众序号/名称，逗号分隔，例如 1,2,3。")
+    parser.add_argument("--select-pathways", default="", help="Step2 用户选择的关键传播节点序号/账号名/账号ID，逗号分隔。")
+    parser.add_argument("--select-audience-vulns", default="", help="Step3 用户选择的受众脆弱点序号，逗号分隔。")
+    parser.add_argument("--select-pathway-vulns", default="", help="Step4 用户选择的传播节点脆弱点序号，逗号分隔。")
+    parser.add_argument("--strategy-target", choices=["", "audience", "pathway"], default="", help="策略生成目标类型；Step5 只能 audience，Step6 只能 pathway。")
+    parser.add_argument("--select-strategy-audiences", default="", help="Step5 用户选择的受众策略生成对象序号/受众ID/名称，逗号分隔。")
+    parser.add_argument("--select-strategy-pathways", default="", help="Step6 用户选择的传播节点策略生成对象序号/节点ID/账号名，逗号分隔。")
     return parser
 
 
@@ -2758,6 +3169,20 @@ def main(argv: list[str] | None = None) -> int:
             args.downstream_audiences = as_int(config.get("downstream_audiences")) or args.downstream_audiences
         if config.get("downstream_pathways") is not None:
             args.downstream_pathways = as_int(config.get("downstream_pathways")) or args.downstream_pathways
+        if config.get("require_user_assist") is False or config.get("user_assist") is False:
+            args.no_user_assist = True
+        selection_map = {
+            "select_audiences": "select_audiences",
+            "select_pathways": "select_pathways",
+            "select_audience_vulns": "select_audience_vulns",
+            "select_pathway_vulns": "select_pathway_vulns",
+            "select_strategy_audiences": "select_strategy_audiences",
+            "select_strategy_pathways": "select_strategy_pathways",
+            "strategy_target": "strategy_target",
+        }
+        for config_key, arg_name in selection_map.items():
+            if config.get(config_key) is not None:
+                setattr(args, arg_name, config.get(config_key))
 
     cli_config = {
         "api_base": args.api_base,
@@ -2785,6 +3210,14 @@ def main(argv: list[str] | None = None) -> int:
         max_strategies=max(args.max_strategies, 1),
         downstream_audiences=max(args.downstream_audiences, 1),
         downstream_pathways=max(args.downstream_pathways, 1),
+        require_user_assist=not args.no_user_assist,
+        select_audience_refs=split_selection_refs(args.select_audiences),
+        select_pathway_refs=split_selection_refs(args.select_pathways),
+        select_audience_vuln_refs=split_selection_refs(args.select_audience_vulns),
+        select_pathway_vuln_refs=split_selection_refs(args.select_pathway_vulns),
+        select_strategy_audience_refs=split_selection_refs(args.select_strategy_audiences),
+        select_strategy_pathway_refs=split_selection_refs(args.select_strategy_pathways),
+        strategy_target=str(args.strategy_target or ""),
         sync_step_status=not args.no_step_status,
         model_retries=max(args.model_retries, 0),
         verify_ssl=bool(args.verify_ssl),
@@ -2838,6 +3271,8 @@ def main(argv: list[str] | None = None) -> int:
     final_payload = {
         "日志类型": "执行完成",
         "是否成功": "是",
+        "是否等待用户选择": yes_no(has_pending_assist(state)),
+        "待用户选择": state.get("_pendingUserAssist"),
         "执行模式": mode_label(args.mode),
         "状态文件": str(state_path),
         "摘要文件": str(ctx.summary_path) if ctx.summary_path else "",

skills-data/xdfx-split-controller/scripts/xdfx_run_full_steps.py

@@ -15,6 +15,10 @@ import subprocess
 import sys
 from pathlib import Path
 
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+if hasattr(sys.stderr, "reconfigure"):
+    sys.stderr.reconfigure(encoding="utf-8", errors="replace")
 
 STEPS = ("step1", "step2", "step3", "step4", "step5", "step6")
 
@@ -53,7 +57,7 @@ def selected_steps(first: str, last: str) -> list[str]:
     return list(STEPS[start : end + 1])
 
 
-def stream_command(step: str, cmd: list[str]) -> int:
+def stream_command(step: str, cmd: list[str]) -> tuple[int, bool]:
     """Run one step and forward stdout/stderr lines as they are produced."""
     print(f"运行步骤：{STEP_LABELS.get(step, step)}；执行命令：" + " ".join(cmd), flush=True)
     env = os.environ.copy()
@@ -68,10 +72,13 @@ def stream_command(step: str, cmd: list[str]) -> int:
         env=env,
     )
     assert process.stdout is not None
+    waiting_user = False
     for line in process.stdout:
+        if line.startswith("用户协助 "):
+            waiting_user = True
         sys.stdout.buffer.write(line.encode("utf-8", errors="replace"))
         sys.stdout.buffer.flush()
-    return process.wait()
+    return process.wait(), waiting_user
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -83,7 +90,10 @@ def main(argv: list[str] | None = None) -> int:
         # Keep all business flags unchanged so the wrapper behaves like a thin
         # streaming version of repeated xdfx_flow.py --mode stepN calls.
         cmd = [sys.executable, str(flow_script), "--mode", step, *passthrough]
-        code = stream_command(step, cmd)
+        code, waiting_user = stream_command(step, cmd)
+        if waiting_user:
+            print("已发起用户协助，暂停后续步骤；请根据用户选择重跑当前步骤。", flush=True)
+            return code
         if code:
             exit_code = code
             if not args.continue_on_error: