feat(classification): 新增分类表与维度取帖接口，统一归类口径

新增 step_classification 派生表存储五维度分类结果，与前端显示同源
新增 /api/classified_cases 接口支持按维度/路径/名称查询归类案例
重构 category_match.py，将 TOP_K 设为 1 实现单归属，新增全五维度归类双写逻辑
更新项目文档与 README，补充分类匹配相关说明与新接口文档
更新 .gitignore 添加项目相关文档文件
现有接口的请求响应形状未变更，无需修改前端调用

examples/mode_workflow/.gitignore

@@ -8,4 +8,8 @@ __pycache__/
 .cloudflared.log
 .cloudflared.pid
 .server_8772.out
-import_process_knowledge.处理方式.md
+import_process_knowledge.处理方式.md
+归类口径修复与表结构优化方案.md
+工序接口文档.md
+流程执行手册.md
+拆表_3-II_设计.md

examples/mode_workflow/README.md

@@ -1,6 +1,7 @@
 # mode_workflow · 搜索评估 + 工序/工具解构工作台
 
-MySQL 四表(search_process / search_tools / mode_process / mode_tools)为唯一事实源的单页工作台:
+MySQL 四表(search_process / search_tools / mode_process / mode_tools)为唯一事实源,外加归类派生表
+`step_classification`(由 category-match 回写,与显示同源,见 `docs/step_classification_3-I_设计.md`)的单页工作台:
 Dashboard(结果/过程指标可视化)、Dataset(query → 帖子 → 工序/工具解构)、聚类库(占位)。
 顶栏「**Query 规则**」按内容树维度铺词 → Sonnet 评分 → 高亮达标组合 → 一键搜(见下「Query 规则组织器」)。
 
@@ -11,7 +12,7 @@ Dashboard(结果/过程指标可视化)、Dataset(query → 帖子 → 工序/
 
 ```bash
 # 0. 前置:.env 配 MYSQL_* 与 OPEN_ROUTER_API_KEY;pip install -e .
-python db.py init             # 建四张表(幂等);db.py clear 清空数据
+python db.py init             # 建表(幂等,含归类派生表 step_classification);db.py clear 清空数据
 python server.py              # http://localhost:8772
 ```
 
@@ -26,6 +27,7 @@ python server.py              # http://localhost:8772
 | `stages/procedure_extract.py` | 工序解构(LLM 直出)→ mode_process |
 | `stages/tool_extract.py` | 工具解构 → mode_tools |
 | `stages/import_process_knowledge.py` | 已采纳工序(mode_process 最新版)→ 知识导入接口;**读 DB 非本地文件**,采纳口径同 Dashboard(`db.is_adopted_rel`) |
+| `stages/category_match.py` | 工序五维度词 → cat-api 分类匹配(`top_k=1` 单归属);回写 `mode_process.steps[].substanceMatch/formMatch` **并** upsert 派生表 `step_classification`(name+path+score)。`--all` 全量重跑;`/api/classified_cases` 按 `matched_path` 子树取帖(与显示同源,替代 cat-api 树 `knowledge_ids`) |
 | `prompts/` | 工序/工具解构 system prompt(可单独迭代) |
 | `reference/judged_matrix.json` | 内容树(27 动作×50 类型),Dashboard 覆盖度用 |
 | `stages/query_score.py` | Query 正交格评分:tier≥1 格 × 维度上下文 → Sonnet 打分,结果缓存 `.cache/query_score/` |

examples/mode_workflow/db.py

@@ -220,6 +220,100 @@ CREATE TABLE IF NOT EXISTS tools_ingest_log (
 """
 
 
+# step 维度归类结果独立表(3-I,见 docs/step_classification_3-I_设计.md):一行 = 某 case 某版本
+# 某工序某 step 某维度某子项的命中分类。与 mode_process 解耦,归类重跑只覆盖本表、不重写 steps blob;
+# 存 matched_path 全路径,使「某节点及其子树的所有帖」一句 SQL(LIKE 前缀)即可,且与显示同源。
+DDL_STEP_CLASSIFICATION = """
+CREATE TABLE IF NOT EXISTS step_classification (
+  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
+  case_id       VARCHAR(128) NOT NULL,
+  query_id      VARCHAR(32)  NULL COMMENT 'record 时的 post_id,便于溯源',
+  version       VARCHAR(32)  NULL COMMENT '对齐 mode_process 最新真实版,重跑覆盖',
+  procedure_id  VARCHAR(16)  NULL COMMENT 'p1,p2…',
+  step_id       VARCHAR(16)  NULL COMMENT 's1,s2…',
+  dimension     VARCHAR(8)   NOT NULL COMMENT '实质/形式/类型/作用/动作',
+  sub_index     SMALLINT     NOT NULL COMMENT '原值「、」拆分后的子项下标',
+  raw_term      VARCHAR(255) NULL COMMENT '原始词',
+  matched_name  VARCHAR(255) NULL COMMENT '命中分类名(单一最优;无命中不入库)',
+  matched_path  VARCHAR(512) NULL COMMENT '命中分类全路径,如 /表象/视觉/空间/空间环境',
+  matched_id    INT          NULL COMMENT 'cat-api stable_id',
+  score         FLOAT        NULL,
+  match_run_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  UNIQUE KEY uk_step_dim_sub (case_id, version, procedure_id, step_id, dimension, sub_index),
+  KEY idx_case (case_id),
+  KEY idx_dim_name (dimension, matched_name),
+  KEY idx_dim_path (dimension, matched_path)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='step 维度归类结果(与 mode_process 解耦)';
+"""
+
+
+# 工序解构拆表(3-II,见 docs/拆表_3-II_设计.md):process_run → process_procedure → process_step 三层,
+# 与旧 mode_process 并行。step_json 逐字保留原 step → ProcessPayload 还原零漂移;抽取列供 DB 层查询。
+DDL_PROCESS_RUN = """
+CREATE TABLE IF NOT EXISTS process_run (
+  id          BIGINT AUTO_INCREMENT PRIMARY KEY,
+  query_id    VARCHAR(32)   NOT NULL,
+  case_id     VARCHAR(128)  NOT NULL,
+  version     VARCHAR(32)   NOT NULL,
+  is_latest   TINYINT       NOT NULL DEFAULT 0 COMMENT '该 case 最新真实版=1(link_ 恒 0)',
+  platform    VARCHAR(32)   NULL,
+  post_title  VARCHAR(512)  NULL,
+  source      JSON          NULL,
+  model       VARCHAR(64)   NULL,
+  cost_usd    DECIMAL(10,6) NULL,
+  duration_s  FLOAT         NULL,
+  created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  UNIQUE KEY uk_q_case_ver (query_id, case_id, version),
+  KEY idx_case_latest (case_id, is_latest),
+  KEY idx_qid (query_id)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构运行级(一行=一帖一版本)';
+"""
+
+DDL_PROCESS_PROCEDURE = """
+CREATE TABLE IF NOT EXISTS process_procedure (
+  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
+  run_id        BIGINT        NOT NULL,
+  case_id       VARCHAR(128)  NOT NULL,
+  version       VARCHAR(32)   NOT NULL,
+  procedure_id  VARCHAR(16)   NULL,
+  seq           SMALLINT      NOT NULL,
+  name          VARCHAR(255)  NULL,
+  purpose       TEXT          NULL,
+  category      VARCHAR(32)   NULL,
+  declarations  JSON          NULL,
+  type_registry JSON          NULL,
+  tools_used    JSON          NULL,
+  step_count    INT           NULL,
+  UNIQUE KEY uk_run_seq (run_id, seq),
+  KEY idx_run (run_id),
+  KEY idx_case_ver (case_id, version)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构工序级(一行=一个 procedure)';
+"""
+
+DDL_PROCESS_STEP = """
+CREATE TABLE IF NOT EXISTS process_step (
+  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
+  procedure_pk  BIGINT        NOT NULL,
+  case_id       VARCHAR(128)  NOT NULL,
+  version       VARCHAR(32)   NOT NULL,
+  step_id       VARCHAR(16)   NULL,
+  seq           SMALLINT      NOT NULL,
+  kind          VARCHAR(16)   NULL,
+  grp           VARCHAR(16)   NULL COMMENT '原 step.group(group 为保留字)',
+  via           VARCHAR(255)  NULL,
+  effect        VARCHAR(255)  NULL,
+  action        VARCHAR(255)  NULL,
+  substance     VARCHAR(512)  NULL,
+  form          VARCHAR(512)  NULL,
+  step_json     JSON          NOT NULL COMMENT '原 step 逐字 JSON,payload 还原即用它',
+  UNIQUE KEY uk_proc_seq (procedure_pk, seq),
+  KEY idx_procedure (procedure_pk),
+  KEY idx_case_ver (case_id, version),
+  KEY idx_action (action), KEY idx_substance (substance), KEY idx_form (form)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='工序解构步骤级(一行=一个 step)';
+"""
+
+
 def _ensure_column(cur, table, column, column_ddl):
     """给已存在的表幂等补列:列已存在则跳过(MySQL ADD COLUMN 无 IF NOT EXISTS)。
     column_ddl 为 ADD COLUMN 后的完整定义,如 \"source JSON NULL ... AFTER post_title\"。"""
@@ -250,6 +344,10 @@ def init_tables():
             cur.execute(DDL_TOOLS)
             cur.execute(DDL_INGEST_LOG)
             cur.execute(DDL_TOOLS_INGEST_LOG)
+            cur.execute(DDL_STEP_CLASSIFICATION)
+            cur.execute(DDL_PROCESS_RUN)
+            cur.execute(DDL_PROCESS_PROCEDURE)
+            cur.execute(DDL_PROCESS_STEP)
             # 历史库迁移:version 由 VARCHAR(16) 放宽到 32,容纳 link_v_mopN_* 复制版本。
             # MODIFY 幂等(已是 32 则 MySQL 元数据无操作),建表后表必存在,可安全执行。
             for t in ("mode_process", "mode_tools"):
@@ -280,7 +378,7 @@ def init_tables():
                 _ensure_unique_index(cur, t, "uk_q_case_ver_seq",
                                      "query_id, case_id, version, seq")
         print("✅ 建表完成:search_process, search_tools, mode_process, mode_tools, "
-              "knowledge_ingest_log, tools_ingest_log")
+              "knowledge_ingest_log, tools_ingest_log, step_classification")
     finally:
         conn.close()
 
@@ -290,7 +388,8 @@ def clear_tables():
     conn = _conn()
     try:
         with conn.cursor() as cur:
-            for t in ("search_process", "search_tools", "mode_process", "mode_tools"):
+            for t in ("search_process", "search_tools", "mode_process", "mode_tools",
+                      "process_run", "process_procedure", "process_step", "step_classification"):
                 cur.execute(f"TRUNCATE TABLE {t}")
                 print(f"🧹 已清空 {t}")
     finally:
@@ -566,7 +665,7 @@ def fetch_queries(mode="process"):
             for r in cur.fetchall():
                 if is_adopted_rel(r["overall_score"], r["rel"], r["publish_time"], r["repro"]):
                     hits[r["query_id"]] = hits.get(r["query_id"], 0) + 1
-            cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM mode_process GROUP BY query_id")
+            cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM process_run GROUP BY query_id")  # 3-II
             np = {r["query_id"]: r["n"] for r in cur.fetchall()}
             cur.execute("SELECT query_id, COUNT(DISTINCT case_id) AS n FROM mode_tools GROUP BY query_id")
             nt = {r["query_id"]: r["n"] for r in cur.fetchall()}
@@ -595,19 +694,21 @@ def fetch_posts(query_id, mode="process"):
                             FROM {table} WHERE query_id=%s AND {_REAL_POST}
                             ORDER BY overall_score DESC, id""", (query_id,))
             rows = cur.fetchall()
-            cur.execute("SELECT DISTINCT case_id FROM mode_process WHERE query_id=%s", (query_id,))
+            cur.execute("SELECT DISTINCT case_id FROM process_run WHERE query_id=%s", (query_id,))  # 3-II
             hp = {r["case_id"] for r in cur.fetchall()}
             cur.execute("SELECT DISTINCT case_id FROM mode_tools WHERE query_id=%s", (query_id,))
             ht = {r["case_id"] for r in cur.fetchall()}
-            # 已归类(工序):hp 中各 case 最新真实版的工序里有任一步骤含 substanceMatch(与归类回写
-            # 同口径)。库端 LIKE 算、只回 0/1,不拉 steps。聚合逻辑见 _categorized_from_rows
-            # (不能只看 id 最大行——空 steps 的 procedure 行永远不含,会误判)。
+            # 已归类(工序):hp 中各 case 最新真实版的工序里有任一 step 含 substanceMatch(同归类回写口径)。
+            # 3-II:改查三表——run 级聚合「该 run 是否有任一 step_json 含 substanceMatch」,逻辑同 _categorized_from_rows。
             hc = set()
             if hp:
                 ph = ",".join(["%s"] * len(hp))
-                cur.execute(f"""SELECT case_id, version, id,
-                                       (LEFT(version,5)='link_') AS islink, (steps LIKE %s) AS cat
-                                FROM mode_process WHERE case_id IN ({ph})""",
+                cur.execute(f"""SELECT r.case_id, r.version, r.id,
+                                       (LEFT(r.version,5)='link_') AS islink,
+                                       EXISTS(SELECT 1 FROM process_step s
+                                              JOIN process_procedure p ON s.procedure_pk=p.id
+                                              WHERE p.run_id=r.id AND s.step_json LIKE %s) AS cat
+                                FROM process_run r WHERE r.case_id IN ({ph})""",
                             ['%substanceMatch%'] + list(hp))
                 hc = _categorized_from_rows(cur.fetchall())
     finally:
@@ -679,7 +780,7 @@ def fetch_all_posts(mode="process", *, query_ids=None, case_ids=None, adopted_on
                             ORDER BY overall_score DESC, id""", params)
             rows = cur.fetchall()
             # has_process/has_tools 全局判定:跨 query 的「该帖是否已解构」,两张解构表各取一次
-            cur.execute("SELECT DISTINCT case_id FROM mode_process")
+            cur.execute("SELECT DISTINCT case_id FROM process_run")  # 3-II
             hp = {r["case_id"] for r in cur.fetchall()}
             cur.execute("SELECT DISTINCT case_id FROM mode_tools")
             ht = {r["case_id"] for r in cur.fetchall()}
@@ -759,6 +860,9 @@ def replace_process(query_id, case_id, platform, post_title, payload,
                    tools_used, model, version, cost_usd, duration_s, seq)
                 VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                 """, rows)
+            # 3-II 双写:同一事务内覆盖写 process_run/procedure/step(新表为主,mode_process 兜底)
+            _split_write_run(cur, query_id, case_id, platform, post_title, source,
+                             procedures, model, version, cost_usd, duration_s)
         conn.commit()
         return len(procedures)
     except Exception:
@@ -769,36 +873,25 @@ def replace_process(query_id, case_id, platform, post_title, payload,
 
 
 def fetch_process_versions(case_id):
+    # 3-II:版本列表读自三表(n=该版本工序数,口径同旧 mode_process 行数)
     conn = _conn()
     try:
         with conn.cursor() as cur:
-            cur.execute("""SELECT version, COUNT(*) AS n, MAX(model) AS model
-                           FROM mode_process WHERE case_id=%s
-                           GROUP BY version
-                           ORDER BY (LEFT(version,5)='link_') ASC, MAX(id) DESC""", (case_id,))
+            cur.execute("""SELECT r.version, COUNT(p.id) AS n, MAX(r.model) AS model
+                           FROM process_run r
+                           LEFT JOIN process_procedure p ON p.run_id = r.id
+                           WHERE r.case_id=%s
+                           GROUP BY r.version
+                           ORDER BY (LEFT(r.version,5)='link_') ASC, MAX(r.id) DESC""", (case_id,))
             return cur.fetchall()
     finally:
         conn.close()
 
 
 def fetch_process(case_id, version=None):
-    """重建 {case_id, version, model, source, procedures:[...]}。version=None 取最新。"""
-    conn = _conn()
-    try:
-        with conn.cursor() as cur:
-            if version is None:
-                cur.execute("""SELECT version FROM mode_process WHERE case_id=%s
-                               ORDER BY (LEFT(version,5)='link_') ASC, id DESC LIMIT 1""", (case_id,))
-                row = cur.fetchone()
-                if not row:
-                    return None
-                version = row["version"]
-            cur.execute("""SELECT * FROM mode_process WHERE case_id=%s AND version=%s
-                           ORDER BY id""", (case_id, version))
-            rows = cur.fetchall()
-    finally:
-        conn.close()
-    return _proc_payload(case_id, version, rows)
+    """重建 {case_id, version, model, source, procedures:[...]}。version=None 取最新。
+    3-II:读自 process_run/procedure/step(返回形状与旧 mode_process 口径完全一致)。"""
+    return rebuild_process_from_split(case_id, version)
 
 
 def fetch_all_process(case_ids=None, lite=False):
@@ -808,6 +901,7 @@ def fetch_all_process(case_ids=None, lite=False):
     返回 {case_id: _proc_payload(...)};无解构记录的 case_id 不出现在结果里。"""
     if case_ids is not None and not case_ids:
         return {}
+    # 3-II:读自三表,3 条批量 SQL(run → procedure → step),避免按 case 逐次往返(保首屏性能)。
     where, params = "", []
     if case_ids is not None:
         where = " WHERE case_id IN (" + ",".join(["%s"] * len(case_ids)) + ")"
@@ -815,24 +909,382 @@ def fetch_all_process(case_ids=None, lite=False):
     conn = _conn()
     try:
         with conn.cursor() as cur:
-            cur.execute(f"SELECT * FROM mode_process{where} ORDER BY case_id, id", params)
+            cur.execute(f"SELECT * FROM process_run{where} ORDER BY case_id, id", params)
+            runs = cur.fetchall()
+            best = {}  # case_id -> (key, run);口径同 fetch_process(is_latest→真实版→id 最大)
+            for r in runs:
+                c = r["case_id"]
+                k = ((r.get("is_latest") or 0), not str(r["version"]).startswith("link_"), r["id"])
+                if c not in best or k > best[c][0]:
+                    best[c] = (k, r)
+            run_ids = [run["id"] for _k, run in best.values()]
+            procs, steps_by_proc = [], {}
+            if run_ids:
+                ph = ",".join(["%s"] * len(run_ids))
+                cur.execute(f"SELECT * FROM process_procedure WHERE run_id IN ({ph}) "
+                            "ORDER BY run_id, seq, id", run_ids)
+                procs = cur.fetchall()
+                pid = [p["id"] for p in procs]
+                if pid:
+                    ph2 = ",".join(["%s"] * len(pid))
+                    cur.execute(f"SELECT procedure_pk, step_json FROM process_step "
+                                f"WHERE procedure_pk IN ({ph2}) ORDER BY procedure_pk, seq, id", pid)
+                    for s in cur.fetchall():
+                        steps_by_proc.setdefault(s["procedure_pk"], []).append(
+                            _loads(s["step_json"], {}))
+    finally:
+        conn.close()
+    procs_by_run = {}
+    for p in procs:
+        procs_by_run.setdefault(p["run_id"], []).append(p)
+    out = {}
+    for cid, (_k, run) in best.items():
+        rprocs = procs_by_run.get(run["id"], [])
+        if lite:
+            procedures = [{"id": p["procedure_id"], "name": p["name"],
+                           "steps": _lite_steps(steps_by_proc.get(p["id"], []))} for p in rprocs]
+            out[cid] = {"case_id": cid, "version": run["version"],
+                        "title": run["post_title"], "procedures": procedures}
+        else:
+            procedures = [{
+                "id": p["procedure_id"], "name": p["name"], "purpose": p["purpose"],
+                "category": p["category"], "declarations": _loads(p["declarations"]),
+                "type_registry": _loads(p["type_registry"]),
+                "steps": steps_by_proc.get(p["id"], []),
+                "tools_used": _loads(p["tools_used"], [])} for p in rprocs]
+            out[cid] = {"case_id": cid, "version": run["version"], "platform": run["platform"],
+                        "title": run["post_title"], "model": run["model"],
+                        "cost_usd": float(run["cost_usd"]) if run["cost_usd"] is not None else None,
+                        "duration_s": run["duration_s"], "source": _loads(run["source"]),
+                        "procedures": procedures}
+    return out
+
+
+def fetch_all_process_pairs():
+    """全量重跑归类用:每个有解构记录的 case 取其「最新真实版」(link_ 排后、id 最大)
+    所在行的 (query_id, case_id)。选版口径与 fetch_process / fetch_all_process 完全一致,
+    保证回写的版本即前端展示的版本。query_id 作 post_id 供 category-match record。
+    返回 [(query_id, case_id)],按 case_id 排序、按 case 去重。"""
+    conn = _conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SELECT id, query_id, case_id, version FROM process_run ORDER BY case_id, id")
+            rows = cur.fetchall()
+    finally:
+        conn.close()
+    best = {}  # case_id -> ((is_real, id), query_id);取 max,与 fetch_all_process 同口径
+    for r in rows:
+        c = r["case_id"]
+        key = (not str(r["version"]).startswith("link_"), r["id"])
+        if c not in best or key > best[c][0]:
+            best[c] = (key, r["query_id"])
+    return [(qid, c) for c, (_k, qid) in sorted(best.items())]
+
+
+# ── 拆表(3-II):mode_process → process_run/procedure/step ────────────────────────
+
+def _jn(v):
+    """JSON 列搬运规整:先 _loads 再 _j,无论 pymysql 返回 str 还是已解析对象都得到合法 JSON 文本/None。"""
+    return _j(_loads(v))
+
+
+def _t(s, n):
+    """抽取列按字符数截断(只用于可查询列;完整值在 step_json,不丢)。None 原样。"""
+    return s if (s is None or len(str(s)) <= n) else str(s)[:n]
+
+
+def migrate_to_split_tables(truncate=True):
+    """把 mode_process 灌入 process_run/procedure/step(加法,不动 mode_process)。
+    幂等:默认先 TRUNCATE 三表再灌。step_json 逐字保留原 step。返回 {runs, procedures, steps}。"""
+    conn = _conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute("SELECT * FROM mode_process ORDER BY query_id, case_id, version, seq, id")
             rows = cur.fetchall()
     finally:
         conn.close()
-    by_case = {}
+    # 每 case 最新真实版(口径同 fetch_all_process),用于置 is_latest
+    best = {}
+    for r in rows:
+        c = r["case_id"]; k = (not str(r["version"]).startswith("link_"), r["id"])
+        if c not in best or k > best[c][0]:
+            best[c] = (k, r["version"])
+    latest_ver = {c: v for c, (_k, v) in best.items()}
+    # 按 (query_id, case_id, version) 分组 = 一个 run(保序)
+    runs = {}
+    order = []
     for r in rows:
-        by_case.setdefault(r["case_id"], []).append(r)
+        key = (r["query_id"], r["case_id"], r["version"])
+        if key not in runs:
+            runs[key] = []; order.append(key)
+        runs[key].append(r)
+
+    nrun = nproc = nstep = 0
+    conn = _conn()
+    try:
+        conn.begin()
+        with conn.cursor() as cur:
+            if truncate:
+                for t in ("process_step", "process_procedure", "process_run"):
+                    cur.execute(f"TRUNCATE TABLE {t}")
+            for key in order:
+                qid, cid, ver = key
+                prows = sorted(runs[key],
+                               key=lambda r: (r["seq"] if r["seq"] is not None else 0, r["id"]))
+                first = prows[0]
+                is_latest = 1 if (not str(ver).startswith("link_") and latest_ver.get(cid) == ver) else 0
+                cur.execute("""INSERT INTO process_run
+                    (query_id, case_id, version, is_latest, platform, post_title, source,
+                     model, cost_usd, duration_s, created_at)
+                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+                    (qid, cid, ver, is_latest, first["platform"], first["post_title"],
+                     _jn(first["source"]), first["model"], first["cost_usd"], first["duration_s"],
+                     first["created_at"]))   # 保留原解构时间(Dashboard 成本趋势按此)
+                run_id = cur.lastrowid; nrun += 1
+                for pseq, pr in enumerate(prows):
+                    steps = _loads(pr["steps"], [])
+                    cur.execute("""INSERT INTO process_procedure
+                        (run_id, case_id, version, procedure_id, seq, name, purpose, category,
+                         declarations, type_registry, tools_used, step_count)
+                        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+                        (run_id, cid, ver, pr["procedure_id"], pseq, pr["name"], pr["purpose"],
+                         pr["category"], _jn(pr["declarations"]), _jn(pr["type_registry"]),
+                         _jn(pr["tools_used"]), pr["step_count"]))
+                    proc_pk = cur.lastrowid; nproc += 1
+                    for sseq, st in enumerate(steps if isinstance(steps, list) else []):
+                        if not isinstance(st, dict):
+                            st = {"value": st}
+                        cur.execute("""INSERT INTO process_step
+                            (procedure_pk, case_id, version, step_id, seq, kind, grp, via,
+                             effect, action, substance, form, step_json)
+                            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+                            (proc_pk, cid, ver, _t(st.get("id"), 16), sseq,
+                             _t(st.get("kind"), 16), _t(st.get("group"), 16), _t(st.get("via"), 255),
+                             _t(st.get("effect"), 255), _t(st.get("action"), 255),
+                             _t(st.get("substance"), 512), _t(st.get("form"), 512),
+                             _j(st)))
+                        nstep += 1
+        conn.commit()
+    except Exception:
+        conn.rollback(); raise
+    finally:
+        conn.close()
+    return {"runs": nrun, "procedures": nproc, "steps": nstep}
+
+
+def _steps_by_proc(cur, run_id):
+    """取某 run 下所有 step 的 step_json,按 procedure_pk 分组(组内按 seq 序)。"""
+    cur.execute("""SELECT procedure_pk, step_json FROM process_step
+                   WHERE procedure_pk IN (SELECT id FROM process_procedure WHERE run_id=%s)
+                   ORDER BY procedure_pk, seq, id""", (run_id,))
     out = {}
-    for cid, crows in by_case.items():
-        # 选版本:非 link_ 优先(is_real=True 排前),再按 id 最大——口径同 fetch_process
-        best = max(crows, key=lambda r: (not str(r["version"]).startswith("link_"), r["id"]))
-        ver = best["version"]
-        vrows = sorted((r for r in crows if r["version"] == ver),
-                       key=lambda r: (r["seq"] if r["seq"] is not None else 0, r["id"]))
-        out[cid] = _proc_payload(cid, ver, vrows, lite=lite)
+    for s in cur.fetchall():
+        out.setdefault(s["procedure_pk"], []).append(_loads(s["step_json"], {}))
     return out
 
 
+def _split_payload(cur, run, lite=False):
+    """run 行 → ProcessPayload(形状与 _proc_payload 完全一致)。需开放 cursor。无 run 返回 None。"""
+    if not run:
+        return None
+    cur.execute("SELECT * FROM process_procedure WHERE run_id=%s ORDER BY seq, id", (run["id"],))
+    procs = cur.fetchall()
+    sbp = _steps_by_proc(cur, run["id"])
+    case_id, version = run["case_id"], run["version"]
+    if lite:
+        procedures = [{
+            "id": p["procedure_id"], "name": p["name"],
+            "steps": _lite_steps(sbp.get(p["id"], [])),
+        } for p in procs]
+        return {"case_id": case_id, "version": version,
+                "title": run["post_title"], "procedures": procedures}
+    procedures = [{
+        "id": p["procedure_id"], "name": p["name"], "purpose": p["purpose"],
+        "category": p["category"], "declarations": _loads(p["declarations"]),
+        "type_registry": _loads(p["type_registry"]), "steps": sbp.get(p["id"], []),
+        "tools_used": _loads(p["tools_used"], []),
+    } for p in procs]
+    return {"case_id": case_id, "version": version, "platform": run["platform"],
+            "title": run["post_title"], "model": run["model"],
+            "cost_usd": float(run["cost_usd"]) if run["cost_usd"] is not None else None,
+            "duration_s": run["duration_s"],
+            "source": _loads(run["source"]), "procedures": procedures}
+
+
+def _select_run(cur, case_id, version=None, query_id=None):
+    """选 run:version 指定则精确取;否则取最新(is_latest 优先 → 非 link_ → id 最大,
+    口径同旧 fetch_process)。query_id 给定则限定该 query。无则 None。"""
+    where = "case_id=%s"; params = [case_id]
+    if query_id is not None:
+        where += " AND query_id=%s"; params.append(query_id)
+    if version is not None:
+        cur.execute(f"SELECT * FROM process_run WHERE {where} AND version=%s LIMIT 1",
+                    params + [version])
+    else:
+        cur.execute(f"""SELECT * FROM process_run WHERE {where}
+                        ORDER BY is_latest DESC, (LEFT(version,5)='link_') ASC, id DESC
+                        LIMIT 1""", params)
+    return cur.fetchone()
+
+
+def rebuild_process_from_split(case_id, version=None):
+    """从三表重建 ProcessPayload(迁移校验/读侧共用)。version=None 取最新。"""
+    conn = _conn()
+    try:
+        with conn.cursor() as cur:
+            return _split_payload(cur, _select_run(cur, case_id, version))
+    finally:
+        conn.close()
+
+
+def _refresh_is_latest(cur, case_id):
+    """重算某 case 各 run 的 is_latest:最新真实版(非 link_、id 最大)置 1,其余 0。
+    口径同 fetch_process 选版。只有 link_ 时无 1(读侧 rebuild 仍能按排序兜底)。需在事务内调用。"""
+    cur.execute("SELECT id, version FROM process_run WHERE case_id=%s", (case_id,))
+    rows = cur.fetchall()
+    if not rows:
+        return
+    cur.execute("UPDATE process_run SET is_latest=0 WHERE case_id=%s", (case_id,))
+    best = max(rows, key=lambda r: (not str(r["version"]).startswith("link_"), r["id"]))
+    if not str(best["version"]).startswith("link_"):
+        cur.execute("UPDATE process_run SET is_latest=1 WHERE id=%s", (best["id"],))
+
+
+def _split_write_run(cur, query_id, case_id, platform, post_title, source,
+                     procedures, model, version, cost_usd, duration_s):
+    """在三表覆盖写某 (query_id, case_id, version) 的 run/procedure/step(需在调用方事务内)。
+    与 replace_process 写 mode_process 同口径(tools_used 从 steps[].via 去重)。维护 is_latest。"""
+    cur.execute("SELECT id FROM process_run WHERE query_id=%s AND case_id=%s AND version=%s",
+                (query_id, case_id, version))
+    old = cur.fetchone()
+    if old:
+        cur.execute("""DELETE FROM process_step WHERE procedure_pk IN
+                       (SELECT id FROM process_procedure WHERE run_id=%s)""", (old["id"],))
+        cur.execute("DELETE FROM process_procedure WHERE run_id=%s", (old["id"],))
+        cur.execute("DELETE FROM process_run WHERE id=%s", (old["id"],))
+    cur.execute("""INSERT INTO process_run
+        (query_id, case_id, version, is_latest, platform, post_title, source,
+         model, cost_usd, duration_s)
+        VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+        (query_id, case_id, version, 0, platform, (post_title or "")[:500],
+         _j(source), model, cost_usd, duration_s))
+    run_id = cur.lastrowid
+    for pseq, p in enumerate(procedures or []):
+        steps = p.get("steps") or []
+        vias = []
+        for s in steps:
+            v = s.get("via") if isinstance(s, dict) else None
+            if v and v not in vias:
+                vias.append(v)
+        cur.execute("""INSERT INTO process_procedure
+            (run_id, case_id, version, procedure_id, seq, name, purpose, category,
+             declarations, type_registry, tools_used, step_count)
+            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+            (run_id, case_id, version, p.get("id"), pseq, (p.get("name") or "")[:250],
+             p.get("purpose"), p.get("category"), _j(p.get("declarations")),
+             _j(p.get("type_registry")), _j(vias), len(steps)))
+        proc_pk = cur.lastrowid
+        for sseq, st in enumerate(steps):
+            if not isinstance(st, dict):
+                st = {"value": st}
+            cur.execute("""INSERT INTO process_step
+                (procedure_pk, case_id, version, step_id, seq, kind, grp, via,
+                 effect, action, substance, form, step_json)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+                (proc_pk, case_id, version, _t(st.get("id"), 16), sseq,
+                 _t(st.get("kind"), 16), _t(st.get("group"), 16), _t(st.get("via"), 255),
+                 _t(st.get("effect"), 255), _t(st.get("action"), 255),
+                 _t(st.get("substance"), 512), _t(st.get("form"), 512), _j(st)))
+    _refresh_is_latest(cur, case_id)
+
+
+def _split_update_steps(cur, case_id, version, steps_in_order):
+    """归类回写:按工序顺序覆盖某 (case_id, version) 各 procedure 的 step_json(及抽取列)。
+    steps_in_order 与 process_procedure(按 seq) 一一对应。需在事务内。返回更新的 step 行数。"""
+    cur.execute("""SELECT id FROM process_procedure
+                   WHERE case_id=%s AND version=%s ORDER BY seq, id""", (case_id, version))
+    proc_ids = [r["id"] for r in cur.fetchall()]
+    if len(proc_ids) != len(steps_in_order):
+        raise ValueError(f"process_procedure 行数({len(proc_ids)})与工序数({len(steps_in_order)})不一致")
+    n = 0
+    for proc_pk, steps in zip(proc_ids, steps_in_order):
+        cur.execute("DELETE FROM process_step WHERE procedure_pk=%s", (proc_pk,))
+        for sseq, st in enumerate(steps or []):
+            if not isinstance(st, dict):
+                st = {"value": st}
+            cur.execute("""INSERT INTO process_step
+                (procedure_pk, case_id, version, step_id, seq, kind, grp, via,
+                 effect, action, substance, form, step_json)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
+                (proc_pk, case_id, version, _t(st.get("id"), 16), sseq,
+                 _t(st.get("kind"), 16), _t(st.get("group"), 16), _t(st.get("via"), 255),
+                 _t(st.get("effect"), 255), _t(st.get("action"), 255),
+                 _t(st.get("substance"), 512), _t(st.get("form"), 512), _j(st)))
+            n += 1
+    return n
+
+
+# ── step_classification(归类独立表,3-I)────────────────────────────────────────
+
+def replace_step_classification(case_id, version, records):
+    """覆盖写某 (case_id, version) 的 step 维度归类结果(DELETE+INSERT 原子,语义同 replace_process)。
+    records:[{query_id, procedure_id, step_id, dimension, sub_index, raw_term,
+              matched_name, matched_path, matched_id, score}](只含有命中的子项)。
+    同版本重跑幂等、洗净旧归属。返回写入条数。"""
+    conn = _conn()
+    try:
+        conn.begin()
+        with conn.cursor() as cur:
+            cur.execute("DELETE FROM step_classification WHERE case_id=%s AND version=%s",
+                        (case_id, version))
+            if records:
+                rows = [(
+                    case_id, r.get("query_id"), version,
+                    r.get("procedure_id"), r.get("step_id"),
+                    r.get("dimension"), r.get("sub_index"), r.get("raw_term"),
+                    r.get("matched_name"), r.get("matched_path"),
+                    r.get("matched_id"), r.get("score"),
+                ) for r in records]
+                cur.executemany("""
+                INSERT INTO step_classification
+                  (case_id, query_id, version, procedure_id, step_id, dimension,
+                   sub_index, raw_term, matched_name, matched_path, matched_id, score)
+                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                """, rows)
+        conn.commit()
+        return len(records)
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
+
+
+def fetch_classified_cases(dimension, path=None, name=None):
+    """维度联动取帖:返回某维度下命中「该节点(及其整棵子树)」的 distinct case_id 集合。
+      - path:给定分类全路径,取该节点 **及其子树**(matched_path = path 或 LIKE 'path/%')。
+      - name:给定分类名精确匹配(matched_name = name)。
+    path 与 name 至少给一个;都给则取并集口径以 path 为主(name 作补充 OR)。
+    与 mode_process.steps[].substanceMatch 同源(同一次归类双写),不依赖 cat-api 树。"""
+    conds, params = [], [dimension]
+    if path:
+        conds.append("(matched_path = %s OR matched_path LIKE %s)")
+        params += [path, path.rstrip("/") + "/%"]
+    if name:
+        conds.append("matched_name = %s")
+        params.append(name)
+    if not conds:
+        return set()
+    where = "dimension=%s AND (" + " OR ".join(conds) + ")"
+    conn = _conn()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(f"SELECT DISTINCT case_id FROM step_classification WHERE {where}", params)
+            return {r["case_id"] for r in cur.fetchall()}
+    finally:
+        conn.close()
+
+
 def fetch_process_by_query(query_id, case_id, version=None):
     """同 fetch_process,但用 (query_id, case_id) 精确定位某 query 下该帖的工序
     (category-match 用:post_id=query_id / knowledge_id=case_id)。
@@ -840,20 +1292,9 @@ def fetch_process_by_query(query_id, case_id, version=None):
     conn = _conn()
     try:
         with conn.cursor() as cur:
-            if version is None:
-                cur.execute("""SELECT version FROM mode_process WHERE query_id=%s AND case_id=%s
-                               ORDER BY (LEFT(version,5)='link_') ASC, id DESC LIMIT 1""",
-                            (query_id, case_id))
-                row = cur.fetchone()
-                if not row:
-                    return None
-                version = row["version"]
-            cur.execute("""SELECT * FROM mode_process WHERE query_id=%s AND case_id=%s AND version=%s
-                           ORDER BY seq, id""", (query_id, case_id, version))
-            rows = cur.fetchall()
+            return _split_payload(cur, _select_run(cur, case_id, version, query_id=query_id))
     finally:
         conn.close()
-    return _proc_payload(case_id, version, rows)
 
 
 def update_process_steps_by_query(query_id, case_id, version, steps_in_order):
@@ -874,6 +1315,8 @@ def update_process_steps_by_query(query_id, case_id, version, steps_in_order):
             for row_id, steps in zip(ids, steps_in_order):
                 cur.execute("UPDATE mode_process SET steps=%s WHERE id=%s", (_j(steps), row_id))
                 n += cur.rowcount
+            # 3-II 双写:同步更新 process_step.step_json(同事务)
+            _split_update_steps(cur, case_id, version, steps_in_order)
         conn.commit()
         return n
     except Exception:
@@ -902,6 +1345,8 @@ def update_process_steps(case_id, version, steps_in_order):
             for row_id, steps in zip(ids, steps_in_order):
                 cur.execute("UPDATE mode_process SET steps=%s WHERE id=%s", (_j(steps), row_id))
                 n += cur.rowcount
+            # 3-II 双写:同步更新 process_step.step_json(同事务)
+            _split_update_steps(cur, case_id, version, steps_in_order)
         conn.commit()
         return n
     except Exception:
@@ -933,16 +1378,25 @@ def fetch_categorized_cases(case_ids, mode="process"):
     供前端判断「是否已全部归类 → 提示重新归类」。仅工序方向有意义(mode_process)。"""
     if not case_ids:
         return set()
-    table = _mode_table(mode)
     ph = ",".join(["%s"] * len(case_ids))
     conn = _conn()
     try:
         with conn.cursor() as cur:
-            # 拉各行的 (case_id, version, id, islink, cat);steps LIKE 在库端算,不拉 steps 大字段。
-            cur.execute(f"""SELECT case_id, version, id,
-                                   (LEFT(version,5)='link_') AS islink, (steps LIKE %s) AS cat
-                            FROM {table} WHERE case_id IN ({ph})""",
-                        ['%substanceMatch%'] + list(case_ids))
+            if mode == "process":
+                # 3-II:run 级聚合「该 run 是否有任一 step_json 含 substanceMatch」
+                cur.execute(f"""SELECT r.case_id, r.version, r.id,
+                                       (LEFT(r.version,5)='link_') AS islink,
+                                       EXISTS(SELECT 1 FROM process_step s
+                                              JOIN process_procedure p ON s.procedure_pk=p.id
+                                              WHERE p.run_id=r.id AND s.step_json LIKE %s) AS cat
+                                FROM process_run r WHERE r.case_id IN ({ph})""",
+                            ['%substanceMatch%'] + list(case_ids))
+            else:
+                table = _mode_table(mode)
+                cur.execute(f"""SELECT case_id, version, id,
+                                       (LEFT(version,5)='link_') AS islink, (steps LIKE %s) AS cat
+                                FROM {table} WHERE case_id IN ({ph})""",
+                            ['%substanceMatch%'] + list(case_ids))
             rows = cur.fetchall()
     finally:
         conn.close()
@@ -1102,7 +1556,24 @@ def fetch_extract(mode, case_id, version=None):
     """一次取版本列表 + 解构详情,复用同一条池连接、最少往返。
     返回 {versions, data, missing}。mode: process / tools。"""
     is_proc = mode != "tools"
-    mtable = _mode_table("process" if is_proc else "tools")
+    if is_proc:
+        # 3-II:工序方向版本列表 + 详情都读自三表(形状不变)
+        conn = _conn()
+        try:
+            with conn.cursor() as cur:
+                cur.execute("""SELECT r.version, COUNT(p.id) AS n, MAX(r.model) AS model
+                               FROM process_run r
+                               LEFT JOIN process_procedure p ON p.run_id = r.id
+                               WHERE r.case_id=%s GROUP BY r.version
+                               ORDER BY (LEFT(r.version,5)='link_') ASC, MAX(r.id) DESC""", (case_id,))
+                versions = cur.fetchall()
+                target = version or (versions[0]["version"] if versions else None)
+                payload = _split_payload(cur, _select_run(cur, case_id, target)) if target else None
+        finally:
+            conn.close()
+        return {"versions": versions, "data": payload, "missing": payload is None}
+    # 工具方向:仍读 mode_tools(本期不拆)
+    mtable = _mode_table("tools")
     conn = _conn()
     try:
         with conn.cursor() as cur:
@@ -1111,7 +1582,6 @@ def fetch_extract(mode, case_id, version=None):
                             GROUP BY version
                             ORDER BY (LEFT(version,5)='link_') ASC, MAX(id) DESC""", (case_id,))
             versions = cur.fetchall()
-            # 详情:把"取最新版本"折进同一条 SQL,版本指定时直接用;省一次往返。
             target = version or (versions[0]["version"] if versions else None)
             rows = []
             if target is not None:
@@ -1120,7 +1590,7 @@ def fetch_extract(mode, case_id, version=None):
                 rows = cur.fetchall()
     finally:
         conn.close()
-    payload = (_proc_payload if is_proc else _tools_payload)(case_id, target, rows)
+    payload = _tools_payload(case_id, target, rows)
     return {"versions": versions, "data": payload, "missing": payload is None}
 
 
@@ -1131,7 +1601,7 @@ def fetch_extract(mode, case_id, version=None):
 def latest_real_version(case_id, mode="process"):
     """该 case 是否已有「真实」解构(任意 query;link_* 是复制品,不算源)。
     返回最新一行 {"version","query_id"} 或 None。给解构前去重判定用。"""
-    table = _mode_table(mode)
+    table = "process_run" if mode == "process" else _mode_table(mode)  # 3-II:工序读三表的 run
     conn = _conn()
     try:
         with conn.cursor() as cur:
@@ -1176,6 +1646,18 @@ def link_process(query_id, case_id, mode="process"):
                 cur.execute(
                     f"INSERT INTO {table} ({','.join(cols)}) VALUES ({','.join(['%s']*len(cols))})",
                     [row[k] for k in cols])
+            # 3-II 双写(仅工序方向):把 link_ 复制版同步写进三表
+            if mode == "process" and rows:
+                srows = sorted(rows, key=lambda r: (r.get("seq") if r.get("seq") is not None else 0))
+                f = srows[0]
+                procedures = [{
+                    "id": r.get("procedure_id"), "name": r.get("name"), "purpose": r.get("purpose"),
+                    "category": r.get("category"), "declarations": _loads(r.get("declarations")),
+                    "type_registry": _loads(r.get("type_registry")), "steps": _loads(r.get("steps"), []),
+                } for r in srows]
+                _split_write_run(cur, query_id, case_id, f.get("platform"), f.get("post_title"),
+                                 _loads(f.get("source")), procedures, f.get("model"), newver,
+                                 0, f.get("duration_s"))
             return len(rows)
     finally:
         conn.close()
@@ -1210,7 +1692,7 @@ def fetch_adopted_process_cases(query_id=None):
     sql = (f"SELECT DISTINCT s.case_id, s.overall_score, s.publish_time, "
            f"{_REL_SQL} AS rel, {_REPRO_SQL} AS repro "
            "FROM search_process s "
-           "JOIN (SELECT DISTINCT case_id FROM mode_process) m ON s.case_id = m.case_id")
+           "JOIN (SELECT DISTINCT case_id FROM process_run) m ON s.case_id = m.case_id")  # 3-II
     params = ()
     if query_id:
         sql += " WHERE s.query_id=%s"
@@ -1396,17 +1878,25 @@ def fetch_dashboard_rows():
             for p in st:
                 p["mode"] = "tools"
             posts += st
-            # 成本/耗时按全部版本计;steps 仅最新版需要 → 非最新版只回 NULL,省传输。
-            cur.execute("""SELECT p.id, p.case_id, p.version, p.cost_usd, p.duration_s, p.created_at,
-                                  CASE WHEN p.version = m.maxv THEN p.steps END AS steps
-                           FROM mode_process p
-                           JOIN (SELECT t.case_id, t.version AS maxv FROM mode_process t
-                                 JOIN (SELECT case_id, MAX(id) AS mid FROM mode_process
-                                       WHERE LEFT(version,5) <> 'link_' GROUP BY case_id) x
-                                   ON t.id = x.mid) m
-                             ON p.case_id = m.case_id
-                           ORDER BY p.id""")
-            procs = cur.fetchall()
+            # 3-II:成本/耗时按 run(=case+version)取自 process_run;steps 仅最新版(is_latest)需要,
+            # 其余 run steps=[] 省传输。返回形状与旧 mode_process 段一致(id/case_id/version/cost/dur/created_at/steps)。
+            cur.execute("""SELECT id, case_id, version, cost_usd, duration_s, created_at, is_latest
+                           FROM process_run ORDER BY id""")
+            run_rows = cur.fetchall()
+            latest_ids = [r["id"] for r in run_rows if r["is_latest"]]
+            steps_by_run = {}
+            if latest_ids:
+                ph = ",".join(["%s"] * len(latest_ids))
+                cur.execute(f"""SELECT p.run_id, s.step_json FROM process_step s
+                                JOIN process_procedure p ON s.procedure_pk = p.id
+                                WHERE p.run_id IN ({ph})
+                                ORDER BY p.run_id, p.seq, s.seq, s.id""", latest_ids)
+                for row in cur.fetchall():
+                    steps_by_run.setdefault(row["run_id"], []).append(_loads(row["step_json"], {}))
+            procs = [{"id": r["id"], "case_id": r["case_id"], "version": r["version"],
+                      "cost_usd": r["cost_usd"], "duration_s": r["duration_s"],
+                      "created_at": r["created_at"],
+                      "steps": steps_by_run.get(r["id"], [])} for r in run_rows]
             cur.execute("""SELECT id, case_id, version, tool_name, substance_scope,
                                   form_scope, cost_usd, duration_s, created_at
                            FROM mode_tools""")
@@ -1432,7 +1922,8 @@ def check():
     conn = _conn()
     try:
         with conn.cursor() as cur:
-            for t in ("search_process", "search_tools", "mode_process", "mode_tools"):
+            for t in ("search_process", "search_tools", "mode_process", "mode_tools",
+                      "process_run", "process_procedure", "process_step", "step_classification"):
                 cur.execute(f"SELECT COUNT(*) AS n FROM {t}")
                 print(f"{t}: {cur.fetchone()['n']} 行")
     finally:

examples/mode_workflow/docs/step_classification_3-I_设计.md

@@ -0,0 +1,126 @@
+# 归类独立成表(step_classification)· 3-I 设计
+
+> 来源:`归类口径修复与表结构优化方案.md` 第 6 节的 ④(归类级表)。
+> 范围:**只做 3-I**——新增一张 `step_classification` 表 + 回写改造 + 查询接口,**不动 `mode_process` 结构**(第 6.2 ①②③ 的 run/procedure/step 规范化 = 3-II,延后)。
+> 目标:让「维度联动取帖」与「工序解构表里显示的归类值」**同源**,从根上消除第 1 节的「树多归属 ≠ 表单归属」,且不再依赖 cat-api 树的 `knowledge_ids` / `top_k`。
+
+---
+
+## 1. 现状回顾(本项目侧)
+
+- `stages/category_match.py`:对一帖五维度词(实质/形式/类型/作用/动作)调 cat-api `/api/v1/category-match`,`enrich_steps()` 只把**实质/形式**的「单一最优 name」回写进 `mode_process.steps[].substanceMatch / formMatch`(JSON blob 内)。
+- cat-api 侧 `record=True` 把命中节点记进分类库(= 树的 `knowledge_ids`)。`top_k` 已在批次 0 改为 `1`。
+- 前端(**aigc-web**,非本项目)维度联动:点分类树节点 → 用该节点 cat-api `knowledge_ids` 取帖 → 与帖里显示的 `substanceMatch` 口径不一致。
+
+痛点:归类结果埋在 `steps` JSON 里,DB 层**无法按归类值查询/索引**;前端只能依赖 cat-api 树,导致多归属不一致。
+
+---
+
+## 2. 新增表 DDL(MySQL,加入 `db.py`)
+
+一行 = 某 case 某版本某工序某 step 某维度某子项的归类结果。
+
+```sql
+CREATE TABLE IF NOT EXISTS step_classification (
+  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
+  case_id       VARCHAR(128) NOT NULL,
+  query_id      VARCHAR(32)  NULL  COMMENT 'record 时的 post_id,便于溯源',
+  version       VARCHAR(32)  NULL  COMMENT '对齐 mode_process 最新真实版,重跑覆盖',
+  procedure_id  VARCHAR(16)  NULL  COMMENT 'p1,p2…',
+  step_id       VARCHAR(16)  NULL  COMMENT 's1,s2…',
+  dimension     VARCHAR(8)   NOT NULL COMMENT '实质/形式/类型/作用/动作',
+  sub_index     SMALLINT     NOT NULL COMMENT '原值「、」拆分后的子项下标(等长等序)',
+  raw_term      VARCHAR(255) NULL  COMMENT '原始词',
+  matched_name  VARCHAR(255) NULL  COMMENT '命中分类名(单一最优;无命中不入库)',
+  matched_path  VARCHAR(512) NULL  COMMENT '命中分类全路径,如 /表象/视觉/空间/空间环境',
+  matched_id    INT          NULL  COMMENT 'cat-api stable_id',
+  score         FLOAT        NULL,
+  match_run_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  UNIQUE KEY uk_step_dim_sub (case_id, version, procedure_id, step_id, dimension, sub_index),
+  KEY idx_case (case_id),
+  KEY idx_dim_name (dimension, matched_name),
+  KEY idx_dim_path (dimension, matched_path)   -- matched_path LIKE '前缀%' 走索引
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='step 维度归类结果(与 mode_process 解耦)';
+```
+
+设计要点:
+- **只入「有命中」的子项**(matched_name 非空)。无命中子项不入库——维度联动只关心命中的;`steps[].substanceMatch` 仍保留「无」占位供前端按下标对齐,二者职责分开。
+- 用**自然键** `(case_id, version, procedure_id, step_id, dimension, sub_index)` 唯一,不引 `process_step.id` 外键——这样 3-I 完全独立于 3-II;将来 3-II 落地再加 FK。
+- 写入用 **DELETE(by case_id,version)+INSERT** 原子覆盖(同 `replace_process`),重跑幂等、不留旧归属。
+
+---
+
+## 3. 回写改造(`stages/category_match.py`)
+
+现 `_build_match_lookup` 只取 name。新增「全字段」lookup 与「记录构造」,**五维度全覆盖**(现仅实质/形式):
+
+```text
+_build_match_full(resp) -> {(term, source_type): {name, path, id, score}}
+    择优规则不变:精确同名(name==term)优先,再比 score。
+
+build_classification(procedures, resp, *, case_id, query_id, version) -> [record...]
+    遍历每 proc/step,对五维度逐子项(按「、」拆,_split_values)查 lookup:
+      实质 ← step.substance        形式 ← step.form
+      作用 ← step.effect           动作 ← step.action
+      类型 ← step.inputs[].type(多 input 顺序展开)
+    命中则产出一行 record(含 procedure_id=proc.id, step_id=step.id, dimension,
+    sub_index, raw_term, matched_*),无命中跳过。
+```
+
+`process_one()` 流程在「`enrich_steps` 回写 steps」之后,**新增一步**:
+
+```text
+recs = build_classification(procedures, resp, case_id, query_id, version)
+db.replace_step_classification(case_id, version, recs)   # DELETE+INSERT 原子覆盖
+```
+
+- `enrich_steps`(写 `substanceMatch/formMatch` 到 steps)**保留不动** → 前端工序表显示不受影响。
+- 即:一次归类,**双写**——steps 内显示值(name) + step_classification(name+path+score,五维)。
+
+---
+
+## 4. 读侧接口
+
+`db.py` 新增:
+
+```text
+fetch_classified_cases(dimension, path=None, name=None) -> set[case_id]
+    SELECT DISTINCT case_id FROM step_classification
+    WHERE dimension=%s AND ( matched_path = %s OR matched_path LIKE CONCAT(%s,'/%') )
+    -- 给定 path 取「该节点及其整棵子树」的帖;或给定 name 精确取。
+```
+
+`server.py` 新增路由:
+
+```text
+GET /api/classified_cases?dimension=实质&path=/表象/视觉/空间/空间环境
+   -> {"dimension","path","cases":[...],"count":N}
+```
+
+aigc-web 维度联动据此把「取 case 的来源」从 cat-api `node.knowledge_ids` 切到本接口
+(→ 与 `substanceMatch` 同源,彻底解决树多归属)。**此前端改动属 aigc-web,3-I 不含,作为协同项单列。**
+
+---
+
+## 5. 数据填充(与重跑合并,不单独迁移)
+
+- step_classification 的正确数据(含 path/score)只能来自**重跑** category-match(批次 4):旧 `steps` 里只有 name、无 path,回填无意义。
+- 故 3-I **不写迁移脚本**;建表后表为空,**批次 4 重跑即填充**(每帖 `replace_step_classification`)。这与「重跑放最后」一致。
+
+---
+
+## 6. 落地清单(3-I)
+
+1. `db.py`:`DDL_STEP_CLASSIFICATION` + `init_tables()` 建表 + `replace_step_classification()` 写 + `fetch_classified_cases()` 读。
+2. `stages/category_match.py`:`_build_match_full()` + `build_classification()`,`process_one()` 调 `replace_step_classification`。
+3. `server.py`:`GET /api/classified_cases`。
+4. 跑 `python db.py init`(幂等建新表)。
+5. (协同/后续)aigc-web 维度联动切到 `/api/classified_cases`。
+6. (批次 4)重跑填充 step_classification。
+
+---
+
+## 7. 风险与回退
+
+- 纯**新增**:新表 + 新函数 + 新路由 + category_match 多一步写;不改任何现有读写口径与表结构 → 旧链路完全不受影响,可回退(删表/去掉那一步写)。
+- 重跑前表为空,不影响现有功能。

examples/mode_workflow/server.py

@@ -676,6 +676,17 @@ class Handler(BaseHTTPRequestHandler):
                 cids = [c for c in qs.get("case_ids", "").split(",") if c.strip()]
                 done = db.fetch_categorized_cases(cids, qs.get("mode", "process")) if cids else set()
                 self._json({"categorized": sorted(done), "total": len(cids)})
+            elif u.path == "/api/classified_cases":
+                # 维度联动取帖(3-I):按 step_classification.matched_path 前缀取某节点(及其子树)
+                # 的 distinct case,与帖里显示的 substanceMatch 同源,替代 cat-api 树的 knowledge_ids。
+                dim = qs.get("dimension", "实质")
+                path = qs.get("path") or None
+                name = qs.get("name") or None
+                if not path and not name:
+                    return self._err("需提供 path 或 name", 400)
+                cases = db.fetch_classified_cases(dim, path=path, name=name)
+                self._json({"dimension": dim, "path": path, "name": name,
+                            "cases": sorted(cases), "count": len(cases)})
             elif u.path == "/api/post":
                 # 单帖详情(正文/配图/评估全量):列表已瘦身,详情按需取;带 ETag/304
                 r = db.fetch_post(qs.get("query_id", ""), qs.get("case_id", ""),

examples/mode_workflow/stages/category_match.py

@@ -54,7 +54,7 @@ import db  # noqa: E402
 # ── 配置(后续按需调整)──────────────────────────────────────────────────────────
 CATEGORY_MATCH_API = os.environ.get("CATEGORY_MATCH_API", "http://47.236.83.130:8300").rstrip("/")
 MATCH_ENDPOINT = "/api/v1/category-match"
-TOP_K = int(os.environ.get("CATEGORY_MATCH_TOP_K", "10"))          # 后续再调
+TOP_K = int(os.environ.get("CATEGORY_MATCH_TOP_K", "1"))           # 单归属:每个词只记 1 个最优节点(避免树多归属,见 归类口径修复方案.md 第4节)
 MIN_SCORE = float(os.environ.get("CATEGORY_MATCH_MIN_SCORE", "0.8"))  # 后续再调
 RECORD = True
 API_TIMEOUT = float(os.environ.get("CATEGORY_MATCH_TIMEOUT", "60"))   # 下游单帖可达 30s+,默认放宽到 60
@@ -159,6 +159,63 @@ def _build_match_lookup(resp: dict) -> dict:
     return {k: v[1] for k, v in best.items()}
 
 
+def _build_match_full(resp: dict) -> dict:
+    """{(term, source_type): {"name","path","id","score"}}。择优规则同 _build_match_lookup
+    (精确同名优先,再比 score),但保留全字段(path/score/stable_id)供 step_classification 落库。"""
+    best = {}  # key -> (rank, full)
+    for item in (resp.get("items") or []):
+        term = item.get("term")
+        for m in (item.get("matches") or []):
+            st, name, score = m.get("source_type"), m.get("name"), m.get("score") or 0
+            if term is None or st is None or name is None:
+                continue
+            key = (term, st)
+            rank = (name == term, score)
+            if key not in best or rank > best[key][0]:
+                best[key] = (rank, {"name": name, "path": m.get("path"),
+                                    "id": m.get("stable_id"), "score": m.get("score")})
+    return {k: v[1] for k, v in best.items()}
+
+
+def _dim_terms(step: dict):
+    """产出 [(dimension, [子项,...]), ...](五维度,顺序固定)。类型把多 input 的 type 顺序展开。
+    每个原值按「、」拆(_split_values),用于 step_classification 按子项落 sub_index。"""
+    type_terms = []
+    for inp in (step.get("inputs") or []):
+        type_terms.extend(_split_values(_s(inp.get("type"))))
+    return [
+        (ST_SUBSTANCE, _split_values(_s(step.get("substance")))),
+        (ST_FORM,      _split_values(_s(step.get("form")))),
+        (ST_TYPE,      type_terms),
+        (ST_EFFECT,    _split_values(_s(step.get("effect")))),
+        (ST_ACTION,    _split_values(_s(step.get("action")))),
+    ]
+
+
+def build_classification(procedures: List[dict], resp: dict, *, query_id: str, version: str) -> list:
+    """构造 step_classification 记录(五维度全覆盖):遍历每 proc/step 逐子项查命中分类(全字段),
+    只产出**有命中**的子项(无命中跳过——steps 内仍有「无」占位供前端按下标对齐)。
+    sub_index 为该 (step, 维度) 内子项的 0-based 下标。供 db.replace_step_classification 落库。"""
+    lookup = _build_match_full(resp)
+    recs = []
+    for proc in (procedures or []):
+        pid = proc.get("id")
+        for step in (proc.get("steps") or []):
+            sid = step.get("id")
+            for dim, terms in _dim_terms(step):
+                for idx, term in enumerate(terms):
+                    m = lookup.get((term, dim))
+                    if not m:
+                        continue
+                    recs.append({
+                        "query_id": query_id, "procedure_id": pid, "step_id": sid,
+                        "dimension": dim, "sub_index": idx, "raw_term": term,
+                        "matched_name": m.get("name"), "matched_path": m.get("path"),
+                        "matched_id": m.get("id"), "score": m.get("score"),
+                    })
+    return recs
+
+
 # 子项无命中时的占位符:保证 substanceMatch/formMatch 与原值子项「等长等序」,
 # 前端才能按下标正确配对(否则丢项/去重会导致 龙舟→人物 这类错位)。
 NO_MATCH = "无"
@@ -262,9 +319,14 @@ async def process_one(client: httpx.AsyncClient, query_id: str, case_id: str,
             db.update_process_steps, case_id, version,
             [p.get("steps") or [] for p in procedures])
 
+        # 3-I 双写:归类结果(name+path+score,五维)落 step_classification,与显示同源
+        recs = build_classification(procedures, resp, query_id=query_id, version=version)
+        cls_written = await asyncio.to_thread(
+            db.replace_step_classification, case_id, version, recs)
+
         out = {**base, "ok": True, "version": version, "items_sent": len(items),
-               "rows_updated": rows_updated, "recorded": resp.get("recorded"),
-               "attempts": attempts}
+               "rows_updated": rows_updated, "classified": cls_written,
+               "recorded": resp.get("recorded"), "attempts": attempts}
         if include_response:
             out["response"] = resp
             out["procedures"] = procedures
@@ -331,22 +393,22 @@ async def category_match_batch(req: BatchReq):
     return {"total": len(results), "ok": ok, "failed": len(results) - ok, "results": results}
 
 
-def _cli_run(query_id: str, case_ids: List[str]) -> int:
-    """CLI 归类:对一个 query 下的若干 case 跑归类,实时打印进度(供 server.py 起子进程、
-    前端轮询日志)。返回退出码:全成功=0,有失败=1(便于任务状态判定)。"""
-    pairs = [(query_id, c) for c in case_ids]
+def _run_pairs(pairs: list) -> int:
+    """对一批 (query_id, case_id) 跑归类,实时打印进度(供 server.py 起子进程、前端轮询日志)。
+    返回退出码:全成功=0,有失败=1(便于任务状态判定)。--run / --all 共用。"""
     if not pairs:
         print("无 case 可归类"); return 0
-    print(f"开始归类:query_id={query_id}  {len(pairs)} 帖  (下游 {CATEGORY_MATCH_API}{MATCH_ENDPOINT})", flush=True)
+    print(f"开始归类:{len(pairs)} 帖  (下游 {CATEGORY_MATCH_API}{MATCH_ENDPOINT}, "
+          f"top_k={TOP_K} min_score={MIN_SCORE} record={RECORD})", flush=True)
     n = len(pairs)
     cnt = {"i": 0}
 
     def _progress(_idx, r):
         cnt["i"] += 1
         tag = "✓ OK  " if r.get("ok") else "✗ FAIL"
-        extra = (f"items={r.get('items_sent')} rows_updated={r.get('rows_updated')} attempts={r.get('attempts')}"
+        extra = (f"items={r.get('items_sent')} rows={r.get('rows_updated')} cls={r.get('classified')} attempts={r.get('attempts')}"
                  if r.get("ok") else (r.get("error") or r.get("note") or ""))
-        print(f"[{cnt['i']}/{n}] {tag} {r.get('case_id')}  {extra}", flush=True)
+        print(f"[{cnt['i']}/{n}] {tag} q={r.get('query_id')} {r.get('case_id')}  {extra}", flush=True)
 
     results = asyncio.run(gather_pairs(pairs, on_each=_progress))
     ok = sum(1 for r in results if r.get("ok"))
@@ -354,6 +416,19 @@ def _cli_run(query_id: str, case_ids: List[str]) -> int:
     return 0 if ok == n else 1
 
 
+def _cli_run(query_id: str, case_ids: List[str]) -> int:
+    """CLI 归类:对一个 query 下的若干 case 跑归类。"""
+    return _run_pairs([(query_id, c) for c in case_ids])
+
+
+def _cli_run_all() -> int:
+    """CLI 全量归类:对库里所有有解构记录的 case(各取最新真实版)跑归类。
+    用于改 TOP_K 后或清空 cat-api 分类库后重建树(见 归类口径修复方案.md 第4.2节)。"""
+    pairs = db.fetch_all_process_pairs()
+    print(f"全量归类:库内 {len(pairs)} 个有解构的 case(各取最新真实版)", flush=True)
+    return _run_pairs(pairs)
+
+
 if __name__ == "__main__":
     import argparse
     ap = argparse.ArgumentParser(description="category-match:FastAPI 服务 或 CLI 归类(--run)")
@@ -361,10 +436,15 @@ if __name__ == "__main__":
                     default=int(os.environ.get("CATEGORY_MATCH_PORT", "8780")),
                     help="服务端口(不带 --run 时生效)")
     ap.add_argument("--run", action="store_true", help="CLI 归类模式:跑完即退出,不起服务")
+    ap.add_argument("--all", action="store_true",
+                    help="全量重跑:对库里所有有解构的 case(各取最新真实版)跑归类,跑完即退出")
     ap.add_argument("--query-id", help="--run 用:post_id")
     ap.add_argument("--case-ids", help="--run 用:逗号分隔的 case_id(knowledge_id)")
     args = ap.parse_args()
 
+    if args.all:
+        sys.exit(_cli_run_all())
+
     if args.run:
         cids = [c.strip() for c in (args.case_ids or "").split(",") if c.strip()]
         if not args.query_id or not cids:

examples/mode_workflow/工序接口文档.md

@@ -3,12 +3,40 @@
 > 来源：`examples/mode_workflow`（`server.py` + `db.py`）。本文档只整理**工序方向**（`mode=process`）相关接口。
 > 工具方向（`mode=tools`）多数接口同形，仅把 `process→tools`、`mode_process→mode_tools`、`search_process→search_tools` 替换即可，文末附差异说明。
 
+---
+
+## 变更记录（Changelog）
+
+### 2026-06-27 · 归类口径修复 + 工序解构拆表（对 aigc-web **无破坏性**）
+
+> 本轮后端改动**不改变任何现有接口的请求/响应形状**（已 347/347 逐字节校验）。aigc-web 现有调用全部继续可用；仅 1 项**建议采用**的新接口。
+
+| 类型 | 接口/字段 | aigc-web 动作 |
+|---|---|---|
+| ⭐ 新增（建议采用） | `GET /api/classified_cases?dimension=&path=`（§3） | 维度联动取帖改用它，**替代**先前「按外部分类树节点 `knowledge_ids` 取帖」。与帖里显示的 `substanceMatch` 同源，根治「树多归属 ≠ 显示单归属」 |
+| 补文档（接口早已支持） | `GET /api/all_posts` 的 `case_list` 参数（§2） | 配合上面:`classified_cases` 拿 `cases` → `all_posts?case_list=<cases>` 取帖 |
+| 说明 | `ProcessPayload.steps[].substanceMatch/formMatch`（§3） | 仍照常返回（展示用），无需改；权威归类源已是 `step_classification` |
+| 内部（无感知） | 工序解构存储从 `mode_process` 宽表拆为 `process_run/procedure/step` 三层 | **无需任何改动**，响应形状不变 |
+
+**推荐取帖链路**（维度联动）：
+```
+GET /api/classified_cases?dimension=实质&path=/表象/视觉/空间/空间环境
+  → { "cases": ["gzh_…", …], "count": 46 }
+GET /api/all_posts?mode=process&case_list=gzh_…,gzh_…
+  → { "posts": [ … ] }
+```
+
+---
+
 ## 0. 基础约定
 
 - **Base URL**：`http://<host>:8772`（`server.py` 的 `PORT=8772`）。
-- **数据源**：MySQL 四表为唯一事实源，工序方向用其中两张：
+- **数据源**：MySQL 为唯一事实源，工序方向相关的表：
   - `search_process`：每行 = 一个 `(query, 帖子)`，搜索 + LLM 评估结果。
-  - `mode_process`：每行 = 一个解构出的**工序**（一帖可解出多个工序；`steps` 等嵌套结构存 JSON 列）。
+  - `process_run / process_procedure / process_step`：工序解构的**物理存储**（运行级/工序级/步骤级三层规范化）。所有工序接口的数据均由这三表重建。
+  - `mode_process`：**兜底镜像**（写侧仍双写,待观察后废弃),**不再被任何接口读取**。历史上「一行=一个工序」的宽表。
+  - `step_classification`：归类（category-match）结果独立表，一行 = 某 step 某维度某子项的命中分类（含 `matched_name/matched_path/score`，五维度）。**维度联动取帖的权威源**，与帖里显示的 `substanceMatch` 同源（见 `/api/classified_cases`）。
+  > 说明：工序解构存储已于 2026-06-27 规范化为上述三层(去元数据冗余、steps 可结构化查询)。**本文档所有响应形状保持不变**（`ProcessPayload` 等由新表逐字节还原,已 347/347 全量校验）。**aigc-web 无需任何改动。**
 - **方向参数**：凡带 `mode` 的接口，工序方向传 `mode=process`（缺省即 `process`）。
 - **响应格式**：统一 `application/json; charset=utf-8`。出错返回 `{"error": "<msg>"}` + 对应 HTTP 状态码（400/404/500）。
 - **缓存**：部分 GET 带 `ETag`，命中返回 `304`；`/api/dashboard`、`/api/queries` 另有 60s 内存缓存（解构任务完成时作废）。
@@ -119,6 +147,7 @@
 | `page_size` | 否 | `100` | 每页条数（上限 500） |
 | `adopted` | 否 | `0` | `1`/`true` 只返回采纳帖（`is_adopted_rel` 口径） |
 | `distinct` | 否 | `0` | `1`/`true` 按 `case_id` 去重（同帖被多 query 搜到时只保留 `overall_score` 最高的一行） |
+| `case_list` | 否 | 全部 | **按 `case_id` 取帖**（知识归类联动用：先 `/api/classified_cases` 拿某分类节点的 case，再传这里取帖）。写法同 `query_list`（逗号 / 重复 / JSON 数组）；`[]` 空列表 → 空结果 |
 
 `query_list` 三种等价写法（在 SQL 层 `WHERE query_id IN(...)` 过滤，先过滤再去重/分页；命中不到的 id 自动忽略）：
 - 逗号分隔：`?query_list=q0031,q0032`
@@ -214,12 +243,39 @@
 ```
 
 > `steps[].action` 的叶子（按 `/` 取最后一段）× `inputs/outputs[].type` 用于 Dashboard 内容树覆盖度命中统计。
+> `steps[]` 还会带 `substanceMatch` / `formMatch`（实质/形式归类名，按子项「、」拼接，未命中占位「无」）。
+> **注**：这两个展示值的**权威归类源是 `step_classification` 表**（含全路径/分数，五维度）；维度联动「取某分类节点下的帖子」请走 `/api/classified_cases`（与显示同源），不要再依赖外部分类树的 `knowledge_ids`。
 
 ### GET /api/extract_prompt?mode=process
 取当前工序解构 system prompt 原文（供“重新解构·编辑 Prompt”弹框预填）。
 
 **响应**：`{"prompt": "<prompts/procedure_extract_system.md 全文>"}`
 
+### GET /api/classified_cases?dimension=实质&path=/表象/视觉/空间/空间环境
+**维度联动取帖**：返回某维度下命中「该分类节点**及其整棵子树**」的去重 `case_id`。
+数据源是 `step_classification`（与帖里显示的 `substanceMatch` 同源），**替代**先前「按外部分类树节点 `knowledge_ids` 取帖」的口径，从根上消除「树多归属 ≠ 显示单归属」的不一致。
+
+**Query 参数**
+| 参数 | 必填 | 缺省 | 说明 |
+|---|---|---|---|
+| `dimension` | 否 | `实质` | 维度：`实质` / `形式` / `类型` / `作用` / `动作` |
+| `path` | 二选一 | — | 分类**全路径**，取该节点及其子树（`matched_path = path` 或 `LIKE 'path/%'`） |
+| `name` | 二选一 | — | 分类名精确匹配（`matched_name = name`） |
+
+> `path` 与 `name` 至少给一个，都不给返回 `400 {"error":"需提供 path 或 name"}`。两者都给取并集。
+
+**响应**
+```jsonc
+{
+  "dimension": "实质",
+  "path": "/表象/视觉/空间/空间环境",
+  "name": null,
+  "cases": ["gzh_...", "xhs_..."],   // 命中该节点(含子树)的去重 case_id
+  "count": 46
+}
+```
+> 典型用法：① 调本接口拿 `cases` → ② `GET /api/all_posts?case_list=<cases>` 取帖列表（见 §2）。
+
 ---
 
 ## 4. 发起工序解构（异步）