# 作者记忆维护

[author-memory.md](author-memory.md) 的少见时刻补充：记一条、确认、替换、忘掉和优先级仍按那份协议，本文件只在下列情况读——作者说「整理作者记忆」，或回执 `warnings`／查询 `omitted_ids` 提示超编；写入因 `作者画像.md` 写满失败；书根就是工作区，或工具报 `state.book`、单书布局错误；要做存量迁移 `migrate`、多事件原子 `commit`、派生视图 `check`；冲突候选要落定；碰到升级前留下的旧条目。

作者记忆借鉴“原始证据 → 候选 → 已确认画像 → 变更记录”的记忆管道，但把决定权留给作者。

## 文件

```text
{工作区}/.story/作者记忆/          # 项目级 store：global / genre / workflow 条目，编号 AP
├── _author-memory-state.json  # 唯一结构化权威
├── 作者画像.md               # 仅 active，供作者查看与管理
├── 待确认.md                 # pending / conflict，不参与约束
└── 变更记录.md               # 最近 100 次、最新在前的事务记录
{书}/.story/作者记忆/            # 书级 store：只存这本书的 book 条目，编号 BP，同样四个文件
```

三个 Markdown 文件都从 state 确定性生成，禁止手改；完整历史保留在 state，变更记录只展示最近 100 次。`作者画像.md` 是人类管理视图，普通写作 agent 不整份注入，而是调用 `query` 取得本次相关的紧凑上下文。

## 任务映射表

各 skill 入口的 `query` 命令按此表选 kind；写入时的预算提醒也按这四类任务组合估算。

| 任务 | query kinds | 注入位置 |
|---|---|---|
| 正文初稿 / 续写 | `prose_style` + `story_design` | 主会话与实际正文 agent |
| 去 AI 味 / 改写 | `prose_style` | 主会话与实际改写 agent |
| 设定 / 大纲 | `story_design` + `workflow` + `interaction` | 主会话，不传正文 agent |
| 审稿 | `delivery` + `interaction` + 必要的 `prose_style` | 主会话，不降低 rubric |

审稿匹配项只用于交付格式、协作方式和“作者有意采用的表达选择”说明；问题严重度和 PASS/FAIL 仍由 rubric 决定。

## 注入预算与容量

- **写入不因注入预算失败**：`record` / `commit` 照常成功、给回执；工具按上表四类任务组合估算最坏查询情形（全局条目＋各 scope 维度最重的单一切片，切片按大小写无关归并、轻重按写作时真正读到的字段算，与真实查询同一把尺），装不进 2048 字节的组合在返回的 `warnings` 里点名将被略过的条目及其断言首句。
- 写入落盘后另一级 store 读不出来（书目录不存在、`--book` 与书级记录不符等）也照常给回执，`warnings` 注明本次提醒没算上它。写书级条目时「本书＋全局」按实际条目精确计算；写项目级条目时只看得到项目级 store，顺手传 `--book-root` 就把当前这本书也算进提醒。
- 查询按 **重要度 → 本书例外 → 最近更新** 排序装填（同一范围的条目必在同一 store，「最近」按该 store 的修订号比，不跨 store 比较），先丢的恒是重要度较低的条目——`importance` 决定超编时谁留在 prompt 里。装不下的条目跳过而不中断（一条长的不挡后面的短条），漏下的 ID 按同一优先级报进 `omitted_ids`（最多列 20 条，`omitted` 是真实总数）。
- 注入预算之外还有一道硬上限：`作者画像.md` 超过 12288 字节时写入会直接失败并要求先整理。active 条目攒到几十上百条才会碰到（远在注入预算之后），碰到就走「整理作者记忆」；`forget` 这类减量操作在满编时照常可用。

## 整理作者记忆

作者说「整理作者记忆」，或回执 `warnings`／查询 `omitted_ids` 提示超编、写入因画像写满失败时：读项目级与当前书的 `作者画像.md`（每条都标了范围、重要度、把握和确认次数，重要度就是超编时的去留依据），提出合并同义条（`replace` 多合一）、退役过时条（`forget`）、给错标成 `high` 的条目下调重要度、把超长断言压缩成一句话的提案；项目级画像里还有「本书：」条目时，「对该书运行 `migrate --book-root`」列为默认提案项。清单用原话逐条列给作者确认（编号只放括号里），确认后按 store 各汇成一份 `commit` 事务提交（一份事务只写一个 store）。合并时保住每条的否定词、限定词和适用范围——合不动就退役其中一条，不要靠删限定词把两条凑成一条。整理只由作者发起或确认，不自动执行。

## 冲突候选

冲突候选（`conflict`）不能绕过旧规则直接 `decide=activate`。作者选新说法：用 `replace`，`old_ids` 同时列旧 active 条目和这条冲突候选，新条目直接 active、两条旧的标 `superseded`；作者留旧规则：对候选 `decide=reject`。旧条目被 `replace` / `forget` 撤下后，它不再是任何候选的冲突对象，冲突对象清空的候选退回 `pending`。

## 单书布局

书根就是工作区（`--book-root` 与 `--workspace` 同一目录）时，书级 store 改住 `{工作区}/.story/作者记忆/书级/`，与项目级各自一份 state；首次建立的书名优先取项目级存量本书条目里唯一的书名，再取目录名。旧版曾把书级 state 写在项目级位置，此后项目级读写都报 `state.book`；带 `--book-root {工作区}` 运行任一命令（含 `query`）会先把它原样移进 `书级/`，不改内容。这个目录其实是某个工作区里的一本书时（上一层叫 `长篇/` 或 `短篇/`，或某个祖先有 `.active-book` 或项目级 state），工具直接报错、不动任何文件，按报错改传 `--workspace`。

## 存量迁移

不做双读：升级前写进项目级 store 的 book 条目不再参与查询与预算估算，也不再接受新的 book 写入；它们仍在 `作者画像.md` 里可见、可 `decide` / `forget`。对每本书运行一次 `migrate --book-root {书目录}` 即可整批搬回来：断言、证据、确认次数、重要度原样保留，换成 `BP` 编号，原 `AP` 条目标 `superseded` 并注明去向；与全局条目的冲突关系在迁移后不再成立，这类候选退回 `pending`。书级每个源条目一笔事务，重跑只补没做完的一半。「整理作者记忆」看到项目级画像里还有「本书：」条目时，把迁移列为默认提案项。

## 旧条目

- 升级前写下的长断言不受 120 字节新建上限约束：原样重申它会**强化**原条目（确认次数 +1），不会因超长被拒；只有真正新建条目才校验 120 字节。
- 存量 state 里推断类旧来源的条目照常可读、可确认、可退役；新写入仍只接受 `explicit_user`、`accepted_suggestion`、`manual`。

## 其他命令

先依次尝试 `python3`、`python`、`py -3` 找到 Python 3，再从当前 skill 根运行本地副本（`record` / `query` 见 author-memory.md）：

```text
{PYTHON} {当前 skill 根}/scripts/author_memory_commit.py init    --workspace {工作区} [--book-root {书目录}]
{PYTHON} {当前 skill 根}/scripts/author_memory_commit.py commit  --workspace {工作区} [--book-root {书目录}] --input {工作区}/.story/work/作者记忆-事务.json
{PYTHON} {当前 skill 根}/scripts/author_memory_commit.py migrate --workspace {工作区} --book-root {书目录}
{PYTHON} {当前 skill 根}/scripts/author_memory_commit.py check   --workspace {工作区} [--book-root {书目录}]
```

- `commit`：高级批量入口，只在需要把多个动作绑定成一次原子提交时用（如整理作者记忆）。顶层传 `schema_version`、唯一 `transaction_id`、当前 `expected_state_revision` 和含 1–32 项的 `operations`（每项与单事件的 `operation` 同形）。一份事务只写一个 store；先在内存完成 schema、引用、容量和所有视图校验，操作按数组顺序应用，任一步失败则整份事务零写入，最后原子替换 state。过期修订会在任何写入前失败。事务文件在成功前必须保留，成功后删除；显式记忆请求按 author-memory.md「回执怎么告诉作者」转告。
- `migrate`：把项目级 store 里某本书的存量 book 条目整批搬进 `--book-root` 的书级 store，幂等，中途失败直接重跑；返回 `migrated`（源→新编号），没有存量时为空。
- `check`：从 state 重建并逐字核验所有派生视图；传 `--book-root` 时两级一起核验。
- `init`：显式初始化 store；平常不需要，首次 `record` 会随事务创建。

# 作者记忆协议

作者记忆保存跨会话复用的创作偏好，不保存小说世界里的事实，决定权留给作者。本文件管最常见的时刻：作者说出一条偏好，或要确认、替换、忘掉某条。少见情况读 [author-memory-maintenance.md](author-memory-maintenance.md)：整理作者记忆与超编、画像写满、单书布局与 `state.book` 报错、存量迁移、多事件 `commit`、`check`、冲突候选落定、升级前的旧条目。

## 边界与优先级

加载优先级从高到低：

1. 安全、用户授权范围、明确的平台交付要求、字数与文件协议；句长、视角、修辞和标点偏好不属于不可覆盖的硬门禁；
2. 用户在当前请求中的明确要求；
3. 当前书的 `设定/文风.md`、题材定位、细纲和其他项目设定；
4. 作者记忆中的本书偏好；
5. 作者记忆中的题材、流程和全局偏好；
6. 对标素材、通用方法和默认值。

