Merge branch 'dhk-1' into 'master'

代码修改

See merge request knowledge-base-bj/deerflow-server!167

frontend-web/src/open-canvas/pages/AIWritingPage.tsx

@@ -414,6 +414,8 @@ function AIWritingBody({
   const routeBase = useAIWritingRouteBase();
   const agentRoute = useAIWritingAgentRouteState(routeBase);
   const isAdmin = getStoredAuth()?.system_role === 'admin';
+  const externalNotebookMode =
+    studioToken != null || initialNotebookId != null || initialDocumentIds != null;
 
   // 需求 0616 #5：左栏点「笔记本 / 所有空间」→ 在右侧内嵌打开对应页，替换右侧内容
   // （写作布局不卸载，仅 display:none 隐藏，避免 WritingSetupChat 重挂丢线程）。
@@ -432,12 +434,12 @@ function AIWritingBody({
     }
   }, [agentRoute.active]);
 
-  const agentsActive = sidebarTab === 'agents';
+  const agentsActive = !externalNotebookMode && sidebarTab === 'agents';
 
   // 研究模板侧栏在「写作配置(idle)」与「历史回看」两态都展示（点历史记录后保留侧栏、
   // 历史内容显示到右侧区域）；写作进行中收起；嵌入态(studioToken)始终不展示。
   const showSidebar =
-    !studioToken && (state.status === 'idle' || isHistoryMode || agentsActive);
+    !externalNotebookMode && (state.status === 'idle' || isHistoryMode || agentsActive);
   const compactAppearanceActions = compactChrome ? (
     <AIWritingCompactAppearanceActions />
   ) : null;

offline-backend-20260512/backend/CLAUDE.md

@@ -273,7 +273,7 @@ The AI-writing graph (`packages/harness/deerflow/agents/ai_writing/`) has four s
 
 Seeding mirrors the roundtable pattern: verified `config.yaml` + `SOUL.md` ship in `app/gateway/routers/_ai_writing_seed_assets/<id>/`; `_ai_writing_seed.py::ensure_ai_writing_functional_agents()` copies them to `.deer-flow/agents/` when missing (never overwrites local edits), called from the startup lifespan (before `_sync_legacy_agents`, so the agents appear in `/api/agents` on first boot) and self-heals from `GET /api/ai-writing/sessions` + `GET /api/ai-writing/article-types`. Tests: `tests/test_ai_writing_seed.py`, `tests/test_ai_writing_agent_runner.py`.
 
-**Skill-driven retrieval (检索类型).** `material_source` has three values: `general` (通用检索, default), `knowledge_base` (知识库检索), and `notebook` (我的空间); legacy `skill` is normalized to `general` by the researcher node. **`knowledge_base` 直连内网 ES 接口（= 旧版「内网直连检索」）** — it bypasses skill orchestration entirely and calls `IntranetSearchProvider` over `ai_writing.intranet_search_url` (concurrent over the generated keywords; `intranet_knowledge_base`/`intranet_search_timeout` apply). **An empty/unconfigured `intranet_search_url` does NOT disable it** — the empty string is passed straight to `IntranetSearchProvider`, which falls back to its built-in default endpoint (`intranet_stub._DEFAULT_ENDPOINT`), exactly like the legacy `get_search_provider(search_provider=intranet)` path. So 内网直连检索 works out of the box without configuring an address (configure `intranet_search_url` only to point at a different endpoint). This is the "弱模型识别不了技能" escape hatch — the old no-skill retrieval path restored as an explicit option (weak models like deepseek-chat often fail to invoke configured skills). Tests: `tests/test_ai_writing_knowledge_base.py`. General retrieval is orchestrated from the **skills configured on the `ai-writing-researcher` agent** (`config.yaml → skills`, editable from the agent management page): the built-in `knowledge-base-search` skill (知识库检索) runs the native keyword fan-out web/intranet search (the old default path), while any other configured skill searches with a `description`-targeted query; skills run in configured order with early-stop at `max_materials`, and an empty/unresolvable skills config falls back to the built-in knowledge search so retrieval always works. The skill itself is seeded from `_ai_writing_seed_assets/skills/knowledge-base-search/SKILL.md` into `skills/public/` (never overwrites), and a pre-existing researcher `config.yaml` lacking the `skills` key is patched in place with the default (explicit `skills: []` is respected). **PAUSE-1 input append flow**: when the user types natural language into the material-confirm input (`re_search`/legacy `enable_skill_search` + `user_query`), the node matches the most relevant configured skill via `SkillPicker(candidates=…)` (skipped when only one skill is configured), searches with the user's words, and **appends** the new materials to the existing package (URL-dedup, original order preserved, no LLM re-rank/truncation, 100-item hard cap). **Skill-failure fault tolerance**: a configured skill that times out / raises / has all its keyword searches fail (vs. legitimately returning 0 results) no longer just silently moves on — the researcher node emits a `skill_search_step` of type `skill_error` (carrying the error `reason`), and if the configured skills collected **no** new materials at all it runs a `fallback` pass over other available skills (`_fallback_candidate_skills`: the built-in `knowledge-base-search` native path first, then other enabled catalog skills not yet tried) instead of erroring out. The frontend `SkillStepsCard` renders `skill_error` (rose) and `fallback` (amber) so the timeline shows "技能报错 → 改用其他技能搜集". Tests: `tests/test_ai_writing_material_source.py`.
+**Skill-driven retrieval (检索类型).** `material_source` has three values: `general` (通用检索, default), `knowledge_base` (知识库检索), and `notebook` (我的空间); legacy `skill` is normalized to `general` by the researcher node. **`knowledge_base` 直连内网 ES 接口（= 旧版「内网直连检索」）** — it bypasses skill orchestration entirely and calls `IntranetSearchProvider` over `ai_writing.intranet_search_url` (concurrent over the generated keywords; `intranet_knowledge_base`/`intranet_search_timeout` apply). The client uses `trust_env=False` and defaults `intranet_verify_ssl=false`, so internal HTTPS endpoints with self-signed/internal-CA certificates do not fail on certificate verification; set `ai_writing.intranet_verify_ssl: true` only for environments that require strict certificate checks. **An empty/unconfigured `intranet_search_url` does NOT disable it** — the empty string is passed straight to `IntranetSearchProvider`, which falls back to its built-in default endpoint (`intranet_stub._DEFAULT_ENDPOINT`), exactly like the legacy `get_search_provider(search_provider=intranet)` path. So 内网直连检索 works out of the box without configuring an address (configure `intranet_search_url` only to point at a different endpoint). This is the "弱模型识别不了技能" escape hatch — the old no-skill retrieval path restored as an explicit option (weak models like deepseek-chat often fail to invoke configured skills). Tests: `tests/test_ai_writing_knowledge_base.py`. General retrieval is orchestrated from the **skills configured on the `ai-writing-researcher` agent** (`config.yaml → skills`, editable from the agent management page): the built-in `knowledge-base-search` skill (知识库检索) runs the native keyword fan-out web/intranet search (the old default path), while any other configured skill searches with a `description`-targeted query; skills run in configured order with early-stop at `max_materials`, and an empty/unresolvable skills config falls back to the built-in knowledge search so retrieval always works. The skill itself is seeded from `_ai_writing_seed_assets/skills/knowledge-base-search/SKILL.md` into `skills/public/` (never overwrites), and a pre-existing researcher `config.yaml` lacking the `skills` key is patched in place with the default (explicit `skills: []` is respected). **PAUSE-1 input append flow**: when the user types natural language into the material-confirm input (`re_search`/legacy `enable_skill_search` + `user_query`), the node matches the most relevant configured skill via `SkillPicker(candidates=…)` (skipped when only one skill is configured), searches with the user's words, and **appends** the new materials to the existing package (URL-dedup, original order preserved, no LLM re-rank/truncation, 100-item hard cap). **Skill-failure fault tolerance**: a configured skill that times out / raises / has all its keyword searches fail (vs. legitimately returning 0 results) no longer just silently moves on — the researcher node emits a `skill_search_step` of type `skill_error` (carrying the error `reason`), and if the configured skills collected **no** new materials at all it runs a `fallback` pass over other available skills (`_fallback_candidate_skills`: the built-in `knowledge-base-search` native path first, then other enabled catalog skills not yet tried) instead of erroring out. The frontend `SkillStepsCard` renders `skill_error` (rose) and `fallback` (amber) so the timeline shows "技能报错 → 改用其他技能搜集". Tests: `tests/test_ai_writing_material_source.py`.
 
 **Skill-call timeout** is hardcoded **300s (5 min)** — `researcher.py` `_SKILL_SEARCH_TIMEOUT_SECONDS` (per-skill wrapper) and `_skill_agent.py` `_REAL_AGENT_TIMEOUT_SECONDS` (whole real-agent skill run). The intranet ES fallback (`IntranetSearchProvider`) gets a longer **600s (10 min)** default (`intranet_stub.py`; overridable via `ai_writing.intranet_search_timeout`) — that interface is the slow one, so our own fallback timeout is generous to avoid being blamed for "检不出".
 

offline-backend-20260512/backend/README.md

@@ -135,6 +135,10 @@ FastAPI application providing REST endpoints for frontend integration:
 
 Custom agents are indexed in the `agents` database table. The `id` column is the stable runtime identifier and filesystem directory name (`.deer-flow/agents/{id}/SOUL.md`); `name` is display-only, may be Chinese, and may be duplicated. Rows with `user_id = NULL` are built-in agents, user rows are private unless `published = true`, and users can only update or delete agents they created.
 
+### AI Writing Intranet Retrieval
+
+AI 写作的「知识库检索」会直连 `ai_writing.intranet_search_url` 内网 ES/知识库接口，并默认跳过代理环境变量和 HTTPS 证书校验，适配自签名或内部 CA 证书的内网服务。需要强校验时可在 `config.yaml` 里设置 `ai_writing.intranet_verify_ssl: true`。
+
 ### IM Channels
 
 The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])` and updates a single in-thread card in place.

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py

@@ -27,13 +27,15 @@ from types import SimpleNamespace
 
 from langchain_core.runnables import RunnableConfig
 
-from ..search import get_search_provider, SearchResult
+from ..prompts.researcher_prompts import (
+    EXTRACT_KEYWORDS_SYSTEM,
+    EXTRACT_KEYWORDS_USER,
+    RANK_MATERIALS_SYSTEM,
+    RANK_MATERIALS_USER,
+)
+from ..search import SearchResult, get_search_provider
 from ..search.skill_search import SkillPicker
 from ..state import AIWritingState, Material, MaterialPackage
-from ..prompts.researcher_prompts import (
-    EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER,
-    RANK_MATERIALS_SYSTEM, RANK_MATERIALS_USER,
-)
 from ._agent_runner import (
     KEYWORDS_READY_MARKER,
     MATERIALS_RANKED_MARKER,
@@ -620,6 +622,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
             # 内网慢检索超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大，
             # 默认 600s，且最低 300s。
             timeout=_intranet_timeout_seconds(ai_writing_config),
+            verify_ssl=ai_writing_config.get("intranet_verify_ssl", False),
         )
         endpoint_label = intranet_url or getattr(provider, "endpoint", "内置默认内网端点")
         intranet_debug = {
@@ -628,6 +631,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
             "endpoint": endpoint_label,
             "knowledge_base": ai_writing_config.get("intranet_knowledge_base", "samples"),
             "timeout": _intranet_timeout_seconds(ai_writing_config),
+            "verify_ssl": getattr(provider, "verify_ssl", False),
             "status": "running",
             "queries": [],
             "calls": [],
@@ -780,7 +784,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
                     timeout=_SKILL_SEARCH_TIMEOUT_SECONDS,
                 )
                 return (results or []), ""
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 msg = f"检索超时（>{_SKILL_SEARCH_TIMEOUT_SECONDS}s）"
                 logger.warning("skill[%s] search timed out after %ss", skill_name, _SKILL_SEARCH_TIMEOUT_SECONDS)
                 search_errors.append(f"skill[{skill_name}]: {msg}")
@@ -923,13 +927,14 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
         intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip()
         # 诊断开关：开启后即使技能成功也照样跑完内网并记录完整结果（用于验证内网接口是否真能通）。
         intranet_debug_force = bool(ai_writing_config.get("intranet_search_debug", False))
-        intranet_task: "asyncio.Task | None" = None
+        intranet_task: asyncio.Task | None = None
         intranet_holder: dict = {}
         intranet_debug = {
             "configured": bool(intranet_url),
             "endpoint": intranet_url or "(未配置 ai_writing.intranet_search_url)",
             "knowledge_base": ai_writing_config.get("intranet_knowledge_base", "samples"),
             "timeout": _intranet_timeout_seconds(ai_writing_config),
+            "verify_ssl": bool(ai_writing_config.get("intranet_verify_ssl", False)),
             "debug_force": intranet_debug_force,
             "status": "pending" if intranet_url else "not_configured",
             "queries": [],          # 每个检索词：{query, count, error?}
@@ -952,6 +957,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
                     # 兜底接口超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大，
                     # 默认 600s（10 分钟），且最低 300s。
                     timeout=_intranet_timeout_seconds(ai_writing_config),
+                    verify_ssl=ai_writing_config.get("intranet_verify_ssl", False),
                 )
                 intranet_holder["provider"] = intranet
                 fb_keywords = keywords[:keyword_count] or [intent]

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/__init__.py

@@ -1,8 +1,8 @@
 from .base import BaseSearchProvider, SearchResult
-from .internet_search import InternetSearchProvider
-from .intranet_stub import IntranetSearchProvider, normalize_intranet_timeout
 from .demo_search import DemoSearchProvider
 from .duckduckgo_search import DuckDuckGoSearchProvider
+from .internet_search import InternetSearchProvider
+from .intranet_stub import IntranetSearchProvider, normalize_intranet_timeout, normalize_intranet_verify_ssl
 from .skill_search import SkillSearchProvider
 
 
@@ -14,6 +14,7 @@ def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
       intranet_search_url: http://10.106.57.180:7863/knowledge_base/search_docs_es
       intranet_knowledge_base: samples
       intranet_search_timeout: 300 # 内网检索超时秒数，最少 300 秒
+      intranet_verify_ssl: false   # 内网 HTTPS 自签名证书场景默认关闭证书校验
       ddg_region: "cn-zh"         # DuckDuckGo 区域，默认 cn-zh
 
     internet 模式：DuckDuckGo 为主搜索（免 API Key），失败时 fallback 到 demo 演示素材。
@@ -26,6 +27,7 @@ def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
             base_url=cfg.get("intranet_search_url", ""),
             knowledge_base_name=cfg.get("intranet_knowledge_base", "samples"),
             timeout=normalize_intranet_timeout(cfg.get("intranet_search_timeout")),
+            verify_ssl=normalize_intranet_verify_ssl(cfg.get("intranet_verify_ssl")),
         )
     if provider == "tavily":
         return _FallbackSearchProvider(InternetSearchProvider(), DemoSearchProvider())

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 
 import logging
 import time
-from typing import List
 
 from .base import BaseSearchProvider, SearchResult, simplify_query
 
@@ -15,6 +14,8 @@ _DEFAULT_KB = "samples"
 # 内网检索最少等待 5 分钟；该接口经常慢返回，低于这个值会造成素材收集误判为空。
 _MIN_TIMEOUT_SECONDS = 300
 _DEFAULT_TIMEOUT_SECONDS = 600
+# 内网 HTTPS 检索服务常使用自签名/内部 CA 证书，默认跳过证书校验避免 SSL 证书链错误。
+_DEFAULT_VERIFY_SSL = False
 # 单条原始响应体最大保留长度，避免把超大响应整段灌进前端时间线
 _RAW_SNIPPET_LIMIT = 2000
 
@@ -28,6 +29,23 @@ def normalize_intranet_timeout(timeout: object, default: int = _DEFAULT_TIMEOUT_
     return max(value, _MIN_TIMEOUT_SECONDS)
 
 
+def normalize_intranet_verify_ssl(value: object, default: bool = _DEFAULT_VERIFY_SSL) -> bool:
+    """返回内网检索是否校验 SSL 证书，兼容 config.yaml 里的 bool / 字符串写法。"""
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return default
+    if isinstance(value, (int, float)):
+        return bool(value)
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"1", "true", "yes", "on"}:
+            return True
+        if normalized in {"0", "false", "no", "off"}:
+            return False
+    return default
+
+
 def _pick(d: dict, keys: tuple) -> str:
     """从结果项里按候选字段名取第一个非空值（两套接口字段名不同，做并集兼容）。"""
     for k in keys:
@@ -62,12 +80,14 @@ class IntranetSearchProvider(BaseSearchProvider):
         base_url: str = "",
         knowledge_base_name: str = _DEFAULT_KB,
         timeout: int = _DEFAULT_TIMEOUT_SECONDS,
+        verify_ssl: bool = _DEFAULT_VERIFY_SSL,
         **_ignored,
     ):
         self.endpoint = base_url or _DEFAULT_ENDPOINT
         self.knowledge_base_name = knowledge_base_name or _DEFAULT_KB
         # 内网 ES 接口很慢，兜底检索超时默认 10 分钟，且无论配置如何至少 5 分钟。
         self.timeout = normalize_intranet_timeout(timeout)
+        self.verify_ssl = normalize_intranet_verify_ssl(verify_ssl)
         # 收集本次运行的检索异常，供 researcher 节点取出并提示用户（ES 波动等）
         self.errors: list[str] = []
         # 解析失败时收集的原始响应体，供前端「错误返回结果」折叠展示
@@ -82,7 +102,7 @@ class IntranetSearchProvider(BaseSearchProvider):
         if snippet:
             self.raw_responses.append(snippet)
 
-    async def search(self, query: str, max_results: int = 5, **kwargs) -> List[SearchResult]:
+    async def search(self, query: str, max_results: int = 5, **kwargs) -> list[SearchResult]:
         results = await self._search_once(query, max_results)
         # 0 结果兜底：复合检索词常检不出（如「特朗普竞选」），换成更核心的检索词重试一次
         # （「特朗普竞选」→「特朗普」）。仅在首检为空、且能简化出不同的词时才重试。
@@ -93,7 +113,7 @@ class IntranetSearchProvider(BaseSearchProvider):
                 results = await self._search_once(simpler, max_results)
         return results
 
-    async def _search_once(self, query: str, max_results: int = 5) -> List[SearchResult]:
+    async def _search_once(self, query: str, max_results: int = 5) -> list[SearchResult]:
         import httpx
 
         payload = {
@@ -112,12 +132,13 @@ class IntranetSearchProvider(BaseSearchProvider):
             "status": "pending",
             "ok": False,
             "count": 0,
+            "verify_ssl": self.verify_ssl,
         }
         # 排查用：每次真实调用都打一行请求明细（端点 / 知识库 / 检索词 / 超时），
         # 内网复现时直接看后端日志即可知道"到底请求了哪个地址、用什么参数"。
         logger.info(
-            "[intranet-search] POST endpoint=%s kb=%s timeout=%ss query=%r",
+            "[intranet-search] POST endpoint=%s kb=%s timeout=%ss verify_ssl=%s query=%r",
-            self.endpoint, self.knowledge_base_name, self.timeout, query,
+            self.endpoint, self.knowledge_base_name, self.timeout, self.verify_ssl, query,
         )
         try:
             # trust_env=False：**关键**。内网 ES 是内网地址，但本部署的模型走公网，
@@ -125,7 +146,8 @@ class IntranetSearchProvider(BaseSearchProvider):
             # 会把这个内网请求也套用公网代理 → 代理到不了内网 IP → 连接被重置/空回复，
             # 表现为「接口异常」。强制忽略代理 env、直连内网（与 multi_agent.py 的回环
             # 调用同款修法）。
-            async with httpx.AsyncClient(timeout=self.timeout, trust_env=False) as client:
+            # verify=False：内网 HTTPS 常见自签名/内部 CA 证书；这里仅跳过本内网检索端点的证书校验。
+            async with httpx.AsyncClient(timeout=self.timeout, trust_env=False, verify=self.verify_ssl) as client:
                 resp = await client.post(self.endpoint, json=payload)
                 call["http_status"] = getattr(resp, "status_code", None)
                 resp.raise_for_status()
@@ -210,7 +232,7 @@ class IntranetSearchProvider(BaseSearchProvider):
             "[intranet-search] OK endpoint=%s query=%r http_status=%s raw_items=%s",
             self.endpoint, query, call.get("http_status"), len(items) if isinstance(items, list) else 0,
         )
-        results: List[SearchResult] = []
+        results: list[SearchResult] = []
         for i, item in enumerate(items[:max_results]):
             if not isinstance(item, dict):
                 continue

offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py

@@ -6,7 +6,7 @@ from datetime import UTC, datetime
 from typing import Any
 from uuid import uuid4
 
-from sqlalchemy import and_, case, delete, func, or_, select
+from sqlalchemy import and_, case, delete, func, literal, not_, or_, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
@@ -19,6 +19,7 @@ from deerflow.persistence.agents.model import (
     AgentRow,
     AgentSkillRow,
 )
+from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
 from deerflow.persistence.tags.model import TagAssignmentRow
 
 
@@ -36,6 +37,30 @@ class AgentRepository(AgentStore):
     def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
         self._sf = session_factory
 
+    @staticmethod
+    def _business_chain_agent_exists(agent_id_col):
+        return (
+            select(RoundtableChainRow.id)
+            .where(
+                or_(
+                    RoundtableChainRow.seats.like(literal('%"agent_id": "') + agent_id_col + literal('"%')),
+                    RoundtableChainRow.seats.like(literal('%"agent_id":"') + agent_id_col + literal('"%')),
+                )
+            )
+            .exists()
+        )
+
+    @classmethod
+    def _valid_favorite_filter(cls):
+        return or_(
+            AgentFavoriteRow.origin == "user",
+            and_(
+                AgentFavoriteRow.origin == "position",
+                AgentRow.user_id.is_not(None),
+                not_(cls._business_chain_agent_exists(AgentFavoriteRow.agent_id)),
+            ),
+        )
+
     @staticmethod
     def _visible_filter(user_id: str):
         """Visibility for single-record fetches (``get_visible``).
@@ -53,13 +78,21 @@ class AgentRepository(AgentStore):
         - favorited — the "我的" listing keeps favorited agents even after
           the owner unpublishes them, so reads keep honoring the favorite.
         """
-        fav_subq = select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
+        valid_favorite_exists = (
+            select(AgentFavoriteRow.id)
+            .where(
+                AgentFavoriteRow.user_id == user_id,
+                AgentFavoriteRow.agent_id == AgentRow.id,
+                AgentRepository._valid_favorite_filter(),
+            )
+            .exists()
+        )
         return or_(
             AgentRow.user_id.is_(None),
             AgentRow.user_id == user_id,
             AgentRow.published.is_(True),
             AgentRow.is_private.is_(True),
-            AgentRow.id.in_(fav_subq),
+            valid_favorite_exists,
         )
 
     @staticmethod
@@ -269,10 +302,19 @@ class AgentRepository(AgentStore):
                 AgentRow.is_private.is_(False),
             )
         if scope == "mine":
-            # owned OR favorited. A correlated IN-subquery (not a JOIN) keeps a
+            # owned OR valid favorites. Built-in agents and business-chain
-            # single row per agent even when it is both owned and favorited.
+            # seats only enter "mine" when explicitly favorited by the user;
-            fav_subq = select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
+            # the original position-granted behavior stays for ordinary agents.
-            return or_(AgentRow.user_id == user_id, AgentRow.id.in_(fav_subq))
+            valid_favorite_exists = (
+                select(AgentFavoriteRow.id)
+                .where(
+                    AgentFavoriteRow.user_id == user_id,
+                    AgentFavoriteRow.agent_id == AgentRow.id,
+                    self._valid_favorite_filter(),
+                )
+                .exists()
+            )
+            return or_(AgentRow.user_id == user_id, valid_favorite_exists)
         if scope == "private":
             return AgentRow.is_private.is_(True)
         if scope == "builtin":
@@ -441,12 +483,16 @@ class AgentRepository(AgentStore):
         async with self._sf() as session:
             existing = (
                 await session.execute(
-                    select(AgentFavoriteRow.id)
+                    select(AgentFavoriteRow)
                     .where(AgentFavoriteRow.user_id == user_id, AgentFavoriteRow.agent_id == agent_id)
                     .limit(1)
                 )
             ).scalar_one_or_none()
             if existing is not None:
+                if existing.origin != "user" or existing.position_id is not None:
+                    existing.origin = "user"
+                    existing.position_id = None
+                    await session.commit()
                 return
             session.add(AgentFavoriteRow(id=uuid4().hex, user_id=user_id, agent_id=agent_id))
             try:
@@ -470,7 +516,12 @@ class AgentRepository(AgentStore):
     async def list_favorite_ids(self, user_id: str) -> set[str]:
         async with self._sf() as session:
             result = await session.execute(
-                select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
+                select(AgentFavoriteRow.agent_id)
+                .join(AgentRow, AgentRow.id == AgentFavoriteRow.agent_id)
+                .where(
+                    AgentFavoriteRow.user_id == user_id,
+                    self._valid_favorite_filter(),
+                )
             )
             return {row[0] for row in result.all()}
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260701_01_agent_position_favorites.py

@@ -0,0 +1,90 @@
+"""Stop position-granted favorites from collecting built-ins/chain seats.
+
+Built-in agents and agents used as business-chain seats should only enter
+"mine" when a user explicitly clicks the favorite action. This migration
+removes historical automatic ``origin='position'`` rows for those agents while
+preserving user-created favorites (``origin='user'``) and ordinary
+position-granted agent favorites.
+
+Revision ID: 20260701_01
+Revises: 20260625_04
+Create Date: 2026-07-01
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+revision = "20260701_01"
+down_revision = "20260625_04"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    inspector = inspect(op.get_bind())
+    return any(col["name"] == column_name for col in inspector.get_columns(table_name))
+
+
+def upgrade() -> None:
+    if not (
+        _has_table("agent_favorites")
+        and _has_column("agent_favorites", "origin")
+        and _has_column("agent_favorites", "agent_id")
+    ):
+        return
+
+    favorites = sa.table(
+        "agent_favorites",
+        sa.column("agent_id", sa.String(64)),
+        sa.column("origin", sa.String(16)),
+    )
+    cleanup_conditions = []
+    if _has_table("agents") and _has_column("agents", "user_id"):
+        agents = sa.table(
+            "agents",
+            sa.column("id", sa.String(64)),
+            sa.column("user_id", sa.String(64)),
+        )
+        cleanup_conditions.append(
+            favorites.c.agent_id.in_(sa.select(agents.c.id).where(agents.c.user_id.is_(None)))
+        )
+    if _has_table("roundtable_chains") and _has_column("roundtable_chains", "seats"):
+        chains = sa.table(
+            "roundtable_chains",
+            sa.column("seats", sa.Text()),
+        )
+        cleanup_conditions.append(
+            sa.select(sa.literal(1))
+            .select_from(chains)
+            .where(
+                sa.or_(
+                    chains.c.seats.like(sa.literal('%"agent_id": "') + favorites.c.agent_id + sa.literal('"%')),
+                    chains.c.seats.like(sa.literal('%"agent_id":"') + favorites.c.agent_id + sa.literal('"%')),
+                )
+            )
+            .exists()
+        )
+    if not cleanup_conditions:
+        return
+
+    op.get_bind().execute(
+        favorites.delete().where(
+            favorites.c.origin == "position",
+            sa.or_(*cleanup_conditions),
+        )
+    )
+
+
+def downgrade() -> None:
+    # Data cleanup only; removed auto-grants are not reconstructable.
+    pass

offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py

@@ -12,7 +12,8 @@ Resolution rule (per resource type):
 * For a resource type **with** bound tags → the union of resources carrying any
   of those tags.
 * For a resource type **without** bound tags → the *default set*:
-  - ``agent``               → all built-in agents (``user_id IS NULL``)
+  - ``agent``               -> built-ins first, then filtered out of
+    position-granted favorites
   - ``scheduled_task``      → empty (no built-in tasks)
   - ``skill``               → empty (the 广场 still shows every visible skill;
     a position only *adds* its tagged skills to the member's "我的技能")
@@ -24,9 +25,12 @@ Resolution rule (per resource type):
 
 Materialization (``sync_user``) only ever touches rows it created itself
 (``origin='position'``), so a user's own favorites / subscriptions are never
-clobbered. ``skill`` is materialized into ``skill_favorites`` exactly like
+clobbered. ``agent`` keeps the original position-granted behavior for ordinary
-``agent`` → ``agent_favorites``: the tagged skills are *added* to the user's
+tagged agents, but filters out built-ins and business-chain seats; those only
-"我的技能" while the browse-all 广场 stays complete (it is **not** restricted).
+enter "mine" after an explicit user favorite. ``skill`` is still materialized
+into ``skill_favorites`` so tagged skills are added to the user's "mine"
+collection while the browse-all square stays complete (it is **not**
+restricted).
 Only recommended-question visibility is computed live by :meth:`resolve_visible`
 and is *not* materialized. (The admin-configured prompts shown to users ARE the
 recommended questions — there is no separate prompt-template store.)
@@ -34,6 +38,7 @@ recommended questions — there is no separate prompt-template store.)
 
 from __future__ import annotations
 
+import json
 import logging
 import uuid
 from datetime import UTC, datetime
@@ -43,13 +48,15 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
 from deerflow.persistence.positions.base import PositionStore
+from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
 from deerflow.persistence.scheduled_tasks.model import ScheduledTaskSubscriptionRow
 from deerflow.persistence.skills.model import SkillFavoriteRow
 from deerflow.persistence.tags.base import TagStore
 
 logger = logging.getLogger(__name__)
 
-# Materialized into "我的" tables by sync_user.
+# Reconciled by sync_user. ``agent`` preserves ordinary position-granted rows,
+# but filters built-ins and business-chain seats.
 _MATERIALIZED_TYPES = ("agent", "skill", "scheduled_task")
 # Resolved live at query time (visibility filters), never materialized.
 _VISIBILITY_TYPES = ("recommended_question",)
@@ -93,8 +100,12 @@ class PositionSyncService:
         bindings = await self._positions.get_tag_bindings(pid) if pid else {}
         tag_ids = bindings.get(resource_type) or []
         if tag_ids:
-            return set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
+            target = set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
-        return await self._default_set(resource_type, session)
+        else:
+            target = await self._default_set(resource_type, session)
+        if resource_type == "agent":
+            return await self._filter_materializable_agents(target, session)
+        return target
 
     async def _default_set(self, resource_type: str, session: AsyncSession) -> set[str]:
         if resource_type == "agent":
@@ -103,6 +114,37 @@ class PositionSyncService:
         # scheduled_task (and any other materialized type) defaults to empty.
         return set()
 
+    async def _filter_materializable_agents(self, agent_ids: set[str], session: AsyncSession) -> set[str]:
+        if not agent_ids:
+            return set()
+        rows = (
+            await session.execute(
+                select(AgentRow.id).where(
+                    AgentRow.id.in_(agent_ids),
+                    AgentRow.user_id.is_not(None),
+                )
+            )
+        ).all()
+        candidate_ids = {r[0] for r in rows}
+        return candidate_ids - await self._business_chain_agent_ids(session)
+
+    async def _business_chain_agent_ids(self, session: AsyncSession) -> set[str]:
+        rows = (await session.execute(select(RoundtableChainRow.seats))).scalars()
+        agent_ids: set[str] = set()
+        for raw in rows:
+            if not raw:
+                continue
+            try:
+                seats = json.loads(raw) if isinstance(raw, str) else raw
+            except Exception:
+                continue
+            if not isinstance(seats, list):
+                continue
+            for seat in seats:
+                if isinstance(seat, dict) and isinstance(seat.get("agent_id"), str):
+                    agent_ids.add(seat["agent_id"])
+        return agent_ids
+
     # --------------------------------------------------------- materialize
 
     async def sync_user(self, user_id: str) -> None:

offline-backend-20260512/backend/tests/test_agents_backend.py

@@ -1,7 +1,13 @@
 from __future__ import annotations
 
+import json
+from uuid import uuid4
+
 import pytest
+import pytest_asyncio
 import yaml
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 from app.gateway.routers.agents import (
     _backfill_listing_cache,
@@ -11,7 +17,13 @@ from app.gateway.routers.agents import (
 from app.gateway.services import build_run_config
 from deerflow.config import paths as paths_module
 from deerflow.config.agents_config import AgentConfig
-from deerflow.persistence.agents import MemoryAgentStore
+from deerflow.persistence.agents import AgentRepository, MemoryAgentStore
+from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
+from deerflow.persistence.base import Base
+from deerflow.persistence.positions.memory import MemoryPositionStore
+from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
+from deerflow.persistence.tags.memory import MemoryTagStore
+from deerflow.services.position_sync import PositionSyncService
 
 
 @pytest.fixture
@@ -23,6 +35,92 @@ def isolated_paths(tmp_path, monkeypatch):
     monkeypatch.setattr(paths_module, "_paths", None)
 
 
+@pytest_asyncio.fixture
+async def sql_session_factory(tmp_path):
+    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'agent-tests.db'}")
+    async with engine.begin() as conn:
+        # Importing PositionSyncService above registers the agent / skill /
+        # scheduled-task rows it materializes into.
+        await conn.run_sync(Base.metadata.create_all)
+    sf = async_sessionmaker(engine, expire_on_commit=False)
+    try:
+        yield sf
+    finally:
+        await engine.dispose()
+
+
+async def _add_sql_agent(sf, *, agent_id: str, user_id: str | None) -> None:
+    async with sf() as session:
+        session.add(
+            AgentRow(
+                id=agent_id,
+                name=agent_id,
+                description="",
+                user_id=user_id,
+                published=True,
+                is_private=False,
+                square_id="",
+            )
+        )
+        await session.commit()
+
+
+async def _add_sql_favorite(sf, *, user_id: str, agent_id: str, origin: str) -> None:
+    async with sf() as session:
+        session.add(
+            AgentFavoriteRow(
+                id=uuid4().hex,
+                user_id=user_id,
+                agent_id=agent_id,
+                origin=origin,
+                position_id="default" if origin == "position" else None,
+            )
+        )
+        await session.commit()
+
+
+async def _add_sql_chain(sf, *, chain_id: str, agent_ids: list[str]) -> None:
+    async with sf() as session:
+        session.add(
+            RoundtableChainRow(
+                id=chain_id,
+                user_id="chain-owner",
+                title=chain_id,
+                description="",
+                seats=json.dumps(
+                    [{"agent_id": agent_id, "name": agent_id, "description": ""} for agent_id in agent_ids],
+                    ensure_ascii=False,
+                ),
+            )
+        )
+        await session.commit()
+
+
+async def _sql_favorite_rows(sf, user_id: str) -> set[tuple[str, str]]:
+    async with sf() as session:
+        rows = (
+            await session.execute(
+                select(AgentFavoriteRow.agent_id, AgentFavoriteRow.origin).where(
+                    AgentFavoriteRow.user_id == user_id
+                )
+            )
+        ).all()
+    return {(agent_id, origin) for agent_id, origin in rows}
+
+
+async def _default_positions() -> MemoryPositionStore:
+    positions = MemoryPositionStore()
+    await positions.create_position(
+        {
+            "id": "default",
+            "name": "Default",
+            "description": "",
+            "is_default": True,
+        }
+    )
+    return positions
+
+
 @pytest.mark.asyncio
 async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None:
     store = MemoryAgentStore()
@@ -39,6 +137,136 @@ async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility
     assert user_c_ids == {"builtin-agent", "published-b"}
 
 
+@pytest.mark.asyncio
+async def test_position_sync_default_does_not_collect_builtin_agents(sql_session_factory) -> None:
+    await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None)
+    await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None)
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="builtin-auto",
+        origin="position",
+    )
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="builtin-manual",
+        origin="user",
+    )
+
+    sync = PositionSyncService(
+        sql_session_factory,
+        await _default_positions(),
+        MemoryTagStore(),
+    )
+    await sync.sync_user("user-a")
+
+    assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
+        ("builtin-manual", "user")
+    }
+
+
+@pytest.mark.asyncio
+async def test_position_sync_tagged_agents_are_not_auto_favorited(sql_session_factory) -> None:
+    await _add_sql_agent(sql_session_factory, agent_id="builtin-tagged", user_id=None)
+    await _add_sql_agent(sql_session_factory, agent_id="custom-tagged", user_id="owner")
+    await _add_sql_agent(sql_session_factory, agent_id="chain-tagged", user_id="owner")
+    await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-tagged"])
+
+    positions = await _default_positions()
+    tags = MemoryTagStore()
+    await tags.create_tag({"id": "agent-tag", "name": "Agent tag", "tag_type": "agent"})
+    await tags.set_target_tags("agent", "builtin-tagged", ["agent-tag"])
+    await tags.set_target_tags("agent", "custom-tagged", ["agent-tag"])
+    await tags.set_target_tags("agent", "chain-tagged", ["agent-tag"])
+    await positions.set_tag_bindings("default", {"agent": ["agent-tag"]})
+
+    sync = PositionSyncService(sql_session_factory, positions, tags)
+    await sync.sync_user("user-a")
+
+    assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
+        ("custom-tagged", "position")
+    }
+
+
+@pytest.mark.asyncio
+async def test_mine_scope_ignores_auto_agent_favorites_but_keeps_manual_builtin(sql_session_factory) -> None:
+    await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None)
+    await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None)
+    await _add_sql_agent(sql_session_factory, agent_id="custom-auto", user_id="owner")
+    await _add_sql_agent(sql_session_factory, agent_id="chain-auto", user_id="owner")
+    await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-auto"])
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="builtin-auto",
+        origin="position",
+    )
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="builtin-manual",
+        origin="user",
+    )
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="custom-auto",
+        origin="position",
+    )
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="chain-auto",
+        origin="position",
+    )
+
+    store = AgentRepository(sql_session_factory)
+    rows, total = await store.list_paginated(
+        user_id="user-a",
+        scope="mine",
+        search=None,
+        tag_ids=None,
+        page=1,
+        page_size=20,
+    )
+
+    assert total == 2
+    assert {row["id"] for row in rows} == {"builtin-manual", "custom-auto"}
+    assert await store.list_favorite_ids("user-a") == {"builtin-manual", "custom-auto"}
+
+
+@pytest.mark.asyncio
+async def test_manual_favorite_upgrades_existing_auto_builtin_favorite(sql_session_factory) -> None:
+    await _add_sql_agent(sql_session_factory, agent_id="builtin-agent", user_id=None)
+    await _add_sql_favorite(
+        sql_session_factory,
+        user_id="user-a",
+        agent_id="builtin-agent",
+        origin="position",
+    )
+
+    store = AgentRepository(sql_session_factory)
+    assert await store.list_favorite_ids("user-a") == set()
+
+    await store.add_favorite("user-a", "builtin-agent")
+
+    assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
+        ("builtin-agent", "user")
+    }
+    assert await store.list_favorite_ids("user-a") == {"builtin-agent"}
+    rows, total = await store.list_paginated(
+        user_id="user-a",
+        scope="mine",
+        search=None,
+        tag_ids=None,
+        page=1,
+        page_size=20,
+    )
+    assert total == 1
+    assert [row["id"] for row in rows] == ["builtin-agent"]
+
+
 @pytest.mark.asyncio
 async def test_agent_store_only_owner_can_update_or_delete() -> None:
     store = MemoryAgentStore()

offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py

@@ -10,7 +10,9 @@ from types import SimpleNamespace
 import pytest
 
 from deerflow.agents.ai_writing.nodes import researcher as R
+from deerflow.agents.ai_writing.search import get_search_provider
 from deerflow.agents.ai_writing.search.base import SearchResult
+from deerflow.agents.ai_writing.search.intranet_stub import IntranetSearchProvider
 
 
 class _FakeIntranet:
@@ -21,6 +23,7 @@ class _FakeIntranet:
         self.raw_responses = []
         self.calls = []
         self.kw = kw
+        self.verify_ssl = kw.get("verify_ssl", False)
         _FakeIntranet.instances.append(self)
 
     async def search(self, query, max_results=5, **_):
@@ -55,12 +58,24 @@ class _SkillProviderShouldNotRun:
         raise AssertionError("knowledge_base 模式不应调用通用 provider / 技能检索")
 
 
-def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet):
+class _FakeResponse:
+    status_code = 200
+    text = '[{"title":"内网标题","content":"内网正文","url":"https://intranet/doc"}]'
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self):
+        return [{"title": "内网标题", "content": "内网正文", "url": "https://intranet/doc"}]
+
+
+def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet, intranet_verify_ssl=False):
     _FakeIntranet.instances = []
     cfg = {
         "max_materials": 20,
         "intranet_search_url": intranet_url,
         "intranet_knowledge_base": "samples",
+        "intranet_verify_ssl": intranet_verify_ssl,
     }
     monkeypatch.setattr(R, "get_ai_writing_config", lambda: cfg)
     monkeypatch.setattr(R, "use_builtin_agent", lambda *_: False)
@@ -125,6 +140,15 @@ async def test_knowledge_base_no_url_falls_back_to_default_endpoint(monkeypatch)
     assert _FakeIntranet.instances[0].kw.get("base_url") == "", "空地址应原样下传，由 provider 回退默认端点"
 
 
+@pytest.mark.asyncio
+async def test_knowledge_base_passes_verify_ssl_config(monkeypatch):
+    _patch(monkeypatch, intranet_verify_ssl=True)
+    out = await R.researcher_node(_state(), {"configurable": {}})
+
+    assert _FakeIntranet.instances[0].kw.get("verify_ssl") is True
+    assert out["material_package"]["intranet_debug"]["verify_ssl"] is True
+
+
 @pytest.mark.asyncio
 async def test_knowledge_base_does_not_orchestrate_skills(monkeypatch):
     """knowledge_base 不编排技能：通用 provider 一旦被调用就会 raise。"""
@@ -163,3 +187,44 @@ async def test_knowledge_base_failure_is_visible(monkeypatch):
     steps = pkg["skill_search_steps"]
     assert any(step["type"] == "skill_error" for step in steps)
     assert any(event["type"] == "notice" and event.get("raw") for event in out["progress_events"])
+
+
+@pytest.mark.asyncio
+async def test_intranet_search_disables_ssl_verification_for_internal_https(monkeypatch):
+    import httpx
+
+    seen: dict = {}
+
+    class _FakeAsyncClient:
+        def __init__(self, **kwargs):
+            seen.update(kwargs)
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return None
+
+        async def post(self, endpoint, json):
+            seen["endpoint"] = endpoint
+            seen["payload"] = json
+            return _FakeResponse()
+
+    monkeypatch.setattr(httpx, "AsyncClient", _FakeAsyncClient)
+
+    provider = IntranetSearchProvider(base_url="https://intranet.example/search", knowledge_base_name="kb", timeout=300)
+    results = await provider.search("测试", max_results=1)
+
+    assert results
+    assert seen["verify"] is False
+    assert seen["trust_env"] is False
+    assert seen["endpoint"] == "https://intranet.example/search"
+    assert seen["payload"]["knowledge_base_name"] == "kb"
+    assert provider.calls[0]["verify_ssl"] is False
+
+
+def test_intranet_verify_ssl_can_be_reenabled_from_config():
+    provider = get_search_provider({"search_provider": "intranet", "intranet_verify_ssl": "true"})
+
+    assert isinstance(provider, IntranetSearchProvider)
+    assert provider.verify_ssl is True

skills-data/xdfx-full-flow-skill-guide.md

@@ -194,7 +194,8 @@ Step1 受众目标分析完成：当前已入库受众 X 个，本步骤新增
 - 单个受众、账号、传播节点、脆弱点或策略失败，不中断整个流程。
 - 保存接口必须返回 `id`，否则不会进入下游步骤。
 - 受众目标默认入库 3 个，关键传播节点默认入库 3 个。
-- 下游脆弱点/策略默认只选择前 2 个受众和前 2 个传播节点继续，可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- 下游受众脆弱点/策略只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择，默认取前 2 个；没有产生传播节点的受众不得继续分析脆弱点和策略。
+- 下游传播节点脆弱点/策略默认只选择前 2 个已入库传播节点继续，可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
 - 每个被选中的对象默认最多生成 3 个脆弱点、3 个策略。
 - 模型流式 `answer` 会自动拼接。
 - 流式模型接口第一次返回空响应、`y模型没有响应`、`模型没有响应` 或超时类字符串时，会自动再调用一次；仍失败才写入 `errors`。

skills-data/xdfx-single-agent-full-flow/SKILL.md

@@ -86,7 +86,8 @@ python scripts/xdfx_flow.py --mode full --input xdfx-input.json --dry-run
 
 - 受众目标默认入库最优 3 个，可用 `--max-audiences` 调整。
 - 关键传播节点全局去重后默认入库最优 3 个，可用 `--max-pathways` 调整。
-- 下游脆弱点/策略默认只从已入库结果中选择前 2 个受众、前 2 个传播节点继续，可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- 下游受众脆弱点/策略只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择，默认取前 2 个；没有产生传播节点的受众不得继续分析脆弱点和策略。
+- 下游传播节点脆弱点/策略默认从已入库传播节点中选择前 2 个继续；可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
 - 每个被选中的受众/传播节点默认生成最多 3 个脆弱点，可用 `--max-vulns` 调整。
 - 每个被选中的受众/传播节点默认生成最多 3 个策略，可用 `--max-strategies` 调整。
 - 每个保存接口必须拿到返回 `id` 才进入下游。

skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-split-controller/SKILL.md

@@ -77,8 +77,9 @@ python scripts/xdfx_run_full_steps.py --input xdfx-input.json --state xdfx-flow-
 - Step1 的受众目标抽取结果里，`用户编号列表` / `编号id` / `id` 必须对应上一轮 `zhanghao` 账号数组的索引值；脚本会按这个索引依次找到账号并保存受众账号关系。
 - Step2 成功后会写入 `topPathways`。
 - Step1 默认入库最优 3 个受众；Step2 默认入库最优 3 个传播节点。
-- Step3/Step4 默认只选择前 2 个已入库受众/传播节点继续，每个生成最多 3 个脆弱点；可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- Step3/Step5 的受众下游分析只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择，默认取前 2 个；没有产生传播节点的受众不得继续分析脆弱点和策略。
-- Step5/Step6 默认对上述选中的每个受众/传播节点生成最多 3 个策略。
+- Step4/Step6 的传播节点下游分析默认从已入库传播节点中选择前 2 个继续；可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- Step3-Step6 对上述选中的每个受众/传播节点默认生成最多 3 个脆弱点或策略。
 - 各步骤会同步前端 `task/task-action.step`；需要禁用时每条命令追加 `--no-step-status`。
 - 策略步骤会同步写入 channel、narrate、message、策略与受众/传播节点关系表，保证原前端策略表格/编排页可继续读取。
 - 策略步骤保存 `信息形式/模态` 时，必须使用脚本读取 `sys/sys-dict-data` 的 `modelType` 字典并把文字转成 value 逗号串，不能把“图片,视频”等文字直接写入 `task-action-message.modelType`。

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

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-split-flow-skill-guide.md

@@ -134,8 +134,9 @@ python scripts/xdfx_run_full_steps.py --input xdfx-input.json --state xdfx-flow-
 
 - Step1 入库最优 3 个受众。
 - Step2 入库最优 3 个传播节点。
-- Step3/Step4 只从已入库结果中默认选择前 2 个受众/传播节点继续，每个生成最多 3 个脆弱点；可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- Step3/Step5 的受众下游分析只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择，默认取前 2 个；没有产生传播节点的受众不得继续分析脆弱点和策略。
-- Step5/Step6 对上述选中的每个受众/传播节点默认生成最多 3 个策略。
+- Step4/Step6 的传播节点下游分析默认从已入库传播节点中选择前 2 个继续；可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。
+- Step3-Step6 对上述选中的每个受众/传播节点默认生成最多 3 个脆弱点或策略。
 - Step5/Step6 的策略模态必须按前端 `modelType` 字典保存为 value 逗号串；脚本会自动把“图片,视频/文本,图片”等文字转成 `task-action-message.modelType` 需要的值。
 
 总控反馈用户时建议使用这种格式：

skills-data/xdfx-step1-audience/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None:

skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py

@@ -368,6 +368,7 @@ def default_state() -> dict[str, Any]:
         "topAudiences": [],
         "pathwayCandidates": [],
         "topPathways": [],
+        "audiencesWithPathways": [],
         "audienceVulns": {},
         "pathwayVulns": {},
         "audienceStrategies": [],
@@ -2055,6 +2056,88 @@ def selected_accounts_for_item(item: dict[str, Any], *, audience: bool) -> list[
     return [item.get("account") or item]
 
 
+def id_ref_key(value: Any) -> str:
+    number = as_int(value)
+    if number is not None:
+        return str(number)
+    return str(value or "").strip()
+
+
+def audience_name_key(value: Any) -> str:
+    return re.sub(r"\s+", "", str(value or "")).lower()
+
+
+def compact_pathway_ref(pathway: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "pathwayAccountId": pathway.get("pathwayAccountId") or pathway.get("pathwayId"),
+        "accountId": pathway.get("accountId"),
+        "accountName": pathway.get("accountName"),
+    }
+
+
+def add_unique_pathway(target: list[dict[str, Any]], pathway: dict[str, Any]) -> None:
+    key = id_ref_key(pathway.get("pathwayAccountId") or pathway.get("pathwayId") or pathway.get("accountId") or pathway.get("accountName"))
+    if not key:
+        return
+    seen = {
+        id_ref_key(item.get("pathwayAccountId") or item.get("pathwayId") or item.get("accountId") or item.get("accountName"))
+        for item in target
+    }
+    if key not in seen:
+        target.append(pathway)
+
+
+def audience_pathway_pairs(state: dict[str, Any]) -> list[tuple[dict[str, Any], list[dict[str, Any]]]]:
+    """Return only audiences that produced at least one saved key pathway."""
+    id_links: dict[str, list[dict[str, Any]]] = {}
+    name_links: dict[str, list[dict[str, Any]]] = {}
+    for pathway in state.get("topPathways") or []:
+        if not isinstance(pathway, dict):
+            continue
+        for design_id in pathway.get("designIds") or []:
+            key = id_ref_key(design_id)
+            if key:
+                id_links.setdefault(key, []).append(pathway)
+        for name in pathway.get("relatedAudiences") or []:
+            key = audience_name_key(name)
+            if key:
+                name_links.setdefault(key, []).append(pathway)
+
+    pairs: list[tuple[dict[str, Any], list[dict[str, Any]]]] = []
+    for audience_item in state.get("topAudiences") or []:
+        if not isinstance(audience_item, dict):
+            continue
+        linked: list[dict[str, Any]] = []
+        design_key = id_ref_key(audience_item.get("designId"))
+        for pathway in id_links.get(design_key, []):
+            add_unique_pathway(linked, pathway)
+        name_key = audience_name_key(audience_item.get("targetAudience"))
+        for pathway in name_links.get(name_key, []):
+            add_unique_pathway(linked, pathway)
+        if linked:
+            item = {**audience_item, "downstreamPathways": [compact_pathway_ref(pathway) for pathway in linked]}
+            pairs.append((item, linked))
+    return pairs
+
+
+def audience_pathway_summary(state: dict[str, Any]) -> list[dict[str, Any]]:
+    summary: list[dict[str, Any]] = []
+    for audience_item, pathways in audience_pathway_pairs(state):
+        summary.append(
+            {
+                "designId": audience_item.get("designId"),
+                "targetAudience": audience_item.get("targetAudience"),
+                "pathwayCount": len(pathways),
+                "pathways": [compact_pathway_ref(pathway) for pathway in pathways],
+            }
+        )
+    return summary
+
+
+def has_audiences_with_pathways(state: dict[str, Any]) -> bool:
+    return bool(audience_pathway_pairs(state))
+
+
 def downstream_limit(ctx: FlowContext, *, audience: bool) -> int:
     """Limit expensive downstream analysis while keeping 3 upstream rows stored."""
     raw = ctx.downstream_audiences if audience else ctx.downstream_pathways
@@ -2066,22 +2149,35 @@ def downstream_items(ctx: FlowContext, state: dict[str, Any], stage: str, *, aud
     source_key = "topAudiences" if audience else "topPathways"
     id_key = "designId" if audience else "pathwayAccountId"
     selected_key = "selectedDownstreamAudiences" if audience else "selectedDownstreamPathways"
-    items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    all_items = [item for item in (state.get(source_key) or []) if isinstance(item, dict)]
+    if audience:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
+        items = [item for item, _ in audience_pathway_pairs(state)]
+        if not items:
+            state[selected_key] = []
+            record_error(state, stage, "没有已分析出关键传播节点的受众，不能继续分析受众脆弱点或策略；请先重试关键传播节点分析")
+            return []
+    else:
+        items = all_items
     limit = downstream_limit(ctx, audience=audience)
     selected = items[:limit]
-    state[selected_key] = [
+    selected_rows: list[dict[str, Any]] = []
-        {
+    for item in selected:
+        row = {
             "id": item.get(id_key),
             "name": item.get("targetAudience") if audience else item.get("accountName"),
         }
-        for item in selected
+        if audience:
-    ]
+            row["pathwayCount"] = len(item.get("downstreamPathways") or [])
-    if items:
+            row["pathways"] = item.get("downstreamPathways") or []
+        selected_rows.append(row)
+    state[selected_key] = selected_rows
+    if selected:
         record_event(
             state,
             stage,
-            f"下游分析选择 {len(selected)}/{len(items)} 个{'受众' if audience else '传播节点'}",
+            f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}",
-            {"limit": limit, "selected": state[selected_key]},
+            {"limit": limit, "selected": state[selected_key], "candidateCount": len(items), "totalStoredCount": len(all_items)},
         )
     return selected
 
@@ -2183,6 +2279,7 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         record_error(state, stage, "没有已入库受众，无法分析传播节点")
         return
     if state.get("topPathways") and not ctx.dry_run:
+        state["audiencesWithPathways"] = audience_pathway_summary(state)
         record_event(state, stage, "topPathways 已存在，跳过。如需重跑请清空 state 中该字段。")
         return
     all_candidates: list[dict[str, Any]] = []
@@ -2278,9 +2375,10 @@ def run_step2(ctx: FlowContext, state: dict[str, Any]) -> None:
         http_post_json(ctx, state, "save_relation", relation_payload, stage + "/save-relation")
         saved.append(item)
     state["topPathways"] = saved
+    state["audiencesWithPathways"] = audience_pathway_summary(state)
     if saved:
         mark_step_status(ctx, state, ["step-analysis-keypoint", "step-analysis-weakpoint"], stage + "/step-status")
-    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个")
+    record_event(state, stage, f"完成，保存传播节点 {len(saved)} 个，可继续下游分析的受众 {len(state.get('audiencesWithPathways') or [])} 个")
 
 
 def run_step3(ctx: FlowContext, state: dict[str, Any]) -> None: