feat(demand): write validated demand content to mysql

demandagent给下游的迭代需求.md

@@ -2,9 +2,11 @@
 
 ## 1. 结论先行
 
-新版 ContentFindAgent 不能只靠旧版反查机制来恢复 Pattern、Case 和分类树血缘。
+新版 ContentFindAgent 不能再依赖旧 MySQL Pattern 反查机制来恢复 Pattern、Case 和分类树血缘。
 
-旧版能拿到一些候选线索，例如用需求词去查历史 Pattern 元素，再回查 Case 解构点。但这条链路不是精确血缘，也不能稳定闭合到唯一的 Pattern、itemset 或 Case。新版 DemandAgent 的正式证据源改为 PG `open_aigc.public` Pattern V2 主事实表。
+旧 MySQL `topic_pattern_*` 链路已删除弃用，不进入新版 `evidence_pack`，不作为下游回溯依据。旧字段只允许作为迁移前后的字段对照出现，用来说明它们在 PG Pattern V2 中被哪些字段替换。
+
+新版 DemandAgent 的唯一正式 Pattern 证据源是 PG `open_aigc.public` Pattern V2 主事实表。DemandAgent 输出的 `pattern_source_system` 唯一合法值是 `pg_pattern_v2`。
 
 为了新版能从需求出发稳定跑通数据源、Query、判断、游走、资产沉淀和策略学习，DemandAgent 或上游证据层需要把生成需求时已经知道的证据一起传给下游。
 
@@ -14,22 +16,22 @@
 
 | 字段 | 作用 |
 | --- | --- |
-| `pattern_source_system` | 固定为 `pg_pattern_v2`，标记 Pattern 证据来自 PG Pattern V2 主事实 |
+| `pattern_source_system` | DemandAgent 输出中固定且唯一合法值为 `pg_pattern_v2`，标记 Pattern 证据来自 PG Pattern V2 主事实 |
 | `pattern_execution_id` | 标记需求来自哪次 PG `pattern_mining_execution` |
 | `mining_config_id` | 标记来源 Pattern 使用的具体挖掘配置 |
 | `source_kind` | 标记需求来源是聚类结果、Pattern 频繁项集、直接 Case，还是其他来源 |
 | `case_id_type` | 标准化 `caseid` 的语义，例如本链路中应明确为 `post_id` |
 | `source_post_id` | 标记本条需求证据中的原始支撑帖子 |
-| `category_bindings` | 绑定分类节点，例如分类 ID、路径、层级、父节点、实质/形式/意图 |
+| `category_bindings` | 绑定分类节点，例如分类 ID、路径、层级、来源类型、点类型和维度 |
 | `element_bindings` | 绑定具体元素，例如元素名、元素类型、点类型、所属 post |
 | `itemset_ids` | 绑定 Pattern 频繁项集 |
 | `itemset_items[]` | 绑定 itemset 内每个 item 对应的分类节点、路径、点类型和维度 |
 | `support` / `absolute_support` | 记录 Pattern 支持度，判断它是不是稳定组合 |
 | `matched_post_ids` | 记录支撑 Pattern 的原始帖子或视频 |
 | `video_ids` / `case_ids` | 记录可作为 Case seed 的素材 ID；Pattern 来源下二者都是 `post_id` 语义 |
-| `decode_case_ids` | 可选补充 decode 表行 ID，命中不到时可为空 |
+| `decode_case_ids` | PG-only 当前固定为空数组的兼容字段，不参与 Pattern exact 闭合 |
 | `seed_terms` | 记录真正给 Query 使用的下层特征或元素词 |
-| `source_evidence` | ContentFindAgent 最终入库时继承的结构化来源证据 |
+| `source_evidence` | ContentFindAgent 从 `evidence_pack` 继承并追加运行态字段后形成的下游结果证据，不由 DemandAgent 直接输出 |
 | `source_certainty` / `validation_status` | 标记证据是否已经 DB 强校验通过 |
 | `trace_id` / `demand_task_id` / `demand_content_id` | 串起需求生成、内容寻找、判断、沉淀和策略学习 |
 
@@ -67,7 +69,7 @@ MVP 边界：DemandAgent 只输出 PG `pattern_itemset.scope='topic'` 的分类
 
 ### 3.1 DemandItem
 
-DemandAgent 当前生成的 `DemandItem` 字段很窄。
+DemandAgent 当前生成的 `DemandItem` 是 LLM / tool 产生的候选业务需求，不是最终给下游消费的强校验证据。
 
 代码证据：
 
@@ -82,20 +84,21 @@ DemandAgent 当前生成的 `DemandItem` 字段很窄。
 | `reason` | 生成需求的理由 |
 | `desc` | 需求描述，后续写入 `suggestion` |
 | `type` | 来源类型，例如元素、关系、分类 |
+| `evidence_refs` | LLM 声明的证据引用候选，只能作为 DB 强校验输入，不能直接给下游当真实证据 |
 
-这里没有保留 `execution_id`、分类 ID、元素 ID、itemset、support 或 matched posts。
+`evidence_refs` 里即使出现 `itemset_ids/source_post_id/seed_terms`，也只是声明。DemandAgent 必须用 PG DB 重新读取和校验，再升级成 `demand_content.ext_data.evidence_pack`。任一最终字段来自旧 MySQL `topic_pattern_*` 时，该 DemandItem 必须 rejected。
 
 ### 3.2 demand_content
 
-`write_demand_items_to_mysql()` 会把 DemandItem 写入 `demand_content`。
+新版 local MVP 中，DemandAgent 不真实写 `demand_content` DB，而是写本地 `demand_content.json`。未来恢复生产写库时，正式给下游的载体仍是 `demand_content.ext_data.evidence_pack`。
 
 代码证据：
 
-- `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/run.py:330`
-- `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/run.py:590`
-- `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/run.py:635`
+- `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/run.py`
+- `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/evidence_pack_builder.py`
+- `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/local_output_sink.py`
 
-当前写入字段：
+当前本地 `demand_content.json` 字段：
 
 | `demand_content` 字段 | 来源 |
 | --- | --- |
@@ -104,18 +107,18 @@ DemandAgent 当前生成的 `DemandItem` 字段很窄。
 | `reason` | DemandItem.reason |
 | `suggestion` | DemandItem.desc |
 | `score` | 按 `name` 拆分后查权重并求平均 |
-| `ext_data` | JSON，当前包含 `reason`、`desc`、`type`、`video_ids` |
+| `ext_data` | JSON 对象，包含旧业务字段 `reason/desc/type/video_ids`，以及新版强校验字段 `evidence_pack` |
 | `dt` | 当天日期 |
 
 注意：这里的字段名是历史真实表字段 `merge_leve2`，不是 `merge_level2`。CLI 或函数参数可以使用 `merge_level2` 作为变量名，但落到 `demand_content`、本地 JSON、下游读取契约时必须保持 `merge_leve2`。
 
-`video_ids` 在旧逻辑中主要通过 `execution_id + name` 反查出来；本 MVP 对 Pattern itemset 来源会优先采用 DB 校验后的 `evidence_pack.video_ids/matched_post_ids`。相关代码在 `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/run.py:257` 和 `/Users/samlee/Documents/works/DemandAgentNew/examples/demand/db_manager.py:41`。
+`video_ids` 在新版 PG-only 链路中来自 DB 校验后的 `evidence_pack.video_ids/matched_post_ids`。`pattern_execution_id` 指 PG `pattern_mining_execution.id`，不是旧 MySQL execution，也不是 DemandAgent 本地 run id。
 
-但写入 `demand_content` 后，`execution_id` 自身没有留下。下游只看到 `video_ids`，看不到这些 video 是从哪个 Pattern execution、哪个分类节点、哪个 itemset 推出来的。
+真实 DB 中 `demand_content.ext_data` 是 MySQL JSON 字段，可以承载完整 `evidence_pack`；但当前生产样例仍多为 `reason/desc/type/video_ids`，还没有稳定落下新版 PG `evidence_pack`。这就是本次需求要补齐的输出契约。
 
 ### 3.3 dwd_multi_demand_pool_di
 
-DemandAgent 还会尽力写入 `loghubods.dwd_multi_demand_pool_di`。
+新版 local MVP 不写 Hive，只生成本地 `dwd_multi_demand_pool_di.json` 镜像。这个文件不是给 ContentFindAgent 做 exact 回溯的主输入。
 
 代码证据：
 
@@ -137,9 +140,9 @@ DemandAgent 还会尽力写入 `loghubods.dwd_multi_demand_pool_di`。
 | `extend` | 当前主要保留品类 |
 | `dt` | 分区日期 |
 
-这一层也没有保留 Pattern execution、itemset、分类路径、support 或 matched posts。
+这一层不增加 `evidence_pack`，不承载 Pattern exact evidence。ContentFindAgent 的主输入只能是 `demand_content.json` 或未来 DB `demand_content.ext_data.evidence_pack`。
 
-## 4. 为什么旧版反查不够
+## 4. 旧 MySQL Pattern 链路为什么删除弃用
 
 旧版 ContentFindAgent 的 Case 反查工具是 `get_goodcase_topic_point`。
 
@@ -149,7 +152,7 @@ DemandAgent 还会尽力写入 `loghubods.dwd_multi_demand_pool_di`。
 - `/Users/samlee/Documents/works/ContentFindAgent/examples/content_finder/tools/get_goodcase_topic_point.py:96`
 - `/Users/samlee/Documents/works/ContentFindAgent/examples/content_finder/tools/get_goodcase_topic_point.py:144`
 
-旧链路是：
+旧链路曾经是：
 
 ```text
 features
@@ -159,7 +162,7 @@ features
 -> purpose_points / key_points / inspiration_points
 ```
 
-这条链路有三个问题。
+这条链路已删除弃用，原因如下。
 
 | 问题 | 说明 |
 | --- | --- |
@@ -167,40 +170,61 @@ features
 | 只查元素名 | DemandAgent 当前有些需求名其实是分类节点，旧 CFA 只查 `topic_pattern_element.name` 会漏掉 |
 | 不查 itemset | 无法恢复 `itemset_id`、support、absolute support、matched posts |
 
-DemandAgent 自己在生成 `video_ids` 时，用的是 `execution_id + name/category`。旧 CFA 事后反查用的是裸 `name`。两条链路不是同一种机制。
+更关键的是，新版 DemandAgent 已切换为 PG Pattern V2 主事实。旧 MySQL `topic_pattern_element` 即使能查到 post，也不是新版 `evidence_pack` 的合法来源。
+
+因此旧链路不能作为任何兜底链路写入新版主血缘。它只保留在本文档中作为“删除弃用字段如何被 PG 字段替换”的迁移说明。
 
-所以旧版反查只能作为 fallback，用来找候选 Case；不能作为新版的主血缘。
+## 5. 旧 MySQL 字段删除弃用与 PG 替换关系
 
-## 5. 必须补给下游的字段清单
+| 旧 MySQL 表/字段 | 新状态 | 新 PG 替换来源 |
+| --- | --- | --- |
+| `topic_pattern_execution.id` | 删除弃用 | `pattern_mining_execution.id`，即新版 `pattern_execution_id` |
+| `topic_pattern_execution.status` | 删除弃用 | `pattern_mining_execution.status` |
+| `topic_pattern_execution.cluster_name/merge_leve2/platform/account_name/post_limit/min_absolute_support/mining_configs` | 删除弃用为 Pattern 证据字段 | PG execution 是全局 Pattern 快照；业务侧 `merge_leve2/platform` 只保留在 Demand 输出上下文，不作为 Pattern 主事实 |
+| `topic_pattern_mining_config.id` | 删除弃用 | `pattern_mining_config.id`，即 `mining_config_id` |
+| `topic_pattern_mining_config.dimension_mode/target_depth` | 删除弃用 | 从 `pattern_mining_config.params` 解析或派生 |
+| `topic_pattern_category.id` | 删除弃用 | `pattern_mining_category.id` |
+| `topic_pattern_category.path` | 删除弃用 | `pattern_mining_category.path` |
+| `topic_pattern_category.category_nature` | 删除弃用 | `pattern_mining_category.category_nature` 和 `pattern_mining_category.classified_as` |
+| `topic_pattern_element.id/post_id/point_type/point_text/element_type/name/description/category_id/category_path` | 删除弃用 | `pattern_mining_element.id/post_id/point_type/point_text/element_type/name/description/category_id/category_path` |
+| `topic_pattern_itemset.id` | 删除弃用 | `pattern_itemset.id` |
+| `topic_pattern_itemset.mining_config_id` | 删除弃用 | `pattern_itemset.mining_config_id` |
+| `topic_pattern_itemset.support/absolute_support` | 删除弃用 | `pattern_itemset.support/absolute_support` |
+| `topic_pattern_itemset.matched_post_ids` | 删除弃用 | 从 `pattern_itemset_post` 按 `execution_id + itemset_id` 聚合生成 `matched_post_ids` |
+| `topic_pattern_itemset_item.id/itemset_id/point_type/dimension/category_id/category_path/element_name` | 删除弃用 | `pattern_itemset_item.id/itemset_id/point_type/dimension/category_id/category_path/element_name`，并可补充 PG 新字段 `layer/element_id/post_count` |
+
+新版 `evidence_pack` 中不允许出现旧 MySQL 专属来源。字段名可以保留通用业务含义，例如 `itemset_ids`、`matched_post_ids`、`category_bindings`，但它们的值必须来自 PG Pattern V2。
+
+## 6. 必须补给下游的字段清单
 
 | 字段 | 字段意义 | 下游用途 | 为什么不能靠 CFA 自己补 | 缺失风险 |
 | --- | --- | --- | --- | --- |
-| `pattern_source_system` | Pattern 证据来源系统，MVP 固定为 `pg_pattern_v2` | 判断反查目标是 PG Pattern V2 主事实 | CFA 只看需求单无法知道证据来自哪套 Pattern 表 | 把历史兼容层或候选反查误当 PG 主事实 |
+| `pattern_source_system` | Pattern 证据来源系统，DemandAgent 输出中唯一合法值为 `pg_pattern_v2` | 判断反查目标是 PG Pattern V2 主事实 | CFA 只看需求单无法知道证据来自哪套 Pattern 表 | 把旧 MySQL 删除弃用链路误当 PG 主事实 |
 | `pattern_execution_id` | 当前需求来源的 Pattern 执行 ID | 数据源详情、Pattern 回查、策略学习 | CFA 只能按词反查，无法知道当时 execution | 同名词跨 execution 混淆 |
-| `mining_config_id` | 当前 itemset 所属挖掘配置 ID | 区分同一 execution 下不同 Pattern 配置 | CFA 事后只靠 post 反查会得到多个配置下的候选 itemset | 无法确定具体 Pattern 配置 |
+| `mining_config_id` | 当前 itemset 所属挖掘配置 ID | 区分同一 execution 下不同 Pattern 配置 | CFA 事后只靠 post 反查会得到多个配置下的 itemset | 无法确定具体 Pattern 配置 |
 | `source_kind` | 来源类型：聚类结果、Pattern 频繁项集、直接 Case 等 | 决定后续流程：是否进 Query，是否先游走或先判断 | 只看 `type` 不够，不能区分真实来源路径 | 不同来源走错流程 |
 | `case_id_type` | `caseid` 的 ID 语义，Pattern case 链路中应明确为 `post_id` | 稳定对齐 `post_id/channel_content_id/aweme_id` | CFA 只能猜这个 ID 是结果表 id、decode id 还是 post id | join 语义混乱 |
-| `source_post_id` | 本条证据里的原始支撑帖子 ID | 从最终 case 回到上游 matched post | 单个 post 可能支撑很多 itemset，CFA 不能知道哪条是来源 post | 只能候选回查，不能 exact |
-| `category_bindings` | 分类节点绑定，包含 `category_id/path/level/parent_id/source_type` | 所有结果绑定回分类树父节点 | CFA 按元素名查不到分类节点，且同名分类可能多处存在 | 入库后无法解释属于哪个分类 |
-| `element_bindings` | 元素绑定，包含 `name/element_type/point_type/post_id/category_path` | 作为 Query 输入素材、判断回扣证据 | CFA 可以按词查候选元素，但不知道哪个是 DemandAgent 采用的 | Query seed 可能跑偏 |
+| `source_post_id` | 本条证据里的原始支撑帖子 ID | 从最终 case 回到上游 matched post | 单个 post 可能支撑很多 itemset，CFA 不能知道哪条是来源 post | 只能得到多条可能路径，不能 exact |
+| `category_bindings` | 分类节点绑定，当前包含 `itemset_id/itemset_item_id/category_id/category_name/category_path/category_full_path/category_source_type/category_level/point_type/dimension/element_name` | 所有结果绑定回分类树节点 | CFA 按元素名查不到分类节点，且同名分类可能多处存在 | 入库后无法解释属于哪个分类 |
+| `element_bindings` | 元素绑定，当前包含 `itemset_id/itemset_item_id/category_id/point_type/dimension/element_name/matched_element_count/matched_post_count/matched_post_ids/sample_elements` | 作为 Query 输入素材、判断回扣证据 | CFA 可以按词查元素，但不知道哪个是 DemandAgent 采用的 | Query seed 可能跑偏 |
 | `itemset_ids` | Pattern 频繁项集 ID | 回查 Pattern 组合和稳定性 | CFA 旧链路不查 itemset | 不能证明 Pattern 是组合，不是散词 |
 | `itemset_items[]` | itemset 内每个 item 对应的 `category_id/category_path/point_type/dimension/element_name` | 直接还原 Pattern chip 到分类树节点 | CFA 只靠 `itemset_id` 还要二次猜 item 和节点 | 不能稳定解释具体命中的分类树节点 |
 | `support` | 相对支持度 | 判断 Pattern 是否稳定 | 需求池当前没有保存 | 无法按稳定性排序或过滤 |
 | `absolute_support` | 绝对支撑素材数 | 判断有多少素材支撑 | 需求池当前没有保存 | 不能区分单例和稳定模式 |
 | `matched_post_ids` | 支撑 itemset 的帖子 ID 列表 | 找 Case、看原始素材、做回扣判断 | `video_ids` 不等价于 itemset 的 matched posts | Case 来源不准，支撑证据不准 |
 | `video_ids` / `case_ids` | 可直接作为 Case seed 的素材 ID；Pattern 来源下固定是 `post_id` 语义 | 数据源 Case、Query 输入、判断证据 | CFA 可偶尔从 decode 表命中，但不稳定 | 如果语义不清，容易把 post_id 和 decode row id 混用 |
-| `decode_case_ids` | 兼容补充字段，PG-only MVP 可为空数组 | 后续如需要再挂 decode 行 ID | 并非所有 Pattern matched post 都需要 decode row ID 才能闭合 | 不能把 decode 缺失误判成 Pattern 证据失败 |
+| `decode_case_ids` | PG-only 当前固定为空数组的兼容字段 | 明确 decode row ID 不参与本次 Pattern exact 闭合 | 本次主链路不读旧 decode-case 表 | 不能把 decode 缺失误判成 Pattern 证据失败 |
 | `seed_terms` | 真正用于 Query 的下层特征或元素词 | Query 输入素材 | `name` 可能是分类、关系或宽泛表达，不一定适合作 Query | Query 输入不稳定 |
-| `source_evidence` | 下游最终结果继承的结构化证据 | 写入 `demand_find_content_result` 或 source edge artifact | DemandAgent 只能给需求证据，无法替下游结果落库 | 最终结果又断在 `process_trace` 文本 |
+| `source_evidence` | 下游最终结果继承的结构化证据，不是 DemandAgent 输出字段 | 写入 `demand_find_content_result` 或 source edge artifact | DemandAgent 只能给 `evidence_pack`，无法替下游结果落库 | 最终结果又断在 `process_trace` 文本 |
 | `source_certainty` | 证据可信度，进入下游时必须为 `db_validated` | 下游判断是否能 exact 回溯 | CFA 无法判断 LLM 声明是否已被 DB 校验 | 把候选证据误当真实证据 |
 | `validation_status` | DemandAgent 证据校验状态，进入需求池时必须为 `passed` | 下游只消费通过校验的任务 | CFA 事后补查无法恢复生成时的拒绝原因 | 失败证据流入下游 |
 | `trace_id` | 贯穿下游任务的 trace | 策略学习、失败复盘 | CFA 可创建自己的 trace，但不能补上游生成过程 | 上下游链路断开 |
 | `demand_task_id` | DemandAgent 侧任务 ID | 回查生成日志和任务状态 | CFA 无法凭需求名可靠找到唯一任务 | 无法定位上游运行 |
 | `demand_content_id` | 下游领取的需求 ID | CFA 主任务绑定 | 这个字段 CFA 已能拿到，但应进入统一证据包 | Trace 不统一 |
 
-## 6. 检查过程与证据
+## 7. 检查过程与证据
 
-### 6.1 DemandAgent 文档审计
+### 7.1 DemandAgent 文档审计
 
 `/Users/samlee/Documents/works/DemandAgent工程文档集/demand-agent-analysis.md:432` 已写明 DemandItem 到 `demand_content` 的映射：
 
@@ -216,7 +240,7 @@ DemandAgent 自己在生成 `video_ids` 时，用的是 `execution_id + name/cat
 
 历史文档曾把 `workflow_decode_task_result` 作为辅助回填逻辑；PG-only MVP 不再依赖它闭合 Pattern 证据。Agent 输出主要靠 Prompt 约束，最终是否进入需求池由 DemandAgent 代码按 PG DB 强校验决定。
 
-### 6.2 DemandAgent 代码审计
+### 7.2 DemandAgent 代码审计
 
 DemandAgent 新版代码里直接读取 PG Pattern V2 的分类、元素、itemset 和支撑帖子字段。
 
@@ -231,7 +255,7 @@ DemandAgent 新版代码里直接读取 PG Pattern V2 的分类、元素、items
 | 表 | 关键字段 |
 | --- | --- |
 | `pattern_mining_execution` | `id`, `status`, `topic_itemset_count`, `post_count`, `category_count`, `element_count` |
-| `pattern_mining_config` | `id`, `execution_id`, `scope`, `params` |
+| `pattern_mining_config` | `id`, `execution_id`, `algorithm`, `scope`, `name`, `params` |
 | `pattern_itemset` | `id`, `execution_id`, `mining_config_id`, `scope`, `support`, `absolute_support`, `dimensions` |
 | `pattern_itemset_post` | `execution_id`, `itemset_id`, `post_id` |
 | `pattern_itemset_item` | `itemset_id`, `point_type`, `dimension`, `category_id`, `category_path`, `element_name` |
@@ -240,7 +264,7 @@ DemandAgent 新版代码里直接读取 PG Pattern V2 的分类、元素、items
 
 也就是说，PG 主事实表已经具备 DemandAgent 输出下游证据链所需字段。新版问题不是“上游没有”，而是 DemandAgent 必须把这些字段校验后写进 `demand_content.ext_data.evidence_pack`，并要求下游继续继承。
 
-### 6.3 旧 ContentFindAgent 代码审计
+### 7.3 旧 ContentFindAgent 代码审计
 
 旧 CFA 的 `get_goodcase_topic_point` 只按 `topic_pattern_element.name` 精确匹配。
 
@@ -252,43 +276,46 @@ FROM topic_pattern_element
 WHERE name = %s
 ```
 
-这个 SQL 不带 `execution_id`、`category_id`、`itemset_id`，也不查 `topic_pattern_category`。因此它无法还原 DemandAgent 生成需求时的真实 Pattern 上下文。
+这个 SQL 不带 `execution_id`、`category_id`、`itemset_id`，也不查分类树节点。更重要的是，它读取的是已删除弃用的旧 MySQL `topic_pattern_element`，不能进入新版 PG `evidence_pack`。
 
-### 6.4 真实 DB 只读抽样
+### 7.4 真实 DB 只读抽样
 
 本轮只做了只读查询，没有写 DB，也没有输出 secret。
 
-DB 侧验证结论：
+PG Pattern V2 主事实表只读校对结论：
 
-| 验证项 | 结果 |
+| PG 表 | 行数 | 字段结论 |
 | --- | --- |
-| `demand_content` 行数 | 29284 |
-| 最近样本 `ext_data` | 有 `reason/desc/type/video_ids` |
-| 最新 10 条需求 | 共 104 个 `video_id` |
-| 这 104 个 `video_id` 命中 `workflow_decode_task_result` | 0 |
-| 这 104 个 `video_id` 命中 `topic_pattern_element.post_id` | 0 |
-| 最近 5000 条非空 `ext_data` | 4920 条有 `video_ids` |
-| distinct `video_id` | 922 |
-| 能命中 `workflow_decode_task_result` 的 `video_id` | 38 |
+| `pattern_mining_execution` | 7 | 有 `id/status/snapshot_date/is_current/is_retired/post_filter/post_count/classify_execution_id/category_count/element_count/topic_itemset_count/topic_element_itemset_count` |
+| `pattern_mining_config` | 45 | 有 `id/execution_id/algorithm/scope/name/params/transaction_count/result_count/input_bucket_key` |
+| `pattern_mining_category` | 25816 | 有 `id/execution_id/source_stable_id/source_type/name/description/path/level/parent_id/category_nature/classified_as` |
+| `pattern_mining_element` | 1950755 | 有 `id/execution_id/post_id/source_table/source_element_id/element_type/name/category_id/category_path/point_type/point_text/topic_point_id` |
+| `pattern_itemset` | 449216 | 有 `id/execution_id/mining_config_id/scope/support/absolute_support/dimensions/is_cross_point/is_cross_layer`，没有 `matched_post_ids` 列 |
+| `pattern_itemset_item` | 1808629 | 有 `id/itemset_id/layer/point_type/dimension/category_id/category_path/element_name/element_id/post_count` |
+| `pattern_itemset_post` | 3910262 | 有 `itemset_id/post_id/execution_id`，是新版 `matched_post_ids` 的唯一来源 |
 
-这说明 `ext_data.video_ids` 可以作为候选 Case seed，但不是稳定闭合通道。
+需求池真实表只读校对结论：
 
-对照样本里，`demand_content.id=48563` 的 12 个 video_id 中有 2 个命中 decode Case；但同一需求词能命中多个 Pattern execution，且没有任何 itemset 的 `matched_post_ids` 与整组 `video_ids` 精确相等。
+| 表 | 行数 | 字段 |
+| --- | --- | --- |
+| `demand_content` | 29888 | `id/merge_leve2/name/reason/suggestion/ext_data/score/dt/create_time/update_time` |
+| `demand_task` | 1987 | `id/execution_id/name/platform/status/log/create_time/update_time` |
 
-所以真实 DB 也支持这个结论：当前字段不能精确闭合到唯一 Pattern、itemset 或 Case。
+`demand_content.ext_data` 是 JSON 字段，可以承载完整 `evidence_pack`。当前生产样例仍主要是 `desc/reason/type/video_ids`，没有稳定出现新版 `pg_pattern_v2` 的 `evidence_pack`，所以这仍是待补齐的输出契约。
 
-### 6.5 subagent 交叉验证
+### 7.5 subagent 交叉验证
 
-本轮拆了两个 subagent 交叉验证。
+本轮拆了三个 subagent 交叉验证。
 
 | subagent | 验证范围 | 结论 |
 | --- | --- | --- |
-| DemandAgent 代码与文档侧 | DemandItem、`demand_content`、`dwd_multi_demand_pool_di`、Pattern 模型字段 | DemandAgent 当前下游字段很窄，大量 Pattern lineage 在写需求池时丢失 |
-| 历史 DB 只读侧 | `demand_content`、`workflow_decode_task_result`、旧 `topic_pattern_*` 抽样闭合 | 只能做到候选回查，不能唯一闭合 Pattern / itemset / Case；新版正式证据源改为 PG Pattern V2 |
+| PG/MySQL DB 字段侧 | PG Pattern V2、旧 MySQL `topic_pattern_*`、需求池表字段 | PG V2 能支撑新版证据链；旧 MySQL 字段必须删除弃用，只作为迁移对照 |
+| DemandAgent 代码侧 | PG repository、tool adapter、`evidence_pack_builder`、local output | 当前主路径读取 PG Pattern V2 并写本地 JSON，`evidence_pack.pattern_source_system` 为 `pg_pattern_v2` |
+| ContentFindAgentNew 下游侧 | `source_context.ext_data.evidence_pack`、`source_evidence`、校验器 | 下游需要继承 DemandAgent 的 PG `evidence_pack`，再追加运行态 `source_evidence` 字段 |
 
-两个方向结论一致。
+三个方向结论一致。
 
-## 7. 建议的交付方式
+## 8. 建议的交付方式
 
 V1 建议先不要大改表结构，先扩展 `demand_content.ext_data`。
 
@@ -320,8 +347,8 @@ V1 建议先不要大改表结构，先扩展 `demand_content.ext_data`。
     ],
     "category_bindings": [],
     "element_bindings": [],
-    "support": 0.04539396315170521,
-    "absolute_support": 579,
+    "support": 0.000075,
+    "absolute_support": 1,
     "matched_post_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
     "video_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
     "case_ids": ["02454b8dab4c13c9fa21042bdc92d122"],
@@ -341,17 +368,17 @@ V1 建议先不要大改表结构，先扩展 `demand_content.ext_data`。
 | 阶段 | 建议 |
 | --- | --- |
 | V1 必须 | `pattern_source_system`, `source_kind`, `case_id_type`, `source_post_id`, `pattern_execution_id`, `mining_config_id`, `itemset_ids`, `itemset_items[]`, `category_bindings`, `element_bindings`, `support`, `absolute_support`, `matched_post_ids`, `video_ids/case_ids`, `seed_terms`, `trace_id`, `source_certainty`, `validation_status` |
-| V1 可选补充 | `decode_case_ids`，用于保存能命中 `workflow_decode_task_result` 的 decode 行 ID |
-| V1.1 增强 | 独立证据表、下游完整候选池判断日志、topic_element 后续分类树补齐方案 |
+| V1 兼容字段 | `decode_case_ids`，PG-only 当前固定为空数组；如后续重新引入 decode row ID，需要另做下游契约 |
+| V1.1 增强 | 独立证据表、下游完整判断日志、topic_element 后续分类树补齐方案 |
 | 后续再拆表 | 如果 `evidence_pack` 变大，再考虑独立证据表或关系表 |
 
 注意：本次 MVP 只下发 `source_kind=pattern_itemset` 且 PG `scope='topic'` 的需求；如果来源不是该类 Pattern 频繁项集，暂不进入 exact 输出。
 
-### 7.1 DemandAgent 必须保证的真实性边界
+### 8.1 DemandAgent 必须保证的真实性边界
 
-给下游 ContentFindAgent 的任务证据链必须真实。LLM 只能在 `evidence_refs` 里声明候选引用，不能直接决定最终证据。
+给下游 ContentFindAgent 的任务证据链必须真实。LLM 只能在 `evidence_refs` 里声明引用，不能直接决定最终证据。
 
-当前 MVP 中，LLM 生成需求时仍只写 `evidence_refs` 候选引用；DemandAgent 代码会基于真实 DB 只读校验并升级为 `evidence_pack`。推荐 `evidence_refs` 形态如下：
+当前 MVP 中，LLM 生成需求时仍只写 `evidence_refs` 引用声明；DemandAgent 代码会基于 PG DB 只读校验并升级为 `evidence_pack`。推荐 `evidence_refs` 形态如下：
 
 ```json
 {
@@ -384,11 +411,13 @@ DemandAgent 代码必须按真实 DB 校验并补齐 `evidence_pack`。对于 `s
 | `element_bindings` | 必须能用 `execution_id + source_post_id + category_id + point_type + dimension/element_type + element_name(如有)` 精确绑定到 PG `pattern_mining_element`，且限定 `source_table='post_decode_topic_point_element'` |
 | `source_certainty` / `validation_status` | 进入 `demand_content` 时只能是 `db_validated` / `passed` |
 
-任一必需字段查不到、跨 execution、post 不在 matched posts、category 不在当前 execution、source post 精确 element binding 不成立，当前 DemandItem 都失败，不写入 `demand_content`。失败项只进入本地 `rejected_demand_items.json`，Agent 继续生成下一条需求。
+在当前官方 `local_json` MVP 路径中，任一必需字段查不到、跨 execution、post 不在 matched posts、category 不在当前 execution、source post 精确 element binding 不成立，当前 DemandItem 都失败，不写入本地 `demand_content.json`。失败项只进入本地 `rejected_demand_items.json`，Agent 继续生成下一条需求。
+
+任一最终证据字段来自旧 MySQL `topic_pattern_*` 时，同样必须 rejected。新版 exact evidence 不允许 MySQL 兜底。
 
-对 `source_kind=pattern_itemset` 的需求，`case_ids` 按 `post_id` 语义直接等于已校验的 `matched_post_ids`，因为 PG Pattern V2 的主支撑关系来自 `pattern_itemset_post`；可为空的是兼容字段 `decode_case_ids`。`source_post_id/itemset_ids/itemset_items/mining_config_id/matched_post_ids/support/absolute_support` 不可为空且必须 DB 校验通过。
+对 `source_kind=pattern_itemset` 的需求，`case_ids` 按 `post_id` 语义直接等于已校验的 `matched_post_ids`，因为 PG Pattern V2 的主支撑关系来自 `pattern_itemset_post`；兼容字段 `decode_case_ids` 当前固定为空数组。`source_post_id/itemset_ids/itemset_items/mining_config_id/matched_post_ids/support/absolute_support` 不可为空且必须 DB 校验通过。
 
-### 7.2 本地 MVP 输出方式
+### 8.2 本地 MVP 输出方式
 
 本次 MVP 不真实写 `demand_task`、`demand_content`、Hive、PG 或 MySQL，只通过 `examples.demand.run_existing_execution_local` 读取已有 PG `pattern_mining_execution.id` 并把输出写到本地：
 
@@ -409,15 +438,17 @@ examples/demand/test_output_data/{run_id}/
 
 其中只有 `demand_content.json[].ext_data.evidence_pack` 是给下游 exact evidence 的新增载体；`dwd_multi_demand_pool_di.json` 和 `feature_point_data.json` 只做原业务字段镜像，不增加 `evidence_pack`。
 
-### 7.3 ContentFindAgent 接到需求后必须做什么
+### 8.3 ContentFindAgent 接到需求后必须做什么
 
 ContentFindAgent 后续领取 `demand_content` 时，不能只取 `id/name/suggestion/score/merge_leve2/dt`，还必须解析 `ext_data.evidence_pack`。
 
 它需要把 `source_kind/source_post_id/pattern_execution_id/mining_config_id/itemset_ids/itemset_items/category_bindings/matched_post_ids/source_certainty` 注入数据源、Query、判断、游走、入库上下文。最终结果必须保存结构化 `source_evidence` 到 `demand_find_content_result` 或 sidecar/source edge artifact。
 
+ContentFindAgentNew 生成 `source_evidence` 时，需要继承 DemandAgent 的 `evidence_pack`，并追加运行态字段：`origin_source/immediate_source/origin_edge_id/query_id/query/generation_type/candidate_aweme_id/candidate_relation/source_edge_ids`。这些字段不由 DemandAgent 预写完整。当前 CFA validator 已强校验 `origin_source/immediate_source/origin_edge_id/candidate_aweme_id`，并校验 `source_evidence` 与 `evidence_pack` 的 `pattern_source_system/pattern_execution_id/source_post_id/case_id_type/mining_config_id/source_certainty/validation_status/itemset_ids/itemset_items/matched_post_ids/seed_terms` 保持一致；`query_id/query/generation_type/candidate_relation/source_edge_ids` 当前由生成逻辑追加，后续如要作为硬门槛，需要在 CFA validator 中继续加严。
+
 如果新找到的 `aweme_id/post_id` 不在 DemandAgent 的 `source_post_id/matched_post_ids` 中，只能标记为 `derived` 或 `candidate`，不能声称它 exact 属于某个 Pattern 节点。除非下游额外完成新 post 的解构和分类绑定，否则 DemandAgent 的证据只能证明“这个需求来源于某 Pattern”，不能证明“这个新 post 属于这个 Pattern”。
 