按表达维度取最窄适用要求：低优先级只补缺项，不与高优先级要求并列执行。通用 references 自称“必须/禁用”不改变此顺序；审稿不因作者有意采用的表达本身扣分，真实可读性与因果问题仍照常评价。

作者记忆不能把本书事实写进 `.story/作者记忆/`，不能覆盖当前请求，不能降低审稿 rubric，也不能让去 AI 味改动剧情意图。小说事实继续由各书的 `追踪/` 和 `设定/` 管理。

## 存放与路由

两级 store，记忆随书走：`{工作区}/.story/作者记忆/` 存 global / genre / workflow 条目（编号 `AP`），`{书}/.story/作者记忆/` 只存本书的 book 条目（编号 `BP`）。每级各有 `作者画像.md`（生效条目）和 `待确认.md`（候选，不参与约束），都从 state 生成，禁止手改；不存在时写作、审稿、去味照常继续，首次 `record` 自动创建。

- `--workspace` 必须显式传，指创作工作区根——承载多本书、`.active-book`、`长篇/`、`短篇/` 或 `拆文库/` 的那一层；已有记忆时，是项目级 state（不带 `book` 字段）所在的最近祖先。`长篇/`、`短篇/` 下的书目录永远不当 `--workspace`，也不要把用户主目录当默认工作区。
- `--book-root` 是当前书的项目目录（`.active-book` 指向、或含 `设定/`、`正文/` 的那一层，如 `{工作区}/长篇/{书名}/`）；书名默认取书级 state 记的名字，首次取目录名，`--book` 可覆盖。书根就是工作区时读维护文件「单书布局」。
- ID 前缀就是 store：`decide` / `forget` 看 `item_id`（`AP` 项目级，`BP` 书级），`remember` / `replace` 看 `scope.level`（`book` 书级，其余项目级）。书级操作必须传 `--book-root`，没传直接报错，不会退而写进项目级。
- `replace` 与 `conflicts_with` 不能跨 store：本书例外按优先级覆盖全局规则，不算冲突，直接 `remember` 为 book 条目；要把全局规则改成本书规则，拆成 `forget` ＋ `remember` 两个事件。

## 查询

各 skill 入口已写好本任务的 `query` 命令（长篇正文由组装脚本代查）；没写命令的长篇设定、大纲等任务查 `story_design` + `workflow` + `interaction`，结果只给主会话、不传正文 agent。四类任务的映射表见维护文件。state 存在才查（两级都不存在时返回空结果、零写入）；结果合并项目级与 `--book-root` 所指书级（不传就拿不到本书条目），`--kind` 必传，输出不超过 2048 字节。

普通创作只做一次本地 `query`，完整画像、证据、候选和 journal 不进 prompt。查询项是低优先级倾向，不是逐条打卡清单：自然吸收，不复述画像、不刻意提高词面命中率，不为命中牺牲连贯、节奏、字数或本书既定笔调。**`omitted_ids` 非空＝记忆超编**，不是「没有更多了」：转告作者并建议「整理作者记忆」，不得改读完整画像规避预算。待确认项不进 prompt，也不为确认它们中断任务；只在作者主动查看、候选积累到适合回顾的节点，或新偏好与 active 条目冲突时集中呈现。

## 记不记、记成什么

不装记录全部用户消息的 prompt hook，不在作者没开口时观察他，只记作者明确表达的偏好。是否属于长期习惯由 agent 判断，拿不准就只执行不记录；作者可明说“记住：……”，以回执验收。

| 输入证据 | 处理 |
|---|---|
| “以后都这样”“我一直习惯……”等直接、稳定、范围清楚的原话 | `active`，`source=explicit_user` |
| 用户明确接受助手提出的长期做法 | `active`，`source=accepted_suggestion` |
| 作者原话像长期偏好但范围或稳定性含糊 | `pending`，取当前最窄合理范围；待确认只来自作者自己的话 |
| 同类修改反复出现、从成稿或操作轨迹看出的模式 | 不记录、不推断；作者没开口的偏好不进记忆 |
| “这一章别……”“这次给我……”等一次性要求 | 只执行，不记录 |
| 角色、时间线、伏笔、世界观、当前剧情走向 | 写项目设定/追踪，不写作者记忆 |
| 助手自己生成的文字、默认模板、工具告警、rubric 结论 | 不自我学习 |

保留否定词、限定词和适用范围：`quote` 写原话，`assertion` 只做不改变语义的紧凑归纳，**新建条目限一句话（≤120 字节，约 40 个字）**，写不下就压缩措辞、不切限定词；背景写进 `reason`（不进 prompt），不另开字段。

**一条偏好就是一条记录，例外和限定不许拆出去单列。** 「以后少用破折号，对话里也别用，除非表示打断」整条写成「破折号少用、对话里也不用，只在表示打断时保留」：超编时条目逐条被丢，拆开就可能只丢掉例外，把作者说过的限定变成绝对禁令。只有原话塞了**几条互不依赖**的偏好（如「多用短句」＋「章末留钩子」）才拆。

范围：“本书 / 这个角色 / 这次连载” → `book`；“都市文 / 这类题材” → `genre`；交稿、检查、确认节奏等操作习惯 → `workflow`；“以后 / 一贯 / 我习惯”且无更窄限定 → `global`；含糊但可能稳定 → 最窄合理范围并置 `pending`。

类型：`prose_style`、`story_design`、`workflow`、`delivery`、`interaction`。置信度与重要度均为 `low | medium | high`；超编时先丢重要度低的，按偏好的实际分量填，不要一律 `high`。`source` 只接受 `explicit_user`、`accepted_suggestion`、`manual`，工具拒绝推断类来源。

## 确认、替换、忘掉与冲突

- 同一类型、范围、归纳文本再出现，脚本强化原条目（累加证据与确认次数），不重复建条。
- 新偏好与同一 store 的 active 条目矛盾：以 `conflict` 记候选，`conflicts_with` 列冲突 ID，本轮仍按当前要求执行；本书例外与全局规则不算冲突。冲突候选不能直接 activate，落定见维护文件「冲突候选」。
- pending 用 `decide=activate|reject`。同一范围的规则改版用 `replace`，新条目启用、旧条目标 `superseded`；只有作者明确撤销或改变旧规则范围才跨范围替换。
- 作者说“忘掉 / 这不再是我的习惯”用 `forget`，保留历史证据但不再加载。active 条目的语义不可原地偷改，语义变化必须 replace，历史才可审计。

## 回执怎么告诉作者

回复就两行纯文本，不加代码块或引用格式：第一行用一句人话说记住了什么、管哪本书或哪类场合，如「记住了：《{书名}》的对话一律用「」，以后写这本书都照这个来；想改随时说。」；第二行是机器回执作凭证，如「技术备注：Author Memory Receipt: r1 · BP001」。

- 确认、替换、忘掉同理：「好，这条生效了：……」「换成了：……，原来的「……」不再用」「忘掉了：……」。只进待确认时说「这条先记在待确认里，你说"确认"才生效」；有冲突时用原话说明跟哪条旧习惯冲突。
- `warnings` / `omitted_ids` 不原样贴：说「你的习惯攒得有点多，写正文时这几条可能顾不上：「……」」，并建议说「整理作者记忆」。不提字节、prompt、kind、scope；编号只能跟着原话出现。

## 运行工具

依次尝试 `python3`、`python`、`py -3` 找到 Python 3，从当前 skill 根运行本地副本：

```text
{PYTHON} {当前 skill 根}/scripts/author_memory_commit.py record --workspace {工作区} [--book-root {书目录}] --input {工作区}/.story/work/作者记忆-事件.json
{PYTHON} {当前 skill 根}/scripts/author_memory_commit.py query  --workspace {工作区} --book-root {书目录} --kind {类型}（必传，可重复） [--genre {题材}] [--workflow {流程}]
```

- 子命令都可加 `--book {书名}`；写某本书时一律带 `--book-root`。事件 JSON 写在 `{工作区}/.story/work/`（不写系统 `/tmp`），成功后删掉；book 条目的 `scope.value` 填书名（书级 store 已记的名字，首次取目录名）。
- 明确的“记住 / 确认 / 替换 / 忘掉”都走单事件 `record`：自动读该 store 当前修订、首次自动初始化，不手工读修订号或拼多操作事务。同一 `event_id` 同内容幂等返回原回执，内容不同则失败；返回的 `store` / `book` 说明写到了哪一级。
- 成功才有 `Author Memory Receipt: rN · APxxx`，没有回执不得声称“已经记住”。**写入不因注入预算失败**：`warnings` 只是提醒（另一级 store 读不出来也在这里注明），按上节转告；有回执就是已记住，不要换 `event_id` 重试。

## 事件格式

新增或强化（`record` 输入；book 范围传 `--book-root`）：

```json
{
  "schema_version": 1,
  "event_id": "conversation-2026-08-25-message-42",
  "operation": {
    "action": "remember",
    "preference": {
      "kind": "prose_style",
      "scope": {"level": "global", "value": null},
      "assertion": "对话尽量短，用动作承接情绪，不用大段解释",
      "quote": "以后对话都短一点，情绪放动作里，别让角色长篇解释。",
      "source_ref": "conversation:2026-08-25",
      "source": "explicit_user",
      "confidence": "high",
      "importance": "high",
      "status": "active",
      "reason": "用户以“以后”明确声明长期偏好",
      "conflicts_with": []
    }
  }
}
```

