框架: B1-B3工具层交付——import skill(三种章题+卷前缀/自适应格式/重贴去重/目录空章弹出/证据制尾截/孤题如实报;批量executemany+keepalive防半死连接);embed skill(Qwen3-8B 1024维/批量重试/content_hash幂等);search skill(余弦召回+admin|work双授权面+aiContext字段裁剪);parse_ingest(五型归型/字段合同校验/15连字脱敏机械拒/任务状态机)

.claude/skills/embed/SKILL.md

@@ -0,0 +1,34 @@
+---
+name: embed
+description: New-API 嵌入封装——Qwen3-Embedding-8B、dimensions=1024、禁系统代理、批量+失败重试；输入知识行（draft/entity）批量嵌入并写 example_knowledge_embedding，content_hash 幂等不重嵌。B2/B3 的向量生产端。
+---
+
+# embed —— 嵌入封装（New-API 网关）
+
+对应 muse API 面：AI 网关（嵌入）。通道事实见 [`db/连接信息.md`](../../../db/连接信息.md)：BASE `http://100.64.0.8:3000`、模型 `Qwen/Qwen3-Embedding-8B`、请求体 `"dimensions":1024`（实测生效）、**禁系统代理**（`trust_env=False`）。
+
+## 用法
+
+```bash
+# 批量嵌入所有待嵌知识草稿行（默认：muse_knowledge_draft 中 status='pending' 且尚无嵌入行的）
+.venv/bin/python .claude/skills/embed/scripts/embed_drafts.py
+
+# 指定 work（=参考书拆书批次）或限量
+.venv/bin/python .claude/skills/embed/scripts/embed_drafts.py --work-id 3 --limit 100
+
+# 自由文本试嵌（调试/B3 查询端复用同实现）
+.venv/bin/python .claude/skills/embed/scripts/embed_drafts.py --probe "机甲近战的节奏控制"
+```
+
+## 合同
+
+- **嵌入文本构造**：`【型】名称：一句话摘要\n字段正文摘选`（draft_payload 的 embed_text 字段优先；无则按固定拼接），与检索端 query 语义对齐。
+- **幂等**：sha256(嵌入文本+模型) 为 `content_hash`，已存在则跳过（uk: tenant+hash+model）。
+- **批量**：每请求 ≤16 条文本；失败整批重试 2 次（指数退避），仍失败逐条降级重试，坏行记错并继续（不断批）。
+- **落库**：`example_knowledge_embedding(draft_id, content_hash, embed_text, model, dimensions=1024, embedding)`；draft 确认落 entity 后由 confirm 流程回填 entity_id（嵌入行不动）。
+- 汇报：新嵌 N、跳过（幂等）M、失败 K 及原因。
+
+## 红线
+
+- 调用必须 `trust_env=False`（系统代理会假 502）；令牌用 `MUSE_AI_NEW_API_TOKEN`（勿用管理令牌，打 /v1 报无效）。
+- 只嵌知识行内容，不嵌参考书原文全文（原文私有库不进向量面——脱敏边界在 B2 拆书层保证）。

.claude/skills/embed/scripts/embed_drafts.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+"""embed skill：知识行批量嵌入（New-API / Qwen3-Embedding-8B / 1024 维）。
+
+合同见同 skill SKILL.md；通道事实见 db/连接信息.md。失败原样报错不静默。
+"""
+import hashlib
+import json
+import sys
+import time
+
+import click
+import psycopg
+import requests
+
+DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
+       "?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
+BASE = "http://100.64.0.8:3000"
+TOKEN = "sk-DyVqO3lDmEvQZ3PqGpbNaaaHZHhbh0xaHRIiynhYSmVlLHl2"  # MUSE_AI_NEW_API_TOKEN(勿用管理令牌)
+MODEL = "Qwen/Qwen3-Embedding-8B"
+DIM = 1024
+TENANT, ACTOR = 1, "1"
+BATCH = 16
+
+
+def _session():
+    """禁系统代理的会话（系统代理会假 502）。"""
+    s = requests.Session()
+    s.trust_env = False
+    s.headers["Authorization"] = f"Bearer {TOKEN}"
+    return s
+
+
+def embed_texts(sess, texts):
+    """调 New-API /v1/embeddings；整批重试 2 次后逐条降级。返回 (向量列表, 失败索引集)。"""
+    def call(batch):
+        r = sess.post(f"{BASE}/v1/embeddings", json={
+            "model": MODEL, "input": batch, "dimensions": DIM}, timeout=120)
+        r.raise_for_status()
+        data = r.json()["data"]
+        return [d["embedding"] for d in sorted(data, key=lambda d: d["index"])]
+
+    for attempt in range(3):
+        try:
+            return call(texts), set()
+        except Exception as e:
+            if attempt < 2:
+                time.sleep(2 ** attempt)
+                continue
+            # 整批三败 → 逐条降级，坏行记错不断批
+            vecs, bad = [], set()
+            for i, t in enumerate(texts):
+                try:
+                    vecs.append(call([t])[0])
+                except Exception as ee:
+                    vecs.append(None)
+                    bad.add(i)
+                    click.echo(f"  [失败] 第{i}条: {ee}", err=True)
+            return vecs, bad
+
+
+def build_embed_text(payload: dict) -> str:
+    """嵌入文本构造：payload 自带 embed_text 优先；否则固定拼接（与检索端语义对齐）。"""
+    if payload.get("embed_text"):
+        return payload["embed_text"]
+    t = payload.get("型") or payload.get("target_type", "")
+    name = payload.get("名称", "")
+    brief = payload.get("一句话摘要", "")
+    fields = payload.get("字段") or {}
+    body = "\n".join(f"{k}：{v}" for k, v in fields.items() if v and k not in ("名称", "一句话摘要"))
+    return f"【{t}】{name}：{brief}\n{body}"[:4000]
+
+
+@click.command()
+@click.option("--work-id", type=int, help="限定拆书批次的 work（draft.work_id=0 为全局行，用 source_id 关联参考书）")
+@click.option("--limit", type=int, default=0, help="最多处理条数（0=不限）")
+@click.option("--probe", help="自由文本试嵌（打印维度与前 5 维，不落库）")
+def main(work_id, limit, probe):
+    sess = _session()
+    if probe:
+        vecs, bad = embed_texts(sess, [probe])
+        if bad:
+            raise click.ClickException("试嵌失败")
+        v = vecs[0]
+        click.echo(f"维度={len(v)} 前5维={[round(x, 4) for x in v[:5]]}")
+        return
+
+    with psycopg.connect(DSN) as conn:
+        # 待嵌=pending 草稿且无嵌入行
+        sql = """SELECT d.id, d.draft_payload FROM muse_knowledge_draft d
+                 WHERE d.tenant_id=%s AND d.deleted=FALSE AND d.status='pending'
+                   AND NOT EXISTS (SELECT 1 FROM example_knowledge_embedding e
+                                   WHERE e.tenant_id=%s AND e.draft_id=d.id AND e.deleted=FALSE)"""
+        args = [TENANT, TENANT]
+        if work_id is not None:
+            sql += " AND d.source_id=%s"
+            args.append(work_id)
+        sql += " ORDER BY d.id"
+        if limit:
+            sql += f" LIMIT {int(limit)}"
+        rows = conn.execute(sql, args).fetchall()
+        click.echo(f"待嵌草稿: {len(rows)} 条")
+
+        done = skip = fail = 0
+        for i in range(0, len(rows), BATCH):
+            chunk = rows[i:i + BATCH]
+            texts, metas = [], []
+            for did, payload in chunk:
+                text = build_embed_text(payload or {})
+                h = hashlib.sha256(f"{text}|{MODEL}".encode()).hexdigest()
+                if conn.execute(
+                        "SELECT 1 FROM example_knowledge_embedding WHERE tenant_id=%s AND content_hash=%s AND model=%s",
+                        (TENANT, h, MODEL)).fetchone():
+                    skip += 1  # 幂等：同文同模型不重嵌
+                    continue
+                texts.append(text)
+                metas.append((did, h, text))
+            if not texts:
+                continue
+            vecs, bad = embed_texts(sess, texts)
+            for j, (did, h, text) in enumerate(metas):
+                if j in bad:
+                    fail += 1
+                    continue
+                conn.execute(
+                    """INSERT INTO example_knowledge_embedding
+                           (draft_id, content_hash, embed_text, model, dimensions, embedding,
+                            creator, updater, tenant_id)
+                       VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
+                       ON CONFLICT (tenant_id, content_hash, model) DO NOTHING""",
+                    (did, h, text, MODEL, DIM, json.dumps(vecs[j]), ACTOR, ACTOR, TENANT))
+                done += 1
+            conn.commit()
+            click.echo(f"  进度 {min(i + BATCH, len(rows))}/{len(rows)}（新嵌{done} 跳过{skip} 失败{fail}）")
+        click.echo(f"完成：新嵌 {done}、幂等跳过 {skip}、失败 {fail}")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (psycopg.Error, requests.RequestException) as e:
+        click.echo(f"[错误] {type(e).__name__}: {e}", err=True)
+        sys.exit(1)

.claude/skills/import/SKILL.md

@@ -0,0 +1,47 @@
+---
+name: import
+description: 参考书/旧稿 txt 导入解析——回目正则静态分章后落库（作品行+章行+正文 block+参考书档案），对账表即审查面。B1 参考书导入与 C8 用户旧稿复用同一通道；LLM 不参与，纯确定性工具。
+---
+
+# import —— 导入解析（静态分章入库）
+
+对应 muse API 面：导入解析旅程（产品-03 §3.7）。落库走 psycopg 直连（凭据同 db skill），写入约定见 [`db/表映射.md`](../../../db/表映射.md)。
+
+## 用法
+
+```bash
+# 预演：只解析打印对账，不落库
+.venv/bin/python .claude/skills/import/scripts/import_novel.py --dry-run "../小说清单/机动风暴_骷髅精灵.txt"
+
+# 正式导入（已存在同名作品则拒绝，--force 软删旧行重导）
+.venv/bin/python .claude/skills/import/scripts/import_novel.py "../小说清单/超神机械师_齐佩甲.txt"
+.venv/bin/python .claude/skills/import/scripts/import_novel.py --force "../小说清单/机动风暴_骷髅精灵.txt"
+```
+
+## 分章规则（按覆盖的真实格式）
+
+- **恒启用**：`第X章 标题`（中文/阿拉伯数字，含「两/零/〇」）；`第X卷 第Y章 标题` 与无空格变体（取章开章，卷号入章行快照）；纯卷行（有卷无章）不开章只计数。
+- **自适应启用**（书内命中 ≥50 次才生效，防误切）：裸阿拉伯 `001 标题` 式；裸中文数字 `一百零二 标题` 式。
+- **修复规则**：同题重现一律不开新章（盗版重贴水印，丢标题行、正文归当前章）；尾部章号大幅回落（<前文峰值一半且位于文件末 5%）→ 该处起截断丢弃；全文过 `html.unescape` 解实体；章号解析失败（如「一八五十一」错写）→ 继承前章号+1 并计数。
+
+## 落库（一书一事务）
+
+| 表 | 写入 |
+|---|---|
+| muse_content_work | 一行：title/genre='科幻'/status='completed'/owner_user_id=1/字数章数/import_status='imported'/parse_status='pending' |
+| muse_content_chapter | 一章一行：order_no=顺序号、title=章题（≤200 字）、outline_snapshot 存 {卷号,解析章号} |
+| muse_content_block | 一章一 block：order_no=1、block_type='scene'、content_text=正文、word_count |
+| muse_knowledge_base | 幂等 ensure 两行：『参考书私有库』(kb_type='user') 与『公共范式库』(kb_type='global') |
+| muse_knowledge_document | 每书一行挂私有库：file_hash=sha256、storage_ref=源文件相对路径 |
+| example_reference_work | 参考书档案：作者/声明章数/导入章数/字符数/notes（修复统计） |
+| muse_content_import_task | 审计一行：command_id=import-<hash16> 幂等，source_snapshot=对账 JSON |
+
+## 审查面
+
+- `--dry-run` 与正式导入都打印**对账表**：声明章数 vs 导入章数、丢弃重复题数、尾部截断行、卷分隔数、章号异常数、总字数，另附首/中/末三章题目与正文首行抽样。
+- 导入后用 db skill 抽查：`query "SELECT order_no,title FROM muse_content_chapter WHERE work_id=… ORDER BY order_no LIMIT 5"`。
+
+## 红线
+
+- 原文只入私有库（work/chapter/block + 私有 kb 档案），公共面（拆书范式）另走 B2 且脱敏；本 skill 不做任何内容改写（除实体解码与噪音标题行修复）。
+- 同名作品不覆盖：必须显式 `--force`（软删旧行，审计可溯）。

.claude/skills/import/scripts/import_novel.py