-## 8. 下游使用方式
+## 9. 下游使用方式
 
 | 下游阶段 | 怎么使用这些字段 |
 | --- | --- |
@@ -429,9 +460,9 @@ ContentFindAgent 后续领取 `demand_content` 时，不能只取 `id/name/sugge
 | 资产清洗沉淀 | 入库前保留来源路径、判断结果、分类或元素绑定、平台调性说明、trace 和结构化 `source_evidence` |
 | 策略学习 | 用 trace 回看哪类来源、哪类 seed、哪类规则包有效 |
 
-## 9. 最终判断
+## 10. 最终判断
 
-旧版 ContentFindAgent 可以做补充反查，但不能承担新版主血缘。
+旧版 ContentFindAgent 的 MySQL Pattern 反查已删除弃用，不能承担新版主血缘，也不能作为新版 `evidence_pack` 的兜底来源。
 
 如果新版要求精准找到 Pattern、Case、分类或元素绑定，并让所有入库资产可追溯，DemandAgent 必须把生成需求时已经知道的证据一并传给下游。
 

examples/demand/mysql_demand_content_sink.py

@@ -0,0 +1,219 @@
+"""MySQL sink for DB-validated DemandAgent demand_content rows.
+
+This module is intentionally narrow: it only writes the test demand_content
+table. It must not create demand_task rows, Hive rows, Pattern rows, or any
+other side effects.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+from dataclasses import dataclass, field
+from typing import Any
+from urllib.parse import unquote, urlparse
+
+import pymysql
+
+
+@dataclass
+class MySQLDemandContentWriteResult:
+    inserted_count: int = 0
+    skipped_count: int = 0
+    inserted_ids: list[int] = field(default_factory=list)
+    skipped: list[dict[str, Any]] = field(default_factory=list)
+
+
+def _env_value(*names: str, default: str = "") -> str:
+    for name in names:
+        value = os.getenv(name)
+        if value is not None and str(value).strip():
+            return str(value).strip()
+    return default
+
+
+def _resolve_mysql_config() -> dict[str, Any]:
+    dsn = _env_value("DEMAND_CONTENT_MYSQL_DSN")
+    if dsn:
+        parsed = urlparse(dsn)
+        return {
+            "host": parsed.hostname or "127.0.0.1",
+            "port": int(parsed.port or 3306),
+            "user": unquote(parsed.username or ""),
+            "password": unquote(parsed.password or ""),
+            "database": unquote(parsed.path.lstrip("/") or ""),
+        }
+
+    return {
+        "host": _env_value("DEMAND_CONTENT_MYSQL_HOST", "CONTENT_SUPPLY_DB_HOST", default="127.0.0.1"),
+        "port": int(_env_value("DEMAND_CONTENT_MYSQL_PORT", "CONTENT_SUPPLY_DB_PORT", default="3306")),
+        "user": _env_value("DEMAND_CONTENT_MYSQL_USER", "CONTENT_SUPPLY_DB_USER", default="content_rw"),
+        "password": _env_value("DEMAND_CONTENT_MYSQL_PASSWORD", "CONTENT_SUPPLY_DB_PASSWORD"),
+        "database": _env_value(
+            "DEMAND_CONTENT_MYSQL_DB",
+            "CONTENT_SUPPLY_DB_NAME",
+            default="content-deconstruction-supply",
+        ),
+    }
+
+
+def _connect():
+    cfg = _resolve_mysql_config()
+    missing = [key for key in ("host", "user", "database") if not cfg.get(key)]
+    if missing:
+        raise RuntimeError(f"MySQL demand_content config missing: {missing}")
+    return pymysql.connect(
+        host=cfg["host"],
+        port=int(cfg["port"]),
+        user=cfg["user"],
+        password=cfg["password"],
+        database=cfg["database"],
+        charset="utf8mb4",
+        cursorclass=pymysql.cursors.DictCursor,
+        autocommit=False,
+    )
+
+
+def _as_ext_data(value: Any) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return copy.deepcopy(value)
+    if isinstance(value, str) and value.strip():
+        parsed = json.loads(value)
+        if isinstance(parsed, dict):
+            return parsed
+    return {}
+
+
+def _dedupe_key(row: dict[str, Any], ext_data: dict[str, Any]) -> tuple[str, str, str]:
+    evidence_pack = ext_data.get("evidence_pack") or {}
+    source_post_id = str(evidence_pack.get("source_post_id") or "")
+    return (
+        str(row.get("name") or ""),
+        source_post_id,
+        "",
+    )
+
+
+def _row_exists_for_run(cursor, *, run_label: str, row: dict[str, Any], ext_data: dict[str, Any]) -> bool:
+    if not run_label:
+        return False
+    name, source_post_id, _ = _dedupe_key(row, ext_data)
+    cursor.execute(
+        """
+        SELECT id
+        FROM demand_content
+        WHERE JSON_UNQUOTE(JSON_EXTRACT(ext_data, '$.run_label')) = %s
+          AND name = %s
+          AND JSON_UNQUOTE(JSON_EXTRACT(ext_data, '$.evidence_pack.source_post_id')) = %s
+        LIMIT 1
+        """,
+        (run_label, name, source_post_id),
+    )
+    return cursor.fetchone() is not None
+
+
+def _validate_row(row: dict[str, Any], ext_data: dict[str, Any]) -> None:
+    evidence_pack = ext_data.get("evidence_pack")
+    if not isinstance(evidence_pack, dict):
+        raise ValueError("missing ext_data.evidence_pack")
+    expected = {
+        "pattern_source_system": "pg_pattern_v2",
+        "source_kind": "pattern_itemset",
+        "case_id_type": "post_id",
+        "source_certainty": "db_validated",
+        "validation_status": "passed",
+    }
+    for field, value in expected.items():
+        if evidence_pack.get(field) != value:
+            raise ValueError(f"invalid evidence_pack.{field}: {evidence_pack.get(field)!r}")
+
+    required = [
+        "source_post_id",
+        "pattern_execution_id",
+        "mining_config_id",
+        "itemset_ids",
+        "itemset_items",
+        "category_bindings",
+        "element_bindings",
+        "support",
+        "absolute_support",
+        "matched_post_ids",
+        "video_ids",
+        "case_ids",
+        "seed_terms",
+        "trace_id",
+    ]
+    for field in required:
+        value = evidence_pack.get(field)
+        if value is None or value == "" or value == []:
+            raise ValueError(f"missing evidence_pack.{field}")
+
+    matched_post_ids = [str(post_id) for post_id in evidence_pack.get("matched_post_ids") or []]
+    if str(evidence_pack["source_post_id"]) not in set(matched_post_ids):
+        raise ValueError("evidence_pack.source_post_id must be in matched_post_ids")
+    if [str(post_id) for post_id in evidence_pack.get("video_ids") or []] != matched_post_ids:
+        raise ValueError("evidence_pack.video_ids must equal matched_post_ids")
+    if [str(post_id) for post_id in evidence_pack.get("case_ids") or []] != matched_post_ids:
+        raise ValueError("evidence_pack.case_ids must equal matched_post_ids")
+    if len(matched_post_ids) < int(evidence_pack["absolute_support"]):
+        raise ValueError("evidence_pack.matched_post_ids must cover absolute_support")
+    if not row.get("merge_leve2") or not row.get("name") or not row.get("dt"):
+        raise ValueError("missing demand_content merge_leve2/name/dt")
+
+
+def write_demand_content_rows(
+    rows: list[dict[str, Any]],
+    *,
+    run_label: str = "",
+) -> MySQLDemandContentWriteResult:
+    result = MySQLDemandContentWriteResult()
+    if not rows:
+        return result
+
+    conn = _connect()
+    try:
+        with conn.cursor() as cursor:
+            for row in rows:
+                ext_data = _as_ext_data(row.get("ext_data"))
+                if run_label:
+                    ext_data["run_label"] = run_label
+                _validate_row(row, ext_data)
+
+                if _row_exists_for_run(cursor, run_label=run_label, row=row, ext_data=ext_data):
+                    result.skipped_count += 1
+                    result.skipped.append({"name": row.get("name"), "reason": "duplicate_run_label_key"})
+                    continue
+
+                cursor.execute(
+                    """
+                    INSERT INTO demand_content
+                      (merge_leve2, name, reason, suggestion, ext_data, score, dt)
+                    VALUES
+                      (%s, %s, %s, %s, %s, %s, %s)
+                    """,
+                    (
+                        row.get("merge_leve2"),
+                        row.get("name"),
+                        row.get("reason"),
+                        row.get("suggestion"),
+                        json.dumps(ext_data, ensure_ascii=False),
+                        row.get("score"),
+                        row.get("dt"),
+                    ),
+                )
+                demand_content_id = int(cursor.lastrowid)
+                ext_data.setdefault("evidence_pack", {})["demand_content_id"] = demand_content_id
+                cursor.execute(
+                    "UPDATE demand_content SET ext_data=%s WHERE id=%s",
+                    (json.dumps(ext_data, ensure_ascii=False), demand_content_id),
+                )
+                result.inserted_count += 1
+                result.inserted_ids.append(demand_content_id)
+        conn.commit()
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
+    return result

examples/demand/run.py

@@ -57,18 +57,26 @@ CUSTOM_TOOL_MODULES = {
 }
 
 LOCAL_JSON_MODE = "local_json"
+MYSQL_DEMAND_CONTENT_MODE = "mysql_demand_content"
 LOCAL_ENTRYPOINT_ENV = "DEMAND_LOCAL_ENTRYPOINT"
 LOCAL_ENTRYPOINT_NAME = "run_existing_execution_local"
+MYSQL_ENTRYPOINT_ENV = "DEMAND_MYSQL_ENTRYPOINT"
+MYSQL_ENTRYPOINT_NAME = "run_existing_execution_mysql"
 
 
 def _is_local_json_mode() -> bool:
     return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == LOCAL_JSON_MODE
 
 
+def _is_mysql_demand_content_mode() -> bool:
+    return os.getenv("DEMAND_OUTPUT_MODE", "").strip().lower() == MYSQL_DEMAND_CONTENT_MODE
+
+
 def _raise_db_write_disabled(action: str) -> None:
     raise RuntimeError(
-        f"{action} 已在 PG Pattern V2 local MVP 中禁用；"
-        "请通过 DEMAND_OUTPUT_MODE=local_json 和 run_existing_execution_local 写本地 JSON"
+        f"{action} 已在 PG Pattern V2 MVP 中禁用；"
+        "请通过 run_existing_execution_local 写本地 JSON，或通过 "
+        "run_existing_execution_mysql 只写测试 MySQL demand_content"
     )
 
 
@@ -83,6 +91,17 @@ def _require_local_entrypoint() -> None:
         )
 
 
+def _require_mysql_entrypoint() -> None:
+    if not _is_mysql_demand_content_mode():
+        return
+    entrypoint = os.getenv(MYSQL_ENTRYPOINT_ENV, "").strip()
+    if entrypoint != MYSQL_ENTRYPOINT_NAME:
+        raise RuntimeError(
+            "DEMAND_OUTPUT_MODE=mysql_demand_content 只能通过 "
+            "examples.demand.run_existing_execution_mysql 入口运行"
+        )
+
+
 def _default_local_output_root(execution_id: int) -> Path:
     ts = datetime.now(ZoneInfo("Asia/Shanghai")).strftime("%Y%m%d_%H%M%S")
     return Path(__file__).parent / "test_output_data" / f"execution_{execution_id}_{ts}"
@@ -549,9 +568,35 @@ def write_demand_items_to_mysql(
         trace_id: Optional[str] = None,
         task_id: Optional[int] = None,
 ) -> int:
-    """Disabled production demand_content writer retained for legacy imports."""
-    del execution_id, merge_level2, trace_id, task_id
-    _raise_db_write_disabled("生产 write_demand_items_to_mysql")
+    """Write DB-validated demand_content rows to the test MySQL sink only."""
+    if not _is_mysql_demand_content_mode():
+        _raise_db_write_disabled("生产 write_demand_items_to_mysql")
+
+    from examples.demand.mysql_demand_content_sink import write_demand_content_rows
+
+    rows, demand_items, rejected_items = _build_demand_content_rows(
+        execution_id=execution_id,
+        merge_level2=merge_level2,
+        trace_id=trace_id,
+        task_id=task_id,
+        serialize_ext_data=False,
+        require_evidence=True,
+    )
+    run_label = os.getenv("DEMAND_RUN_LABEL", "").strip()
+    result = write_demand_content_rows(rows, run_label=run_label)
+    log(
+        "[mysql_demand_content] 写入测试 MySQL demand_content 完成，"
+        f"demand_items={len(demand_items)}, inserted={result.inserted_count}, "
+        f"rejected={len(rejected_items)}, skipped={result.skipped_count}"
+    )
+    if rejected_items:
+        log(
+            "[mysql_demand_content] rejected_items="
+            + json.dumps(rejected_items, ensure_ascii=False, default=str)
+        )
+    if result.inserted_ids:
+        log(f"[mysql_demand_content] inserted_ids={result.inserted_ids}")
+    return result.inserted_count
 
 
 def write_demand_items_to_local_json(
@@ -832,7 +877,7 @@ async def run_once(
                 pass
         if _is_local_json_mode():
             _finish_local_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
-        else:
+        elif not _is_mysql_demand_content_mode():
             _finish_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)
         _write_local_run_manifest(
             execution_id=execution_id,
@@ -867,8 +912,8 @@ async def main(
         execution_id: Optional[int] = None,
         task_id: Optional[int] = None,
 ) -> dict:
-    if not _is_local_json_mode():
-        _raise_db_write_disabled("非 local_json 运行入口")
+    if not (_is_local_json_mode() or _is_mysql_demand_content_mode()):
+        _raise_db_write_disabled("非受控运行入口")
 
     if _is_local_json_mode():
         _require_local_entrypoint()
@@ -876,6 +921,13 @@ async def main(
             raise ValueError("local_json 模式必须传入已有 execution_id，不能触发 prepare 写库流程")
         _ensure_local_output_paths(execution_id)
 
+    if _is_mysql_demand_content_mode():
+        _require_mysql_entrypoint()
+        if execution_id is None:
+            raise ValueError("mysql_demand_content 模式必须传入已有 execution_id，不能触发 prepare 写库流程")
+        if str(cluster_name).strip() == "全局树":
+            raise ValueError("mysql_demand_content 模式只写普通 demand_content，不支持全局树 Hive 输出")
+
     if execution_id is None:
         execution_id = get_execution_id_by_merge_level2(cluster_name)
     if not execution_id:

examples/demand/run_existing_execution_mysql.py

@@ -0,0 +1,76 @@
+"""Cloud MySQL entrypoint for an existing PG Pattern V2 execution.
+
+This entrypoint intentionally writes only content-deconstruction-supply.demand_content.
+It never runs prepare/mining and does not write demand_task, Hive, PG Pattern, or
+legacy MySQL Pattern tables.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+from pathlib import Path
+
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+
+def _configure_mysql_env(run_label: str | None) -> None:
+    os.environ["DEMAND_OUTPUT_MODE"] = "mysql_demand_content"
+    os.environ["DEMAND_MYSQL_ENTRYPOINT"] = "run_existing_execution_mysql"
+    if run_label:
+        os.environ["DEMAND_RUN_LABEL"] = run_label
+
+
+def _validate_execution_success(execution_id: int) -> None:
+    from examples.demand.db_manager import query_execution_for_evidence
+
+    execution = query_execution_for_evidence(execution_id)
+    if not execution:
+        raise ValueError(f"execution_id={execution_id} 不存在，MySQL 写入入口只允许跑已有 execution")
+    if str(execution.get("status") or "").strip().lower() != "success":
+        raise ValueError(
+            f"execution_id={execution_id} status={execution.get('status')}，不是 success，拒绝生成需求"
+        )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Run DemandAgent against an existing PG Pattern V2 execution_id and write only MySQL demand_content."
+    )
+    parser.add_argument("--execution-id", type=int, required=True, help="Existing PG pattern_mining_execution.id")
+    parser.add_argument("--merge-level2", required=True, help="Current merge_leve2 / cluster name")
+    parser.add_argument("--platform-type", default="piaoquan", help="Platform label")
+    parser.add_argument("--count", type=int, default=5, help="Approximate demand count requested from the prompt")
+    parser.add_argument("--run-label", default=None, help="Label stored at ext_data.run_label for verification")
+    return parser.parse_args()
+
+
+async def async_main() -> dict:
+    args = parse_args()
+    _configure_mysql_env(args.run_label)
+    _validate_execution_success(args.execution_id)
+
+    from examples.demand.run import main
+
+    result = await main(
+        cluster_name=args.merge_level2,
+        platform_type=args.platform_type,
+        count=args.count,
+        execution_id=args.execution_id,
+        task_id=None,
+    )
+    result["run_label"] = args.run_label
+    return result
+
+
+def main() -> None:
+    result = asyncio.run(async_main())
+    print(json.dumps(result, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

examples/demand/tests/test_mysql_demand_content_sink.py

@@ -0,0 +1,143 @@
+import json
+import sys
+import types
+import unittest
+from unittest.mock import patch
+
+
+class _FakeCursor:
+    def __init__(self):
+        self.statements = []
+        self.lastrowid = 321
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def execute(self, sql, params=None):
+        self.statements.append((sql, params))
+        return 1
+
+    def fetchone(self):
+        return None
+
+
+class _FakeConnection:
+    def __init__(self):
+        self.cursor_obj = _FakeCursor()
+        self.committed = False
+        self.rolled_back = False
+        self.closed = False
+
+    def cursor(self):
+        return self.cursor_obj
+
+    def commit(self):
+        self.committed = True
+
+    def rollback(self):
+        self.rolled_back = True
+
+    def close(self):
+        self.closed = True
+
+
+def _install_fake_pymysql():
+    fake_pymysql = types.ModuleType("pymysql")
+    fake_pymysql.cursors = types.SimpleNamespace(DictCursor=object)
+    fake_pymysql.connect = lambda **kwargs: _FakeConnection()
+    sys.modules["pymysql"] = fake_pymysql
+
+
+def _valid_row():
+    evidence_pack = {
+        "pattern_source_system": "pg_pattern_v2",
+        "source_kind": "pattern_itemset",
+        "case_id_type": "post_id",
+        "source_certainty": "db_validated",
+        "validation_status": "passed",
+        "source_post_id": "p1",
+        "pattern_execution_id": 581,
+        "mining_config_id": 7,
+        "itemset_ids": [11],
+        "itemset_items": [{"itemset_id": 11, "category_id": 22, "category_path": "A>B"}],
+        "category_bindings": [{"category_id": 22, "category_path": "A>B"}],
+        "element_bindings": [{"post_id": "p1", "category_id": 22, "element_name": "x"}],
+        "support": 0.5,
+        "absolute_support": 2,
+        "matched_post_ids": ["p1", "p2"],
+        "video_ids": ["p1", "p2"],
+        "case_ids": ["p1", "p2"],
+        "seed_terms": ["x"],
+        "trace_id": "trace-1",
+    }
+    return {
+        "merge_leve2": "PG Pattern V2 需求测试",
+        "name": "测试需求",
+        "reason": "原因",
+        "suggestion": "建议",
+        "score": 1.0,
+        "dt": "20260606",
+        "ext_data": {"evidence_pack": evidence_pack},
+    }
+
+
+class MySQLDemandContentSinkTest(unittest.TestCase):
+    def setUp(self):
+        _install_fake_pymysql()
+
+    def test_insert_updates_evidence_pack_with_real_mysql_id(self):
+        from examples.demand import mysql_demand_content_sink as sink
+
+        fake_conn = _FakeConnection()
+        with patch.object(sink, "_connect", return_value=fake_conn):
+            result = sink.write_demand_content_rows([_valid_row()], run_label="batch01")
+
+        self.assertEqual(result.inserted_count, 1)
+        self.assertEqual(result.inserted_ids, [321])
+        self.assertTrue(fake_conn.committed)
+        update_statements = [
+            params
+            for sql, params in fake_conn.cursor_obj.statements
+            if sql.strip().startswith("UPDATE demand_content")
+        ]
+        self.assertEqual(len(update_statements), 1)
+        updated_ext_data = json.loads(update_statements[0][0])
+        self.assertEqual(updated_ext_data["run_label"], "batch01")
+        self.assertEqual(updated_ext_data["evidence_pack"]["demand_content_id"], 321)
+
+    def test_rejects_rows_without_required_pg_evidence(self):
+        from examples.demand import mysql_demand_content_sink as sink
+
+        row = _valid_row()
+        row["ext_data"]["evidence_pack"]["pattern_source_system"] = "mysql_topic_pattern"
+
+        with patch.object(sink, "_connect", return_value=_FakeConnection()):
+            with self.assertRaisesRegex(ValueError, "pattern_source_system"):
+                sink.write_demand_content_rows([row], run_label="batch01")
+
+    def test_rejects_rows_with_mismatched_case_or_video_ids(self):
+        from examples.demand import mysql_demand_content_sink as sink
+
+        row = _valid_row()
+        row["ext_data"]["evidence_pack"]["case_ids"] = ["p1"]
+
+        with patch.object(sink, "_connect", return_value=_FakeConnection()):
+            with self.assertRaisesRegex(ValueError, "case_ids"):
+                sink.write_demand_content_rows([row], run_label="batch01")
+
+    def test_rejects_rows_when_support_is_not_covered_by_posts(self):
+        from examples.demand import mysql_demand_content_sink as sink
+
+        row = _valid_row()
+        row["ext_data"]["evidence_pack"]["absolute_support"] = 3
+
+        with patch.object(sink, "_connect", return_value=_FakeConnection()):
+            with self.assertRaisesRegex(ValueError, "absolute_support"):
+                sink.write_demand_content_rows([row], run_label="batch01")
+
+
+if __name__ == "__main__":
+    unittest.main()