待确认用 `"status": "pending"`；冲突候选用 `conflict` 并填同一 store 的 active ID。确认、替换、忘掉时，把下列对象换进新事件的 `operation`（`BP` 编号传 `--book-root`）；`replace.preference` 字段同上但不传 `status`、`conflicts_with`，新条目直接 active：

```json
{"action":"decide","item_id":"AP002","decision":"activate","quote":"对，这就是我的长期习惯。","reason":"作者明确确认"}
{"action":"replace","old_ids":["AP001"],"preference":{"kind":"prose_style","scope":{"level":"global","value":null},"assertion":"以后对话允许更长的试探，但避免解释设定","quote":"……","source_ref":"conversation:2026-08-25","source":"explicit_user","confidence":"high","importance":"high","reason":"作者明确替换原有全局规则，不是新增本书例外"}}
{"action":"forget","item_id":"AP003","quote":"忘掉这个偏好。","reason":"作者明确撤回"}
```

#!/usr/bin/env python3
"""Maintain evidence-backed author preferences and deterministic Markdown views.

The language model supplies compact semantic transactions. This tool validates
and applies them in memory, renders every derived view, and writes the JSON state
last as the commit point. Author memory lives in two kinds of store: the
project-level store under the workspace holds global / genre / workflow items
(`AP` ids); each book keeps its own book-level store under the book directory
(`BP` ids) so memory travels with the book. Both stay separate from each book's
story-continuity tracking. When the book root is the workspace itself, the
book-level store moves into a `书级/` subdirectory so the two never share a file.
"""

from __future__ import annotations

import argparse
import copy
import hashlib
import json
import os
import stat
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


INPUT_SCHEMA_VERSION = 1
STATE_SCHEMA_VERSION = 1
STATE_MAX_BYTES = 2 * 1024 * 1024
PROFILE_MAX_BYTES = 12288
PENDING_MAX_BYTES = 12288
JOURNAL_MAX_BYTES = 24576
QUERY_MAX_BYTES = 2048
ASSERTION_MAX_BYTES = 120  # 新建条目的断言限一句话；解释进 reason（不进 query 载荷）。
OMITTED_IDS_MAX = 20  # omitted_ids 封顶，omitted 保留真实总数——漏项列表不许把载荷本身挤炸。
LEGACY_ASSERTION_MAX_BYTES = 768  # 存量条目的读取上限；强化老条目不受新上限约束。

# query 的输出是要原样贴进执行 agent prompt 的注入载荷，QUERY_MAX_BYTES
# 是它在 prompt 里的注意力预算，不该放大。防「作者以为载入了、实际被静默
# 截断挤掉」靠三层：①新建条目的断言限 ASSERTION_MAX_BYTES，从源头短（强化
# 已有条目不受限，否则存量长断言再也无法被确认，只会派生重复条目）；②写入
# 端按下列任务组合（与 references/author-memory-maintenance.md 的映射表同包跟版）估算
# 最坏查询情形——全局条目＋各 scope 维度上最重的单一切片（一次查询只带一
# 个 book/genre/workflow，不同书的条目不会同现；切片按 casefold 归并，与
# same_scope_value 同一口径，轻重按 compact 字节＋列表分隔符算，与真实载荷
# 同一把尺），装不下时在返回的 warnings 里点名将被略过的条目、指向「整理作
# 者记忆」，写入本身永不因注入预算失败；③查询按 重要度→本书例外→最近更新
# 排序装填，被略过的恒是重要度较低的条目，漏下的 ID 按同一优先级顺序报进
# omitted_ids。
QUERY_COMBOS: dict[str, tuple[str, ...]] = {
    "正文初稿/续写": ("prose_style", "story_design"),
    "去AI味/改写": ("prose_style",),
    "设定/大纲": ("story_design", "workflow", "interaction"),
    "审稿": ("delivery", "interaction", "prose_style"),
}

KINDS = ("prose_style", "story_design", "workflow", "delivery", "interaction")
KIND_TITLES = {
    "prose_style": "文风与表达",
    "story_design": "故事设计",
    "workflow": "创作流程",
    "delivery": "交付格式",
    "interaction": "协作方式",
}
SCOPE_LEVELS = ("global", "genre", "book", "workflow")
STATUSES = ("active", "pending", "conflict", "rejected", "superseded")
CONFIDENCE_LEVELS = ("low", "medium", "high")
IMPORTANCE_LEVELS = ("low", "medium", "high")
# 作者记忆只记作者明确表达的偏好。repeated_correction / inferred_pattern 两条
# 由 agent 主动推断写入的管道已经移除（#436）：它们只在攒待确认清单的审阅负
# 担，不是记忆质量；文档也明说不装全量消息 hook，隐式捕获本就承诺不了完整
# 性。SOURCES 里保留这两个值只为存量 state 仍能通过校验、仍能 decide/forget，
# 新写入一律按 WRITE_SOURCES 校验。
SOURCES = (
    "explicit_user",
    "accepted_suggestion",
    "repeated_correction",
    "inferred_pattern",
    "manual",
)
WRITE_SOURCES = ("explicit_user", "accepted_suggestion", "manual")
RANK = {"low": 0, "medium": 1, "high": 2}

# 两级 store（#435）：项目级存 global/genre/workflow，ID 前缀 AP，位于
# {工作区}/.story/作者记忆/；书级只存该书的 book 条目，ID 前缀 BP，位于
# {书}/.story/作者记忆/，书归档、迁移时记忆随书走。ID 前缀就是路由键——
# decide/forget 看 item_id 前缀，remember/replace 看 scope.level，一份事务
# 只写一个 store。项目级 store 里升级前写入的存量 book 条目不再参与查询与估算，
# 用 migrate 搬进书目录后才回来——不做双读，双读会让迁移永远没人做。
STORE_PREFIX = {"project": "AP", "book": "BP"}
ID_PREFIXES = tuple(STORE_PREFIX.values())
# 单书布局（书根就是工作区）下书级 store 的子目录：两级 store 的默认落点在这种
# 布局里是同一个 state 文件，书级改住 {工作区}/.story/作者记忆/书级/。
SINGLE_ROOT_BOOK_DIR = "书级"


class AuthorMemoryError(ValueError):
    """Expected validation or state error."""


def require(condition: bool, message: str) -> None:
    if not condition:
        raise AuthorMemoryError(message)


def as_mapping(value: object, label: str) -> dict[str, Any]:
    require(isinstance(value, dict), f"{label} must be a JSON object")
    return value


def as_list(value: object, label: str) -> list[Any]:
    require(isinstance(value, list), f"{label} must be a JSON array")
    return value


def as_int(value: object, label: str, *, minimum: int = 0) -> int:
    require(isinstance(value, int) and not isinstance(value, bool), f"{label} must be an integer")
    require(value >= minimum, f"{label} must be >= {minimum}")
    return value


def require_known_keys(mapping: dict[str, Any], allowed: set[str], label: str) -> None:
    unknown = set(mapping) - allowed
    require(not unknown, f"{label} contains unsupported fields: {', '.join(sorted(unknown))}")


def clean_text(value: object, label: str, *, max_bytes: int = 768) -> str:
    require(isinstance(value, str), f"{label} must be a string")
    cleaned = " ".join(value.replace("|", "｜").split())
    require(bool(cleaned), f"{label} must not be empty")
    require(len(cleaned.encode("utf-8")) <= max_bytes, f"{label} exceeds {max_bytes} bytes")
    return cleaned


def optional_text(value: object, label: str, *, max_bytes: int = 768) -> str | None:
    if value is None:
        return None
    return clean_text(value, label, max_bytes=max_bytes)


def choice(value: object, allowed: tuple[str, ...], label: str) -> str:
    require(isinstance(value, str) and value in allowed, f"{label} must be one of: {', '.join(allowed)}")
    return value


def is_item_id(value: object) -> bool:
    return (
        isinstance(value, str)
        and len(value) >= 3
        and value[:2] in ID_PREFIXES
        and value[2:].isdecimal()
        and int(value[2:]) >= 1
    )


def id_number(item_id: str) -> int:
    return int(item_id[2:])


def id_store(item_id: str) -> str:
    return "book" if item_id.startswith(STORE_PREFIX["book"]) else "project"


def clean_id_list(value: object, label: str, *, maximum: int = 32) -> list[str]:
    raw = as_list(value, label)
    require(len(raw) <= maximum, f"{label} may contain at most {maximum} items")
    result: list[str] = []
    for index, item in enumerate(raw):
        item_id = clean_text(item, f"{label}[{index}]", max_bytes=32)
        require(is_item_id(item_id), f"{label}[{index}] is not an author-memory id")
        if item_id not in result:
            result.append(item_id)
    return result


def emit(document: object, *, error: bool = False) -> None:
    payload = json.dumps(document, ensure_ascii=False, sort_keys=True)
    stream = sys.stderr if error else sys.stdout
    stream.flush()
    stream.buffer.write((payload + "\n").encode("utf-8"))
    stream.buffer.flush()


def json_payload(document: object) -> str:
    return json.dumps(document, ensure_ascii=False, indent=2, sort_keys=True) + "\n"


def read_json(path: Path) -> object:
    try:
        require(path.stat().st_size <= STATE_MAX_BYTES, f"{path} exceeds {STATE_MAX_BYTES} bytes")
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise AuthorMemoryError(f"unable to read JSON {path}: {exc}") from exc


def atomic_write_text(path: Path, payload: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    mode = stat.S_IMODE(path.stat().st_mode) if path.exists() else 0o644
    fd, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
    temporary = Path(temporary_name)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(temporary, mode)
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def write_if_changed(path: Path, payload: str) -> None:
    try:
        if path.read_text(encoding="utf-8") == payload:
            return
    except FileNotFoundError:
        pass
    atomic_write_text(path, payload)


# ---------------------------------------------------------------------------
# Stores
# ---------------------------------------------------------------------------


class Store:
    """一个 state 文件的落点：项目级（工作区）或书级（书目录）。"""

    __slots__ = ("kind", "root", "book")

    def __init__(self, kind: str, root: Path, book: str | None) -> None:
        self.kind = kind
        self.root = root
        self.book = book

    @property
    def prefix(self) -> str:
        return STORE_PREFIX[self.kind]

    @property
    def state_path(self) -> Path:
        return self.root / "_author-memory-state.json"

    @property
    def label(self) -> str:
        return "项目级" if self.kind == "project" else f"书级（{self.book}）"


def project_store(workspace: Path) -> Store:
    return Store("project", workspace.resolve() / ".story" / "作者记忆", None)


def peek_book_name(state_path: Path) -> str | None:
    """不校验整份 state，只取书名——书级 store 一旦建立，书名以 state 为准，
    书目录改名不影响。"""
    if not state_path.exists():
        return None
    document = read_json(state_path)
    name = document.get("book") if isinstance(document, dict) else None
    return name if isinstance(name, str) and name.strip() else None


def enclosing_workspace(workspace: Path) -> Path | None:
    """--workspace 指到的其实是多书工作区里的一本书时，返回真正的创作工作区。

    认三种迹象，任一成立即是：①工作区的上一层目录叫 长篇 / 短篇（story-setup 的书
    目录约定，返回再上一层）；②某个祖先目录有 .active-book；③某个祖先目录有项目级
    store（不带 book 字段的 state）——但工作区自己已有项目级 store 时不认这一条，最
    近的项目级 store 就是它自己，免得主目录里一份误建的 store 挡住正常的单书工作区。
    读不出来的祖先 state 不算迹象。"""
    resolved = workspace.resolve()
    if resolved.parent.name in {"长篇", "短篇"}:
        return resolved.parent.parent
    own = project_store(resolved).state_path
    owns_project_store = own.exists() and peek_book_name(own) is None
    for ancestor in resolved.parents:
        if (ancestor / ".active-book").exists():
            return ancestor
        if owns_project_store:
            continue
        state_path = project_store(ancestor).state_path
        try:
            document = read_json(state_path) if state_path.is_file() else None
        except AuthorMemoryError:
            document = None
        if isinstance(document, dict) and "book" not in document:
            return ancestor
    return None


def same_directory(first: Path, second: Path) -> bool:
    # samefile 认得大小写不敏感文件系统（macOS APFS 默认）上只差大小写的同一目录。
    if first.exists() and second.exists():
        return os.path.samefile(first, second)
    return first.resolve() == second.resolve()


def is_single_root(workspace: Path, book_root: Path | None) -> bool:
    """单书布局：书根就是工作区（正文/、大纲/、追踪/ 直接在工作区根）。

    同一目录还可能是多书工作区里的一本书被误传成了 --workspace（书目录自己也含
    .story/作者记忆/）。那时按单书布局处理会把书级 state 挪进 书级/、在原处补一份
    空的项目级 state，此后正确的调用全部失败，所以一律报错、零写入。"""
    if book_root is None or not same_directory(book_root, workspace):
        return False
    outer = enclosing_workspace(workspace)
    require(
        outer is None,
        f"{workspace} 是创作工作区 {outer} 里的一本书，不是单书工作区：--workspace 应传 {outer}，"
        f"这个目录放 --book-root",
    )
    return True


def book_memory_root(workspace: Path, book_root: Path) -> Path:
    root = book_root.resolve() / ".story" / "作者记忆"
    # 单书布局下两级 store 的默认落点是同一个 state 文件；书级改住子目录，
    # 两份 state、两套派生视图、两条修订线仍各自独立。
    return root / SINGLE_ROOT_BOOK_DIR if is_single_root(workspace, book_root) else root


def sole_legacy_book_name(project: Store) -> str | None:
    """项目级 store 里存量 book 条目只指向一本书时返回该书名（单书布局的升级默认）。"""
    if not project.state_path.exists():
        return None
    document = read_json(project.state_path)
    items = document.get("items") if isinstance(document, dict) else None
    names: dict[str, str] = {}
    for item in (items.values() if isinstance(items, dict) else ()):
        scope = item.get("scope") if isinstance(item, dict) else None
        if (
            isinstance(scope, dict) and scope.get("level") == "book"
            and isinstance(scope.get("value"), str) and scope["value"].strip()
            and item.get("status") in {"active", "pending", "conflict"}
        ):
            names.setdefault(scope["value"].casefold(), scope["value"])
    return next(iter(names.values())) if len(names) == 1 else None


def book_store(workspace: Path, book_root: Path, book: str | None) -> Store:
    require(book_root.exists() and book_root.is_dir(), f"book root does not exist: {book_root}")
    resolved = book_root.resolve()
    root = book_memory_root(workspace, book_root)
    name = optional_text(book, "book", max_bytes=180)
    stored = peek_book_name(root / "_author-memory-state.json")
    if name is None:
        name = stored
    elif stored is not None:
        require(
            name.casefold() == stored.casefold(),
            f"--book「{name}」与 {root} 里记录的书「{stored}」不一致",
        )
    if name is None and is_single_root(workspace, book_root):
        # 单书工作区的目录名常常不是书名；升级前的本书条目只指向一本书时，以它为准，
        # 否则 migrate 会按目录名找不到存量、静默迁移零条。
        name = sole_legacy_book_name(project_store(workspace))
    if name is None:
        name = clean_text(resolved.name, "book root name", max_bytes=180)
    return Store("book", root, name)


def relocate_misplaced_book_state(workspace: Path, book_root: Path | None) -> None:
    """单书布局的自愈：旧版把书级 state（带 state.book）写在了项目级位置，此后项目级
    读写一律失败。带 --book-root {工作区} 运行任一命令时，把它原子移进书级子目录，
    再在项目级位置补一份空 state 重建视图。state 内容不变，不推进任何修订。"""
    if not is_single_root(workspace, book_root):
        return
    project = project_store(workspace)
    misplaced = peek_book_name(project.state_path)
    if misplaced is None:
        return
    target = Store("book", book_memory_root(workspace, book_root), misplaced)
    require(
        not target.state_path.exists(),
        f"{project.state_path} 与 {target.state_path} 都是书级 state，无法自动归位；"
        f"保留修订较新的一份移到 {target.state_path}，另一份备份后移走再重跑",
    )
    state = validate_state(read_json(project.state_path), store=target)
    target.root.mkdir(parents=True, exist_ok=True)
    os.replace(project.state_path, target.state_path)
    write_snapshot(target, state)
    write_snapshot(project, empty_state())


def resolve_target_store(kind: str, workspace: Path, book_root: Path | None, book: str | None) -> Store:
    if kind == "project":
        return project_store(workspace)
    require(
        book_root is not None,
        "book 级条目须传 --book-root {书目录}——记忆随书存放在 {书}/.story/作者记忆/，不再写进工作区的项目级 store",
    )
    return book_store(workspace, book_root, book)


def empty_state(book: str | None = None) -> dict[str, Any]:
    state: dict[str, Any] = {
        "schema_version": STATE_SCHEMA_VERSION,
        "state_revision": 0,
        "next_item_number": 1,
        "items": {},
        "journal": [],
        "applied_transactions": {},
    }
    if book is not None:
        state["book"] = book
    return state


# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------


def normalize_scope(value: object, label: str) -> dict[str, str | None]:
    scope = as_mapping(value, label)
    require_known_keys(scope, {"level", "value"}, label)
    level = choice(scope.get("level"), SCOPE_LEVELS, f"{label}.level")
    raw_value = scope.get("value")
    if level == "global":
        require(raw_value is None, f"{label}.value must be null for global scope")
        normalized_value = None
    else:
        normalized_value = clean_text(raw_value, f"{label}.value", max_bytes=180)
    return {"level": level, "value": normalized_value}


def normalize_evidence(value: object, label: str) -> dict[str, str | None]:
    evidence = as_mapping(value, label)
    require_known_keys(evidence, {"quote", "source_ref"}, label)
    return {
        "quote": clean_text(evidence.get("quote"), f"{label}.quote", max_bytes=768),
        "source_ref": optional_text(evidence.get("source_ref"), f"{label}.source_ref", max_bytes=240),
    }


def normalize_item(value: object, label: str) -> dict[str, Any]:
    item = as_mapping(value, label)
    allowed = {
        "id", "kind", "scope", "assertion", "confidence", "importance", "status", "source",
        "reason", "conflicts_with", "confirmation_count", "evidence", "created_revision",
        "updated_revision", "superseded_by",
    }
    require_known_keys(item, allowed, label)
    item_id = clean_text(item.get("id"), f"{label}.id", max_bytes=32)
    require(is_item_id(item_id), f"{label}.id is invalid")
    evidence = [normalize_evidence(entry, f"{label}.evidence[{index}]") for index, entry in enumerate(as_list(item.get("evidence"), f"{label}.evidence"))]
    require(bool(evidence), f"{label}.evidence must not be empty")
    status = choice(item.get("status"), STATUSES, f"{label}.status")
    conflicts = clean_id_list(item.get("conflicts_with"), f"{label}.conflicts_with")
    superseded_by = optional_text(item.get("superseded_by"), f"{label}.superseded_by", max_bytes=32)
    if superseded_by is not None:
        require(is_item_id(superseded_by), f"{label}.superseded_by is invalid")
    return {
        "id": item_id,
        "kind": choice(item.get("kind"), KINDS, f"{label}.kind"),
        "scope": normalize_scope(item.get("scope"), f"{label}.scope"),
        "assertion": clean_text(item.get("assertion"), f"{label}.assertion", max_bytes=LEGACY_ASSERTION_MAX_BYTES),
        "confidence": choice(item.get("confidence"), CONFIDENCE_LEVELS, f"{label}.confidence"),
        "importance": choice(item.get("importance"), IMPORTANCE_LEVELS, f"{label}.importance"),
        "status": status,
        "source": choice(item.get("source"), SOURCES, f"{label}.source"),
        "reason": clean_text(item.get("reason"), f"{label}.reason", max_bytes=480),
        "conflicts_with": conflicts,
        "confirmation_count": as_int(item.get("confirmation_count"), f"{label}.confirmation_count", minimum=1),
        "evidence": evidence,
        "created_revision": as_int(item.get("created_revision"), f"{label}.created_revision", minimum=1),
        "updated_revision": as_int(item.get("updated_revision"), f"{label}.updated_revision", minimum=1),
        "superseded_by": superseded_by,
    }


def scope_fits_store(scope: dict[str, str | None], book: str | None) -> bool:
    if book is None:
        return True  # 项目级：存量 book 条目仍合法，只为老库能通过校验并被 migrate 搬走
    return scope["level"] == "book" and (scope["value"] or "").casefold() == book.casefold()


def validate_state(value: object, *, store: Store) -> dict[str, Any]:
    state = as_mapping(value, "state")
    allowed = {"schema_version", "state_revision", "next_item_number", "items", "journal", "applied_transactions", "book"}
    require_known_keys(state, allowed, "state")
    require(state.get("schema_version") == STATE_SCHEMA_VERSION, f"state.schema_version must be {STATE_SCHEMA_VERSION}")
    if store.kind == "book":
        book = clean_text(state.get("book"), "state.book", max_bytes=180)
        require(store.book is not None and book.casefold() == store.book.casefold(), f"state.book「{book}」与目标书「{store.book}」不一致")
    else:
        require(
            "book" not in state,
            "project-level state must not carry state.book：这份其实是书级 state。--workspace 指到了"
            "长篇/、短篇/ 下的书目录时，改传创作工作区根、书目录放 --book-root；只有书根就是工作区"
            "（单书布局）时，带 --book-root {工作区} 重跑任一命令即自动移进 .story/作者记忆/书级/",
        )
        book = None
    revision = as_int(state.get("state_revision"), "state.state_revision")
    next_number = as_int(state.get("next_item_number"), "state.next_item_number", minimum=1)
    raw_items = as_mapping(state.get("items"), "state.items")
    items: dict[str, Any] = {}
    max_number = 0
    for raw_id, raw_item in raw_items.items():
        normalized = normalize_item(raw_item, f"state.items.{raw_id}")
        require(raw_id == normalized["id"], f"state.items key {raw_id} does not match item id")
        require(raw_id.startswith(store.prefix), f"state.items.{raw_id} does not belong to the {store.label} store（前缀应为 {store.prefix}）")
        require(scope_fits_store(normalized["scope"], book), f"state.items.{raw_id} scope does not belong to book「{book}」")
        max_number = max(max_number, id_number(raw_id))
        require(normalized["created_revision"] <= normalized["updated_revision"] <= revision, f"state.items.{raw_id} revision is ahead of state")
        items[raw_id] = normalized
    require(next_number > max_number, "state.next_item_number must be greater than every allocated item id")
    for item_id, item in items.items():
        for conflict_id in item["conflicts_with"]:
            require(conflict_id in items and conflict_id != item_id, f"state.items.{item_id} has an invalid conflict id")
        if item["superseded_by"] is not None:
            require(item["superseded_by"] in items and item["superseded_by"] != item_id, f"state.items.{item_id} has an invalid superseded_by id")
        if item["status"] == "active":
            require(not item["conflicts_with"], f"active item {item_id} cannot retain conflicts")
        if item["status"] == "pending":
            require(not item["conflicts_with"], f"pending item {item_id} cannot retain conflicts")
        if item["status"] == "conflict":
            require(bool(item["conflicts_with"]), f"conflict item {item_id} must reference an active item")
            require(all(items[conflict_id]["status"] == "active" for conflict_id in item["conflicts_with"]), f"conflict item {item_id} must reference only active items")
        if item["status"] != "superseded":
            require(item["superseded_by"] is None, f"only superseded item {item_id} may set superseded_by")
    journal = as_list(state.get("journal"), "state.journal")
    require(len(journal) == revision, "state.journal length must equal state.state_revision")
    journal_revisions: dict[str, int] = {}
    for index, entry in enumerate(journal):
        mapping = as_mapping(entry, f"state.journal[{index}]")
        require_known_keys(mapping, {"revision", "transaction_id", "committed_at", "summaries"}, f"state.journal[{index}]")
        entry_revision = as_int(mapping.get("revision"), f"state.journal[{index}].revision", minimum=1)
        require(entry_revision == index + 1, f"state.journal[{index}].revision must be {index + 1}")
        transaction_id = clean_text(mapping.get("transaction_id"), f"state.journal[{index}].transaction_id", max_bytes=128)
        require(transaction_id not in journal_revisions, f"state.journal repeats transaction_id {transaction_id}")
        journal_revisions[transaction_id] = entry_revision
        clean_text(mapping.get("committed_at"), f"state.journal[{index}].committed_at", max_bytes=64)
        summaries = as_list(mapping.get("summaries"), f"state.journal[{index}].summaries")
        require(bool(summaries), f"state.journal[{index}].summaries must not be empty")
        for summary_index, summary in enumerate(summaries):
            clean_text(summary, f"state.journal[{index}].summaries[{summary_index}]", max_bytes=768)
    transactions = as_mapping(state.get("applied_transactions"), "state.applied_transactions")
    require(set(transactions) == set(journal_revisions), "state.applied_transactions must match state.journal transaction ids")
    for transaction_id, record in transactions.items():
        clean_text(transaction_id, "state.applied_transactions key", max_bytes=128)
        mapping = as_mapping(record, f"state.applied_transactions.{transaction_id}")
        require_known_keys(mapping, {"revision", "digest", "item_ids"}, f"state.applied_transactions.{transaction_id}")
        transaction_revision = as_int(mapping.get("revision"), f"state.applied_transactions.{transaction_id}.revision", minimum=1)
        require(transaction_revision == journal_revisions[transaction_id], f"state.applied_transactions.{transaction_id}.revision does not match journal")
        digest = clean_text(mapping.get("digest"), f"state.applied_transactions.{transaction_id}.digest", max_bytes=64)
        require(len(digest) == 64 and all(char in "0123456789abcdef" for char in digest), f"state.applied_transactions.{transaction_id}.digest is invalid")
        item_ids = clean_id_list(mapping.get("item_ids"), f"state.applied_transactions.{transaction_id}.item_ids")
        require(bool(item_ids), f"state.applied_transactions.{transaction_id}.item_ids must not be empty")
        require(all(item_id in items for item_id in item_ids), f"state.applied_transactions.{transaction_id}.item_ids references an unknown item")
    result = {
        "schema_version": STATE_SCHEMA_VERSION,
        "state_revision": revision,
        "next_item_number": next_number,
        "items": items,
        "journal": copy.deepcopy(journal),
        "applied_transactions": copy.deepcopy(transactions),
    }
    if book is not None:
        result["book"] = book
    return result


def load_state(store: Store) -> dict[str, Any] | None:
    if not store.state_path.exists():
        return None
    return validate_state(read_json(store.state_path), store=store)


def normalize_preference(value: object, label: str, *, allow_status: bool) -> dict[str, Any]:
    preference = as_mapping(value, label)
    allowed = {"kind", "scope", "assertion", "quote", "source_ref", "source", "confidence", "importance", "reason"}
    if allow_status:
        allowed |= {"status", "conflicts_with"}
    require_known_keys(preference, allowed, label)
    raw_source = preference.get("source")
    require(
        raw_source not in {"repeated_correction", "inferred_pattern"},
        f"{label}.source「{raw_source}」已不再写入：作者记忆只记作者明确表达的偏好，"
        f"不从重复修改或成稿轨迹推断；范围含糊的原话用 explicit_user 并置 status=pending",
    )
    source = choice(raw_source, WRITE_SOURCES, f"{label}.source")
    status = choice(preference.get("status"), ("active", "pending", "conflict"), f"{label}.status") if allow_status else "active"
    conflicts = clean_id_list(preference.get("conflicts_with", []), f"{label}.conflicts_with") if allow_status else []
    if status == "active":
        require(not conflicts, f"{label}.conflicts_with must be empty for active status")
    elif status == "conflict":
        require(bool(conflicts), f"{label}.conflicts_with is required for conflict status")
    else:
        require(not conflicts, f"{label}.conflicts_with is only valid for conflict status")
    return {
        "kind": choice(preference.get("kind"), KINDS, f"{label}.kind"),
        "scope": normalize_scope(preference.get("scope"), f"{label}.scope"),
        # 这里按存量上限收；ASSERTION_MAX_BYTES 只在真正新建条目时校验
        # （require_new_item_assertion），好让存量长断言仍能被强化。
        "assertion": clean_text(preference.get("assertion"), f"{label}.assertion", max_bytes=LEGACY_ASSERTION_MAX_BYTES),
        "quote": clean_text(preference.get("quote"), f"{label}.quote", max_bytes=768),
        "source_ref": optional_text(preference.get("source_ref"), f"{label}.source_ref", max_bytes=240),
        "source": source,
        "confidence": choice(preference.get("confidence"), CONFIDENCE_LEVELS, f"{label}.confidence"),
        "importance": choice(preference.get("importance"), IMPORTANCE_LEVELS, f"{label}.importance"),
        "status": status,
        "reason": clean_text(preference.get("reason"), f"{label}.reason", max_bytes=480),
        "conflicts_with": conflicts,
    }


def normalize_transaction(value: object) -> dict[str, Any]:
    transaction = as_mapping(value, "transaction")
    require_known_keys(transaction, {"schema_version", "transaction_id", "expected_state_revision", "operations"}, "transaction")
    require(transaction.get("schema_version") == INPUT_SCHEMA_VERSION, f"transaction.schema_version must be {INPUT_SCHEMA_VERSION}")
    transaction_id = clean_text(transaction.get("transaction_id"), "transaction.transaction_id", max_bytes=128)
    operations = as_list(transaction.get("operations"), "transaction.operations")
    require(1 <= len(operations) <= 32, "transaction.operations must contain 1-32 operations")
    normalized_operations: list[dict[str, Any]] = []
    for index, raw_operation in enumerate(operations):
        label = f"transaction.operations[{index}]"
        operation = as_mapping(raw_operation, label)
        action = operation.get("action")
        if action == "remember":
            require_known_keys(operation, {"action", "preference"}, label)
            normalized_operations.append({"action": action, "preference": normalize_preference(operation.get("preference"), f"{label}.preference", allow_status=True)})
        elif action == "decide":
            require_known_keys(operation, {"action", "item_id", "decision", "quote", "reason"}, label)
            normalized_operations.append({
                "action": action,
                "item_id": clean_id_list([operation.get("item_id")], f"{label}.item_id", maximum=1)[0],
                "decision": choice(operation.get("decision"), ("activate", "reject"), f"{label}.decision"),
                "quote": clean_text(operation.get("quote"), f"{label}.quote", max_bytes=768),
                "reason": clean_text(operation.get("reason"), f"{label}.reason", max_bytes=480),
            })
        elif action == "replace":
            require_known_keys(operation, {"action", "old_ids", "preference"}, label)
            old_ids = clean_id_list(operation.get("old_ids"), f"{label}.old_ids")
            require(bool(old_ids), f"{label}.old_ids must not be empty")
            normalized_operations.append({"action": action, "old_ids": old_ids, "preference": normalize_preference(operation.get("preference"), f"{label}.preference", allow_status=False)})
        elif action == "forget":
            require_known_keys(operation, {"action", "item_id", "quote", "reason"}, label)
            normalized_operations.append({
                "action": action,
                "item_id": clean_id_list([operation.get("item_id")], f"{label}.item_id", maximum=1)[0],
                "quote": clean_text(operation.get("quote"), f"{label}.quote", max_bytes=768),
                "reason": clean_text(operation.get("reason"), f"{label}.reason", max_bytes=480),
            })
        else:
            raise AuthorMemoryError(f"{label}.action must be one of: remember, decide, replace, forget")
    return {
        "schema_version": INPUT_SCHEMA_VERSION,
        "transaction_id": transaction_id,
        "expected_state_revision": as_int(transaction.get("expected_state_revision"), "transaction.expected_state_revision"),
        "operations": normalized_operations,
    }


def normalize_record_event(value: object) -> dict[str, Any]:
    event = as_mapping(value, "event")
    require_known_keys(event, {"schema_version", "event_id", "operation"}, "event")
    require(event.get("schema_version") == INPUT_SCHEMA_VERSION, f"event.schema_version must be {INPUT_SCHEMA_VERSION}")
    event_id = clean_text(event.get("event_id"), "event.event_id", max_bytes=120)
    normalized = normalize_transaction({
        "schema_version": INPUT_SCHEMA_VERSION,
        "transaction_id": f"record:{event_id}",
        "expected_state_revision": 0,
        "operations": [event.get("operation")],
    })
    return {"event_id": event_id, "operation": normalized["operations"][0]}


# ---------------------------------------------------------------------------
# Routing: which store does an operation belong to?
# ---------------------------------------------------------------------------


def scope_store_kind(scope: dict[str, str | None]) -> str:
    return "book" if scope["level"] == "book" else "project"


def store_kind_label(kind: str) -> str:
    return "书级（BP）" if kind == "book" else "项目级（AP）"


def operation_store_kind(operation: dict[str, Any], label: str) -> str:
    action = operation["action"]
    if action in {"decide", "forget"}:
        return id_store(operation["item_id"])
    preference = operation["preference"]
    kind = scope_store_kind(preference["scope"])
    if action == "remember":
        for conflict_id in preference["conflicts_with"]:
            require(
                id_store(conflict_id) == kind,
                f"{label}.conflicts_with 只能引用同一 store 的条目（{conflict_id} 在{store_kind_label(id_store(conflict_id))}）："
                f"本书例外不算与全局规则冲突，直接 remember 为 book 条目即可",
            )
        return kind
    for old_id in operation["old_ids"]:
        require(
            id_store(old_id) == kind,
            f"{label} replace 不能跨 store：{old_id} 在{store_kind_label(id_store(old_id))}，新条目范围属于{store_kind_label(kind)}；"
            f"跨 store 改版拆成 forget＋remember，存量 book 条目先 migrate",
        )
    return kind


def transaction_store_kind(transaction: dict[str, Any]) -> str:
    kinds = {
        operation_store_kind(operation, f"transaction.operations[{index}]")
        for index, operation in enumerate(transaction["operations"])
    }
    require(
        len(kinds) == 1,
        "一份事务只能写一个 store：项目级（global/genre/workflow 范围或 AP 编号）与书级（book 范围或 BP 编号）的操作要分开提交",
    )
    return kinds.pop()


# ---------------------------------------------------------------------------
# Transactions
# ---------------------------------------------------------------------------


def transaction_digest(transaction: dict[str, Any]) -> str:
    canonical = json.dumps(transaction, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def fingerprint(preference: dict[str, Any]) -> str:
    """同一条偏好的身份：kind＋scope＋断言，scope.value 与 same_scope_value 同样按
    casefold 比——否则「Urban」「urban」会各建一条，两条同断言一起挤进 prompt。"""
    scope = preference["scope"]
    value = {
        "kind": preference["kind"],
        "scope": {"level": scope["level"], "value": None if scope["value"] is None else scope["value"].casefold()},
        "assertion": preference["assertion"].casefold(),
    }
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def require_new_item_assertion(preference: dict[str, Any]) -> None:
    """新建条目的断言限一句话。强化已有条目走不到这里——存量长断言必须还能
    被确认，否则作者重申老偏好只会派生一条重复条目，库反而更挤。"""
    size = len(preference["assertion"].encode("utf-8"))
    require(
        size <= ASSERTION_MAX_BYTES,
        f"新条目的 assertion {size} 字节，超出 {ASSERTION_MAX_BYTES} 字节上限——"
        f"断言限一句话，需要解释的背景写进 reason。若这是对已有条目的重申，"
        f"原样使用该条目的 assertion 即可强化（不受本上限约束）；"
        f"若确实是几条互不依赖的偏好，才拆成几条分别记录。",
    )


def require_scope_fits_store(state: dict[str, Any], scope: dict[str, str | None]) -> None:
    book = state.get("book")
    if book is None:
        require(
            scope["level"] != "book",
            "book 级条目须传 --book-root {书目录}——记忆随书存放在 {书}/.story/作者记忆/，不再写进工作区的项目级 store",
        )
    else:
        require(
            scope_fits_store(scope, book),
            f"book 条目的范围「{scope['value']}」与本书 store「{book}」不一致——只有这本书的条目才住在这个书目录",
        )


def next_item_id(state: dict[str, Any]) -> str:
    prefix = STORE_PREFIX["book"] if state.get("book") is not None else STORE_PREFIX["project"]
    item_id = f"{prefix}{state['next_item_number']:03d}"
    state["next_item_number"] += 1
    return item_id


def allocate_item(state: dict[str, Any], preference: dict[str, Any], revision: int) -> dict[str, Any]:
    require_new_item_assertion(preference)
    require_scope_fits_store(state, preference["scope"])
    return {
        "id": next_item_id(state),
        "kind": preference["kind"],
        "scope": copy.deepcopy(preference["scope"]),
        "assertion": preference["assertion"],
        "confidence": preference["confidence"],
        "importance": preference["importance"],
        "status": preference["status"],
        "source": preference["source"],
        "reason": preference["reason"],
        "conflicts_with": list(preference["conflicts_with"]),
        "confirmation_count": 1,
        "evidence": [{"quote": preference["quote"], "source_ref": preference["source_ref"]}],
        "created_revision": revision,
        "updated_revision": revision,
        "superseded_by": None,
    }


def best_level(first: str, second: str) -> str:
    return first if RANK[first] >= RANK[second] else second


def add_evidence(item: dict[str, Any], quote: str, source_ref: str | None) -> None:
    evidence = {"quote": quote, "source_ref": source_ref}
    if evidence not in item["evidence"]:
        item["evidence"].append(evidence)


def require_item(state: dict[str, Any], item_id: str, label: str) -> dict[str, Any]:
    require(item_id in state["items"], f"{label} references unknown item {item_id}")
    return state["items"][item_id]


def apply_remember(state: dict[str, Any], preference: dict[str, Any], revision: int) -> str:
    require_scope_fits_store(state, preference["scope"])
    for conflict_id in preference["conflicts_with"]:
        conflict = require_item(state, conflict_id, "remember")
        require(conflict["status"] == "active", f"remember conflict {conflict_id} must be active")
    preference_fingerprint = fingerprint(preference)
    for item in state["items"].values():
        if item["status"] not in {"active", "pending", "conflict"} or fingerprint(item) != preference_fingerprint:
            continue
        require(not (item["status"] == "conflict" and preference["status"] == "active"), f"conflict item {item['id']} must be resolved with replace or rejected")
        require(not (item["status"] == "active" and preference["status"] == "conflict"), f"active item {item['id']} cannot be recategorized as its own conflict")
        add_evidence(item, preference["quote"], preference["source_ref"])
        item["confirmation_count"] += 1
        item["confidence"] = best_level(item["confidence"], preference["confidence"])
        item["importance"] = best_level(item["importance"], preference["importance"])
        item["updated_revision"] = revision
        item["reason"] = preference["reason"]
        if item["status"] == "pending" and preference["status"] == "active":
            item["status"] = "active"
        elif item["status"] == "pending" and preference["status"] == "conflict":
            item["status"] = "conflict"
            item["conflicts_with"] = list(preference["conflicts_with"])
        elif item["status"] == "conflict" and preference["status"] == "conflict":
            item["conflicts_with"] = sorted(set(item["conflicts_with"]) | set(preference["conflicts_with"]))
        return f"强化 {item['id']}：{item['assertion']}"
    item = allocate_item(state, preference, revision)
    state["items"][item["id"]] = item
    return f"新增 {item['id']}（{item['status']}）：{item['assertion']}"


def apply_decide(state: dict[str, Any], operation: dict[str, Any], revision: int) -> str:
    item = require_item(state, operation["item_id"], "decide")
    require(item["status"] in {"pending", "conflict"}, f"decide requires pending/conflict item, got {item['status']}")
    if operation["decision"] == "activate":
        require(item["status"] == "pending" and not item["conflicts_with"], "conflict candidates must be activated with replace")
        item["status"] = "active"
        verb = "确认"
    else:
        item["status"] = "rejected"
        verb = "拒绝"
    add_evidence(item, operation["quote"], None)
    item["reason"] = operation["reason"]
    item["updated_revision"] = revision
    return f"{verb} {item['id']}：{item['assertion']}"


def release_conflicts(state: dict[str, Any], removed_ids: set[str], revision: int) -> int:
    """被撤下的 active 条目不再是任何候选的冲突对象；冲突对象清空的候选退回 pending。"""
    released = 0
    for candidate in state["items"].values():
        if candidate["status"] != "conflict":
            continue
        retained = [item_id for item_id in candidate["conflicts_with"] if item_id not in removed_ids]
        if retained == candidate["conflicts_with"]:
            continue
        candidate["conflicts_with"] = retained
        candidate["updated_revision"] = revision
        if not retained:
            candidate["status"] = "pending"
            released += 1
    return released


def apply_replace(state: dict[str, Any], operation: dict[str, Any], revision: int) -> str:
    old_items = [require_item(state, item_id, "replace") for item_id in operation["old_ids"]]
    for item in old_items:
        require(item["status"] in {"active", "conflict", "pending"}, f"replace target {item['id']} is already {item['status']}")
    replacement = allocate_item(state, operation["preference"], revision)
    replacement["status"] = "active"
    replacement["conflicts_with"] = []
    state["items"][replacement["id"]] = replacement
    for item in old_items:
        item["status"] = "superseded"
        item["superseded_by"] = replacement["id"]
        item["updated_revision"] = revision
    released = release_conflicts(state, {item["id"] for item in old_items}, revision)
    replaced = ", ".join(item["id"] for item in old_items)
    suffix = f"；{released} 个其他冲突候选退回待确认" if released else ""
    return f"用 {replacement['id']} 替代 {replaced}：{replacement['assertion']}{suffix}"


def apply_forget(state: dict[str, Any], operation: dict[str, Any], revision: int) -> str:
    item = require_item(state, operation["item_id"], "forget")
    require(item["status"] in {"active", "pending", "conflict"}, f"forget target {item['id']} is already {item['status']}")
    item["status"] = "superseded"
    item["superseded_by"] = None
    item["reason"] = operation["reason"]
    item["updated_revision"] = revision
    add_evidence(item, operation["quote"], None)
    released = release_conflicts(state, {item["id"]}, revision)
    suffix = f"；{released} 个冲突候选退回待确认" if released else ""
    return f"忘记 {item['id']}：{item['assertion']}{suffix}"


def committed_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def seal_transaction(
    updated: dict[str, Any],
    transaction_id: str,
    digest: str,
    summaries: list[str],
    *,
    store: Store,
) -> dict[str, Any]:
    """把本次修订写进 journal / applied_transactions，并整份校验。"""
    revision = updated["state_revision"] + 1
    updated["state_revision"] = revision
    updated["journal"].append({
        "revision": revision,
        "transaction_id": transaction_id,
        "committed_at": committed_now(),
        "summaries": summaries,
    })
    item_ids = sorted(
        (item_id for item_id, item in updated["items"].items() if item["updated_revision"] == revision),
        key=id_number,
    )
    require(bool(item_ids), "transaction did not update any author-memory item")
    updated["applied_transactions"][transaction_id] = {
        "revision": revision,
        "digest": digest,
        "item_ids": item_ids,
    }
    return validate_state(updated, store=store)


def apply_transaction(state: dict[str, Any], transaction: dict[str, Any], digest: str, *, store: Store) -> tuple[dict[str, Any], list[str]]:
    applied = state["applied_transactions"].get(transaction["transaction_id"])
    if applied is not None:
        require(applied["digest"] == digest, "transaction_id was already used with different content")
        return state, [f"事务已应用于修订 {applied['revision']}，本次为幂等重放"]
    require(transaction["expected_state_revision"] == state["state_revision"], f"stale state revision: expected {transaction['expected_state_revision']}, current {state['state_revision']}")
    updated = copy.deepcopy(state)
    revision = updated["state_revision"] + 1
    summaries: list[str] = []
    for operation in transaction["operations"]:
        if operation["action"] == "remember":
            summaries.append(apply_remember(updated, operation["preference"], revision))
        elif operation["action"] == "decide":
            summaries.append(apply_decide(updated, operation, revision))
        elif operation["action"] == "replace":
            summaries.append(apply_replace(updated, operation, revision))
        else:
            summaries.append(apply_forget(updated, operation, revision))
    return seal_transaction(updated, transaction["transaction_id"], digest, summaries, store=store), summaries


# ---------------------------------------------------------------------------
# Derived views
# ---------------------------------------------------------------------------


def scope_label(scope: dict[str, str | None]) -> str:
    if scope["level"] == "global":
        return "全局"
    labels = {"genre": "题材", "book": "本书", "workflow": "流程"}
    return f"{labels[scope['level']]}：{scope['value']}"


def store_caption(state: dict[str, Any]) -> str:
    book = state.get("book")
    if book is None:
        return "项目级记忆（全局、题材、流程）；各书的书级记忆住在各自书目录的 .story/作者记忆/。"
    return f"本书「{book}」的书级记忆；全局、题材、流程记忆住在工作区的 .story/作者记忆/。"


def render_profile(state: dict[str, Any]) -> str:
    lines = [
        "# 作者画像",
        "",
        "<!-- 由 author_memory_commit.py 生成，请勿手改；修改请提交事务。 -->",
        "",
        f"> 状态修订：{state['state_revision']}。{store_caption(state)}仅列出已确认偏好；当前明确要求、本书设定与硬性门禁优先。",
        "",
    ]
    active = [item for item in state["items"].values() if item["status"] == "active"]
    for kind in KINDS:
        lines.extend([f"## {KIND_TITLES[kind]}", ""])
        items = sorted((item for item in active if item["kind"] == kind), key=lambda item: id_number(item["id"]))
        if not items:
            lines.extend(["- 暂无", ""])
            continue
        for item in items:
            # 必须显示 importance：它决定超编时谁留在 prompt 里，而「整理作者
            # 记忆」只以本文件为输入——不显示就无从判断该退役哪条。
            lines.append(
                f"- **{item['id']}**〔{scope_label(item['scope'])}｜重要 {item['importance']}"
                f"｜把握 {item['confidence']}｜确认 {item['confirmation_count']} 次〕{item['assertion']}"
            )
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def render_pending(state: dict[str, Any]) -> str:
    lines = [
        "# 待确认的作者习惯",
        "",
        "<!-- 由 author_memory_commit.py 生成，请勿手改；修改请提交事务。 -->",
        "",
        f"> 状态修订：{state['state_revision']}。{store_caption(state)}待确认项不参与创作约束，也不应打断当前任务。",
        "",
    ]
    items = sorted((item for item in state["items"].values() if item["status"] in {"pending", "conflict"}), key=lambda item: id_number(item["id"]))
    if not items:
        lines.extend(["暂无待确认项。", ""])
    for item in items:
        lines.extend([
            f"## {item['id']} · {'冲突' if item['status'] == 'conflict' else '待确认'}",
            "",
            f"- 候选习惯：{item['assertion']}",
            f"- 范围：{scope_label(item['scope'])}",
            f"- 原话：“{item['evidence'][-1]['quote']}”",
            f"- 依据：{item['reason']}",
            f"- 置信度 / 重要度：{item['confidence']} / {item['importance']}",
        ])
        if item["conflicts_with"]:
            lines.append(f"- 冲突对象：{', '.join(item['conflicts_with'])}")
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def render_journal(state: dict[str, Any]) -> str:
    lines = [
        "# 作者记忆变更记录",
        "",
        "<!-- 由 author_memory_commit.py 生成，请勿手改；最近记录在前。 -->",
        "",
        f"> {store_caption(state)}",
        "",
    ]
    if not state["journal"]:
        lines.extend(["暂无变更。", ""])
    for entry in reversed(state["journal"][-100:]):
        lines.extend([f"## r{entry['revision']} · {entry['committed_at']}", "", f"- 事务：`{entry['transaction_id']}`"])
        lines.extend(f"- {summary}" for summary in entry["summaries"])
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"


def render_views(state: dict[str, Any]) -> dict[str, str]:
    views = {
        "作者画像.md": render_profile(state),
        "待确认.md": render_pending(state),
        "变更记录.md": render_journal(state),
    }
    limits = {"作者画像.md": PROFILE_MAX_BYTES, "待确认.md": PENDING_MAX_BYTES, "变更记录.md": JOURNAL_MAX_BYTES}
    for name, payload in views.items():
        require(len(payload.encode("utf-8")) <= limits[name], f"{name} exceeds {limits[name]} bytes; consolidate old memory first")
    return views


def write_snapshot(store: Store, state: dict[str, Any]) -> None:
    views = render_views(state)
    state_payload = json_payload(state)
    require(len(state_payload.encode("utf-8")) <= STATE_MAX_BYTES, f"_author-memory-state.json exceeds {STATE_MAX_BYTES} bytes")
    for name, payload in views.items():
        write_if_changed(store.root / name, payload)
    # State is the authority and therefore the last commit point.
    write_if_changed(store.state_path, state_payload)


# ---------------------------------------------------------------------------
# Query & budget
# ---------------------------------------------------------------------------


def same_scope_value(item_value: str | None, requested: str | None) -> bool:
    return requested is not None and item_value is not None and item_value.casefold() == requested.casefold()


def summarize_assertion(assertion: str, *, limit: int = 14) -> str:
    return assertion if len(assertion) <= limit else assertion[:limit] + "…"


def compact_item(item: dict[str, Any]) -> dict[str, Any]:
    """query 载荷只带这四个字段——估算与真实输出必须同一把尺。"""
    return {"id": item["id"], "kind": item["kind"], "scope": item["scope"], "assertion": item["assertion"]}


def compact_bytes(item: dict[str, Any]) -> int:
    return len(json.dumps(compact_item(item), ensure_ascii=False).encode("utf-8"))


SCOPE_RANK = {"book": 0, "genre": 1, "workflow": 2, "global": 3}


def query_sort_key(item: dict[str, Any]) -> tuple[int, int, int, int, int]:
    """重要度→本书例外→最近更新→确认次数→编号。

    重要度必须排在 scope 之前：超编时先丢的应当是不重要的条目，而不是「凡
    全局一律先丢」。scope 在前会让任意数量的 low 本书琐事挤掉 high 的全局
    铁律——那恰恰是作者最不愿意丢的那一类。同重要度之内才按本书例外优先。
    两个 store 的修订号互不可比，但 book 条目只来自书级 store、其余只来自项目
    级，同 scope 必同 store，所以按修订号比「最近更新」在合并后仍成立。
    """
    return (
        -RANK[item["importance"]],
        SCOPE_RANK[item["scope"]["level"]],
        -item["updated_revision"],
        -item["confirmation_count"],
        id_number(item["id"]),
    )


def fit_items(
    sorted_items: list[dict[str, Any]],
    revision: int,
    *,
    extra: dict[str, Any] | None = None,
) -> tuple[dict[str, Any], list[str]]:
    """按 query 输出信封把条目装进 QUERY_MAX_BYTES：装不下的跳过而不中断
    （一条长的不挡后面的短条），漏下的 ID 报进 omitted_ids（封顶
    OMITTED_IDS_MAX 条，omitted 保留真实总数）。返回 (结果文档, 全部漏下 ID)。

    漏项恒按候选优先级排序，不按被丢弃的先后：收尾回吐的条目优先级高于循环
    里跳过的，若按追加顺序排，omitted_ids 的封顶正好会把最该报的那条切掉。
    `extra` 是同样计入信封的附加字段（book_revision 等）。
    """
    result: dict[str, Any] = {
        "ok": True,
        "command": "query",
        "initialized": True,
        "revision": revision,
        "items": [],
        "omitted": 0,
        "omitted_ids": [],
    }
    result.update(extra or {})
    order = {item["id"]: index for index, item in enumerate(sorted_items)}
    kept: list[dict[str, Any]] = []
    dropped: set[str] = set()

    def ordered_omitted() -> list[str]:
        return sorted(dropped, key=order.__getitem__)

    def envelope_bytes() -> int:
        omitted = ordered_omitted()
        result["items"] = [compact_item(item) for item in kept]
        result["omitted"] = len(omitted)
        result["omitted_ids"] = omitted[:OMITTED_IDS_MAX]
        return len((json.dumps(result, ensure_ascii=False, sort_keys=True) + "\n").encode("utf-8"))

    for item in sorted_items:
        kept.append(item)
        if envelope_bytes() > QUERY_MAX_BYTES:
            kept.pop()
            dropped.add(item["id"])
    # omitted 计数落定后包可能恰好贴边超出一两个字节，回吐条目直到装下。
    while kept and envelope_bytes() > QUERY_MAX_BYTES:
        dropped.add(kept.pop()["id"])
    envelope_bytes()
    return result, ordered_omitted()


def build_query_result(
    candidates: list[dict[str, Any]],
    project_state: dict[str, Any] | None,
    book_state: dict[str, Any] | None,
) -> tuple[dict[str, Any], list[str]]:
    """真实查询与写入端估算共用的唯一信封构造：候选已按优先级排好，这里补上
    book_revision 等附加字段再装填。两处信封差一个字段就是一次假阴性——回执
    说没事、query 照样丢条。"""
    extra: dict[str, Any] = {}
    if book_state is not None:
        extra["book_revision"] = book_state["state_revision"]
    revision = project_state["state_revision"] if project_state is not None else 0
    return fit_items(candidates, revision, extra=extra)


def merged_query(
    project_state: dict[str, Any] | None,
    book_state: dict[str, Any] | None,
    kinds: set[str],
    requested_scopes: dict[str, str | None],
) -> tuple[dict[str, Any], list[str]]:
    """真实查询的纯函数部分：项目级按 kind／scope 过滤（存量 book 条目不参与，
    migrate 后才回来），书级整个 store 就是这本书的，只按 status／kind 过滤；
    合并后按 重要度→本书例外→最近更新 装填。"""

    def relevant(item: dict[str, Any]) -> bool:
        if item["status"] != "active" or item["kind"] not in kinds:
            return False
        level = item["scope"]["level"]
        if level == "global":
            return True
        return level != "book" and same_scope_value(item["scope"]["value"], requested_scopes[level])

    candidates: list[dict[str, Any]] = []
    if project_state is not None:
        candidates.extend(item for item in project_state["items"].values() if relevant(item))
    if book_state is not None:
        candidates.extend(active_of_kinds(book_state, tuple(kinds)))
    candidates.sort(key=query_sort_key)
    return build_query_result(candidates, project_state, book_state)


def slice_weight(items: list[dict[str, Any]]) -> int:
    """切片在真实载荷里的占位：compact 字节＋每条在 JSON 数组里的分隔符。

    只比 compact 字节会挑错切片——条目多、单条短的切片字节和更小，实际占位
    却更大，于是估算判「装得下」而真实查询溢出（warnings 假阴性）。
    """
    return sum(compact_bytes(item) + 2 for item in items)


def active_of_kinds(state: dict[str, Any] | None, kinds: tuple[str, ...]) -> list[dict[str, Any]]:
    if state is None:
        return []
    return [item for item in state["items"].values() if item["status"] == "active" and item["kind"] in kinds]


def worst_case_items(
    project_state: dict[str, Any] | None,
    book_state: dict[str, Any] | None,
    kinds: tuple[str, ...],
) -> list[dict[str, Any]]:
    """某个任务组合的最坏查询候选：全局条目＋各 scope 维度上最重的单一切片。

    一次查询只带一个 book/genre/workflow，不同书的条目不会同现，所以按切片
    取最重而不是全加起来，多书工作区才不会被粗算误伤。切片按 casefold 归并，
    与 same_scope_value 同一口径——否则只差大小写的同名书在这里算两个切