@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""import skill：参考书/旧稿 txt 静态分章导入（B1/C8 共用，LLM 不参与）。
+
+分章与修复规则、落库映射见同 skill 的 SKILL.md；写入约定见 db/表映射.md。
+失败原样抛错不静默（公约）。
+"""
+import hashlib
+import html
+import json
+import pathlib
+import re
+import sys
+
+import click
+import psycopg
+from psycopg.types.json import Jsonb
+
+DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
+       "?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
+# keepalive 防 Tailscale 半死连接（2026-07-13 实测：逐行插入两万次往返曾卡死 16 分钟）
+TENANT, ACTOR, OWNER = 1, "1", 1  # 实验写入约定：系统主账号
+
+CN_DIGITS = {"零": 0, "〇": 0, "一": 1, "二": 2, "两": 2, "三": 3, "四": 4,
+             "五": 5, "六": 6, "七": 7, "八": 8, "九": 9}
+CN_UNITS = {"十": 10, "百": 100, "千": 1000}
+
+# 章题格式（named group: vol=卷号, no=章号, title=题名）
+PAT_VOL_CH = re.compile(  # 第X卷 第Y章 标题（含无空格变体）——恒启用
+    r'^\s*第\s*(?P<vol>[零〇一二两三四五六七八九十百千0-9]+)\s*卷\s*'
+    r'第\s*(?P<no>[零〇一二两三四五六七八九十百千0-9]+)\s*[章回]\s*(?P<title>\S.*)?$')
+PAT_CH = re.compile(  # 第X章 标题——恒启用
+    r'^\s*第\s*(?P<no>[零〇一二两三四五六七八九十百千0-9]+)\s*[章回]\s*(?P<title>\S.*)?$')
+PAT_VOL_ONLY = re.compile(  # 纯卷行（有卷无章）——不开章
+    r'^\s*第\s*[零〇一二两三四五六七八九十百千0-9]+\s*卷\s*(?P<title>\S.*)?$')
+PAT_NUM = re.compile(r'^\s*(?P<no>\d{3,4})\s+(?P<title>\S.*)$')  # 001 标题——自适应
+PAT_CN_BARE = re.compile(  # 一百零二 标题——自适应（限长防误切正文）
+    r'^\s*(?P<no>[零〇一二两三四五六七八九十百千]{1,8})[ 　]+(?P<title>\S.{0,28})$')
+
+
+def cn2int(s: str):
+    """中文数字→整数；混写/错写返回 None（调用方 fallback 前章+1）。"""
+    s = s.strip()
+    if s.isdigit():
+        return int(s)
+    total, section, num = 0, 0, 0
+    for ch in s:
+        if ch in CN_DIGITS:
+            num = CN_DIGITS[ch]
+        elif ch in CN_UNITS:
+            u = CN_UNITS[ch]
+            section += (num or 1) * u
+            num = 0
+        else:
+            return None
+    total = section + num
+    return total or None
+
+
+def detect_adaptive(lines):
+    """第一遍扫描：统计自适应格式命中数，≥50 次才启用（防普通正文误切）。"""
+    n_num = sum(1 for ln in lines if PAT_NUM.match(ln))
+    n_cn = sum(1 for ln in lines if PAT_CN_BARE.match(ln) and not PAT_CH.match(ln) and not PAT_VOL_ONLY.match(ln))
+    return n_num >= 50, n_cn >= 50
+
+
+def match_title(line, use_num, use_cn):
+    """按启用格式集识别章题行；返回 (卷号, 章号, 题名) 或 None。"""
+    m = PAT_VOL_CH.match(line)
+    if m:
+        return cn2int(m["vol"]), cn2int(m["no"]), (m["title"] or "").strip()
+    m = PAT_CH.match(line)
+    if m:
+        return None, cn2int(m["no"]), (m["title"] or "").strip()
+    if PAT_VOL_ONLY.match(line):
+        return "VOL_ONLY", None, None  # 纯卷行标记
+    if use_num:
+        m = PAT_NUM.match(line)
+        if m:
+            return None, int(m["no"]), m["title"].strip()
+    if use_cn:
+        m = PAT_CN_BARE.match(line)
+        if m:
+            no = cn2int(m["no"])
+            if no is not None:  # 数字非法则不认为是章题（正文行）
+                return None, no, m["title"].strip()
+    return None
+
+
+def parse_book(path: pathlib.Path):
+    """解析一本书 → (meta, chapters, stats)。修复规则见 SKILL.md。"""
+    raw = path.read_text(encoding="utf-8", errors="replace")
+    raw = html.unescape(raw)  # 解 &ldquo; 等实体
+    lines = raw.split("\n")
+
+    # 头部 meta：# 《书名》 / # 作者: X / # 共 N 章
+    title = author = None
+    declared = None
+    for ln in lines[:8]:
+        m = re.match(r'^#\s*《(.+?)》', ln)
+        if m:
+            title = m.group(1)
+        m = re.match(r'^#?\s*书名[:：]\s*(\S+)', ln)
+        if m:
+            title = title or m.group(1)
+        m = re.match(r'^#?\s*作者[:：]\s*(\S+)', ln)
+        if m:
+            author = author or m.group(1)
+        m = re.match(r'^#\s*共\s*(\d+)\s*章', ln)
+        if m:
+            declared = int(m.group(1))
+    if not title or not author:  # 文件名兜底：书名_作者.txt 或 书名(作者).txt
+        stem = path.stem
+        m = re.match(r'^(.+?)[(（](.+?)[)）]\s*$', stem)
+        if m:
+            title = title or m.group(1).strip()
+            author = author or m.group(2).strip()
+        else:
+            title = title or stem.split("_")[0]
+            author = author or (stem.split("_")[-1] if "_" in stem else None)
+
+    use_num, use_cn = detect_adaptive(lines)
+
+    # 第二遍：切章
+    chapters = []          # [{seq,vol,no,title,lines:[...]}]
+    seen_titles = {}       # normalized 全题 → 首现章 idx（同题重现不开新章）
+    cur = None
+    stats = {"丢弃重复题行": 0, "纯卷行": 0, "章号解析失败": 0, "目录空章弹出": 0}
+    last_no = 0
+    for ln in lines:
+        hit = match_title(ln, use_num, use_cn)
+        if hit:
+            vol, no, t = hit
+            if vol == "VOL_ONLY":
+                stats["纯卷行"] += 1
+                continue
+            # 目录残留修复：上一题行至此无任何正文 → 那是目录行，弹出空章并注销其题名，
+            # 让后文的真章（同题）能正常开章（否则真章被判重、边界丢失）
+            if cur is not None and not any(l.strip() for l in cur["lines"]):
+                seen_titles.pop(cur["_norm"], None)
+                chapters.pop()
+                stats["目录空章弹出"] += 1
+                stats.setdefault("空题名样例", []).append(cur["title"][:20])
+                if len(stats["空题名样例"]) > 8:
+                    stats["空题名样例"] = stats["空题名样例"][:8] + ["…"]
+            norm = re.sub(r'\s+', '', ln.strip())
+            if norm in seen_titles:
+                stats["丢弃重复题行"] += 1  # 盗版重贴：丢标题行，正文归当前章
+                continue
+            if no is None:
+                no = last_no + 1
+                stats["章号解析失败"] += 1
+            seen_titles[norm] = len(chapters)
+            cur = {"seq": len(chapters) + 1, "vol": vol, "no": no,
+                   "title": t or f"第{no}章", "lines": [], "_norm": norm}
+            chapters.append(cur)
+            last_no = no
+            continue
+        if cur is not None:
+            cur["lines"].append(ln)
+
+    # 尾部残留截断（证据制，防误伤）：
+    # 候选=尾部连续「章号 < 鲁棒峰值(90分位,防错打大号污染)一半」的段，且长度≤全书20%；
+    # 护栏1：段内末两章带完结标记（终章/全书完/大结局/（终）/（完））→ 是末卷重新计号的正文，保留；
+    # 护栏2：段内题文与前文章题文匹配率≥50% → 判为早期章节重贴残留，截断；否则保守保留。
+    if len(chapters) > 20:
+        nos = sorted(c["no"] for c in chapters)
+        robust_peak = nos[int(len(nos) * 0.9)]
+        run_start = None
+        for i in range(len(chapters) - 1, -1, -1):
+            if chapters[i]["no"] < robust_peak * 0.5:
+                run_start = i
+            else:
+                break
+        if run_start is not None and (len(chapters) - run_start) <= len(chapters) * 0.2:
+            run = chapters[run_start:]
+            end_marker = re.compile(r'终章|全书完|大结局|（终）|\(终\)|（完）|\(完\)|完本')
+            if any(end_marker.search(c["title"]) for c in run[-2:]):
+                stats["尾部低号段保留(带完结标记)"] = len(run)
+            else:
+                def tnorm(s):
+                    return re.sub(r'[\s，。！？—…·、,.!?()（）]+', '', s)
+                earlier = {tnorm(c["title"]) for c in chapters[:run_start]}
+                hits = sum(1 for c in run if tnorm(c["title"]) in earlier)
+                if hits >= len(run) * 0.5:
+                    stats["尾部截断章"] = len(run)
+                    stats["尾部截断起"] = run[0]["title"]
+                    chapters = chapters[:run_start]
+                else:
+                    stats["尾部低号段保留(题文不重)"] = len(run)
+
+    for c in chapters:
+        c["text"] = "\n".join(c["lines"]).strip()
+        del c["lines"]
+        c.pop("_norm", None)
+    dropped = [c["title"][:24] for c in chapters if not c["text"]]
+    if dropped:  # 源文件孤题无正文（常见：盗版尾部只剩末章标题）——诚实报告
+        stats["无正文题名丢弃"] = dropped[:6] + (["…"] if len(dropped) > 6 else [])
+    chapters = [c for c in chapters if c["text"]]
+    for i, c in enumerate(chapters, 1):
+        c["seq"] = i
+
+    meta = {"title": title, "author": author, "declared": declared,
+            "file": path.name, "chars": len(raw),
+            "自适应格式": {"裸阿拉伯": use_num, "裸中文数字": use_cn}}
+    return meta, chapters, stats
+
+
+def report(meta, chapters, stats):
+    """对账表（审查面）。"""
+    print(f"《{meta['title']}》 作者:{meta['author'] or '?'}  源:{meta['file']}")
+    print(f"  声明章数:{meta['declared'] or '无'}  导入章数:{len(chapters)}  总字符:{meta['chars']:,}")
+    print(f"  自适应格式:{meta['自适应格式']}  修复统计:{stats}")
+    # 章号 vs 顺序号偏差（重号/跳号计数，纯对账不修正）
+    mismatch = sum(1 for c in chapters if c["no"] != c["seq"])
+    print(f"  章号≠顺序号: {mismatch} 章（断更补号/重号常见，仅供参考）")
+    for tag, c in [("首", chapters[0]), ("中", chapters[len(chapters) // 2]), ("末", chapters[-1])]:
+        first_line = next((l for l in c["text"].split("\n") if l.strip()), "")[:40]
+        print(f"  [{tag}] #{c['seq']} 《{c['title'][:30]}》 {len(c['text'])}字 | {first_line}…")
+
+
+def ensure_kbs(conn):
+    """幂等 ensure 两个知识库行：私有参考书库 + 公共范式库。返回 (私有id, 公共id)。"""
+    ids = {}
+    for name, ktype, desc in [("参考书私有库", "user", "参考书全本原文（仅供拆书，不对作品侧开放）"),
+                              ("公共范式库", "global", "拆书产出的脱敏范式（管理员确认后可绑定）")]:
+        row = conn.execute(
+            "SELECT id FROM muse_knowledge_base WHERE tenant_id=%s AND name=%s AND deleted=FALSE",
+            (TENANT, name)).fetchone()
+        if row:
+            ids[name] = row[0]
+        else:
+            ids[name] = conn.execute(
+                """INSERT INTO muse_knowledge_base (name, description, kb_type, owner_user_id, status,
+                       creator, updater, tenant_id)
+                   VALUES (%s,%s,%s,%s,'active',%s,%s,%s) RETURNING id""",
+                (name, desc, ktype, OWNER, ACTOR, ACTOR, TENANT)).fetchone()[0]
+    return ids["参考书私有库"], ids["公共范式库"]
+
+
+def import_book(path: pathlib.Path, force: bool):
+    meta, chapters, stats = parse_book(path)
+    report(meta, chapters, stats)
+    if not chapters:
+        raise click.ClickException("解析出 0 章，拒绝入库")
+    file_hash = hashlib.sha256(path.read_bytes()).hexdigest()
+    command_id = f"import-{file_hash[:16]}"
+    total_words = sum(len(re.sub(r'\s', '', c['text'])) for c in chapters)
+
+    with psycopg.connect(DSN) as conn:
+        exist = conn.execute(
+            "SELECT id FROM muse_content_work WHERE tenant_id=%s AND title=%s AND deleted=FALSE",
+            (TENANT, meta["title"])).fetchone()
+        if exist and not force:
+            raise click.ClickException(f"作品《{meta['title']}》已存在(id={exist[0]})，重导请加 --force")
+        if exist and force:  # 软删旧行（work/chapter/block/档案），审计可溯
+            wid = exist[0]
+            conn.execute("UPDATE muse_content_work SET deleted=TRUE, updater=%s WHERE id=%s", (ACTOR, wid))
+            conn.execute("UPDATE muse_content_chapter SET deleted=TRUE, updater=%s WHERE tenant_id=%s AND work_id=%s", (ACTOR, TENANT, wid))
+            conn.execute("UPDATE muse_content_block SET deleted=TRUE, updater=%s WHERE tenant_id=%s AND work_id=%s", (ACTOR, TENANT, wid))
+            conn.execute("UPDATE example_reference_work SET deleted=TRUE, updater=%s WHERE tenant_id=%s AND work_id=%s", (ACTOR, TENANT, wid))
+            print(f"  --force: 旧作品 id={wid} 及章/块/档案已软删")
+
+        kb_private, _kb_public = ensure_kbs(conn)
+
+        work_id = conn.execute(
+            """INSERT INTO muse_content_work (owner_user_id, title, description, genre, status,
+                   import_status, parse_status, word_count, chapter_count, creator, updater, tenant_id)
+               VALUES (%s,%s,%s,'科幻','completed','imported','pending',%s,%s,%s,%s,%s) RETURNING id""",
+            (OWNER, meta["title"], f"参考书导入（拆书用）；作者:{meta['author'] or '?'}",
+             total_words, len(chapters), ACTOR, ACTOR, TENANT)).fetchone()[0]
+
+        # 批量两阶段（executemany 走 pipeline，一书仅数次网络往返；此前逐行两万往返曾被半死连接卡死）
+        with conn.cursor() as cur:
+            cur.executemany(
+                """INSERT INTO muse_content_chapter (work_id, title, order_no, status, outline_snapshot,
+                       creator, updater, tenant_id)
+                   VALUES (%s,%s,%s,'published',%s,%s,%s,%s)""",
+                [(work_id, c["title"][:200], c["seq"],
+                  Jsonb({"解析章号": c["no"], "卷号": c["vol"]}), ACTOR, ACTOR, TENANT) for c in chapters])
+            id_map = dict(cur.execute(
+                "SELECT order_no, id FROM muse_content_chapter WHERE tenant_id=%s AND work_id=%s AND deleted=FALSE",
+                (TENANT, work_id)).fetchall())
+            cur.executemany(
+                """INSERT INTO muse_content_block (work_id, chapter_id, order_no, block_type, title,
+                       content_text, word_count, creator, updater, tenant_id)
+                   VALUES (%s,%s,1,'scene',%s,%s,%s,%s,%s,%s)""",
+                [(work_id, id_map[c["seq"]], c["title"][:500], c["text"],
+                  len(re.sub(r'\s', '', c["text"])), ACTOR, ACTOR, TENANT) for c in chapters])
+
+        conn.execute(
+            """INSERT INTO muse_knowledge_document (kb_id, title, file_name, file_size, mime_type, file_hash,
+                   storage_ref, scan_status, parse_status, author, creator, updater, tenant_id)
+               VALUES (%s,%s,%s,%s,'text/plain',%s,%s,'completed','pending',%s,%s,%s,%s)""",
+            (kb_private, meta["title"], meta["file"], path.stat().st_size, file_hash,
+             str(path), meta["author"], ACTOR, ACTOR, TENANT))
+
+        conn.execute(
+            """INSERT INTO example_reference_work (work_id, author, source_file, declared_chapter_count,
+                   imported_chapter_count, char_count, parse_status, notes, creator, updater, tenant_id)
+               VALUES (%s,%s,%s,%s,%s,%s,'pending',%s,%s,%s,%s)""",
+            (work_id, meta["author"], meta["file"], meta["declared"], len(chapters), meta["chars"],
+             json.dumps({"修复统计": stats, "自适应格式": meta["自适应格式"]}, ensure_ascii=False),
+             ACTOR, ACTOR, TENANT))
+
+        snapshot = {"file": meta["file"], "declared": meta["declared"], "imported": len(chapters),
+                    "chars": meta["chars"], "words": total_words, "修复统计": stats}
+        conn.execute(
+            """INSERT INTO muse_content_import_task (work_id, owner_user_id, source_type, source_snapshot,
+                   status, command_id, creator, updater, tenant_id)
+               VALUES (%s,%s,'txt',%s,'succeeded',%s,%s,%s,%s)
+               ON CONFLICT (tenant_id, command_id)
+               DO UPDATE SET work_id=EXCLUDED.work_id, source_snapshot=EXCLUDED.source_snapshot, updater=EXCLUDED.updater""",
+            (work_id, OWNER, Jsonb(snapshot), command_id, ACTOR, ACTOR, TENANT))
+        conn.commit()
+    print(f"  ✅ 已入库 work_id={work_id}（章 {len(chapters)}、块 {len(chapters)}、档案 1、import_task {command_id}）\n")
+
+
+@click.command()
+@click.argument("files", nargs=-1, required=True, type=click.Path(exists=True, path_type=pathlib.Path))
+@click.option("--dry-run", is_flag=True, help="只解析打印对账，不落库")
+@click.option("--force", is_flag=True, help="同名作品已存在时软删旧行重导")
+def main(files, dry_run, force):
+    """参考书/旧稿 txt 静态分章导入。"""
+    for p in files:
+        if dry_run:
+            meta, chapters, stats = parse_book(p)
+            report(meta, chapters, stats)
+            print()
+        else:
+            import_book(p, force)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except psycopg.Error as e:
+        click.echo(f"[db错误] {type(e).__name__}: {e}", err=True)
+        sys.exit(1)

.claude/skills/parse-book/scripts/parse_ingest.py

@@ -0,0 +1,224 @@
+#!/usr/bin/env python3
+"""parse-book 配套确定性脚本：拆书产物校验入库 + 任务状态机（B2）。
+
+职责边界：extractor(LLM) 只产结构化 JSON 文件，不碰库；本脚本做机械校验后写库——
+字段 key 合法性（对库内字段合同）、五型归型、出处必填、**脱敏红线 15 连字检测**（硬阻断）。
+状态全在库（example_parse_task），断点续跑与幂等按章。
+"""
+import hashlib
+import json
+import pathlib
+import re
+import sys
+
+import click
+import psycopg
+from psycopg.types.json import Jsonb
+
+DSN = ("postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
+       "?keepalives=1&keepalives_idle=15&keepalives_interval=5&keepalives_count=3")
+TENANT, ACTOR = 1, "1"
+PATTERN_TYPES = {"craft", "combat", "emotion", "scene_pattern", "trope"}  # 拍板①：首轮只拆五型
+NGRAM = 15  # 脱敏红线：≥15 连续字与原文重合=违规（parse-book skill）
+
+
+def chapter_of(conn, work_id, order_no):
+    """取章 id 与正文。"""
+    row = conn.execute(
+        """SELECT c.id, b.content_text FROM muse_content_chapter c
+           JOIN muse_content_block b ON b.chapter_id=c.id AND b.deleted=FALSE
+           WHERE c.tenant_id=%s AND c.work_id=%s AND c.order_no=%s AND c.deleted=FALSE""",
+        (TENANT, work_id, order_no)).fetchone()
+    if not row:
+        raise click.ClickException(f"章不存在: work={work_id} order={order_no}")
+    return row
+
+
+def field_contract(conn, ttype):
+    """库内字段合同：型 → 合法字段 key 集合。"""
+    rows = conn.execute(
+        """SELECT f.field_key FROM muse_meta_field f
+           JOIN muse_meta_schema_version sv ON sv.id=f.schema_version_id
+           JOIN muse_meta_schema s ON s.active_version_id=sv.id
+           WHERE s.tenant_id=%s AND s.schema_key=%s AND f.deleted=FALSE""",
+        (TENANT, ttype)).fetchall()
+    return {r[0] for r in rows}
+
+
+def leak_check(card_texts, source_text):
+    """脱敏机械检查：卡内任一文本值含与原文 ≥NGRAM 连续字重合 → 返回违规片段。"""
+    src = re.sub(r'\s', '', source_text)
+    grams = {src[i:i + NGRAM] for i in range(0, max(0, len(src) - NGRAM + 1))}
+    for t in card_texts:
+        tt = re.sub(r'\s', '', str(t))
+        for i in range(0, max(0, len(tt) - NGRAM + 1)):
+            if tt[i:i + NGRAM] in grams:
+                return tt[i:i + NGRAM]
+    return None
+
+
+def set_task(conn, work_id, chapter_id, **cols):
+    """推进任务状态机（attempt 自增）。"""
+    sets = ", ".join(f"{k}=%s" for k in cols)
+    conn.execute(
+        f"""UPDATE example_parse_task SET {sets}, attempt_count=attempt_count+1, updater=%s
+            WHERE tenant_id=%s AND work_id=%s AND chapter_id=%s""",
+        (*cols.values(), ACTOR, TENANT, work_id, chapter_id))
+
+
+@click.group()
+def cli():
+    """拆书入库与任务状态机"""
+
+
+@cli.command("init-tasks")
+@click.option("--work-id", type=int, required=True)
+@click.option("--from", "from_", type=int, default=1, show_default=True)
+@click.option("--to", type=int, required=True)
+def init_tasks(work_id, from_, to):
+    """按章建任务行（幂等），并把参考书档案 parse_scope/parse_status 置为拆书中。"""
+    with psycopg.connect(DSN) as conn:
+        chs = conn.execute(
+            """SELECT id, order_no FROM muse_content_chapter
+               WHERE tenant_id=%s AND work_id=%s AND order_no BETWEEN %s AND %s AND deleted=FALSE
+               ORDER BY order_no""", (TENANT, work_id, from_, to)).fetchall()
+        n = 0
+        for ch_id, _no in chs:
+            conn.execute(
+                """INSERT INTO example_parse_task (work_id, chapter_id, creator, updater, tenant_id)
+                   VALUES (%s,%s,%s,%s,%s)
+                   ON CONFLICT (tenant_id, work_id, chapter_id) DO NOTHING""",
+                (work_id, ch_id, ACTOR, ACTOR, TENANT))
+            n += 1
+        conn.execute(
+            """UPDATE example_reference_work SET parse_scope=%s, parse_status='parsing', updater=%s
+               WHERE tenant_id=%s AND work_id=%s""",
+            (Jsonb({"from": from_, "to": to}), ACTOR, TENANT, work_id))
+        conn.commit()
+        click.echo(f"任务行就绪: work={work_id} 章 {from_}–{to}（{n} 行）")
+
+
+@cli.command()
+@click.option("--work-id", type=int, required=True)
+@click.option("--chapter-order", type=int, required=True)
+@click.option("--file", "file_", type=click.Path(exists=True), required=True)
+def scaffold(work_id, chapter_order, file_):
+    """脚手架入库：{细纲, 实体:[{型,名称,一句话摘要,备注?}]}；比例约束校验（3–5%，超标拒绝）。"""
+    data = json.loads(pathlib.Path(file_).read_text())
+    with psycopg.connect(DSN) as conn:
+        ch_id, src = chapter_of(conn, work_id, chapter_order)
+        outline = (data.get("细纲") or "").strip()
+        if not outline:
+            raise click.ClickException("细纲为空")
+        ratio = len(re.sub(r'\s', '', outline)) / max(1, len(re.sub(r'\s', '', src)))
+        if ratio > 0.08:  # 拍板值 3–5%，8% 为机械硬顶（防摘要化伪装结构化）
+            set_task(conn, work_id, ch_id, scaffold_status="failed",
+                     error_message=f"细纲比例超标 {ratio:.1%}>8%")
+            conn.commit()
+            raise click.ClickException(f"细纲比例 {ratio:.1%} 超标（>8%），已记失败退回重解析")
+        ents = data.get("实体") or []
+        for e in ents:
+            if not e.get("名称") or not e.get("型"):
+                raise click.ClickException(f"实体缺 名称/型: {e}")
+        conn.execute(
+            """INSERT INTO example_parse_scaffold (work_id, chapter_id, outline_text, entities,
+                   creator, updater, tenant_id)
+               VALUES (%s,%s,%s,%s,%s,%s,%s)
+               ON CONFLICT (tenant_id, chapter_id)
+               DO UPDATE SET outline_text=EXCLUDED.outline_text, entities=EXCLUDED.entities,
+                             deleted=FALSE, updater=EXCLUDED.updater""",
+            (work_id, ch_id, outline, Jsonb(ents), ACTOR, ACTOR, TENANT))
+        set_task(conn, work_id, ch_id, scaffold_status="done", error_message=None)
+        conn.commit()
+        click.echo(f"scaffold✓ work={work_id} ch#{chapter_order}: 细纲{len(outline)}字({ratio:.1%}) 实体{len(ents)}")
+
+
+@cli.command()
+@click.option("--work-id", type=int, required=True)
+@click.option("--chapter-order", type=int, required=True)
+@click.option("--file", "file_", type=click.Path(exists=True), required=True)
+def patterns(work_id, chapter_order, file_):
+    """范式卡入库：[{型∈五型, 名称, 一句话摘要, 字段{…}, 出处{书名,回目,定位}}] → draft(pending)。
+    机械硬阻断：归型合法、字段 key 合法（库内合同）、出处必填、15 连字脱敏检测。"""
+    cards = json.loads(pathlib.Path(file_).read_text())
+    if not isinstance(cards, list):
+        raise click.ClickException("patterns 文件须为卡片数组")
+    with psycopg.connect(DSN) as conn:
+        ch_id, src = chapter_of(conn, work_id, chapter_order)
+        contracts = {t: field_contract(conn, t) for t in PATTERN_TYPES}
+        book = conn.execute("SELECT title FROM muse_content_work WHERE id=%s", (work_id,)).fetchone()[0]
+        # 幂等：重跑本章 = 软删本章旧 draft
+        conn.execute(
+            """UPDATE muse_knowledge_draft SET deleted=TRUE, updater=%s
+               WHERE tenant_id=%s AND source_type='parse_book' AND source_id=%s
+                 AND draft_payload->>'章序'=%s AND deleted=FALSE""",
+            (ACTOR, TENANT, work_id, str(chapter_order)))
+        ok, rejected = 0, []
+        for i, c in enumerate(cards):
+            t = c.get("型")
+            reasons = []
+            if t not in PATTERN_TYPES:
+                reasons.append(f"型不合法:{t}（首轮只拆五型）")
+            if not c.get("名称"):
+                reasons.append("缺名称")
+            src_ref = c.get("出处") or {}
+            if not (src_ref.get("书名") and src_ref.get("回目")):
+                reasons.append("出处不完整（需书名+回目）")
+            fields = c.get("字段") or {}
+            if t in contracts:
+                illegal = set(fields) - contracts[t]
+                if illegal:
+                    reasons.append(f"字段越合同:{sorted(illegal)}")
+            texts = [c.get("名称"), c.get("一句话摘要"), *fields.values()]
+            leak = leak_check([x for x in texts if x], src)
+            if leak:
+                reasons.append(f"脱敏违规(≥{NGRAM}连字重合):「{leak}」")
+            if reasons:
+                rejected.append({"卡": c.get("名称") or f"#{i}", "原因": reasons})
+                continue
+            payload = {"型": t, "名称": c["名称"], "一句话摘要": c.get("一句话摘要", ""),
+                       "字段": fields, "出处": src_ref, "目标库": "公共范式库",
+                       "章序": chapter_order, "来源": f"拆书@{book}", "状态": "草稿"}
+            cid = f"parse-{work_id}-{chapter_order}-{i}-" + hashlib.sha256(
+                json.dumps(payload, ensure_ascii=False, sort_keys=True).encode()).hexdigest()[:8]
+            conn.execute(
+                """INSERT INTO muse_knowledge_draft (work_id, draft_type, draft_payload, status,
+                       source_type, source_id, command_id, creator, updater, tenant_id)
+                   VALUES (0,'entity',%s,'pending','parse_book',%s,%s,%s,%s,%s)
+                   ON CONFLICT (tenant_id, command_id) WHERE command_id IS NOT NULL DO NOTHING""",
+                (Jsonb(payload), work_id, cid, ACTOR, ACTOR, TENANT))
+            ok += 1
+        set_task(conn, work_id, ch_id, pattern_status="done" if not rejected else "done",
+                 error_message=None if not rejected else f"拒卡{len(rejected)}: " + json.dumps(rejected, ensure_ascii=False)[:900])
+        conn.commit()
+        click.echo(f"patterns✓ work={work_id} ch#{chapter_order}: 入库{ok} 拒{len(rejected)}")
+        for r in rejected:
+            click.echo(f"  [拒] {r['卡']}: {'; '.join(r['原因'])}")
+
+
+@cli.command()
+@click.option("--work-id", type=int)
+def progress(work_id):
+    """进度统计（审查面）。"""
+    with psycopg.connect(DSN) as conn:
+        where = " AND t.work_id=%s" if work_id else ""
+        args = [TENANT] + ([work_id] if work_id else [])
+        rows = conn.execute(f"""
+            SELECT w.title, count(*) FILTER (WHERE t.scaffold_status='done') AS s_done,
+                   count(*) FILTER (WHERE t.pattern_status='done') AS p_done, count(*) AS total,
+                   (SELECT count(*) FROM muse_knowledge_draft d
+                    WHERE d.tenant_id=%s AND d.source_type='parse_book' AND d.source_id=t.work_id
+                      AND d.deleted=FALSE) AS drafts
+            FROM example_parse_task t JOIN muse_content_work w ON w.id=t.work_id
+            WHERE t.tenant_id=%s AND t.deleted=FALSE{where}
+            GROUP BY w.title, t.work_id ORDER BY w.title""", [TENANT] + args).fetchall()
+        for r in rows:
+            click.echo(f"{r[0]:<12} 脚手架 {r[1]}/{r[3]}  范式 {r[2]}/{r[3]}  草稿卡 {r[4]}")
+
+
+if __name__ == "__main__":
+    try:
+        cli()
+    except psycopg.Error as e:
+        click.echo(f"[db错误] {type(e).__name__}: {e}", err=True)
+        sys.exit(1)

.claude/skills/search/SKILL.md

@@ -0,0 +1,33 @@
+---
+name: search
+description: 向量检索——创作意图→embed→pgvector 余弦召回→授权过滤（仅已确认+已绑定，或管理面含草稿）→aiContext 字段裁剪→带相似度分的结果集。B3 检索验证与 C4 PG 版 read-context 的取数端。
+---
+
+# search —— 知识向量检索
+
+对应 muse API 面：知识检索。查询嵌入与知识行嵌入同模型同维（Qwen3-Embedding-8B / 1024，经 embed skill 同实现），距离=余弦（`<=>`，HNSW 索引）。
+
+## 用法
+
+```bash
+# 管理面检索（B3 验证：对 draft 面召回，含草稿行）
+.venv/bin/python .claude/skills/search/scripts/search.py "主角被围攻时反杀的打斗写法" --scope admin --top 8
+
+# 作品面检索（C4：仅已确认 entity + 该作品已绑定的库；work-id 必填）
+.venv/bin/python .claude/skills/search/scripts/search.py "师徒决裂的情感铺垫" --scope work --work-id 8 --top 5
+
+# 按型过滤 + 用途裁剪（按 aiContext 只回显该用途可见字段）
+.venv/bin/python .claude/skills/search/scripts/search.py "倒计时紧迫感" --type craft --purpose generation
+```
+
+## 合同
+
+- **两个授权面**：`--scope admin`=治理/优化环用，召回 draft+entity 全量（含草稿）；`--scope work`=创作链路用，只回 `status='active'` 的 entity 且其 kb 经 `muse_knowledge_binding` 绑定到 `--work-id`（绑定≠写入、解绑即消失的活体）。
+- **字段裁剪**：`--purpose`（generation/planning/detection/extraction）按库内 `visibility_policy.policy_snapshot.fieldAiContext` 裁剪字段后再回显；被裁字段名列入尾部「裁剪回显」。
+- **输出**：卡片式（相似度分+型+名称+可见字段+出处），`--json` 给程序消费。
+- 相似度分=1−余弦距离，四舍五入 4 位；默认 top 5。
+
+## 红线
+
+- 作品面查询严禁绕过绑定与状态过滤（授权语义在查询层强制，不靠调用方自觉）；
+- 检索不回原文正文（知识行本身已脱敏；出处只有书名+回目+一句话定位）。

.claude/skills/search/scripts/search.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""search skill：知识向量检索（余弦召回 + 授权过滤 + aiContext 字段裁剪）。
+
+合同见同 skill SKILL.md。查询嵌入与知识行同模型同维（复用 embed skill 的实现）。
+"""
+import json
+import pathlib
+import sys
+
+import click
+import psycopg
+
+# 复用 embed skill 的通道实现（同模型同维，语义对齐）
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2] / "embed" / "scripts"))
+from embed_drafts import _session, embed_texts  # noqa: E402
+
+DSN = "postgresql://root:f6710e2d0294eb1c10e26a805a64bc54@100.64.0.8:5433/muse-example"
+TENANT = 1
+
+
+def load_ai_context(conn):
+    """读库内 23 型的字段级 aiContext 细则：{target_type: {field: true/false/[用途]}}。"""
+    rows = conn.execute(
+        """SELECT s.target_type, v.policy_snapshot->'fieldAiContext'
+           FROM muse_meta_schema s
+           JOIN muse_meta_schema_version sv ON sv.id = s.active_version_id
+           JOIN muse_meta_visibility_policy v ON v.schema_version_id = sv.id
+           WHERE s.tenant_id=%s AND s.deleted=FALSE""", (TENANT,)).fetchall()
+    return {t: (m or {}) for t, m in rows}
+
+
+def visible(ai_rule, purpose):
+    """aiContext 判定：true 全用途可见；false 不可见；[用途] 仅列出的可见；无规则默认可见。"""
+    if ai_rule is None:
+        return True
+    if isinstance(ai_rule, bool):
+        return ai_rule
+    return purpose in ai_rule
+
+
+@click.command()
+@click.argument("intent")
+@click.option("--scope", type=click.Choice(["admin", "work"]), default="admin", show_default=True,
+              help="admin=治理面(含草稿); work=作品面(仅已确认+已绑定)")
+@click.option("--work-id", type=int, help="scope=work 时必填")
+@click.option("--type", "ttype", help="限定型（如 craft/combat/emotion/scene_pattern/trope）")
+@click.option("--purpose", default="generation", show_default=True,
+              type=click.Choice(["generation", "planning", "detection", "extraction"]))
+@click.option("--top", default=5, show_default=True)
+@click.option("--json", "as_json", is_flag=True)
+def main(intent, scope, work_id, ttype, purpose, top, as_json):
+    if scope == "work" and not work_id:
+        raise click.ClickException("--scope work 必须带 --work-id（授权过滤依赖绑定关系）")
+
+    vecs, bad = embed_texts(_session(), [intent])
+    if bad:
+        raise click.ClickException("查询嵌入失败")
+    qvec = json.dumps(vecs[0])
+
+    with psycopg.connect(DSN) as conn:
+        ai_rules = load_ai_context(conn)
+        if scope == "admin":
+            # 治理面：draft(pending/confirmed)+entity 全量
+            sql = """SELECT 'draft' AS src, d.id, d.draft_payload AS payload, d.status,
+                            1 - (e.embedding <=> %s::vector) AS score
+                     FROM example_knowledge_embedding e
+                     JOIN muse_knowledge_draft d ON d.id = e.draft_id
+                     WHERE e.tenant_id=%s AND e.deleted=FALSE AND d.deleted=FALSE"""
+            args = [qvec, TENANT]
+        else:
+            # 作品面：仅 active entity 且其 kb 已绑定到该作品（授权在查询层强制）
+            sql = """SELECT 'entity' AS src, en.id,
+                            jsonb_build_object('型', en.entity_type, '名称', en.normalized_name,
+                                               '一句话摘要', en.description, '字段', en.attributes) AS payload,
+                            en.status, 1 - (e.embedding <=> %s::vector) AS score
+                     FROM example_knowledge_embedding e
+                     JOIN muse_knowledge_entity en ON en.id = e.entity_id
+                     JOIN muse_knowledge_binding b ON b.kb_id = en.kb_id AND b.work_id = %s
+                          AND b.binding_status='active' AND b.deleted=FALSE AND b.tenant_id=%s
+                     WHERE e.tenant_id=%s AND e.deleted=FALSE AND en.deleted=FALSE AND en.status='active'"""
+            args = [qvec, work_id, TENANT, TENANT]
+        if ttype:
+            sql += (" AND d.draft_payload->>'型' = %s" if scope == "admin"
+                    else " AND en.entity_type = %s")
+            args.append(ttype)
+        sql += " ORDER BY score DESC LIMIT %s"
+        args.append(top)
+        rows = conn.execute(sql, args).fetchall()
+
+    results = []
+    for src, rid, payload, status, score in rows:
+        p = payload or {}
+        t = p.get("型", "?")
+        rules = ai_rules.get(t, {})
+        fields = p.get("字段") or {}
+        vis = {k: v for k, v in fields.items() if visible(rules.get(k), purpose)}
+        cut = sorted(set(fields) - set(vis))
+        results.append({"来源": f"{src}#{rid}", "型": t, "名称": p.get("名称"), "状态": status,
+                        "相似度": round(float(score), 4), "一句话摘要": p.get("一句话摘要"),
+                        "可见字段": vis, "出处": p.get("出处"), "裁剪回显": cut})
+
+    if as_json:
+        click.echo(json.dumps(results, ensure_ascii=False, indent=1))
+        return
+    for i, r in enumerate(results, 1):
+        click.echo(f"── {i}. [{r['相似度']}] {r['型']} · {r['名称']}（{r['状态']}，{r['来源']}）")
+        click.echo(f"   摘要: {r['一句话摘要']}")
+        for k, v in (r["可见字段"] or {}).items():
+            click.echo(f"   {k}: {str(v)[:120]}")
+        if r["出处"]:
+            click.echo(f"   出处: {r['出处']}")
+        if r["裁剪回显"]:
+            click.echo(f"   [裁剪回显·{purpose} 不可见] {','.join(r['裁剪回显'])}")
+    if not results:
+        click.echo("（无召回）")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except psycopg.Error as e:
+        click.echo(f"[db错误] {type(e).__name__}: {e}", err=True)
+        sys.exit(1)