diff --git a/frontend-web/src/open-canvas/pages/AIWritingPage.tsx b/frontend-web/src/open-canvas/pages/AIWritingPage.tsx index 4c56e940..2cd88cd4 100644 --- a/frontend-web/src/open-canvas/pages/AIWritingPage.tsx +++ b/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 ? ( ) : null; diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md index 1cb9bcd3..b8eba50b 100644 --- a/offline-backend-20260512/backend/CLAUDE.md +++ b/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//`; `_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 "检不出". diff --git a/offline-backend-20260512/backend/README.md b/offline-backend-20260512/backend/README.md index 40e7e03b..229d1cf0 100644 --- a/offline-backend-20260512/backend/README.md +++ b/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. diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py index b53b5955..a91d8e0e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py +++ b/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] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/__init__.py index 68a0f36f..f58de660 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/__init__.py +++ b/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()) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py index 3f5fcce1..4398933d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py +++ b/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", - self.endpoint, self.knowledge_base_name, self.timeout, query, + "[intranet-search] POST endpoint=%s kb=%s timeout=%ss verify_ssl=%s query=%r", + 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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py index 4eeddb1f..c1e47b44 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py @@ -6,20 +6,21 @@ 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 from deerflow.config.agent_name_desensitize import source_terms_for_needle from deerflow.persistence.agents.base import AgentStore -from deerflow.persistence.agents.model import ( - AgentExtrasRow, - AgentFavoriteRow, - AgentPinRow, - AgentRow, - AgentSkillRow, -) -from deerflow.persistence.tags.model import TagAssignmentRow +from deerflow.persistence.agents.model import ( + AgentExtrasRow, + AgentFavoriteRow, + AgentPinRow, + AgentRow, + AgentSkillRow, +) +from deerflow.persistence.roundtable_chains.model import RoundtableChainRow +from deerflow.persistence.tags.model import TagAssignmentRow def _row_to_dict(row: AgentRow) -> dict[str, Any]: @@ -32,9 +33,33 @@ def _row_to_dict(row: AgentRow) -> dict[str, Any]: return data -class AgentRepository(AgentStore): - def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: - self._sf = session_factory +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): @@ -52,15 +77,23 @@ class AgentRepository(AgentStore): ``published=false``, so this clause cannot lean on ``published``; - 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) - 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 = ( + 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), + valid_favorite_exists, + ) @staticmethod def _listing_filter(user_id: str): @@ -268,11 +301,20 @@ class AgentRepository(AgentStore): or_(AgentRow.user_id.is_(None), AgentRow.published.is_(True)), AgentRow.is_private.is_(False), ) - if scope == "mine": - # owned OR favorited. A correlated IN-subquery (not a JOIN) keeps a - # single row per agent even when it is both owned and favorited. - fav_subq = select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id) - return or_(AgentRow.user_id == user_id, AgentRow.id.in_(fav_subq)) + if scope == "mine": + # owned OR valid favorites. Built-in agents and business-chain + # seats only enter "mine" when explicitly favorited by the user; + # the original position-granted behavior stays for ordinary agents. + 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": @@ -437,18 +479,22 @@ class AgentRepository(AgentStore): # --- Favorites ("收藏") --- - async def add_favorite(self, user_id: str, agent_id: str) -> None: - async with self._sf() as session: - existing = ( - await session.execute( - select(AgentFavoriteRow.id) - .where(AgentFavoriteRow.user_id == user_id, AgentFavoriteRow.agent_id == agent_id) - .limit(1) - ) - ).scalar_one_or_none() - if existing is not None: - return - session.add(AgentFavoriteRow(id=uuid4().hex, user_id=user_id, agent_id=agent_id)) + async def add_favorite(self, user_id: str, agent_id: str) -> None: + async with self._sf() as session: + existing = ( + await session.execute( + 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: await session.commit() except IntegrityError: @@ -467,12 +513,17 @@ class AgentRepository(AgentStore): await session.commit() return (result.rowcount or 0) > 0 - 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) - ) - return {row[0] for row in result.all()} + 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) + .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()} async def delete_favorites_for_agent(self, agent_id: str) -> None: async with self._sf() as session: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260701_01_agent_position_favorites.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260701_01_agent_position_favorites.py new file mode 100644 index 00000000..264a12bf --- /dev/null +++ b/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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py b/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py index 2a6e86e8..23b4a8d9 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/services/position_sync.py +++ b/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 -``agent`` → ``agent_favorites``: the tagged skills are *added* to the user's -"我的技能" while the browse-all 广场 stays complete (it is **not** restricted). +clobbered. ``agent`` keeps the original position-granted behavior for ordinary +tagged agents, but filters out built-ins and business-chain seats; those only +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)) - return await self._default_set(resource_type, session) + target = set(await self._tags.list_targets_for_tags(resource_type, tag_ids)) + 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: diff --git a/offline-backend-20260512/backend/tests/test_agents_backend.py b/offline-backend-20260512/backend/tests/test_agents_backend.py index 57345641..f87df609 100644 --- a/offline-backend-20260512/backend/tests/test_agents_backend.py +++ b/offline-backend-20260512/backend/tests/test_agents_backend.py @@ -1,33 +1,131 @@ -from __future__ import annotations - -import pytest -import yaml - -from app.gateway.routers.agents import ( - _backfill_listing_cache, - _load_config_extras, - _write_agent_config, -) -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 __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, + _load_config_extras, + _write_agent_config, +) +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 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 -def isolated_paths(tmp_path, monkeypatch): - """Redirect get_paths() at a fresh temp directory for the duration of one test.""" - monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) - monkeypatch.setattr(paths_module, "_paths", None) - yield tmp_path - monkeypatch.setattr(paths_module, "_paths", None) +def isolated_paths(tmp_path, monkeypatch): + """Redirect get_paths() at a fresh temp directory for the duration of one test.""" + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + monkeypatch.setattr(paths_module, "_paths", None) + yield tmp_path + 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() - chinese_name = "\u4e2d\u6587\u52a9\u624b" - +async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None: + store = MemoryAgentStore() + chinese_name = "\u4e2d\u6587\u52a9\u624b" + await store.ensure_builtin({"id": "builtin-agent", "name": chinese_name, "description": "built in"}) await store.create({"id": "private-a", "name": chinese_name, "description": "private", "user_id": "user-a"}) await store.create({"id": "published-b", "name": chinese_name, "description": "published", "user_id": "user-b", "published": True}) @@ -35,14 +133,144 @@ async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility user_a_ids = {agent["id"] for agent in await store.list_visible("user-a")} assert user_a_ids == {"builtin-agent", "private-a", "published-b"} - user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")} - assert user_c_ids == {"builtin-agent", "published-b"} - - -@pytest.mark.asyncio -async def test_agent_store_only_owner_can_update_or_delete() -> None: - store = MemoryAgentStore() - + user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")} + 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() + await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"}) await store.create({"id": "private-a", "name": "Owned", "user_id": "user-a"}) diff --git a/offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py b/offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py index 01ed878f..2dadb6f3 100644 --- a/offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py +++ b/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 diff --git a/skills-data/xdfx-full-flow-skill-guide.md b/skills-data/xdfx-full-flow-skill-guide.md index 014a6300..ec7ed043 100644 --- a/skills-data/xdfx-full-flow-skill-guide.md +++ b/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`。 diff --git a/skills-data/xdfx-single-agent-full-flow/SKILL.md b/skills-data/xdfx-single-agent-full-flow/SKILL.md index 2cdce48e..b0a45773 100644 --- a/skills-data/xdfx-single-agent-full-flow/SKILL.md +++ b/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` 才进入下游。 diff --git a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py b/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-single-agent-full-flow/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-split-controller/SKILL.md b/skills-data/xdfx-split-controller/SKILL.md index 1ee180ab..71ad1d51 100644 --- a/skills-data/xdfx-split-controller/SKILL.md +++ b/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。 -- Step5/Step6 默认对上述选中的每个受众/传播节点生成最多 3 个策略。 +- Step3/Step5 的受众下游分析只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择，默认取前 2 个；没有产生传播节点的受众不得继续分析脆弱点和策略。 +- 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`。 diff --git a/skills-data/xdfx-split-controller/scripts/xdfx_flow.py b/skills-data/xdfx-split-controller/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-split-controller/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-split-flow-skill-guide.md b/skills-data/xdfx-split-flow-skill-guide.md index 62d52a6e..f09c6203 100644 --- a/skills-data/xdfx-split-flow-skill-guide.md +++ b/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。 -- Step5/Step6 对上述选中的每个受众/传播节点默认生成最多 3 个策略。 +- Step3/Step5 的受众下游分析只能从“已在 Step2 分析出并关联关键传播节点”的受众中选择，默认取前 2 个；没有产生传播节点的受众不得继续分析脆弱点和策略。 +- Step4/Step6 的传播节点下游分析默认从已入库传播节点中选择前 2 个继续；可用 `--downstream-audiences` / `--downstream-pathways` 调成 1 或 2。 +- Step3-Step6 对上述选中的每个受众/传播节点默认生成最多 3 个脆弱点或策略。 - Step5/Step6 的策略模态必须按前端 `modelType` 字典保存为 value 逗号串；脚本会自动把“图片,视频/文本,图片”等文字转成 `task-action-message.modelType` 需要的值。 总控反馈用户时建议使用这种格式： diff --git a/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py b/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step1-audience/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py b/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step2-pathway/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py b/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step3-audience-vulns/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py b/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step4-pathway-vulns/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py b/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step5-audience-strategy/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: diff --git a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py b/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py index d9580e2f..8e62e51e 100644 --- a/skills-data/xdfx-step6-pathway-strategy/scripts/xdfx_flow.py +++ b/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 items: + if audience: + row["pathwayCount"] = len(item.get("downstreamPathways") or []) + 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 '传播节点'}", - {"limit": limit, "selected": state[selected_key]}, + f"下游分析选择 {len(selected)}/{len(items)} 个{'已产生传播节点的受众' if audience else '传播节点'}", + {"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: