interface:
  display_name: "短剧创作"
  short_description: "基于文件系统初始化、恢复、路由、可视化管理并交付短剧项目"
  default_prompt: "使用 $short-drama 初始化、查看、继续、恢复或打开 Dashboard 管理我的文件系统短剧项目。"

# 观众此刻可以知道什么

背影、门缝、画外声、浅景深和遮挡会决定观众何时知道一个人是谁、看见证据或发现门后有人。
需要同时防两种错：该给的信息被漂亮构图挡住；不该给的信息被镜头提前泄露。

## 一条事实一条判断

对每个有揭示时机的事实，在对应 `SHOT-...` 下写清：

- **事实来源**：来自哪个场景 ID、对白、动作或画面文字；
- **本镜权限**：现在展示，还是暂时扣住；
- **可见/可听载体**：观众具体能看到或听到什么；
- **释放时机**：若暂时扣住，在哪个可见事件后释放；
- **保护方法**：靠遮挡、景深、画外声、光线或构图怎样实现；
- **戏剧理由**：为什么这个时机改变理解或压力。

同一镜头里“来者认识主角”和“来者是谁”是两件事：可以让旧称呼先从画外传来，同时让来者的脸
留在暗处。揉成一句“稍后揭示”，下游就无法分别处理声音和身份。

扣住信息本身没有价值；说不出何时释放，就说明还没完成设计。审查时同时核对原文此刻允许透露什么，
以及本镜必须让观众明白什么。

# 契约与所有权

## 工作流不是瀑布

```text
开发? -> 剧本 -> 视觉设定 -> 图片提示词
                    \-> 分镜 -> 视频提示词
                    \-> 明确确认后的媒体生产
```

可以从任一已有材料直接进入；Look Development 和审查都按需发生。每个 owner 只修改自己的 Markdown，
引用上游可见事实，但不静默改写它们。

## 输出语言

| 内容 | 有项目配置时 | 独立任务默认 |
|---|---|---|
| 创作者可读说明与审查 | `short-drama.json#/language` | 用户当前语言 |
| 图片、关键帧的可复制正文 | `short-drama.json#/format/prompt_language` | `en` |
| 视频的可复制正文 | 目标视频模型档案的 `production_profile.choices.video_prompt_language`；未声明时回退 `format.prompt_language` | 根据目标视频模型确认；未声明时使用用户当前语言 |

创作者说明、通用提示词语言和视频模型提示词语言互相独立；画面中的对白和可读文字来自剧本/视觉事实，
不由描述语言推断。

## 稳定身份与引用

创作文档条目使用创作者可见的稳定 ID：`EP001-SC001`、`IMG-...`、`SHOT-...`、`MOTION-...`。其中 `IMG-...` 专属《图片提示词.md》的提示词标题，不代表已有图片。显示名可以改变，身份不因此改变。跨文档条目引用直接写这些 ID；不写哈希、内部路径、隐藏状态或第二套 record ID。

真实输入参考图不是图片提示词条目，所以每张图使用独立稳定的 `REF-...` 槽位，槽位内写显式顺序、可见项目相对路径、中文名称、用途和控制范围。它不使用 `IMG-...`；路径只用 `/` 分隔，不写绝对路径、Windows 盘符/网络路径或项目外路径。

图片由创作者在项目外准备、生成时自行挂载时没有项目内路径可写，改用 `PLAN-...` 槽位：其余字段与 `REF-...` 相同，定位符写这张图对应的 `IMG-...` 板子或 `SHOT-...` 冻结关键帧。这里出现 `IMG-...` 不是把提示词条目当成图片，而是回答「这张图画的是哪一条已决定的事实」；`PLAN-...` 不声称项目里有文件，也因此不能作为生产输入。

## 规则分级

- `structural_invariant`：本地可证明的引用、ID、算术或显式矛盾；
- `reviewed_invariant`：需要结合语义证据判断的义务；
- `craft_default`：通常有效，但创作者可以说明理由覆盖；
- `taste_option`：表达选择，不能单独阻断。

字数、镜头数、情绪曲线和案例配方不是通用结构门槛。

## 信任与文件安全

普通创作不联网。实际媒体生产只允许项目外 adapter 读取环境凭据，并始终经过
`preview -> explicit confirm -> run`。不要把私有输入、凭据、绝对路径或非公开观察写入可交付文档。

# 五份创作文档

这是一份建议格式，不是 schema。优先让人一眼读懂；只有跨文档引用和投产定位需要稳定 ID。

## `剧本.md`

- 一级标题必须以集号开头，例如 `# EP001 集名`。
- 每场用 `## EP001-SC001 内 · 地点 · 时间/天气`（空间性质也可写 `外` 或 `内外`）。
- 对白使用 `角色（可选表演提示）：台词`；画外音、画外对白、声音、可见文字、连续性和转场分别使用 `[VO]`、`[OS]`、`[SFX]`、`[画面文字]`、`[连续性]`、`[转场]`，不要把含冒号的动作叙述误写成对白。
- 不为每段台词生成 block ID，不另写节拍表、索引或录音表；用户明确需要时再做。

## `视觉设定.md`

按人物、造型、地点和道具分节，每一项自己是一个二级标题。每项只写后续生成真正需要保持的识别锚点、
当前状态与变化原因，不要抄剧情摘要，也不要把瞬时姿势或镜头角度升级成新资产。

```markdown
## 人物 · 江辰
- 识别锚点：窄长眼、左眉尾旧疤、肩背略绷。
- 本集造型：深灰旧西装，右袖口有雨水深痕。
- 状态变化：EP001-SC003 后右手掌沾血。
- 画面代称：Jiangchen
- 连续性锁：LOCK-JIANGCHEN-SUIT《江辰深灰旧西装》（镜头：全集；图片提示词项：IMG-JIANGCHEN-SHEET）· 锁面：dark-grey worn wool suit
```

二级标题固定写成 `## <人物|造型|地点|道具> · <名称>`；分镜的「视觉依据」按这个标题定位条目。

**画面代称**是提示词正文里直接叫到这一条的名字，按项目提示词语言写；多个写法用 `、` 分隔，
大小写必须与正文一致。条目自身的中文名称始终算一个代称，所以中文项目通常不必写这一行。
`creator_markdown_check.py` 用它机械核对「冻结关键帧里出现的名字，视觉依据必须覆盖」。

提示词正文语言不是中文时，被某镜「视觉依据」引用过的**人物**条目必须给出结论：写
`画面代称：<正文里的拼写>`，或在正文从不点名它时写 `画面代称：无`。少了这一行，中文条目名在英文正文里
永远匹配不上，这条核对就静默失效——那正是 #94 报告的形状。地点和道具通常在正文里只被描述、不被点名，
不作强制。

识别锚点写给创作者读；**连续性锁**写给执行端逐字带走。跨镜出现、观众看得出不一致、又不随剧情
改变的可见事实（服装主色与形制、道具颜色材质、发型长度、伤口位置）才上锁，会变的部分留在
`状态变化`。一行一把锁：

- `LOCK-...` 在本集唯一，与 `IMG-...`、`REF-...`、`SHOT-...` 互不混用；
- `镜头：` 写 `全集` 或 `、` 分隔的 `SHOT-...` 列表，两者不能混写；
- `；图片提示词项：` 可选，写 `、` 分隔的 `IMG-...` 列表；
- `· 锁面：` 之后是被逐字带走的最小名词短语，用项目提示词语言写，不含标点、动作和剧情。

生效范围内的冻结关键帧、`MOTION-...` 正文和被点名的 `IMG-...` 正文都必须含有这段锁面（忽略
大小写）；`creator_markdown_check.py` 会机械核对。没有锁的项目不受影响。

## `分镜.md`

每镜一个二级标题。镜头 ID 在本集内唯一并保持稳定；修订镜头内容时不要无故换 ID。

```markdown
## SHOT-EP001-001 · 门外停步
- 来源：EP001-SC001
- 时长：4s
- 目的：先让观众看见他不敢进门，再揭示门内争吵。
- 景别/机位：中近景，门框外侧，轻微低机位。
- 起点：江辰右手悬在门把上方。
- 唯一动作：门内玻璃落地声触发江辰收回右手，视线从门把转向门缝。
- 终点：江辰右手垂回身侧，视线落在门缝。
- 图片提示词项：IMG-JIANGCHEN-SHEET《江辰角色板》（控制：身份、体态、本集造型）；IMG-CORRIDOR-NIGHT《夜间走廊场景板》（控制：空间地理、门位、光向）。
- 输入参考图：无（待补参考图：江辰本集造型、夜间走廊地理、本镜起始帧）。
- 视觉依据：《视觉设定.md》·人物「江辰」（控制：身份、体态、本集造型）；地点「旧走廊」（控制：空间地理、出入口、光向）。

### 冻结关键帧提示词
> 按项目提示词语言编写的冻结瞬间正文……
```

只写有镜头职责的镜头。关键帧提示词必须与该镜“起点”一致；终点才出现的文字、动作或状态不得
提前进入静帧，也不要把整段动作塞进静帧。

三类依据不得混用：

- **图片提示词项**：《图片提示词.md》中可见二级标题的 `IMG-...` 条目，只证明提示词存在；
- **输入参考图**：创作者已提供，或经确认生产且当前可读的真实图片；每张图用稳定 `REF-...` 槽位绑定显式顺序、可见项目相对路径、中文名称、用途、控制与不得控制范围；
- **视觉依据**：《视觉设定.md》中的文字事实，不冒充图片。

**「视觉依据」是每镜必写字段，也是这一格画面的人物、地点与道具清单。** 冻结关键帧写完之后，
按成稿画面逐项回填，语法固定：

```markdown
- 视觉依据：《视觉设定.md》·人物「江辰」（控制：身份、造型）；地点「旧走廊」（控制：空间地理、出入口、光向）。
```

- `《视觉设定.md》·` 只写一次，条目之间用 `；` 连接，条目名称必须能在《视觉设定.md》里找到同名标题；
- 画面里需要认出身份、造型或地理的人物、地点和道具都要在列，小比例或背景人物只要需要认出身份仍在范围内；
- 画外的人声、不可识别的局部和不需要保持一致的临时物件不在列，也不要为凑齐而写进来；
- 「控制」写这一条在本镜真正决定什么，只写这一格画面承担得起的范围——手部特写不写「本集造型」；
- 本镜确实没有需要保持一致的可见条目时写 `无`，这不是省略的借口：关键帧里点名的条目仍会被机械核对。

关键帧正文提到某个条目的名字、但它本镜确实不在画面里时（物主已离场、名字只出现在屏幕文字上），
在同一字段末尾追加画外清单，如实说明它被提到而不在场：

```markdown
- 视觉依据：《视觉设定.md》·地点「教室」（控制：课桌排列、窗位）；道具「旧书包」（控制：磨损、颜色）；画外：人物「小明」。
```

某个条目的名字在正文里根本不可靠时——`道具「手机」` 会撞上「手机店」，`人物「小雨」` 会撞上「下着小雨」——
在《视觉设定.md》该条目写 `画面代称：无`，机械核对不再按名字找它；这一镜是否画到它，仍由「视觉依据」如实回答。

每个 `REF-...` 槽位的语法是：

```markdown
REF-<slot>（顺序：<n>）· <项目相对路径>《<中文名称>》（用途：<用途>；控制：<范围>；不得控制：<范围>）
```

`用途` 回答「这一张在本镜决定什么」，取值只能是
`身份`、`造型状态`、`地理`、`构图`、`尺度`、`效果`、`起始帧`、`结束帧`、`风格` 之一，逐条含义见
[参考角色](reference-roles.md)。一张图承担两件事时拆成两个槽位，不写「全参考」。一镜最多一张
`起始帧`、一张 `结束帧`，且写 `结束帧` 时必须同时有 `起始帧`。有了 `用途`，视频提示词就能直接回答
「本镜要送哪一张分镜起始帧、哪些人物/道具/场景图」，生产端也不必再猜每张图的职责。

创作者在别处出图时，图片不会进项目，`REF-...` 也就无从写起。这时用 `PLAN-...` 槽位，
语法与 `REF-...` 完全一致，只把定位符从项目相对路径换成本项目已有的条目 ID：

```markdown
PLAN-<slot>（顺序：<n>）· <IMG-... 或 SHOT-...>《<中文名称>》（用途：<用途>；控制：<范围>；不得控制：<范围>）
```

`IMG-...` 指《图片提示词.md》的板子，`SHOT-...` 指那一镜的冻结关键帧；两者都必须在当前项目里
解析得到，`IMG-...` 的中文名称还要和它的标题一致。`PLAN-...` 说的是「这张图不在项目里，
生成时由创作者按这个顺序挂载」，与「已有图片」和「待补」都不是同一句话：它不声称文件存在，
也不表示还在等谁。`顺序` 就是创作者在生成界面里挂图的次序，也是可复制正文里 `<Picture N>` 的编号。
一张已经在项目里的图仍然写 `REF-...`；两种槽位可以在同一条声明里混用，`顺序` 在两者之间统一连续编号。

「输入参考图」有四种准备结果，不能把它们混在一起：

- 已绑定 `REF-...`：图片真实存在于项目中，可进入图生/参考生视频，也可交给生产技能执行；
- 已声明 `PLAN-...`：图片由创作者在项目外准备并在生成时自行挂载，视频提示词照常产出；
  本套件没有文件可送，所以这些镜头不能进入生产技能，需要投产时先把真实文件放进项目并改写成 `REF-...`；
- 待补：一张都没有时写 `无（待补参考图：……）`；只缺一部分时保留已有 `REF-...` 或 `PLAN-...`，末尾追加 `；待补参考图：……`。视频提示词尚未就绪；
- `无（创作者已明确选择文生视频）`：创作者在了解一致性影响后明确不用图，可进入文生视频。

`IMG-...` 只用于查找，不靠英文 slug 向创作者解释含义。每个引用都同时写简短中文名称和
控制范围，且 ID 必须在当前《图片提示词.md》标题中存在。中文名称帮助阅读，不取代 ID 定义条目身份。没有该文档或没有合适条目时，分镜仍可
直接从剧本和视觉设定进入，使用下列回退格式，不编造 ID 也不补造提示词阶段：

```markdown
- 图片提示词项：无
- 输入参考图：无（待补参考图：江辰身份、旧走廊地理、本镜起始帧）
- 视觉依据：《视觉设定.md》·人物「江辰」（控制：身份、造型）；地点「旧走廊」（控制：空间地理、出入口、光向）。
```

提示词条目和真实图片是独立的：有 `IMG-...` 条目时，「输入参考图」可以仍为「无」；创作者给了
真实图片时，也不需要为它编造提示词条目：

```markdown
- 图片提示词项：无
- 输入参考图：REF-JIANGCHEN-LOOK（顺序：1）· 输入/参考图/江辰定妆.jpg《江辰定妆照》（用途：造型状态；控制：本集造型、面料质感；不得控制：镜头构图、动作、表情）
- 视觉依据：《视觉设定.md》·人物「江辰」（控制：身份、体态）。
```

只准备了一部分时，已核对的图不丢弃，但必须把缺口留在同一字段中：

```markdown
- 输入参考图：REF-JIANGCHEN-LOOK（顺序：1）· 输入/参考图/江辰定妆.jpg《江辰定妆照》（用途：身份；控制：脸型、体态、本集造型；不得控制：场景地理、构图、动作）；待补参考图：夜间走廊地理、本镜起始帧
```

创作者在自己的工具里出图时，本镜要挂哪几张、按什么次序挂，写成 `PLAN-...`：

```markdown
- 图片提示词项：IMG-JIANGCHEN-SHEET《江辰角色板》（控制：身份、体态）
- 输入参考图：PLAN-SHOT-START（顺序：1）· SHOT-EP001-001《本镜冻结关键帧》（用途：起始帧；控制：起始构图、站位；不得控制：尚未发生的动作、终态）；PLAN-JIANGCHEN（顺序：2）· IMG-JIANGCHEN-SHEET《江辰角色板》（用途：身份；控制：脸型、体态、本集造型；不得控制：构图、动作、表情）
- 视觉依据：《视觉设定.md》·人物「江辰」（控制：身份、体态）；地点「旧走廊」（控制：空间地理、出入口）。
```

两者也可同时存在，但各用各的定位符：

```markdown
- 图片提示词项：IMG-JIANGCHEN-SHEET《江辰角色板》（控制：身份、体态）
- 输入参考图：REF-CORRIDOR-COMPOSITION（顺序：1）· 输入/参考图/走廊构图.jpg《走廊构图参考》（用途：构图；控制：占画比例、留白；不得控制：人物身份、动作、剧情事件）
- 视觉依据：《视觉设定.md》·人物「江辰」（控制：身份、体态）；地点「旧走廊」（控制：空间地理、出入口）。
```

## `图片提示词.md`

每项一个可投产的角色板、场景板、道具板或状态图。标题使用 `IMG-...` ID；正文放在
`### 可复制提示词` 的引用块里，前面只保留用途和参考约束。
这里的 `IMG-...` 是提示词条目 ID，不是已生成图片的记录。

```markdown
## IMG-JIANGCHEN-SHEET · 江辰角色板
- 用途：锁定身份、体态和本集造型。
- 参考：无；若有真实参考图，使用与分镜相同的完整 `REF-...` 格式，说明顺序、项目相对路径、中文名称、用途、控制与不得控制范围。

### 可复制提示词
> 按项目提示词语言编写的正文……
```

## `视频提示词.md`

与分镜一一对应时沿用镜头号并使用 `MOTION-...` ID。若「输入参考图」中的 `REF-...` 与 `PLAN-...` 已全部就绪，使用图生视频，正文
只需重述运动所需的局部起点和防漂移事实；只有「输入参考图：无（创作者已明确选择文生视频）」才使用文生视频，正文必须自足地
包含人物、造型、地点、构图、光线等本镜需要的静态视觉锚点，不能用 `preserve/保持已有外观`
代替模型并未收到的图片。「待补参考图」会阻止最终视频提示词；视频文档中的「输入参考图」必须与对应分镜逐字一致。

可复制正文里被引号或 `<d>` 包起来的中文，必须能在《剧本.md》或《视觉设定.md》里找到同样的字。
提示词负责把已确认的台词送到执行端，不负责替角色补一句更好的台词。

除非剧本或创作者明确要求，视频正文用项目提示词语言直接声明全镜任何时刻都不生成非画内字幕、
caption 和对白文字叠加，不能只写抽象的“文字层为空”，让对白只通过声音与口型发生。旗帜、标牌、
屏幕等画内文字另走已接受的文字政策：`exact_readable` 逐字携带准确
字符、语言、承载面和位置；关闭字幕不能写成会同时抹掉这些画内文字的全局 `no-text`。

```markdown
## MOTION-EP001-001 · 门外停步
- 分镜：SHOT-EP001-001
- 时长：4s
- 生成方式：文生视频
- 输入参考图：无（创作者已明确选择文生视频）。
- 静态视觉锚点：A lean young East Asian man in a dark-grey worn suit stands outside an old corridor door under cold overhead light.
- 起始帧：SHOT-EP001-001 的冻结关键帧
- 状态链：右手悬在门把上方 → 门内玻璃落地触发他收回右手并转移视线 → 右手收回，视线落到门缝
- 终点：右手收回；门内玻璃落地；人物视线转向门缝。

### 可复制提示词
> A lean young East Asian man in a dark-grey worn suit stands outside an old corridor door under cold overhead light. His right hand starts suspended above the handle, then withdraws……
```

有真实输入图时改用「生成方式：图生视频」，并原样携带分镜的完整 REF 声明；静态视觉锚点只写
执行动作所需、且不能仅靠参考图用途推断的可见起点，不把路径或 REF ID 放进可复制正文。

## 共通规则

- 提示词正文必须可直接复制，不放占位符、流程备注或 QA 结论。
- `### 可复制提示词` 下的引用块是**一条**提示词：从第一行到最后一行整段复制进那一次生成。方言把它分成 `subject_definitions` / `overall_soundscape` 这类段落时，段落仍属于同一条正文，不是分几次提交，也不是只交第一段。
- 引用使用文档里看得见的场景/镜头/提示词 ID；不写文件哈希和内部路径。
- 用户没要求的字段不为“完整”而补造。
- 完成或修订五份文档后运行 `python3 {技能目录}/scripts/creator_markdown_check.py 剧集/<EP> --project-root .`，校验可执行的跨文档结构；内容质量仍由创作者审查。

# Creator-first 工作流

## 唯一落盘

每集按需维护五份创作者可读 Markdown：

| 阶段 | owner | 文件 |
|---|---|---|
| 剧本 | `short-drama-write` | `剧集/<EP>/剧本.md` |
| 视觉资产 | `short-drama-assets` | `剧集/<EP>/视觉设定.md` |
| 分镜与冻结关键帧 | `short-drama-storyboard` | `剧集/<EP>/分镜.md` |
| 图片提示词 | `short-drama-image-prompts` | `剧集/<EP>/图片提示词.md` |
| 视频提示词 | `short-drama-video-prompts` | `剧集/<EP>/视频提示词.md` |

没有内容时不预建空文件。一个请求只创建它实际需要的文档；从剧本、视觉设定、分镜或提示词任一
阶段直接进入都合法，不为补齐名义流水线伪造上游。

单集创作主链不落盘 JSON/JSONL、索引、指纹、覆盖表、QA 报告、审核表、交接胶囊或恢复状态。
需要 ID 时写进 Markdown 标题；结构检查和批次状态留在本轮上下文，不成为第六份创作文档。建议
写法见 [五份创作文档](creator-documents.md)。

长篇原著拆解或多集材料识别可以在 `项目开发/` 使用机器索引；它们是用户点名时才建立的可选分析
工作区，不是单集创作真相。其存在不会改变 `剧集/<EP>/` 的五文档布局，也不能要求单集 owner
补造并行结构化产物。

创作者说明跟随项目语言。图片和关键帧正文跟随 `prompt_language`；视频正文优先跟随目标视频模型
档案的 `video_prompt_language` 和 `video_prompt_dialect`，未声明时回退 `prompt_language` 与通用自然语言。
`native_duration_seconds` 在分镜规划时生效，不能等生产提交才发现镜头不合法。没有 `short-drama.json` 时，图片与
关键帧正文默认使用 `en`；视频正文根据目标模型确认，目标也未声明时沿用用户当前语言。描述语言、
对白口语语言和画内文字互相独立，不能用一个推断另一个。

用户在 Brief 或当前请求中已经给出的项目级事实要在初始化时同步到 `short-drama.json`：至少包括已确认的
集数、单集目标时长、画幅和目标生产档案。生产档案只有 `unset` 与 `accepted` 两种项目级状态；下游只读取
已接受档案，不用临时状态代替创作者决定。
同一场次连续生产且目标模型支持续接时，生产档案把连续性路由声明为“上一段实际视频 + 实际尾帧接力”；
`one-shot-per-generation`、单镜容器或多镜容器只描述打包粒度，不能用来取消这条连续性路由。

## 请求范围

用户点名的范围就是本轮范围。“写完这一集”完成整集；“从剧本做到视频提示词”依次完成所点名
阶段。场次、资产组和镜头组只是内部上下文批次，通过后自动继续，不逐批交还控制权，也不为每个
中间文件另开接受、审查或 QA 回合。

只在三处停下：

1. 点名范围已经完成；
2. 缺失事实形成会显著改变剧情或视觉方向的真实创作分叉；
3. 下一步会调用外部生产，必须先展示精确 job 并取得显式确认。

### 一句话把范围说清楚

请求不需要固定话术，技能也不靠关键词匹配。下面几句只是把「一句话里到底要交代哪几件事」写明白，
照抄或改写都可以：

```text
写 EP001 的分镜和每镜冻结关键帧。目标视频模型是 MiniMax H3，先把它写进项目档案。
```

```text
EP001 分镜确认采用，按 MiniMax H3 写视频提示词。人物、场景、道具要跨镜一致，
所以先在项目里找已有的角色图、场景图、道具图和本镜起始帧并绑成参考；缺哪张就列出来，别改成文生视频。
```

```text
EP001 的角色图和场景图在 输入/参考图/ 下，本镜起始帧在 剧集/EP001/制作成果/images/ 下，
请按这些图写 MiniMax H3 的图生视频提示词，逐镜说明送哪一张起始帧、哪些人物/道具/场景图。
```

```text
参考图我自己在 MiniMax 的界面里出，不进项目。EP001 按 H3 写视频提示词，
逐镜告诉我要挂哪几张图、按什么顺序挂、每张管什么。
```

```text
这一集没有也不打算做参考图，明确走文生视频，静态视觉锚点写足。
```

一句请求里值得交代的是四件事：**做哪一集的哪一阶段**、**目标视频模型**、**参考图在哪里或还没有**、
**没有图时怎么办**。前三件缺一件，技能会来问或按已接受档案推进；最后一件从不替创作者假设。
第四件有三个答案：等图、由创作者自己在别处出图并挂载（写成 `PLAN-...`）、或者明确走文生视频；
一个都没说时，就当成还在等图。

结构检查服务于发现错误，不是创作门禁。能在当前文档直接修正的问题直接修正；审查、归档和媒体
生产只有用户点名时开始。安装 selftest、全量 QA 和 demo 验证属于技能维护，不属于普通创作请求。

## 创作质量

简化文件数量不等于降低内容标准。各 owner 按具体问题读取自己的 craft references，检查剧情因果、
人物声音、资产身份、空间连续性、镜头职责、可生成性和提示词边界。规则与路由总表见
[knowhow-index.md](knowhow-index.md)。最终回报只说明已完成范围、关键创作决定、真实未决问题和可选
下一步，不汇报内部流水线噪音。

## 生产边界

图片、视频、TTS 和音乐生产仍遵守 `preview -> explicit confirm -> run`。生产工具可以在隐藏运行目录
保存 job、确认和审计记录；这些是付费/外部副作用的必要边界，不要求模型把创作内容再抄成长期
JSON/JSONL。job 必须把拥有当前提示词的 Markdown 写入 `source`，输出写到
`剧集/<EP>/制作成果/`；任何提示词、参数、输入或输出路径变化都需要重新预览和确认。

# 规则与路由索引

本文件只做两件事：把每类决策路由到**负责的技能**，以及定义规则分级与冲突优先级。
它不复制任何规则正文——每个技能的预检、所有权边界、形态输入与完整规则表，都在该技能
自己的 `references/stage-contract.md` 里。核心不代子技能决定读哪一份资料。

## 目录

- [主题权威路由](#主题权威路由)
- [规则分级](#规则分级)
- [冲突优先级](#冲突优先级)

## 主题权威路由

路由到技能后由它自行加载资料。表中「不适用边界」是该技能**不得**越过的地方。

| 主题 | 负责技能 | 何时读取 | 不适用边界 |
|---|---|---|---|
| 长篇原著切片与逐章功能提取 | `$short-drama-novel-analyze` | 导入小说、连载或多集散稿，需要可追溯的分析层 | 不写改编契约与分集地图，不建资产 |
| 抽样快评一本书值不值得拆 | `$short-drama-novel-analyze` | 全量拆解前先按抽样判断改编密度与风险 | 结论限于抽样范围，不产候选集条目 |
| 改编价值与分集候选 | `$short-drama-novel-analyze` | 判断哪些单元在屏幕上成立、按戏剧结果切候选集 | 候选不是决定；集数由创作者与 develop 定 |
| 故事承诺与冲突引擎 | `$short-drama-develop` | 从点子建立可持续压力与回报 | 不用题材词直接生成剧情 |
| 单集契约与集尾交接 | `$short-drama-develop` | 规划进入状态、局部结果与出去压力 | 不代替剧本场景执行 |
| 单元里被反复运行的机制与耗尽 | `$short-drama-develop` | 单元故事靠同一套机制反复兑现，或中段开始重复、越写越薄 | 不规定运行次数与换代位置；不替单集决定怎么写 |
| 单集容量可行性 | `$short-drama-develop` | 进入场景化前估算各集镜头与时长量级 | 用项目自身比例，不设跨项目门槛，不阻断交付 |
| 连载人物与恢复记忆 | `$short-drama-develop` | 跨集人物策略、信息权限与铺垫债务较重 | 单场任务不必建立全量记忆 |
| 长材料改编 | `$short-drama-develop` | 压缩、合并或把说明转成屏幕载体 | 不复制原句或文风到新剧本 |
| 题材条件与钩子 | `$short-drama-develop` | 需要按压力机制比较题材写法 | 不把案例节拍、数字或标签当公式 |
| 单题材执行细节 | `$short-drama-develop` | 已确定题材、要落到场面颗粒与打法 | 一次只读一张卡；卡片不产生可阻断约束 |
| 前提装置（先知或外部授权） | `$short-drama-develop` | 主角带着已知结果开局，或获得一套外部规则给的能力 | 装置叠加在题材之上，不替代题材卡，也不新建题材 |
| 单形态执行细节 | `$short-drama` | 已确定制作形态、要按表定位一张形态卡 | 一次只读一张卡；形态不改写身份与地理 |
| Look Development 与风格帧 | `$short-drama` | 把视觉方向变成人物、地点和高压力代表帧的可观察规则 | `$short-drama-image-prompts` 只投影提示词；风格不接管身份、地理或剧情 |
| 场景与可表演行动 | `$short-drama-write` | 把单集节拍落实成场景 | 不决定分镜景别和运镜 |
| 对白行动与人物声音 | `$short-drama-write` | 写、改或诊断对白 | 不按句长、口癖数量判权力 |
| 多人场次的可辨识度与主角缺席场 | `$short-drama-write` | 一场里人多到分不清谁在与谁交手，或某场/某集不给主角 | 不决定站位、景别与镜头切换；那属于分镜 |
| 场次声音戏剧 | `$short-drama-write` | 声源、留白或 sound bridge 承担故事信息与压力 | 不替逐镜实现或混音，不让配乐替代表演 |
| 剧本生产标签 | `$short-drama-write` | 规范化对白、动作、VO/OS、文字与连续性 | 不借格式补造剧情 |
| 有限行动单位与精确时限 | `$short-drama-write` | 题面明确给出可核对的次数、轮次、节拍或字面死线 | 按完成状态与题面约定检查，不把定性压力升级为数字或主动增加计数结构 |
| 资产出现证据 | `$short-drama-assets` | 从剧本拆角色、地点、道具与状态 | 不用名词/人名匹配直接建资产 |
| 身份、变体与复用 | `$short-drama-assets` | 决定复用、新变体、新身份或未决 | 相机角度通常不是资产变体 |
| 原著自带的真实品牌与现实实体 | `$short-drama-assets` | 剧本沿用了原著专名，且该专名指向现实存在的品牌、商标、机构或人物 | 不判断合规、不给禁令、不内置清单；只把选择交给创作者并留痕 |
| 资产图片提示词 | `$short-drama-image-prompts` | 把已接受资产投影为单帧可见规格 | 不承载有先后顺序的剧情动作 |
| 场景板光线与跨 View 一致 | `$short-drama-image-prompts` | 同一地点要拆多个观察方向、且这些板会被互相剪接 | 朝向造成的逆光/落影差异不需拉平 |
| 图片提示词定点修改 | `$short-drama-image-prompts` | 修改局部并冻结 preserve set | 不把重写整图伪装成局部编辑 |
| 镜头目的与覆盖 | `$short-drama-storyboard` | 从剧本建立有动机的镜头 | 不为景别多样性硬拆镜 |
| 场次视觉计划与 Coverage Audition | `$short-drama-storyboard` | 关键场次需先比较观众立场、信息时机、空间、摄影与声音运动 | 可选中间层，不拥有剧本事实或 shot 边界，不规定宫格/镜头数 |
| 冻结关键帧 | `$short-drama-storyboard` | 写镜头起点的可冻结瞬间 | 不写动作链或镜头终态 |
| 漫剧关键帧视觉词表 | `$short-drama-storyboard` | 视觉方向已接受为漫剧/二维漫画形态，要为关键帧稳定投影共享画风与当前可读性约束 | 全部是 `taste_option` 选项，不设质量门槛，不替代身份、地理与边界事实 |
| 多人场面调度与交付面遮挡 | `$short-drama-storyboard` | 多人、群体、证据揭示、动态物体场面，或竖屏播放面会叠加遮挡 | 不把一个布局复用于所有题材；遮挡区来自已声明交付方案，不设通用安全框 |
| 视频运动规格 | `$short-drama-video-prompts` | 把已接受镜头边界写成时间变化 | 不改镜头、资产或剧本权威 |
| 交付容器与分段算术 | `$short-drama-video-prompts` | 项目声明了分段或多镜打包的交付方式 | 打包方式不改写镜头边界与可审查性 |
| 交付路由与执行触发词 | `$short-drama-video-prompts` | 项目声明了逐镜以外的路由，或某路由要求固定字样 | 触发词与路由都由创作者档案声明，本套件不自造字样，续接不产生第二个起点权威 |
| 表演动作与时长 | `$short-drama-video-prompts` | 动作、台词与落点可能超载 | 不用每秒动作数设通用阈值 |
| 摄影机、声音与相连边界 | `$short-drama-video-prompts` | 写运镜、环境运动、口型或声音方向 | 不添加无来源的音乐/事件 |
| 目标执行端能力档案 | `$short-drama-video-prompts` | 执行端的时长、参考方式、声音同轨或长度上限会改变正文写法 | 能力由创作者声明不由推断；不内置模型清单，不因收敛改写已接受边界 |
| 跨场与跨镜连续性 | `$short-drama-assets` | 状态变化会传给下一场/镜/集 | 不复制整本 设定集 到每镜 |
| 跨镜可见事实上锁 | `$short-drama-assets` | 服装、道具的颜色/材质在镜与镜之间漂移，参考图管不到 | 锁不变事实，不锁状态、姿势与镜头瞬态；锁面不替创作者选颜色 |
| 制作形态翻译 | `$short-drama` | 把实拍、二维、三维、水墨等方向投影到各阶段 | 风格名或模型代码不是生产决策 |
| 独立证据审查 | `$short-drama-review` | 冻结目标后写 finding 与 verdict | reviewer 不编辑 owner 来源 |
| 项目生产观察与校准 | `$short-drama-review` | 有授权文字观察且需定位到准确 prompt/spec/reference/config | 只在当前项目和版本有效；未观察结果保持未知 |
| 结果处置判断 | `$short-drama-review` | 已有诊断，要决定保留、后期、局部编辑、重新提交还是改写 | 处置先于修订；重复提交同一文本与追加形容词都不算修复 |
| 模板感诊断 | `$short-drama-review` | 至少两个位置出现同构机制或表达损失 | 不用禁词或单次类型惯例定罪 |

## 规则分级

| 分级 | 含义 | 谁能判定 |
|---|---|---|
| `structural_invariant` | 本地可证明的引用、ID、算术或显式状态矛盾 | 校验器可阻断 |
| `reviewed_invariant` | 语义义务，需要引用证据判断 | reviewer |
| `craft_default` | 通常有帮助的做法 | 创作者说明理由即可覆盖 |
| `taste_option` | 创作者的表达选择 | 不得单独阻断交付 |

## 冲突优先级

```text
创作者已接受的事实、结局承诺与改编边界
→ 已接受的单集契约、进入/出去状态与铺垫兑现义务
→ 各技能 stage-contract 中的规则表
→ 题材卡、形态卡与通用叙事手感
```

规则 ID 使用稳定前缀：`STY` 开发、`SCR` 剧本、`AST` 资产、`IMG` 图片提示词、
`SHT` 分镜与关键帧、`VID` 视频提示词、`CON` 连续性、`REV` 审查。ID 全套件唯一；
条目正文由拥有该前缀的技能维护，核心不再复制一份。`CON` 由资产、分镜与视频提示词
共同遵守，三方的 stage-contract 各自列出同一份条目。

# Look Development：把视觉方向变成可比较的代表帧

Look development 不是给全剧粘一个风格标签，也不是让参考图接管角色和剧情。它把创作者
接受的 `visual_direction` / `production_profile` 投影成少量**文本风格帧规格**，让人物、地点与
高压力场景在正式分镜前共用一套可观察视觉语言。本阶段只产出规格和提示词；创作者要求
实际出图时交给 `$short-drama-produce`，并对它展示的精确任务预览明确确认。

## 什么时候值得做

- 制作形态已经选择，但材质、光色、边缘、画面密度或景深倾向仍只停留在形容词；
- 同一项目的人物板、地点板和分镜分别成立，却看不出来自同一作品；
- 某种画风适合静态设定图，但是否保留表演、身份锚点或冲突层次仍未知。

普通项目可以直接使用已接受的视觉方向，不强制增加此阶段。需要时优先比较三类：

1. **人物表现测试帧**：身份锚点、肤质/线条、目光与细小表演是否仍可读；
2. **核心地点测试帧**：空间层级、材质响应、光源逻辑和色彩关系是否可复用；
3. **高压力场景测试帧**：该方向能否承载遮挡、冲突与注意中心，而不只适合静态展示。

## 先写可观察的方向

每个候选方向说明：

- **稳定项**：跨人物、地点和压力场景都必须保留的形状语言、材质处理、色彩层级、阴影边缘、
  景深倾向或画面密度；
- **可变量**：允许按场次变化的冷暖、对比、留白、光比和运动层；
- **叙事职责**：这些选择帮助观众识别什么、感到什么；
- **失效信号**：身份锚点被吞、群像层级打平、冲突场景只剩氛围等可观察风险。

删除风格名称后，如果材质、光、色、边缘和空间写法没有任何变化，这个方向仍只是标签。
创作者选定后，把可观察的稳定项、允许变化和失效信号写入 `视觉设定.md` 的“项目视觉方向”；
候选提示词本身不能反向改写人物身份、地点地理或剧情事实。

## 权限边界

`$short-drama-image-prompts` 只把已接受方向写成 `purpose: lookdev_frame` 的通用图片提示词。
风格参考的 `role` 是 `style`，可控制色彩层级、材质处理、阴影边缘、景深倾向和画面密度；
必须在 `must_not_control` 中保护角色身份、固定场景地理、剧情状态、道具文字和信息揭示。

人物表现帧仍绑定准确人物/造型；地点帧仍绑定准确地点/视角；压力场景仍绑定已接受的场景、
人物和剧本来源。风格帧没有权力发明这些事实，也不能反向覆盖资产或剧本。

## 比较与接受

按当前项目风险选择其中一类或多类；只有确实跨人物、地点或压力轴比较时，才并排展示共同规则、
局部差异、预期叙事效果和未知风险。不固定宫格、帧数或景别比例。若创作者随后提供授权形成的
生产观察，只在准确方向、提示词版本、参考槽位和制作配置范围内修订该项目。没有观察时只报告
文字风险，不声称画面已经成功或失败。

接受的是可观察的视觉选择，不是某个供应商、模型名、参数或任务结果。任何二进制媒体仍留在
外部制作环节，不进入文本交付包。

# 母版、补拍与替代版

| 版本 | 承担什么 | 默认关系 |
|---|---|---|
| 母版 | 当前镜头的全部原文要求 | 当前主要版本 |
| 补拍 | 补一个动作、反应、构图或效果 | 与母版并存，不替代 |
| 替代版 | 同一要求的另一种实现 | 必须明确比较后才能替代 |

缩小提示词范围经常是正确修复。危险不是“变短”，而是把只拍了一部分的版本悄悄当成母版，使未被
它承担的对白、反应、揭示或终点从此无人负责。

## 补拍必须逐项交代去向

在 `视频提示词.md` 中为补拍或替代版使用稳定 `MOTION-...` ID，并写清：

1. 它对应哪个母版；
2. 原文要求逐项由本版本、母版、另一补拍承担，还是必须退回分镜；
3. 本版本的准确起点、终点和保持项；
4. 它是补充还是请求替代。

例如母版要求“甲拍下账页，乙别开视线，丙起身”，补拍只拍手部特写，那么甲的动作由补拍承担，
乙的反应留在母版，丙的动作若两边都没有，就必须退回分镜。不能靠改名让遗漏消失。

补拍默认不替代母版；是否替代由后续审查结合全部故事义务判断，不能由运动提示词自己批准。

# 把视觉方向翻译成制作形态

视觉风格不是贴在提示词前面的标签。先把创作者已接受的 `visual_direction` 和
`production_profile` 拆成可观察、可复核的选择，再让各负责人只投影本阶段需要的部分。
同一故事事实可以有多种制作形态；这里给的是提问维度、构成差异和责任边界，不是固定风格套餐。
逐形态的执行细节放在 `references/form-cards/` 的六张卡里，本文只负责怎么选、
提示词构成随形态改变什么、以及各阶段拿走什么。

## 目录

- [形态卡的规则等级](#形态卡的规则等级)
- [Production-form card](#先写一张-production-form-card)
- [环境物理与身份形态](#先锁环境物理与身份形态)
- [可执行词汇与形态卡](#按形态选择可执行词汇)
- [提示词构成的形态差异](#提示词构成怎样随形态改变)
- [跨阶段传递](#跨阶段传递同一意图)

## 形态卡的规则等级

形态卡与题材卡同级：**默认是 `craft_default`**，创作者说明理由即可覆盖，形态卡自己
不产生 ID，也不能创造新的 `structural_invariant`。卡片里写作“硬规则/硬约束”的条目，
指的是它在投影一条**已经登记**的身份或连续性不变式（见 [knowhow-index.md](knowhow-index.md)），
按那条原规则的等级执行；比例、rig 能力、头身比一类是**本项目的制作约束**，由创作者
写入 `视觉设定.md` 的“项目视觉方向”后才对本项目生效，不是套件层规则。审查者不得单凭形态卡阻断交付。

## 先写一张 production-form card

对当前系列或一段局部变体，记录九项。形态卡里的每一项都要能落到某个阶段的可执行字段；
落不到字段的条目是审美闲聊，先删掉再补：

- **叙事职责**：这种形态要帮助观众更快看懂什么、感到什么，不能只写“高级”“电影感”。
- **身份锚点载体**：靠什么让人物跨镜可辨认——轮廓、线条、比例、结构差异还是材质。
- **连续性载体**：哪些字段必须逐镜传递，哪些可以省；不同形态的答案差别很大。
- **层级拆分**：身份层、环境层、可动层、效果层各包含什么，谁负责同一事实。
- **形与识别**：轮廓、比例、线条/表面、服装与道具的永久识别锚点；哪些可随镜头变化。
- **材质与光色**：物理或绘制材料怎样响应光，色彩关系承担什么信息，不罗列质量词。
- **运动预算**：哪些必须全动作，哪些可用 hold pose、局部循环、视差、特效层或剪辑完成。
- **声音与文字**：表演、声源、旁白和画面文字各自承载什么；可读文字仍服从资产 text policy。
- **变与不变**：跨集稳定项、单场可变项、创作者覆盖和需要先试验的未知项。

风格名称、模型代码或供应商字段不能替代这张卡；风格代码同样不能。不同上游代码可能指向同一种
呈现方向，同一个代码也可能被项目说明改写；若创作者只给了一个风格标签，负责人先提出最小的
可观察澄清，或把未决项保留为候选，不擅自补成一套审美。

判断一条形态描述有没有内容，只需要问：**它改变了哪一个字段的写法**。若把它删掉，
提示词一字不变，那它就只是标签。

## 先锁环境物理与身份形态

有异质身体、身份互换、变形、水下、失重、微缩或非人角色时，先写不能被风格词覆盖的物理卡：

- **身体拓扑**：当前有哪些肢体/形态、明确不存在什么、接触和发力点在哪里；
- **身份占用**：谁的意识/人格占用哪种身体或外形，别人当前如何识别，何时允许变化；
- **环境介质**：水流、浮力、重力、风、黏滞、尺度怎样改变静止、移动、服装/毛发和道具；
- **运动语法**：为当前身体选择悬浮、摆尾、攀附、滑行等动作，删除与物理冲突的陆地动作词；
- **连续性硬项**：身份形态、呼吸/受力状态、持物方式、语言锁与相邻边界怎样逐镜继承；
- **软呈现项**：冷暖、笔触、景深、光效等可以变化，但不得冲掉上述硬项。

这些约束先属于创作者/故事与资产事实，再由分镜和视频提示词投影；不要让提示词阶段
临时发明身体规则。若当前只是普通人物的情绪或服装变化，不额外建立异质物理卡。

## 按形态选择可执行词汇

形态卡是选出来的，不是从题材或上游标签继承的。先按下表定位一张卡，再按该卡写字段；
同一项目可以对不同段落使用不同卡，但每段必须只有一张生效。

| 形态 | 形态卡 | 这种形态最先决定的事 | 最容易丢的东西 |
|---|---|---|---|
| 实拍 | [实拍](form-cards/实拍.md) | 可搭建空间、真实光源、可表演的动作与可收录声源 | 造型状态的逐镜可比较性 |
| 二维动态漫（2D 动态漫 / 限动画） | [二维动态漫](form-cards/二维动态漫.md) | 形状语言、色块与阴影分区、每镜的可动层 | 剪影级身份锚点 |
| 风格化三维（风格化 3D） | [风格化三维](form-cards/风格化三维.md) | 比例、rig 姿势边界、接触与摄影机空间 | 接触点与重量结果 |
| 水墨笔触 | [水墨笔触](form-cards/水墨笔触.md) | 浓淡层级、留白方位、哪些边缘不可散 | 必要地理与手中物 |
| Q 版表达（含 Q 版知识 / 教育短片） | [Q 版表达](form-cards/Q版表达.md) | 比例体系、一镜一因果、符号层与文字预留 | 因果精度与尺度关系 |
| 国漫二次元 | [国漫二次元](form-cards/国漫二次元.md) | 角色间可枚举的结构差异、氛围色边界、口型策略 | 人物之间的区分度 |

混合形态（例如三维角色配绘制背景、实拍加笔触后期）不新建第七张卡，而是取两张卡并
**逐项声明谁负责身份、深度、边缘、光和运动**。两层同时决定同一事实是混合形态最贵的错误。

## 提示词构成怎样随形态改变

形态不改变提示词的段序，改变的是**哪些段必须展开、哪些段可以塌缩、以及某些段的默认值**。
用同一份配方写不同形态时，按下面三条调整。

**一、身份段的载体不同，写法就不同**。 实拍写造型状态而不是长相；二维写剪影与色块；
三维写比例与 rig 边界；水墨写不可散的边缘；Q 版写头身比与放大记号；国漫二次元必须写出
与同剧其他角色的结构差异——这是唯一一个要求“对比着写”的形态，因为精修画风会把角色拉平。

**二、连续性段带走的字段不同**。 逐镜必带项按形态取舍，不是所有项目都携带同一串状态：

| 形态 | 连续性必带 | 通常可省 |
|---|---|---|
| 实拍 | 造型层次与整洁度、湿污伤、手中物、光源时段与方向 | 不出画的内层、无从比较的陈设 |
| 二维动态漫 | 轮廓色块、阴影分区、描边规则、本镜可动层清单 | 写实材质、精确布光角度 |
| 风格化三维 | 比例剪影、材质分区、接触点、轴线与机位高度 | 次要形变、看不见的背面 |
| 水墨笔触 | 不可散的轮廓、浓淡层级、手中物、必要地理、留白方位 | 精确光向、纹理细节 |
| Q 版表达 | 头身比与相对尺度、主色块、放大记号、文字预留位 | 写实材质、背景细节 |
| 国漫二次元 | 脸眼发结构、瞳色与高光形状、固定配件、氛围色与光源色温 | 背景纵深、不承重的装饰 |

**三、运动段的默认值不同**。 实拍默认全动作，要写的是限制；三维默认可全动作，
但摄影机是最贵的一层；二维动态漫与水墨默认 hold，要写的是**哪一层例外**；
Q 版默认一镜一因果；国漫二次元把全动作预算集中给情绪转折镜，其余用循环加效果层撑住。
省掉这一段的后果是可预测的：默认 hold 的形态会被写成全员飘动，默认全动作的形态会被写成
一串没有接触点的动作词。

不要用“加一句风格前缀”的方式处理形态差异。前缀改变的是检索标签，形态改变的是
必须出现和可以省略的字段；只有后者会被下游执行，也只有后者能被审查。

## 跨阶段传递同一意图

| 阶段 | 本阶段新增 | 必须继承 | 不应越权 |
|---|---|---|---|
| develop / 编导简报 | 叙事职责、形态假设、运动预算与未决试验 | 创作者方向、题材与受众承诺 | 供应商字段、逐镜事实 |
| assets | 轮廓、材料、层拆、稳定比例和版本差异 | 身份、地理、文字政策 | 镜头构图和动作终态 |
| image-prompts | 单帧可见的形/材质/层次/光色投影 | 已接受资产与当前用途 | 有先后顺序的动作 |
| storyboard / keyframe | 注意力、层次、遮挡、可动层与冻结边界 | 剧本信息权限、资产绑定 | 用风格词改写人物状态 |
| video-prompts | 运动层、接触、节奏、声源与结果 | shot purpose、起止边界、形态预算 | 回写镜头或资产权威 |
| review | 检查叙事职责是否真的落到可见字段 | 精确 artifact refs 与创作者覆盖 | 用个人偏好替代证据 |

这张表对所有形态成立；每个阶段在本形态下额外要写的那一件事，写在对应形态卡的
“跨阶段传递”小节，取用时只读当前形态那一张，不要把六张卡都装进上下文。

检查时不问“有没有写风格名”，而问：叙事职责是否投影成了本形态可执行的形、层、材质、光、动作或声音；
连续性必带项是否按本形态取舍而不是照抄别的形态；同一决定是否只由一个 owner 写入；
换一个题材或项目时，agent 是否会重新选择而不是复制整套词汇。
Reviewer 要分别引用来源事实、已选择的制作意图和目标 artifact 位置，再区分违约、实现偏差与个人 taste。
若已选择的单镜/形态预算不可执行，回到对应 owner 请求批准，不把拆镜或换形态默认为已经授权的替代。

# 一张参考图可以决定什么

绑定参考图等于告诉下游“哪些地方照着它”。不说明边界，身份参考会越权决定服装、构图、背景或文字。

## 每张参考回答五个问题

1. 它的稳定 `REF-...` 槽位和显式顺序是什么；
2. 它的可见定位符和中文名称是什么：图片在项目里时写项目相对路径，例如
   `输入/参考图/江辰定妆.jpg《江辰定妆照》`；图片由创作者在项目外准备并在生成时自行挂载时，
   写它对应的 `IMG-...` 板子或 `SHOT-...` 冻结关键帧；
3. 本次参考身份、造型状态、地理、构图、尺度、效果、起始帧、结束帧还是风格；
4. 哪些可见内容允许照着写；
5. 哪些身份、文字、人数、道具和空间锚点不能跟着变。

第 3 问在创作文档里有固定落点：`REF-...` 槽位的 `用途` 字段，取值只能是下表左列的九个词之一。
它同时回答创作者的「这一镜要送哪一张分镜起始帧、哪些人物/道具/场景图」，和执行端的「这张图翻成哪个
供应商 role」。同一张图承担两个作用时分开成两个槽位，不写“全参考”。

| 用途 | 可以控制 | 默认不能控制 |
|---|---|---|
| 身份 | 稳定脸型、体态、识别标记 | 临时服装、姿势、场景、文字 |
| 造型状态 | 已决定的服装、伤损、湿污 | 新身份、无来源道具、背景方位 |
| 地理 | 入口、区域、固定锚点、方向关系 | 临时人物、剧情状态 |
| 构图 | 占画比例、视线布局、镜头一侧、留白 | 人物身份、服装、道具文字 |
| 尺度 | 主体之间或主体与环境的相对尺度 | 风格、身份、数量、动作结果 |
| 效果 | 已决定效果的形态、范围、材质 | 非目标对象的消失、复制或变形 |
| 起始帧 | 本镜起始构图与状态 | 尚未发生的动作和终态 |
| 结束帧 | 本镜明确终点 | 新终点或未由镜头承担的结果 |
| 风格 | 色彩层级、材质、阴影边缘、景深、密度 | 身份、固定地理、剧情、道具文字 |

`身份` 和 `造型状态` 的界线是「本集之内会不会变」，不是「算不算衣服」。一张角色板同时固定长相和
本集从头到尾不变的造型时，写一个 `身份` 槽位，并把 `本集造型` 明确列进 `控制`——这是本套件的既定写法。
只有在本集之内会改变的状态（换装、湿透、破损、血污）才另开 `造型状态` 槽位，让它管那一次改变。
判断依据永远是这个槽位自己写下的 `控制` 范围，不是用途名称的字面意思。

道具板落在 `身份` 还是 `造型状态`，按这张图决定的是「它是哪一件」还是「它此刻处在哪个状态」：
固定的形制、配色、裂纹与磕角这类跨镜不变的识别标记写 `身份`；亮屏/熄屏、开合、湿污、破损程度
这类会随剧情改变的写 `造型状态`。同一件道具的两者都需要固定时，拆成两个槽位。

## 观察边界

创作者或权利人给出的可核对说明可以作为文字证据；运行环境获授权看过参考图时，可以报告实际可见
内容。两者都没有时保持“未验证”，不得声称水印、文字、裁切或身份已经通过。负面提示词不能擦掉
参考图已有的像素。

真实参考图使用 `REF-...` 稳定槽位，不使用 `IMG-...`；后一前缀专属《图片提示词.md》的可见提示词标题。
创作者自己挂载的图使用 `PLAN-...` 槽位，语法与 `REF-...` 相同，只把路径换成它对应的 `IMG-...` 或
`SHOT-...` 条目——这五个问题的答案一个不少，只是第 2 问答的是「这张图画的是哪一条已决定的事实」，
而不是「文件在哪里」。上表九个用途对两种槽位同样适用。多张真实图的每个槽位绑定显式顺序、项目相对路径、中文名称、用途、可控制范围和不可控制范围；联系表只是浏览包装，不改变每张图自己的用途。首尾帧默认只使用起点；
确实需要终点帧时，它只能投影 `SHOT-...` 已写明的终点，不能自行发明结果。

观众知情时机见 [观众揭示](audience-reveal.md)，补拍是否能替代母版见
[补拍与替代](pickup-and-alternate.md)。

# 项目定位与安全写入

## 定位项目

优先使用用户给出的项目路径；没有项目时可以直接在用户指定目录完成独立任务。需要项目配置时运行：

```bash
python3 <core>/scripts/project_tool.py init <project> --title "项目名"
```

本文各处写作 `python3`；Windows 原生环境没有这个名字，改用 `py -3` 或 `python`。
套件脚本（含 Dashboard）在 macOS、Linux、WSL 与 Windows 原生都可运行。

创作文档固定写入 `剧集/<EP>/`。只读取当前任务需要的直接输入；不要扫描整个项目寻找可补造的上游。

## 写入纪律

- 创作正文直接写五份 Markdown，修改时保留未受影响内容与稳定可见 ID。
- 输入目录、隐藏运行目录、制作成果和凭据不是创作正文；不要把它们复制进提示词。
- 文本写入使用临时文件加原子替换；外部编辑发生时先重新读取，不静默覆盖。
- Dashboard 只展示和编辑项目文件，不承担创作路由或生产授权。
- 外部生产仍由 `$short-drama-produce` 预览精确任务、取得显式确认后执行。

输出语言、稳定 ID 与信任边界见 [契约与所有权](contract-and-ownership.md)。

#!/usr/bin/env python3
"""Validate the executable cross-document contract of one creator-first episode."""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path, PurePosixPath
from collections.abc import Iterable
from typing import NamedTuple, Optional


MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit(
        "creator_markdown_check.py requires Python {}.{}, running {}.{}".format(
            *MINIMUM_PYTHON, sys.version_info.major, sys.version_info.minor
        )
    )


REQUIRED_DOCUMENTS = (
    "剧本.md",
    "视觉设定.md",
    "图片提示词.md",
    "分镜.md",
    "视频提示词.md",
)
SECTION_RE = re.compile(r"^## ((?:SHOT|MOTION)-[A-Z0-9-]+)\b", re.MULTILINE)
IMG_RE = re.compile(r"\b(IMG-[A-Z0-9-]+)《([^》]+)》（控制：([^）]+)）")
# 用途 is optional in the pattern on purpose: a declaration written before this
# field existed still parses, so the creator gets "REF 缺少用途" instead of the
# generic syntax error that gives no hint about what to add.
REF_RE = re.compile(
    r"(REF-[A-Z0-9-]+)（顺序：([1-9]\d*)）· "
    r"([^；\n]+?\.(?:png|jpe?g|webp))《([^》\n]+)》"
    r"（(?:用途：([^；）\n]+)；)?控制：([^；）]+)；不得控制：([^）]+)）",
    re.IGNORECASE,
)
# The same slot grammar, but the locator is a document entry instead of a file.
# A creator who generates in an external tool never has the file inside the
# project, and the copyable body only ever needed 顺序 and 用途 -- the path is a
# production input, not a prompt input. PLAN says "this picture does not exist
# here; attach it yourself in this order", which is a different claim from
# 「已有图片」 and from 「待补」.
PLAN_RE = re.compile(
    r"(PLAN-[A-Z0-9-]+)（顺序：([1-9]\d*)）· "
    r"((?:IMG|SHOT)-[A-Z0-9-]+)《([^》\n]+)》"
    r"（(?:用途：([^；）\n]+)；)?控制：([^；）]+)；不得控制：([^）]+)）",
    re.IGNORECASE,
)
# One picture answers one question. 「同一张图承担两个作用时分开说明」 in
# references/reference-roles.md is what makes a slot translatable into a
# provider role, so the vocabulary is closed rather than free prose.
REF_PURPOSES = (
    "身份",
    "造型状态",
    "地理",
    "构图",
    "尺度",
    "效果",
    "起始帧",
    "结束帧",
    "风格",
)
# `EP001-SC001` is the documented shape, but a project that scopes ids by season
# writes `S01-EP001-SC001`. Both are one stable scene id, so the pattern takes
# any hyphenated prefix rather than exactly one segment.
SLOT_HEAD_RE = re.compile(
    r"(?:REF|PLAN)-[A-Za-z0-9][A-Za-z0-9-]*（顺序：", re.IGNORECASE
)
SCENE_ID_PATTERN = r"(?:[A-Za-z0-9]+-)+SC[0-9]+"
SCENE_HEADING_RE = re.compile(
    r"^## (" + SCENE_ID_PATTERN + r")\b", re.MULTILINE
)
SCENE_ID_RE = re.compile(SCENE_ID_PATTERN)
# A scene the storyboard decided not to film. SHT-01 has always allowed the
# omission and always required a reason; without a place to write it the
# omission and an oversight look identical in the finished document.
UNFILMED_LINE_RE = re.compile(
    r"^[ \t\u3000]*[-*+][ \t\u3000]*未拍场次[：:](.*)$", re.MULTILINE
)
UNFILMED_ENTRY_RE = re.compile(
    r"(" + SCENE_ID_PATTERN + r")（理由：([^）\n]+)）"
)
# Dialogue reaches the model as a quoted run, either inside the H3 `<d>` tag or
# inside quotation marks. Runs shorter than four Han characters are in-frame
# labels and interjections far more often than they are lines, so the check
# stays off them rather than inventing an escape hatch for signage.
SPOKEN_SPAN_RE = re.compile(r"<d>([^<]*)</d>|[\"“]([^\"”]*)[\"”]|「([^」]*)」")
SPOKEN_RUN_RE = re.compile(r"[\u3400-\u9fff]{4,}")
EXPLICIT_TEXT_TO_VIDEO = "无（创作者已明确选择文生视频）"
PENDING_REFERENCE_SUFFIX_RE = re.compile(r"；待补参考图：[^；。\n]+。?$")
VISUAL_CATEGORIES = ("人物", "造型", "地点", "道具")
# A malformed entry always tries to name something; a heading that is only the
# category word is a section divider ("## 人物"), and rejecting it left the
# document with no way to group entries at all.
VISUAL_SETTING_LINE_RE = re.compile(
    r"^#{2,4}[ \t　]*(?:"
    + "|".join(VISUAL_CATEGORIES)
    + r")(?![ \t　]*$)[^\n]*$",
    re.MULTILINE,
)
VISUAL_SETTING_HEADING_RE = re.compile(
    r"^## (" + "|".join(VISUAL_CATEGORIES) + r") · (.+?)[ \t　]*$", re.MULTILINE
)
# Any other `## <word> · <name>` heading, so a 视觉依据 entry that fails to
# resolve can say "the entry is there, the section word is not one of the four"
# instead of blaming the storyboard for a `视觉设定.md` problem.
OTHER_SETTING_HEADING_RE = re.compile(r"^## ([^\n·]+?) · (.+?)[ \t　]*$", re.MULTILINE)
# 画面代称 is the one field the coverage check depends on, so — like a
# continuity lock — anything that looks like it has to parse rather than
# silently drop out.
# Anchored at the start of the line so a sentence that merely mentions the
# field is prose, not a malformed declaration.
SCREEN_NAME_LINE_RE = re.compile(
    r"^[ \t　]*[-*+]?[ \t　]*画面代称[ \t　]*[：:][^\n]*$", re.MULTILINE
)
SCREEN_NAME_RE = re.compile(
    r"^[ \t　]*[-*+][ \t　]*画面代称[：:](.+)$", re.MULTILINE
)
VISUAL_BASIS_PREFIX = "《视觉设定.md》·"
VISUAL_BASIS_ENTRY_RE = re.compile(
    r"(" + "|".join(VISUAL_CATEGORIES) + r")「([^」\n]+)」（控制：([^）\n]+)）"
)
# A subject a keyframe names but does not show — an owner's abandoned bag, a
# name on a screen, a homonym of an entry name. SHT-22 excludes these from
# coverage, so the field needs a way to say so instead of forcing the creator
# to declare an absent subject as present.
OFFSCREEN_PREFIX = "；画外："
OFFSCREEN_ENTRY_RE = re.compile(
    r"(" + "|".join(VISUAL_CATEGORIES) + r")「([^」\n]+)」"
)
# A declared lock must never become a no-op. Anything that *looks* like a lock
# line -- any list marker, any leading whitespace -- is captured here and then
# has to parse, so a creator who indents the bullet under 识别锚点 gets an error
# instead of silent non-enforcement.
LOCK_LINE_RE = re.compile(r"^[ \t\u3000]*[-*+][ \t\u3000]*连续性锁[：:].*$", re.MULTILINE)
LOCK_RE = re.compile(
    r"^[ \t\u3000]*[-*+][ \t\u3000]*连续性锁：(LOCK-[A-Z0-9-]+)《([^》\n]+)》"
    r"（镜头：([^；）\n]+)"
    r"(?:；图片提示词项：([^；）\n]+))?）"
    r"· 锁面：(.+)$"
)
# The surface has to name what is in the picture. A match glued to a negation
# ("no pale blue sweater") describes what must be absent, so it cannot be the
# evidence that the fact is present.
# Chinese runs without spaces, so the CJK markers cannot require a preceding
# boundary the way the English ones do. A bare 无 is deliberately not a marker:
# 无袖毛衣 describes the garment rather than excluding it.
NEGATION_RE = re.compile(
    r"(?:"
    r"(?:^|[\s,;:(\[/—-])(?:no|not|non|never|without|avoid|excludes?|excluding|"
    r"free\s+of|--?no)(?:\s+(?:a|an|the|any|some))?[\s-]*"
    r"|(?:不要|不得|不能|不应|不含|不出现|没有|未|避免|禁止|排除|去掉|移除)"
    r"(?:出现|包含|存在|带|有)?[\s]*"
    r")$",
    re.IGNORECASE,
)


def _sections(document: str, kind: str) -> dict[str, str]:
    matches = [
        match
        for match in SECTION_RE.finditer(document)
        if match.group(1).startswith(f"{kind}-")
    ]
    return {
        match.group(1): document[
            match.start() : matches[index + 1].start()
            if index + 1 < len(matches)
            else None
        ]
        for index, match in enumerate(matches)
    }


def _fields(section: str, *, owner: str, errors: list[str]) -> dict[str, str]:
    pairs = re.findall(r"^- ([^：\n]+)：(.+)$", section, re.MULTILINE)
    fields: dict[str, str] = {}
    for key, value in pairs:
        # `- **参考**：…` is ordinary Markdown and reads as the same field to a
        # human. Taking the emphasis literally turned it into a different key,
        # so the author got "缺少参考字段" while looking straight at 参考.
        key = re.sub(r"^(?:\*\*|__|\*|_)(.+?)(?:\*\*|__|\*|_)$", r"\1", key.strip())
        if key in fields:
            errors.append(f"{owner}: 字段重复: {key}")
        fields[key] = value
    return fields


def _plain(value: str) -> str:
    return value.strip().rstrip("。")


def _contains_ref_token(value: str) -> bool:
    return "ref-" in value.casefold()


def _contains_plan_token(value: str) -> bool:
    return "plan-" in value.casefold()


def _contains_slot_token(value: str) -> bool:
    return _contains_ref_token(value) or _contains_plan_token(value)


def _is_none(value: str) -> bool:
    # `[^）]+` would stop at the first closing bracket, so a gap list that names
    # an entry in brackets -- 「江晨人物图（IMG-...）」 -- fell through to the REF
    # syntax error, whose obvious repair is deleting the gap record. That is the
    # silent downgrade this contract exists to prevent, so the note takes any
    # content as long as it is bracketed.
    return not _contains_slot_token(value) and bool(
        re.fullmatch(r"无(?:（.+）)?", _plain(value), re.DOTALL)
    )


def _is_explicit_text_to_video(value: str) -> bool:
    return _plain(value) == EXPLICIT_TEXT_TO_VIDEO


def _has_pending_references(value: str) -> bool:
    plain = _plain(value)
    return plain.startswith("无（待补参考图：") or bool(
        PENDING_REFERENCE_SUFFIX_RE.search(value.strip())
    )


def _is_no_external_reference(value: str) -> bool:
    return not _contains_slot_token(value) and bool(
        re.fullmatch(r"无(?:外部参考)?(?:；[^\n]*)?。?", value.strip())
    )


def _copyable_prompt(
    section: str, heading: str = r"可复制(?:通用)?提示词"
) -> Optional[str]:
    markers = list(re.finditer(rf"^### {heading}\s*$", section, re.MULTILINE))
    if len(markers) != 1:
        return None
    body = section[markers[0].end() :]
    following = re.search(r"^###\s+|^##\s+", body, re.MULTILINE)
    if following is not None:
        body = body[: following.start()]
    lines = [line for line in body.splitlines() if line.strip()]
    if not lines or any(not line.startswith(">") for line in lines):
        return None
    prompt = "\n".join(line[1:].lstrip() for line in lines).strip()
    return prompt or None


def _copyable_prompt_cause(
    section: str, heading: str = r"可复制(?:通用)?提示词"
) -> str:
    """Why `_copyable_prompt` returned None, as a suffix for the error.

    "缺少唯一且非空的可复制提示词" is true of four different documents and tells
    the author nothing about which one they wrote. The expensive case is a
    single stray line -- a separator, an HTML comment, a note -- inside an
    otherwise complete block: the prompt is visibly right there, so the message
    reads as a checker bug and the real line goes unlooked at.
    """

    # Name the heading the author actually has to write, not this one's default:
    # the keyframe caller passes a different one, and pointing at the wrong
    # heading is worse than saying nothing.
    label = "### " + re.sub(r"\(\?:([^)]*)\)\?", "", heading).replace("\\", "")
    markers = list(re.finditer(rf"^### {heading}\s*$", section, re.MULTILINE))
    if not markers:
        return f"：没有 `{label}` 小节标题"
    if len(markers) > 1:
        return f"：`{label}` 出现了 {len(markers)} 次，只能有一个"
    body = section[markers[0].end() :]
    following = re.search(r"^###\s+|^##\s+", body, re.MULTILINE)
    if following is not None:
        body = body[: following.start()]
    lines = [line for line in body.splitlines() if line.strip()]
    if not lines:
        return f"：`{label}` 下面是空的"
    intruders = [line for line in lines if not line.startswith(">")]
    if intruders:
        return (
            "：小节内有不以 `>` 开头的行，整块因此不算引用块——"
            f"把它移到小节外，或删掉：{_excerpt(intruders[0])}"
        )
    return "：引用块里只有空白"


def _portable_path(value: str) -> bool:
    if not value or "\\" in value or re.match(r"^[A-Za-z]:", value):
        return False
    parts = value.split("/")
    return not PurePosixPath(value).is_absolute() and not any(
        part in {"", ".", ".."} for part in parts
    )


def _inside(path: Path, root: Path) -> bool:
    resolved = path.resolve()
    return resolved == root or root in resolved.parents


class EntryIndex(NamedTuple):
    """The document entries a `PLAN-...` locator is allowed to name."""

    images: dict[str, str]
    shots: frozenset


def _slot_matches(value: str) -> list:
    """Every `REF-...` and `PLAN-...` slot in `value`, in written order.

    The two patterns cannot match the same span: a REF locator has to be an
    image path and a PLAN locator has to be an `IMG-`/`SHOT-` id.
    """
    found = []
    for kind, pattern in (("REF", REF_RE), ("PLAN", PLAN_RE)):
        for match in pattern.finditer(value):
            found.append((match.start(), match.end(), kind, match.groups()))
    found.sort(key=lambda item: item[0])
    return found


def _references(
    value: str,
    owner: str,
    project_root: Path,
    errors: list[str],
    entries: Optional[EntryIndex] = None,
) -> frozenset:
    """Validate one 输入参考图/参考 declaration; return the slot kinds it uses."""
    if _is_none(value):
        return frozenset()
    reference_value = PENDING_REFERENCE_SUFFIX_RE.sub("", value.strip())
    matches = _slot_matches(reference_value)
    cursor = 0
    separators_are_valid = True
    for index, (start, end, _kind, _groups) in enumerate(matches):
        if reference_value[cursor:start] != ("" if index == 0 else "；"):
            separators_are_valid = False
        cursor = end
    trailing = reference_value[cursor:]
    # Counted as slot heads on the same text the slots were matched in. A bare
    # token count would read the `ref-` in a real path like
    # `输入/参考图/ref-江晨.png` as a second slot and reject a correct declaration.
    declared = len(SLOT_HEAD_RE.findall(reference_value))
    kinds = frozenset(item[2] for item in matches)
    if (
        len(matches) != declared
        or not matches
        or not separators_are_valid
        or trailing not in {"", "。"}
    ):
        # A gap list joined with ； reads as a fourth REF slot and would otherwise
        # be reported as broken REF syntax, whose obvious repair is deleting the
        # gap record -- the silent downgrade this contract exists to prevent.
        if "待补参考图" in value and not _has_pending_references(value):
            errors.append(
                f"{owner}: 待补参考图必须写在最后一个 REF 槽位之后，"
                "缺口之间只用、分隔"
            )
        elif _contains_plan_token(value) and not _contains_ref_token(value):
            errors.append(f"{owner}: 输入参考图必须使用完整 PLAN 语法")
        else:
            errors.append(f"{owner}: 输入参考图必须使用完整 REF 语法")
        return kinds
    groups = [item[3] for item in matches]
    slot_kinds = [item[2] for item in matches]
    slots = [group[0] for group in groups]
    orders = [int(group[1]) for group in groups]
    locators = [group[2] for group in groups]
    purposes = [(group[4] or "").strip() for group in groups]
    # A declaration made only of PLAN slots is not a REF declaration, so its
    # diagnostics must not send the creator looking for a REF line.
    noun = "PLAN" if kinds == frozenset({"PLAN"}) else "REF"
    if len(slots) != len(set(slots)):
        errors.append(f"{owner}: {noun} 槽位重复")
    if len(orders) != len(set(orders)) or sorted(orders) != list(
        range(1, len(orders) + 1)
    ):
        errors.append(f"{owner}: {noun} 顺序必须唯一且从 1 连续编号")
    locator_purposes = list(zip(locators, purposes))
    if len(locator_purposes) != len(set(locator_purposes)):
        errors.append(f"{owner}: {noun} 路径与用途完全重复")
    for slot, purpose in zip(slots, purposes):
        if not purpose:
            errors.append(
                f"{owner}: {noun} 缺少用途: {slot}；用途只能是"
                f"{'、'.join(REF_PURPOSES)}其中一个"
            )
        elif purpose not in REF_PURPOSES:
            errors.append(
                f"{owner}: {noun} 用途不在允许集合内: {slot}（{purpose}）；"
                f"只能是{'、'.join(REF_PURPOSES)}"
            )
    for role in ("起始帧", "结束帧"):
        if purposes.count(role) > 1:
            errors.append(f"{owner}: 同一条目只能有一张{role}参考图")
    if "结束帧" in purposes and "起始帧" not in purposes:
        errors.append(f"{owner}: 绑定结束帧参考图时必须同时绑定起始帧")
    for index, (kind, group) in enumerate(zip(slot_kinds, groups)):
        slot, raw_locator, label = group[0], group[2], group[3]
        may_control, must_not_control = group[5], group[6]
        if kind == "REF":
            if not _portable_path(raw_locator):
                errors.append(
                    f"{owner}: REF 路径不是安全的项目相对路径: {raw_locator}"
                )
            else:
                reference_path = project_root / raw_locator
                if not _inside(reference_path, project_root):
                    errors.append(f"{owner}: REF 路径越出项目根目录: {raw_locator}")
                elif not reference_path.is_file():
                    errors.append(f"{owner}: REF 文件不存在: {raw_locator}")
        else:
            _plan_locator(
                raw_locator, label, owner, slot, purposes[index], entries, errors
            )
        if not re.search(r"[\u4e00-\u9fff]", label):
            errors.append(f"{owner}: {kind} 缺少中文名称: {slot}")
        if not may_control.strip() or not must_not_control.strip():
            errors.append(f"{owner}: {kind} 必须同时声明控制与不得控制: {slot}")
        allowed = {item.strip() for item in re.split(r"[、,，]", may_control)}
        prohibited = {item.strip() for item in re.split(r"[、,，]", must_not_control)}
        if "" in allowed or "" in prohibited or allowed & prohibited:
            errors.append(f"{owner}: {kind} 控制与不得控制范围冲突: {slot}")
    return kinds


def _plan_locator(
    locator: str,
    label: str,
    owner: str,
    slot: str,
    purpose: str,
    entries: Optional[EntryIndex],
    errors: list[str],
) -> None:
    """A planned picture still has to name an entry that exists today.

    The picture itself is the creator's to supply, but what it depicts is
    already decided by an `IMG-...` board or a `SHOT-...` frozen keyframe. A
    locator that resolves to neither is a reference nobody can act on.
    """
    if entries is None:
        return
    if locator.startswith("IMG-"):
        if locator not in entries.images:
            errors.append(f"{owner}: PLAN 指向不存在的 IMG 条目: {locator}")
        elif label != entries.images[locator]:
            errors.append(f"{owner}: PLAN 中文名称与 IMG 标题不一致: {slot}")
    elif locator.startswith("SHOT-"):
        if locator not in entries.shots:
            errors.append(f"{owner}: PLAN 指向不存在的 SHOT 条目: {locator}")
        elif purpose == "起始帧":
            # The picture that opens this shot is this shot's own frozen
            # keyframe. Another shot's keyframe opens a different moment.
            own = re.sub(r"^(?:SHOT|MOTION)-", "SHOT-", owner)
            if own.startswith("SHOT-") and locator != own:
                errors.append(
                    f"{owner}: 起始帧 PLAN 必须指向本镜的冻结关键帧: {locator}"
                )
    else:
        errors.append(
            f"{owner}: PLAN 定位符只能是 IMG-... 或 SHOT-...: {locator}"
        )


def _excerpt(value: str, limit: int = 60) -> str:
    """A short, single-line quote of an offending line for a diagnostic."""
    collapsed = re.sub(r"\s+", " ", value).strip()
    return collapsed if len(collapsed) <= limit else collapsed[: limit - 1] + "…"


def _unique(values: "Iterable[str]") -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        if value not in seen:
            seen.add(value)
            result.append(value)
    return result


def _normalized(value: str) -> str:
    """Fold case and collapse whitespace so a hard-wrapped prompt still matches.

    A copyable prompt is one rendered paragraph; the line breaks the repo's
    Markdown style puts in it are not part of the text a creator wrote.
    """
    return re.sub(r"\s+", " ", value).strip().casefold()


def _wordish(character: str) -> bool:
    return bool(character) and character.isascii() and (
        character.isalnum() or character == "-"
    )


def _carries_surface(prompt: str, surface: str) -> bool:
    """Is this lock surface actually asserted by this prompt?

    Plain containment is not enough. `chipped white enamel mug` must not be
    satisfied by `unchipped ... mug`, and `no pale blue sweater` in a negative
    tail asserts the opposite of the fact the lock exists to hold.
    """
    haystack = _normalized(prompt)
    needle = _normalized(surface)
    if not needle:
        return False
    start = haystack.find(needle)
    while start != -1:
        end = start + len(needle)
        before = haystack[start - 1] if start else ""
        after = haystack[end] if end < len(haystack) else ""
        # Only ASCII words have boundaries to glue across. Chinese is written
        # without spaces, so treating an adjacent CJK character as "glued" would
        # make a Chinese surface impossible to satisfy.
        glued = (
            (_wordish(needle[:1]) and _wordish(before))
            or (_wordish(needle[-1:]) and _wordish(after))
        )
        if not glued and NEGATION_RE.search(haystack[:start]) is None:
            return True
        start = haystack.find(needle, start + 1)
    return False


class VisualEntry(NamedTuple):
    """One `视觉设定.md` entry plus every name a prompt body may call it by."""

    category: str
    name: str
    designators: list[str]

    @property
    def key(self) -> tuple[str, str]:
        return (self.category, self.name)


def _visual_entries(document: str, errors: list[str]) -> list[VisualEntry]:
    """Parse the 人物/造型/地点/道具 entries of one `视觉设定.md`.

    A heading that *looks* like an entry but does not parse would silently drop
    out of the coverage check and then reappear as "视觉依据 names an entry that
    does not exist", so it is rejected here where the cause is visible.
    """
    for line in VISUAL_SETTING_LINE_RE.findall(document):
        if VISUAL_SETTING_HEADING_RE.match(line) is None:
            errors.append(
                "视觉设定.md: 条目标题必须写成 `## <人物|造型|地点|道具> · <名称>`: "
                + _excerpt(line)
            )
    for line in SCREEN_NAME_LINE_RE.findall(document):
        if SCREEN_NAME_RE.match(line) is None:
            errors.append(
                "视觉设定.md: 画面代称必须写成 `- 画面代称：<正文里的拼写>`: "
                + _excerpt(line)
            )
    headings = list(VISUAL_SETTING_HEADING_RE.finditer(document))
    entries: list[VisualEntry] = []
    seen: set[tuple[str, str]] = set()
    for index, heading in enumerate(headings):
        end = (
            headings[index + 1].start()
            if index + 1 < len(headings)
            else len(document)
        )
        category, name = heading.group(1), heading.group(2).strip()
        if not name:
            errors.append(f"视觉设定.md: {category}条目缺少名称")
            continue
        if (category, name) in seen:
            errors.append(f"视觉设定.md: 条目重复: {category}「{name}」")
            continue
        seen.add((category, name))
        declared = SCREEN_NAME_RE.findall(document[heading.end() : end])
        if not declared:
            # No declaration: the entry's own name is the designator, which is
            # what a prompt body written in the project's own language calls it.
            designators = [name]
        elif all(_is_none(_plain(raw)) for raw in declared):
            # `画面代称：无` is the deliberate opt-out for a name too common to
            # match reliably in prose (道具「手机」 against 手机店). It removes the
            # derived name too, otherwise the opt-out would do nothing.
            designators = []
        else:
            designators = []
            for raw in declared:
                value = _plain(raw)
                if _is_none(value):
                    continue
                for item in re.split(r"[、,，]", value):
                    item = item.strip()
                    if item and item not in designators:
                        designators.append(item)
        # Only a *declared* designator earns this diagnostic: the creator chose a
        # spelling the matcher cannot honour and deserves to know. A one-character
        # entry name they never declared is simply not name-matched, the same as
        # any entry in a project whose prompt body is written in another language.
        if declared:
            for designator in designators:
                if len(designator.strip()) < 2:
                    errors.append(
                        f"视觉设定.md: {category}「{name}」的画面代称「{designator}」"
                        "过短，无法在正文中可靠识别；请写成至少两个字符，或写「画面代称：无」"
                    )
        entries.append(VisualEntry(category, name, designators))
    return entries


def _named_entries(
    prompt: str,
    entries: list[VisualEntry],
    *,
    fold_case: bool = False,
    by_entry_name: bool = False,
) -> set[tuple[str, str]]:
    """Which declared entries does this frozen keyframe actually call by name?

    Longer designators win: a keyframe that shows 「江晨手机」 names the prop, and
    the 人物「江晨」 substring inside it is not a second, unrelated claim.
    """
    haystack = re.sub(r"\s+", " ", prompt).strip()
    if fold_case:
        haystack = haystack.casefold()
    owners: dict[str, list[VisualEntry]] = {}
    for entry in entries:
        # 来源 quotes 剧本.md, which is written in the project's own language and
        # calls people by their entry names — not by the 画面代称 a prompt body
        # uses. Matching a Chinese screenplay quote against an English designator
        # would never fire, so that text is matched by entry name instead. An
        # entry that opted out with `画面代称：无` opted out of being found by
        # name at all, and stays out of both.
        needles = (
            [entry.name] if by_entry_name and entry.designators else entry.designators
        )
        for designator in needles:
            needle = re.sub(r"\s+", " ", designator).strip()
            if fold_case:
                needle = needle.casefold()
            # A one-character designator matches far too much prose to be
            # evidence; _visual_entries already reports it.
            if len(needle) < 2:
                continue
            owners.setdefault(needle, []).append(entry)
    named: set[tuple[str, str]] = set()
    # Spans of every occurrence of a longer designator, whether or not it was
    # credited. 「戒指盒」 occupies its characters even when the sentence is
    # 「没有戒指」, so 道具「戒指」 must not be read out of it.
    taken: list[tuple[int, int]] = []
    for needle in sorted(owners, key=len, reverse=True):
        occurrences: list[tuple[int, int]] = []
        start = haystack.find(needle)
        while start != -1:
            end = start + len(needle)
            occurrences.append((start, end))
            start = haystack.find(needle, start + 1)
        for start, end in occurrences:
            before = haystack[start - 1] if start else ""
            after = haystack[end] if end < len(haystack) else ""
            glued = (
                (_wordish(needle[:1]) and _wordish(before))
                or (_wordish(needle[-1:]) and _wordish(after))
            )
            # Strictly longer, so two entries sharing one designator are both
            # credited instead of the first in document order winning.
            inside_longer = any(
                taken_start <= start
                and end <= taken_end
                and taken_end - taken_start > end - start
                for taken_start, taken_end in taken
            )
            if (
                not glued
                and not inside_longer
                and NEGATION_RE.search(haystack[:start]) is None
            ):
                named.update(entry.key for entry in owners[needle])
        taken.extend(occurrences)
    return named


class VisualBasis(NamedTuple):
    """One shot's parsed 视觉依据: what it declares, and what it excludes."""

    parsed: bool
    declared: set[tuple[str, str]]
    offscreen: set[tuple[str, str]]

    @property
    def accounted(self) -> set[tuple[str, str]]:
        return self.declared | self.offscreen


def _resolve_entry(
    key: tuple[str, str],
    known: set[tuple[str, str]],
    other_headings: dict[str, str],
    owner: str,
    errors: list[str],
) -> None:
    if key in known:
        return
    category, name = key
    section = other_headings.get(name)
    if section is not None:
        errors.append(
            f"{owner}: 《视觉设定.md》里有「{name}」，但它的分节词是「{section}」；"
            "条目标题必须写成 `## <人物|造型|地点|道具> · <名称>`"
        )
    else:
        errors.append(
            f"{owner}: 视觉依据指向不存在的《视觉设定.md》条目: {category}「{name}」"
        )


def _check_named_coverage(
    prompt: str,
    owner: str,
    where: str,
    basis: "VisualBasis",
    entries: list[VisualEntry],
    errors: list[str],
    *,
    by_entry_name: bool = False,
) -> None:
    """Every entry this text calls by name is either in frame or declared 画外."""
    # Matching is case-sensitive so an ordinary English word never impersonates a
    # character called May or Will. A body that writes the name in another case
    # would otherwise fall out of the check silently, so it is reported here.
    named = _named_entries(prompt, entries, by_entry_name=by_entry_name)
    folded = _named_entries(
        prompt, entries, fold_case=True, by_entry_name=by_entry_name
    )
    for category, name in sorted(folded - named):
        errors.append(
            f"{owner}: {where}里的名字与画面代称大小写不一致: {category}「{name}」；"
            "同一个名字全集只用一个拼写"
        )
    for category, name in sorted(named):
        if (category, name) not in basis.accounted:
            errors.append(
                f"{owner}: {where}写到{category}「{name}」，视觉依据没有覆盖；"
                "本镜确实看不见时在视觉依据末尾加「；画外："
                f"{category}「{name}」」，正文里这个名字不可靠时在《视觉设定.md》"
                "写「画面代称：无」"
            )


def _visual_basis(
    value: str,
    owner: str,
    entries: list[VisualEntry],
    other_headings: dict[str, str],
    errors: list[str],
) -> VisualBasis:
    """Parse one shot's 视觉依据 and resolve it against `视觉设定.md`."""
    empty = VisualBasis(True, set(), set())
    if _is_none(value):
        return empty
    plain = _plain(value)
    offscreen_raw = ""
    if OFFSCREEN_PREFIX in plain:
        plain, _, offscreen_raw = plain.partition(OFFSCREEN_PREFIX)
    prefixed = plain.startswith(VISUAL_BASIS_PREFIX)
    body = plain[len(VISUAL_BASIS_PREFIX) :] if prefixed else ""
    matches = list(VISUAL_BASIS_ENTRY_RE.finditer(body)) if prefixed else []
    cursor = 0
    separators_are_valid = True
    for index, match in enumerate(matches):
        # Repeating 《视觉设定.md》· before each entry says the same thing and
        # reads naturally; rejecting it would cost a round trip over nothing.
        separator = body[cursor : match.start()]
        allowed = (
            {""}
            if index == 0
            else {"；", "；" + VISUAL_BASIS_PREFIX}
        )
        if separator not in allowed:
            separators_are_valid = False
        cursor = match.end()
    if not prefixed or not matches or not separators_are_valid or body[cursor:]:
        errors.append(
            f"{owner}: 视觉依据必须使用完整语法："
            "《视觉设定.md》·<人物|造型|地点|道具>「<名称>」（控制：<范围>），多项用；连接"
        )
        # Coverage is not reported on top of a parse failure: every entry would
        # be listed as uncovered and bury the one error that matters.
        return VisualBasis(False, set(), set())
    known = {entry.key for entry in entries}
    declared: set[tuple[str, str]] = set()
    for match in matches:
        key = (match.group(1), match.group(2).strip())
        if key in declared:
            errors.append(f"{owner}: 视觉依据条目重复: {key[0]}「{key[1]}」")
        declared.add(key)
        _resolve_entry(key, known, other_headings, owner, errors)
        if not match.group(3).strip():
            errors.append(f"{owner}: 视觉依据缺少控制范围: {key[0]}「{key[1]}」")
    offscreen: set[tuple[str, str]] = set()
    if offscreen_raw:
        offscreen_raw = offscreen_raw.replace(VISUAL_BASIS_PREFIX, "")
        remainder = OFFSCREEN_ENTRY_RE.sub("", offscreen_raw).strip("；、 ")
        offscreen_matches = list(OFFSCREEN_ENTRY_RE.finditer(offscreen_raw))
        if not offscreen_matches or remainder:
            errors.append(
                f"{owner}: 画外清单必须写成 `；画外：<人物|造型|地点|道具>「<名称>」`，多项用；连接"
            )
            return VisualBasis(False, set(), set())
        for match in offscreen_matches:
            key = (match.group(1), match.group(2).strip())
            if key in declared:
                errors.append(
                    f"{owner}: {key[0]}「{key[1]}」同时写进视觉依据和画外清单"
                )
            offscreen.add(key)
            _resolve_entry(key, known, other_headings, owner, errors)
    return VisualBasis(True, declared, offscreen)


class ContinuityLock(NamedTuple):
    """One declared cross-shot lock: the exact surface and where it applies."""

    lock_id: str
    surface: str
    shots: list[str]
    images: list[str]


def _continuity_locks(document: str, errors: list[str]) -> list[ContinuityLock]:
    """Parse the declared continuity locks of one 视觉设定.md."""
    locks: list[ContinuityLock] = []
    seen: set[str] = set()
    for line in LOCK_LINE_RE.findall(document):
        match = LOCK_RE.match(line)
        if match is None:
            errors.append(
                "视觉设定.md: 连续性锁必须使用完整语法: " + _excerpt(line)
            )
            continue
        lock_id, label, scope, image_scope, surface = match.groups()
        if lock_id in seen:
            errors.append(f"{lock_id}: 连续性锁 ID 重复")
            continue
        seen.add(lock_id)
        if not re.search(r"[\u3400-\u9fff]", label):
            errors.append(f"{lock_id}: 连续性锁缺少中文名称")
        surface = _plain(surface)
        if not surface:
            errors.append(f"{lock_id}: 连续性锁缺少锁面")
            continue
        shots = _unique(
            item.strip() for item in re.split(r"[、,，]", _plain(scope)) if item.strip()
        )
        if not shots:
            errors.append(f"{lock_id}: 连续性锁缺少镜头范围")
            continue
        if "全集" in shots and len(shots) != 1:
            errors.append(f"{lock_id}: 连续性锁的镜头范围不能把全集与具体镜头混写")
            continue
        images: list[str] = []
        if image_scope is not None and not _is_none(image_scope):
            images = _unique(
                item.strip()
                for item in re.split(r"[、,，]", _plain(image_scope))
                if item.strip()
            )
            if any(not item.startswith("IMG-") for item in images):
                errors.append(f"{lock_id}: 连续性锁的图片提示词项必须使用 IMG-  ID")
                continue
        locks.append(ContinuityLock(lock_id, surface, shots, images))
    return locks


def _prompt_language(project_root: Path) -> str:
    """The language a copyable prompt body is written in.

    Mirrors the skills' own routing: the project's declared prompt language,
    falling back to `en` when there is no `short-drama.json` -- which is what
    the storyboard skill tells the keyframe author to assume.
    """
    configuration = project_root / "short-drama.json"
    if not configuration.is_file():
        return "en"
    try:
        project = json.loads(configuration.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return "en"
    if not isinstance(project, dict):
        return "en"
    authority = project.get("creator_authority")
    profile = (
        authority.get("production_profile") if isinstance(authority, dict) else None
    )
    if isinstance(profile, dict) and profile.get("status") == "accepted":
        choices = profile.get("choices")
        if isinstance(choices, dict):
            declared = choices.get("video_prompt_language")
            if isinstance(declared, str) and declared:
                return declared
    formats = project.get("format")
    declared = formats.get("prompt_language") if isinstance(formats, dict) else None
    return declared if isinstance(declared, str) and declared else "en"


def _check_language_designators(
    project_root: Path, *, entries: list[VisualEntry], errors: list[str]
) -> None:
    """Every character must be nameable in the prompt body's language.

    Without this, omitting `画面代称` is a silent opt-out of the coverage check
    for exactly the projects that need it -- the keyframe body defaults to `en`
    while `视觉设定.md` is Chinese, which is the shape issue #94 reported.

    This deliberately does not wait until some shot's 视觉依据 references the
    entry. Gating on that made the two halves of the contract land in different
    rounds: a document missing 视觉依据 got one error, and only after fixing it
    did the missing 画面代称 appear.
    """
    if _prompt_language(project_root).casefold().startswith("zh"):
        return
    for entry in entries:
        if entry.category != "人物":
            continue
        if entry.designators == [entry.name]:
            errors.append(
                f"视觉设定.md: 人物「{entry.name}」缺少画面代称；"
                "提示词正文不是中文时，写「画面代称：<正文里的拼写>」，"
                "正文从不点名时写「画面代称：无」"
            )


def _declared_seconds(value: str) -> Optional[float]:
    """The number of seconds a 时长 field declares, however it is spelled.

    `4s`, `4 秒` and `4秒` are the same duration. Returning None for anything
    else keeps an unparseable value out of the comparison instead of turning a
    formatting difference into a false duration conflict.
    """
    match = re.search(r"(\d+(?:\.\d+)?)", value or "")
    return float(match.group(1)) if match else None


def _check_continuity_locks(
    locks: list[ContinuityLock],
    *,
    shots: dict[str, str],
    motion_by_shot: dict[str, tuple[str, str, Optional[str]]],
    image_prompts: dict[str, Optional[str]],
    errors: list[str],
) -> None:
    """Require every declared lock surface to be present where it was scoped."""
    for lock in locks:
        lock_id = lock.lock_id
        surface = lock.surface
        targets = sorted(shots) if lock.shots == ["全集"] else lock.shots
        for shot_id in targets:
            if shot_id not in shots:
                errors.append(f"{lock_id}: 连续性锁指向不存在的镜头: {shot_id}")
                continue
            keyframe = _copyable_prompt(shots[shot_id], heading=r"冻结关键帧提示词")
            if keyframe is None:
                errors.append(f"{lock_id}: {shot_id} 缺少可读的冻结关键帧提示词")
            elif not _carries_surface(keyframe, surface):
                errors.append(f"{lock_id}: {shot_id} 冻结关键帧提示词缺少锁面")
            motion = motion_by_shot.get(shot_id)
            if motion is None:
                continue
            motion_id, _, copyable_prompt = motion
            if copyable_prompt is not None and not _carries_surface(
                copyable_prompt, surface
            ):
                errors.append(f"{lock_id}: {motion_id} 可复制提示词缺少锁面")
        for image_id in lock.images:
            if image_id not in image_prompts:
                errors.append(f"{lock_id}: 连续性锁指向不存在的 IMG 条目: {image_id}")
                continue
            image_prompt = image_prompts[image_id]
            if image_prompt is not None and not _carries_surface(image_prompt, surface):
                errors.append(f"{lock_id}: {image_id} 可复制提示词缺少锁面")


def _storyboard_head(storyboard: str) -> str:
    """The part of 分镜.md that belongs to the episode rather than to a shot."""
    match = re.search(r"^## SHOT-", storyboard, re.MULTILINE)
    return storyboard if match is None else storyboard[: match.start()]


def _shot_sources(
    value: str, owner: str, scenes: "frozenset", errors: list[str]
) -> list:
    """Resolve one shot's 来源 against the scene ids that exist in 剧本.md."""
    plain = _plain(value)
    if not plain:
        errors.append(f"{owner}: 缺少来源字段")
        return []
    # 《镜头手艺》 asks for the scene id plus a short quote of the source line,
    # and a quote contains commas of its own. So the ids are read out of the
    # field rather than by cutting it into pieces on punctuation.
    if SCENE_ID_RE.match(plain) is None:
        errors.append(
            f"{owner}: 来源必须以《剧本.md》的场景 ID 开头: {_excerpt(plain, 24)}"
        )
        return []
    resolved = []
    for scene_id in _unique(SCENE_ID_RE.findall(plain)):
        if scene_id not in scenes:
            errors.append(f"{owner}: 来源场景不在《剧本.md》中: {scene_id}")
        else:
            resolved.append(scene_id)
    return resolved


def _unfilmed_scenes(
    storyboard: str, scenes: "frozenset", errors: list[str]
) -> dict:
    """Scenes 分镜.md declares it deliberately does not film, with reasons."""
    head = _storyboard_head(storyboard)
    declared: dict = {}
    # The line belongs to the episode, not to a shot. Placed inside a SHOT
    # section it would be read as that shot's field and quietly do nothing,
    # which looks exactly like forgetting to write it.
    if len(UNFILMED_LINE_RE.findall(storyboard)) != len(
        UNFILMED_LINE_RE.findall(head)
    ):
        errors.append("分镜.md: 未拍场次要写在第一个 SHOT 之前的正文开头")
    for value in UNFILMED_LINE_RE.findall(head):
        value = value.strip()
        entries = UNFILMED_ENTRY_RE.findall(value)
        remainder = UNFILMED_ENTRY_RE.sub("", value).strip("；;。 ")
        if not entries or remainder:
            errors.append(
                "分镜.md: 未拍场次必须写成 <场景 ID>（理由：……），多项用；分隔"
            )
            continue
        for scene_id, reason in entries:
            if not reason.strip():
                errors.append(f"分镜.md: 未拍场次缺少理由: {scene_id}")
            elif scene_id not in scenes:
                errors.append(f"分镜.md: 未拍场次不在《剧本.md》中: {scene_id}")
            else:
                declared[scene_id] = reason.strip()
    return declared


def _spoken_key(value: str) -> str:
    """Drop punctuation and spacing so quoting choices do not decide identity."""
    return re.sub(r"[^\u3400-\u9fff0-9A-Za-z]", "", value)


def _quoted_speech(prompt: str) -> list:
    # The body is one prompt; a quote that happens to wrap across lines is the
    # same quote. Chinese does not separate words with spaces, so rejoining the
    # lines reconstructs the run exactly.
    joined = prompt.replace("\n", "")
    runs: list = []
    for match in SPOKEN_SPAN_RE.finditer(joined):
        span = next((group for group in match.groups() if group), "")
        # An unpaired quotation mark can pair with one much later and swallow
        # the narration between them. That span is not a quotation, so it is
        # not evidence of an invented line either.
        if len(span) > 200:
            continue
        runs.extend(SPOKEN_RUN_RE.findall(span))
    return runs


def _check_spoken_lines(
    prompt: str, owner: str, sources: "tuple", errors: list[str]
) -> None:
    """Quoted Chinese in a copyable body has to come from an upstream document.

    The video stage already owes the screenplay verbatim dialogue. Checking it
    here is what stops a prompt from writing the line the shot wishes the
    character had said. 视觉设定.md and 分镜.md count as sources too: a location
    or prop entry carries the exact text on a sign, and a shot's own fields
    carry the on-screen text and sound this shot was designed around. What no
    upstream document contains is a line the prompt made up.
    """
    for run in _quoted_speech(prompt):
        key = _spoken_key(run)
        if key and not any(key in source for source in sources):
            errors.append(
                f"{owner}: 可复制提示词的引文在《剧本.md》《视觉设定.md》《分镜.md》里都找不到: "
                f"{_excerpt(run, 24)}"
            )


def validate_episode(episode: Path, project_root: Optional[Path] = None) -> list[str]:
    """Return all deterministic contract errors for ``episode``."""
    episode = episode.resolve()
    project_root = (project_root or episode.parent.parent).resolve()
    errors: list[str] = []
    missing = [name for name in REQUIRED_DOCUMENTS if not (episode / name).is_file()]
    if missing:
        return [f"缺少创作文档: {', '.join(missing)}"]

    images = (episode / "图片提示词.md").read_text(encoding="utf-8")
    storyboard = (episode / "分镜.md").read_text(encoding="utf-8")
    video = (episode / "视频提示词.md").read_text(encoding="utf-8")
    visual = (episode / "视觉设定.md").read_text(encoding="utf-8")
    screenplay = (episode / "剧本.md").read_text(encoding="utf-8")
    scene_headings = SCENE_HEADING_RE.findall(screenplay)
    scenes = frozenset(scene_headings)
    if not scenes:
        errors.append(
            "剧本.md: 没有可解析的场景标题；场景标题写成 ## <场景 ID> 内|外 · 地点 · 时间"
        )
    if len(scene_headings) != len(scenes):
        errors.append("剧本.md: 场景 ID 重复")
    spoken_sources = (
        _spoken_key(screenplay),
        _spoken_key(visual),
        _spoken_key(storyboard),
    )
    unfilmed = _unfilmed_scenes(storyboard, scenes, errors)
    claimed_scenes: set = set()
    locks = _continuity_locks(visual, errors)
    visual_entries = _visual_entries(visual, errors)
    other_headings = {
        match.group(2).strip(): match.group(1).strip()
        for match in OTHER_SETTING_HEADING_RE.finditer(visual)
        if match.group(1).strip() not in VISUAL_CATEGORIES
    }
    shots = _sections(storyboard, "SHOT")
    image_pairs = re.findall(r"^## (IMG-[A-Z0-9-]+) · (.+)$", images, re.MULTILINE)
    image_headings = dict(image_pairs)
    # A planned picture names the board or keyframe that already decides what it
    # depicts, so both indexes have to exist before any declaration is read.
    entries = EntryIndex(images=image_headings, shots=frozenset(shots))
    all_image_headings = re.findall(r"^## (IMG-[A-Z0-9-]+)\b", images, re.MULTILINE)
    if len(image_pairs) != len(all_image_headings):
        errors.append("图片提示词.md: IMG 标题必须包含中文名称")
    if len(image_pairs) != len(image_headings):
        errors.append("图片提示词.md: IMG 标题 ID 重复")
    for image_id, label in image_pairs:
        if not re.search(r"[\u3400-\u9fff]", label):
            errors.append(f"{image_id}: IMG 标题缺少中文名称")
    image_matches = list(re.finditer(r"^## (IMG-[A-Z0-9-]+)\b", images, re.MULTILINE))
    image_prompts: dict[str, Optional[str]] = {}
    for index, match in enumerate(image_matches):
        body = images[
            match.start() : image_matches[index + 1].start()
            if index + 1 < len(image_matches)
            else None
        ]
        reference_value = _fields(body, owner=match.group(1), errors=errors).get(
            "参考", ""
        )
        if not reference_value:
            errors.append(f"{match.group(1)}: 缺少参考字段")
        elif _is_no_external_reference(reference_value):
            pass
        elif "REF-" in reference_value:
            _references(
                reference_value, match.group(1), project_root, errors, entries
            )
        else:
            errors.append(
                f"{match.group(1)}: 参考必须声明无外部参考或使用完整 REF 语法"
            )
        image_prompt = _copyable_prompt(body)
        image_prompts[match.group(1)] = image_prompt
        if image_prompt is None:
            errors.append(
                f"{match.group(1)}: 缺少唯一且非空的可复制提示词"
                + _copyable_prompt_cause(body)
            )

    motions = _sections(video, "MOTION")
    shot_ids = re.findall(r"^## (SHOT-[A-Z0-9-]+)\b", storyboard, re.MULTILINE)
    motion_ids = re.findall(r"^## (MOTION-[A-Z0-9-]+)\b", video, re.MULTILINE)
    if len(shot_ids) != len(set(shot_ids)):
        errors.append("分镜.md: SHOT 标题 ID 重复")
    if len(motion_ids) != len(set(motion_ids)):
        errors.append("视频提示词.md: MOTION 标题 ID 重复")
    named_shots = re.findall(
        r"^## (SHOT-[A-Z0-9-]+) · ([^\n]+)$", storyboard, re.MULTILINE
    )
    named_motions = re.findall(
        r"^## (MOTION-[A-Z0-9-]+) · ([^\n]+)$", video, re.MULTILINE
    )
    if len(named_shots) != len(shot_ids):
        errors.append("分镜.md: SHOT 标题必须包含中文名称")
    if len(named_motions) != len(motion_ids):
        errors.append("视频提示词.md: MOTION 标题必须包含中文名称")
    for shot_id, label in named_shots:
        if not re.search(r"[\u3400-\u9fff]", label):
            errors.append(f"{shot_id}: SHOT 标题缺少中文名称")
    for motion_id, label in named_motions:
        if not re.search(r"[\u3400-\u9fff]", label):
            errors.append(f"{motion_id}: MOTION 标题缺少中文名称")
    if not shots:
        errors.append("分镜.md: 没有 SHOT 条目")
    if not motions:
        errors.append("视频提示词.md: 没有 MOTION 条目")

    motion_by_shot: dict[str, tuple[str, str, Optional[str]]] = {}
    motion_duration: dict[str, str] = {}
    for motion_id, body in motions.items():
        fields = _fields(body, owner=motion_id, errors=errors)
        shot_id = _plain(fields.get("分镜", ""))
        motion_duration[shot_id] = _plain(fields.get("时长", ""))
        copyable_prompt = _copyable_prompt(body)
        if not shot_id:
            errors.append(f"{motion_id}: 缺少分镜字段")
        elif shot_id in motion_by_shot:
            errors.append(f"{motion_id}: 分镜 {shot_id} 被多个 MOTION 引用")
        else:
            motion_by_shot[shot_id] = (motion_id, body, copyable_prompt)
        if motion_id.removeprefix("MOTION-") != shot_id.removeprefix("SHOT-"):
            errors.append(f"{motion_id}: ID 必须与分镜 {shot_id} 一一对应")
        if copyable_prompt is None:
            errors.append(
                f"{motion_id}: 缺少唯一且非空的可复制提示词"
                + _copyable_prompt_cause(body)
            )

    if set(motion_by_shot) != set(shots):
        errors.append("分镜.md/视频提示词.md: SHOT 与 MOTION 未一一对应")

    for shot_id, shot_body in shots.items():
        fields = _fields(shot_body, owner=shot_id, errors=errors)
        source_value = fields.get("来源", "")
        claimed_scenes.update(
            _shot_sources(source_value, shot_id, scenes, errors)
        )
        shot_seconds = _declared_seconds(_plain(fields.get("时长", "")))
        motion_seconds = _declared_seconds(motion_duration.get(shot_id, ""))
        if (
            shot_seconds is not None
            and motion_seconds is not None
            and shot_seconds != motion_seconds
        ):
            # 时长 is the value actually sent to the execution end and the term
            # VID-04/VID-13 arithmetic is built on. A downstream copy that drifts
            # from its accepted upstream is invisible in every other check.
            errors.append(
                f"{shot_id}: 分镜时长 {shot_seconds:g} 秒与视频提示词 "
                f"{motion_seconds:g} 秒不一致；视频提示词只能原样照抄已接受的镜头时长"
            )
        image_value = fields.get("图片提示词项", "")
        if not image_value:
            errors.append(f"{shot_id}: 缺少图片提示词项字段")
        image_refs = IMG_RE.findall(image_value)
        image_remainder = IMG_RE.sub("", image_value).strip("；。 ")
        if not _is_none(image_value) and (
            len(image_refs) != len(re.findall(r"\bIMG-[A-Z0-9-]+\b", image_value))
            or image_remainder
        ):
            errors.append(f"{shot_id}: 图片提示词项语法不完整")
        for image_id, label, _ in image_refs:
            if image_id not in image_headings:
                errors.append(f"{shot_id}: IMG 标题不存在: {image_id}")
            elif label != image_headings[image_id]:
                errors.append(f"{shot_id}: IMG 中文名称与标题不一致: {image_id}")

        shot_input = fields.get("输入参考图", "")
        if not shot_input:
            errors.append(f"{shot_id}: 缺少输入参考图字段")
        _references(shot_input, shot_id, project_root, errors, entries)

        basis_value = fields.get("视觉依据", "")
        if not basis_value:
            errors.append(f"{shot_id}: 缺少视觉依据字段")
            basis = VisualBasis(False, set(), set())
        else:
            basis = _visual_basis(
                basis_value, shot_id, visual_entries, other_headings, errors
            )
        # The keyframe is the only place the frame's contents exist as text, so a
        # shot without one would make the coverage check below vacuous.
        keyframe = _copyable_prompt(shot_body, heading=r"冻结关键帧提示词")
        if keyframe is None:
            errors.append(
                f"{shot_id}: 缺少唯一且非空的冻结关键帧提示词"
                + _copyable_prompt_cause(shot_body, heading=r"冻结关键帧提示词")
            )
        elif basis.parsed:
            _check_named_coverage(
                keyframe, shot_id, "冻结关键帧提示词", basis, visual_entries, errors
            )
        # The closing keyframe is optional — many shots end where they began —
        # but a heading that is present and malformed is a defect, not a shot
        # that quietly has no closing frame.
        if re.search(r"^### 收尾关键帧提示词\s*$", shot_body, re.MULTILINE):
            closing = _copyable_prompt(shot_body, heading=r"收尾关键帧提示词")
            if closing is None:
                errors.append(
                    f"{shot_id}: 收尾关键帧提示词不是唯一且非空的可复制正文"
                    + _copyable_prompt_cause(shot_body, heading=r"收尾关键帧提示词")
                )
            elif basis.parsed:
                _check_named_coverage(
                    closing, shot_id, "收尾关键帧提示词", basis, visual_entries, errors
                )
        if basis.parsed and source_value:
            # A shot's 来源 is its claim on the screenplay: whatever it quotes,
            # this shot is the one that films it. Quoting an action performed by
            # someone the frame never shows takes that action off everyone's
            # list — no other shot claims it, and nothing reports it missing, so
            # it simply never gets filmed. Same rule as the keyframe: in frame,
            # or declared 画外.
            _check_named_coverage(
                source_value,
                shot_id,
                "来源引文",
                basis,
                visual_entries,
                errors,
                by_entry_name=True,
            )

        motion = motion_by_shot.get(shot_id)
        if not motion:
            continue
        motion_id, motion_body, copyable_prompt = motion
        motion_fields = _fields(motion_body, owner=motion_id, errors=errors)
        if copyable_prompt is not None:
            _check_spoken_lines(
                copyable_prompt, motion_id, spoken_sources, errors
            )
        motion_input = motion_fields.get("输入参考图", "")
        _references(motion_input, motion_id, project_root, errors, entries)
        if _plain(motion_input) != _plain(shot_input):
            errors.append(f"{motion_id}: 输入参考图与 {shot_id} 不一致")

        if _has_pending_references(shot_input):
            errors.append(f"{motion_id}: 仍有待补参考图，不能生成最终视频提示词")

        # A `PLAN-...` slot is not a file, but the prompt is still written for
        # image-conditioned generation: the creator attaches those pictures at
        # generation time. What decides 生成方式 is whether the shot sends
        # pictures at all, not whether this suite can read them.
        sends_pictures = not _is_none(shot_input)
        expected_mode = "图生视频" if sends_pictures else "文生视频"
        if _plain(motion_fields.get("生成方式", "")) != expected_mode:
            errors.append(f"{motion_id}: 生成方式应为{expected_mode}")
        if not sends_pictures:
            if not _is_explicit_text_to_video(shot_input):
                errors.append(
                    f"{motion_id}: 无真实输入参考图时不能静默降级为文生视频；"
                    "请先绑定已有图片、列出待补图片，或记录创作者已明确选择文生视频"
                )
            anchor = _plain(motion_fields.get("静态视觉锚点", ""))
            if not anchor or anchor == "无":
                errors.append(f"{motion_id}: 文生视频缺少静态视觉锚点")
            if (
                copyable_prompt is not None
                and anchor
                and anchor != "无"
                and anchor not in copyable_prompt
            ):
                errors.append(f"{motion_id}: 可复制提示词没有包含静态视觉锚点")
            # In text-to-video the anchor carries the appearance the keyframe
            # would otherwise have carried, so it is the same claim about the
            # same frame and answers to the same visual basis. The rest of the
            # motion body is not checked: it may legitimately name an offscreen
            # speaker, which SHT-22 excludes.
            if anchor and anchor

#!/usr/bin/env python3
"""A loopback-only HTTP dashboard for short-drama project text files."""

from __future__ import annotations

import argparse
import contextlib
import errno
import hmac
import hashlib
import importlib.util
import ipaddress
import json
import os
import re
import secrets
import signal
import stat
import subprocess
import sys
import threading
import time
import uuid
import webbrowser
from collections.abc import Iterator, Mapping
from http import HTTPStatus
from http.cookies import SimpleCookie
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path, PurePosixPath
from types import ModuleType
from typing import Any, BinaryIO, Union
from urllib.parse import parse_qs, unquote, urlencode, urlsplit


# Creators run these scripts on whatever interpreter their machine provides, so
# an unsupported version must say so instead of failing inside an import.
MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit(
        "short-drama needs Python {}.{} or newer; this interpreter is {}.{}".format(
            *MINIMUM_PYTHON, sys.version_info.major, sys.version_info.minor
        )
    )

TEXT_EXTENSIONS = frozenset({".md", ".json", ".jsonl", ".txt", ".srt", ".ass"})
MEDIA_EXTENSIONS = frozenset(
    {
        ".png",
        ".jpg",
        ".jpeg",
        ".webp",
        ".gif",
        ".mp4",
        ".webm",
        ".mov",
        ".wav",
        ".mp3",
        ".m4a",
        ".aac",
        ".flac",
        ".opus",
    }
)
MEDIA_TYPES = {
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".webp": "image/webp",
    ".gif": "image/gif",
    ".mp4": "video/mp4",
    ".webm": "video/webm",
    ".mov": "video/quicktime",
    ".wav": "audio/wav",
    ".mp3": "audio/mpeg",
    ".m4a": "audio/mp4",
    ".aac": "audio/aac",
    ".flac": "audio/flac",
    ".opus": "audio/ogg",
}
DEFAULT_MAX_DEPTH = 8
DEFAULT_MAX_NODES = 2_000
DEFAULT_MAX_FILE_BYTES = 2 * 1024 * 1024
DEFAULT_MAX_MEDIA_BYTES = 256 * 1024 * 1024
MAX_JSON_EXPANSION = 6
REQUEST_OVERHEAD_BYTES = 64 * 1024
SKILL_ROOT = Path(__file__).resolve().parents[1]
SESSION_SCHEMA = "1.0"
SESSION_RELATIVE = PurePosixPath(".short-drama/dashboard.json")
SESSION_LOCK_SUFFIX = ".lock"
DETACH_TIMEOUT_SECONDS = 20.0
STOP_TIMEOUT_SECONDS = 10.0
STATIC_ROOT = SKILL_ROOT / "assets/dashboard"

# Windows opens files in text mode unless told otherwise, which would rewrite
# every newline and break the SHA-256 versions the editor round-trips on.
BINARY = getattr(os, "O_BINARY", 0)
NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
# POSIX pins directories by descriptor, which makes traversal free of races by
# construction. Everywhere else the same guarantee is approximated by checking
# each component and re-checking identity; see ``_PathDirectory``.
SECURE_DIR_FD = (
    os.name != "nt"
    and bool(getattr(os, "O_DIRECTORY", 0))
    and bool(NOFOLLOW)
    and os.open in os.supports_dir_fd
)
# Windows refuses to replace a file another process still holds open, and an
# editor or a search indexer is enough to hold one for a moment.
REPLACE_ATTEMPTS = 6


class DashboardError(Exception):
    def __init__(self, status: int, message: str) -> None:
        super().__init__(message)
        self.status = status
        self.message = message


def _is_loopback(host: str) -> bool:
    if host.casefold() == "localhost":
        return True
    try:
        return ipaddress.ip_address(host).is_loopback
    except ValueError:
        return False


def load_project_tool(root: Path) -> ModuleType:
    """Dynamically load the suite's canonical project status implementation."""
    root = root.resolve()
    installed_script = root / "scripts/project_tool.py"
    repository_script = root / "skills/short-drama/scripts/project_tool.py"
    script = installed_script if installed_script.is_file() else repository_script
    spec = importlib.util.spec_from_file_location("dashboard_project_tool", script)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"cannot load project tool: {script}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _version(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _validate_structured_text(path: PurePosixPath, content: str) -> None:
    suffix = path.suffix.casefold()
    if suffix == ".json":
        try:
            json.loads(content)
        except json.JSONDecodeError as exc:
            raise DashboardError(
                HTTPStatus.BAD_REQUEST,
                f"JSON is invalid at line {exc.lineno}, column {exc.colno}",
            ) from exc
        return
    if suffix != ".jsonl":
        return
    for line_number, line in enumerate(content.splitlines(), start=1):
        if not line.strip():
            continue
        try:
            json.loads(line)
        except json.JSONDecodeError as exc:
            raise DashboardError(
                HTTPStatus.BAD_REQUEST,
                f"JSONL line {line_number} is invalid at column {exc.colno}",
            ) from exc


def _is_link_or_reparse(details: os.stat_result) -> bool:
    """Say whether a stat result describes something that must not be traversed.

    ``S_ISLNK`` alone is not enough on Windows: junctions and every other
    reparse point redirect just as well, and need no privilege to create.
    """

    attributes = getattr(details, "st_file_attributes", 0)
    reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
    return stat.S_ISLNK(details.st_mode) or bool(attributes & reparse_flag)


def _entry_is_link(entry: os.DirEntry) -> bool:
    try:
        if entry.is_symlink():
            return True
        if os.name != "nt":
            return False
        return _is_link_or_reparse(entry.stat(follow_symlinks=False))
    except OSError:
        return True


def _identity(details: os.stat_result) -> tuple[int, int]:
    return details.st_dev, details.st_ino


def _same_object(before: tuple[int, int], after: tuple[int, int]) -> bool:
    # A volume that reports no inode gives nothing to compare, so treat the
    # check as inapplicable there instead of failing every read on it.
    if not before[1] or not after[1]:
        return True
    return before == after


class _DescriptorDirectory:
    """A directory pinned by descriptor, the POSIX backend.

    Every operation names a file relative to a descriptor the request already
    holds, so the pin is the directory itself rather than a name that resolves
    to one, and nothing in between can be swapped mid-request.
    """

    __slots__ = ("_fd",)

    contract = (
        "project_status_at",
        "is_protected_project_text",
        "coordinated_project_text_edit_at",
        "project_path_lifecycle_at",
    )

    def __init__(self, descriptor: int) -> None:
        self._fd = descriptor

    @classmethod
    def open_root(cls, path: Path) -> _DescriptorDirectory:
        return cls(os.open(path, os.O_RDONLY | os.O_DIRECTORY | NOFOLLOW))

    def reopen(self) -> _DescriptorDirectory:
        # A directory cursor belongs to the open file description, so dup()
        # would still let concurrent requests advance one another's scan.
        # Opening "." gives every caller an independent description.
        return _DescriptorDirectory(
            os.open(".", os.O_RDONLY | os.O_DIRECTORY | NOFOLLOW, dir_fd=self._fd)
        )

    def child(self, name: str) -> _DescriptorDirectory:
        return _DescriptorDirectory(
            os.open(
                name, os.O_RDONLY | os.O_DIRECTORY | NOFOLLOW, dir_fd=self._fd
            )
        )

    @contextlib.contextmanager
    def scandir(self) -> Iterator[Iterator[os.DirEntry]]:
        with os.scandir(self._fd) as entries:
            yield entries

    def open_regular(self, name: str, flags: int, mode: int = 0o666) -> int:
        return os.open(name, flags | NOFOLLOW | BINARY, mode, dir_fd=self._fd)

    def replace(self, source: str, target: str) -> None:
        os.replace(source, target, src_dir_fd=self._fd, dst_dir_fd=self._fd)

    def unlink(self, name: str) -> None:
        os.unlink(name, dir_fd=self._fd)

    def sync(self) -> None:
        os.fsync(self._fd)

    def close(self) -> None:
        if self._fd >= 0:
            os.close(self._fd)
            self._fd = -1

    def status(self, tool: ModuleType, project_root: str) -> dict[str, Any]:
        return tool.project_status_at(self._fd, project_root=project_root)

    def lifecycle(self, tool: ModuleType, relative: str) -> dict[str, str] | None:
        return tool.project_path_lifecycle_at(self._fd, relative)

    def coordinated_edit(
        self, tool: ModuleType, relative: str, expected_version: str
    ) -> Any:
        return tool.coordinated_project_text_edit_at(
            self._fd, relative, expected_version
        )


class _PathDirectory:
    """A directory pinned by verified path, for platforms without ``openat``.

    Windows cannot name a file relative to an open directory, so the guarantee
    is rebuilt from two checks: every component is inspected with ``os.lstat``,
    which never follows, and refused if it is a link or any other reparse
    point; then the pinned directory's identity is re-checked before each
    operation and an opened file's identity against what was inspected.

    This is weaker than a descriptor -- a swap timed between the two checks is
    not caught. It is the boundary this suite already accepts for Windows
    ``verify``, and the adversary it admits, a local process running as the
    creator, can open these files directly anyway.
    """

    __slots__ = ("_path", "_identity")

    contract = (
        "project_status_from_root",
        "is_protected_project_text",
        "coordinated_project_text_edit",
        "project_path_lifecycle",
    )

    def __init__(self, path: Path, identity: tuple[int, int]) -> None:
        self._path = path
        self._identity = identity

    @classmethod
    def open_root(cls, path: Path) -> _PathDirectory:
        details = os.lstat(path)
        if _is_link_or_reparse(details) or not stat.S_ISDIR(details.st_mode):
            raise OSError(
                errno.ENOTDIR, "not a plain directory", str(path)
            )
        return cls(path, _identity(details))

    def _pinned(self) -> Path:
        details = os.lstat(self._path)
        if _is_link_or_reparse(details) or not stat.S_ISDIR(details.st_mode):
            raise OSError(
                errno.ENOTDIR, "pinned directory is no longer a directory",
                str(self._path),
            )
        if not _same_object(self._identity, _identity(details)):
            raise OSError(
                errno.ENOENT, "pinned directory was replaced", str(self._path)
            )
        return self._path

    def reopen(self) -> _PathDirectory:
        return _PathDirectory(self._pinned(), self._identity)

    def child(self, name: str) -> _PathDirectory:
        return _PathDirectory.open_root(self._pinned() / name)

    @contextlib.contextmanager
    def scandir(self) -> Iterator[Iterator[os.DirEntry]]:
        with os.scandir(self._pinned()) as entries:
            yield entries

    def open_regular(self, name: str, flags: int, mode: int = 0o666) -> int:
        target = self._pinned() / name
        expected: tuple[int, int] | None = None
        if not flags & os.O_CREAT:
            # O_CREAT|O_EXCL already refuses an existing name, so only an open
            # of something that must already be there needs inspecting first.
            details = os.lstat(target)
            if _is_link_or_reparse(details) or not stat.S_ISREG(details.st_mode):
                raise OSError(errno.EPERM, "not a plain file", str(target))
            expected = _identity(details)
        descriptor = os.open(target, flags | BINARY, mode)
        if expected is None:
            return descriptor
        try:
            if not _same_object(expected, _identity(os.fstat(descriptor))):
                raise OSError(
                    errno.ENOENT, "file was replaced while opening", str(target)
                )
        except BaseException:
            os.close(descriptor)
            raise
        return descriptor

    def replace(self, source: str, target: str) -> None:
        root = self._pinned()
        delay = 0.01
        for remaining in range(REPLACE_ATTEMPTS - 1, -1, -1):
            try:
                os.replace(root / source, root / target)
                return
            except PermissionError:
                if not remaining:
                    raise
                time.sleep(delay)
                delay *= 2

    def unlink(self, name: str) -> None:
        os.unlink(self._pinned() / name)

    def sync(self) -> None:
        # Windows has no directory handle to flush; the file itself was already
        # fsynced before the replace, which is the same guarantee the CLI makes.
        return None

    def close(self) -> None:
        return None

    def status(self, tool: ModuleType, project_root: str) -> dict[str, Any]:
        return tool.project_status_from_root(
            self._pinned(), project_root=project_root
        )

    def lifecycle(self, tool: ModuleType, relative: str) -> dict[str, str] | None:
        return tool.project_path_lifecycle(self._pinned(), relative)

    def coordinated_edit(
        self, tool: ModuleType, relative: str, expected_version: str
    ) -> Any:
        return tool.coordinated_project_text_edit(
            self._pinned(), relative, expected_version
        )


Directory = Union[_DescriptorDirectory, _PathDirectory]


def directory_backend() -> type[_DescriptorDirectory] | type[_PathDirectory]:
    return _DescriptorDirectory if SECURE_DIR_FD else _PathDirectory


@contextlib.contextmanager
def _open_parent_directory(
    root: Directory, relative: PurePosixPath
) -> Iterator[tuple[Directory, str]]:
    current = root.reopen()
    try:
        for part in relative.parts[:-1]:
            child = current.child(part)
            current.close()
            current = child
        yield current, relative.name
    finally:
        current.close()


class ProjectStore:
    def __init__(
        self,
        workspace: Path,
        project_tool: ModuleType,
        *,
        max_depth: int = DEFAULT_MAX_DEPTH,
        max_nodes: int = DEFAULT_MAX_NODES,
        max_file_bytes: int = DEFAULT_MAX_FILE_BYTES,
        max_media_bytes: int = DEFAULT_MAX_MEDIA_BYTES,
    ) -> None:
        workspace = workspace.expanduser().resolve(strict=True)
        if not workspace.is_dir():
            raise NotADirectoryError(workspace)
        self.workspace = workspace
        self.project_tool = project_tool
        self.max_depth = max_depth
        self.max_nodes = max_nodes
        self.max_file_bytes = max_file_bytes
        self.max_media_bytes = max_media_bytes
        # JSON can encode one content byte as a six-byte ``\u00xx`` escape.
        # Size the transport envelope from the configured file limit so a
        # valid text file is never rejected only because of JSON escaping.
        self.max_request_bytes = (
            max_file_bytes * MAX_JSON_EXPANSION + REQUEST_OVERHEAD_BYTES
        )
        # Which four entry points this server invokes depends on how the
        # platform can pin a directory: by descriptor, or by verified path.
        self.backend = directory_backend()
        missing = [
            name
            for name in self.backend.contract
            if not callable(getattr(project_tool, name, None))
        ]
        if missing:
            raise RuntimeError(
                "project tool does not support the Dashboard contract: "
                + ", ".join(missing)
            )
        self._workspace = self.backend.open_root(workspace)
        self._write_lock = threading.Lock()

    def close(self) -> None:
        self._workspace.close()

    @staticmethod
    def _project_id(relative: str) -> str:
        return hashlib.sha256(relative.encode("utf-8")).hexdigest()[:20]

    def discover(self) -> tuple[list[dict[str, Any]], list[str]]:
        projects: list[dict[str, Any]] = []
        warnings: list[str] = []
        nodes = 0
        depth_truncated = False
        node_truncated = False

        def walk(directory: Directory, parts: tuple[str, ...], depth: int) -> None:
            nonlocal depth_truncated, node_truncated, nodes
            if nodes >= self.max_nodes:
                return
            try:
                with directory.scandir() as iterator:
                    entries = sorted(iterator, key=lambda item: item.name.casefold())
            except OSError as exc:
                warnings.append(f"无法读取 {'/'.join(parts) or '.'}: {exc}")
                return
            for entry in entries:
                if nodes >= self.max_nodes:
                    node_truncated = True
                    return
                nodes += 1
                if _entry_is_link(entry):
                    continue
                if (
                    entry.is_file(follow_symlinks=False)
                    and entry.name == "short-drama.json"
                ):
                    relative = "/".join(parts) or "."
                    title = "未命名短剧"
                    try:
                        raw, _mode = self._read_regular(directory, entry.name)
                        manifest = json.loads(raw.decode("utf-8"))
                        candidate = manifest.get("title") if isinstance(manifest, dict) else None
                        if isinstance(candidate, str) and candidate.strip():
                            title = " ".join(candidate.split())[:200]
                    except (DashboardError, OSError, UnicodeError, json.JSONDecodeError):
                        pass
                    projects.append(
                        {
                            "id": self._project_id(relative),
                            "path": relative,
                            "title": title,
                        }
                    )
                    continue
                if not entry.is_dir(follow_symlinks=False):
                    continue
                if depth >= self.max_depth:
                    depth_truncated = True
                    continue
                try:
                    child = directory.child(entry.name)
                except OSError:
                    continue
                try:
                    walk(child, (*parts, entry.name), depth + 1)
                finally:
                    child.close()

        # Reopen rather than share: on POSIX a directory cursor belongs to the
        # open file description, so two concurrent discoveries walking one
        # handle would advance each other's scan.
        discovery = self._workspace.reopen()
        try:
            walk(discovery, (), 0)
        finally:
            discovery.close()
        if node_truncated:
            warnings.append(f"项目发现达到节点上限 {self.max_nodes}，结果已截断")
        if depth_truncated:
            warnings.append(f"项目发现达到深度上限 {self.max_depth}，更深目录已截断")
        return projects, warnings

    @contextlib.contextmanager
    def _pinned_project(self, project_id: str) -> Iterator[tuple[Directory, Path]]:
        """Pin the project root and report its creator-facing path.

        Every parent is opened from the pinned workspace with the backend's
        no-follow guarantee, so a directory swapped for a symlink mid-request
        fails instead of redirecting the operation. The yielded path is for
        display only — no caller reads through it.
        """

        selected = next(
            (item for item in self.discover()[0] if item["id"] == project_id), None
        )
        if selected is None:
            raise DashboardError(HTTPStatus.NOT_FOUND, "project not found")
        display = (
            self.workspace
            if selected["path"] == "."
            else self.workspace / selected["path"]
        )
        # Reopen instead of sharing the workspace handle: on POSIX that handle
        # carries a seek position other requests would advance, including when
        # the workspace itself is the selected project.
        directory = self._workspace.reopen()
        marker = -1
        try:
            for part in PurePosixPath(selected["path"]).parts:
                if part == ".":
                    continue
                child = directory.child(part)
                directory.close()
                directory = child
            marker = directory.open_regular("short-drama.json", os.O_RDONLY)
            if not stat.S_ISREG(os.fstat(marker).st_mode):
                raise OSError("project manifest is not a regular file")
        except OSError as exc:
            if marker >= 0:
                os.close(marker)
            directory.close()
            raise DashboardError(
                HTTPStatus.FORBIDDEN, "project root cannot be opened safely"
            ) from exc
        os.close(marker)
        try:
            yield directory, display
        finally:
            directory.close()

    def status(self, project_id: str) -> dict[str, Any]:
        with self._pinned_project(project_id) as (directory, root):
            return directory.status(self.project_tool, str(root))

    @staticmethod
    def _safe_relative(relative: str) -> PurePosixPath:
        raw = unquote(relative).replace("\\", "/")
        pure = PurePosixPath(raw)
        if (
            not raw
            or pure.is_absolute()
            or any(part in ("", ".", "..") for part in pure.parts)
        ):
            raise DashboardError(HTTPStatus.BAD_REQUEST, "unsafe project-relative path")
        return pure

    def _is_protected(self, relative: PurePosixPath) -> bool:
        return bool(
            self.project_tool.is_protected_project_text(relative.as_posix())
        )

    def tree(self, project_id: str) -> dict[str, Any]:
        with self._pinned_project(project_id) as (directory, _root):
            return self._tree_from_root(directory)

    def _tree_from_root(self, directory: Directory) -> dict[str, Any]:
        nodes = 0
        warnings: list[str] = []
        depth_truncated = False
        node_truncated = False
        oversized_text = 0
        oversized_media = 0

        def scan(
            parent: Directory, parts: tuple[str, ...], depth: int
        ) -> list[dict[str, Any]]:
            nonlocal depth_truncated, node_truncated, nodes
            nonlocal oversized_media, oversized_text
            children: list[dict[str, Any]] = []
            try:
                with parent.scandir() as iterator:
                    entries = sorted(
                        iterator,
                        key=lambda item: (
                            not item.is_dir(follow_symlinks=False),
                            item.name.casefold(),
                        ),
                    )
            except OSError as exc:
                warnings.append(f"无法读取 {'/'.join(parts) or '.'}: {exc}")
                return children
            for entry in entries:
                if _entry_is_link(entry):
                    continue
                relative = "/".join((*parts, entry.name))
                if parts == () and entry.name.casefold() == ".short-drama":
                    continue
                if nodes >= self.max_nodes:
                    node_truncated = True
                    break
                nodes += 1
                if entry.is_dir(follow_symlinks=False):
                    node: dict[str, Any] = {
                        "name": entry.name,
                        "path": relative,
                        "type": "directory",
                        "children": [],
                    }
                    if depth < self.max_depth:
                        try:
                            child = parent.child(entry.name)
                        except OSError:
                            continue
                        try:
                            node["children"] = scan(
                                child, (*parts, entry.name), depth + 1
                            )
                        finally:
                            child.close()
                    else:
                        depth_truncated = True
                        node["truncated"] = True
                    children.append(node)
                    continue
                suffix = PurePosixPath(entry.name).suffix.casefold()
                if suffix not in TEXT_EXTENSIONS and suffix not in MEDIA_EXTENSIONS:
                    continue
                try:
                    size = entry.stat(follow_symlinks=False).st_size
                except OSError:
                    continue
                limit = (
                    self.max_file_bytes
                    if suffix in TEXT_EXTENSIONS
                    else self.max_media_bytes
                )
                if size > limit:
                    if suffix in TEXT_EXTENSIONS:
                        oversized_text += 1
                    else:
                        oversized_media += 1
                children.append(
                    {
                        "name": entry.name,
                        "path": relative,
                        "type": "text" if suffix in TEXT_EXTENSIONS else "media",
                        "size": size,
                        "oversize": size > limit,
                        "writable": suffix in TEXT_EXTENSIONS
                        and not self._is_protected(PurePosixPath(relative)),
                    }
                )
            return children

        tree = scan(directory, (), 0)
        if node_truncated:
            warnings.append(f"文件树达到节点上限 {self.max_nodes}，结果已截断")
        if depth_truncated:
            warnings.append(f"文件树达到深度上限 {self.max_depth}，更深目录已截断")
        if oversized_text:
            warnings.append(
                f"{oversized_text} 个文本文件超过大小上限 {self.max_file_bytes} bytes，内容预览已禁用"
            )
        if oversized_media:
            warnings.append(
                f"{oversized_media} 个媒体文件超过大小上限 {self.max_media_bytes} bytes，媒体预览已禁用"
            )
        return {
            "tree": tree,
            "warnings": warnings,
            "limits": {
                "depth": self.max_depth,
                "nodes": self.max_nodes,
                "fileBytes": self.max_file_bytes,
                "mediaBytes": self.max_media_bytes,
            },
        }

    def read_text(self, project_id: str, relative: str) -> dict[str, Any]:
        pure = self._safe_relative(relative)
        if pure.suffix.casefold() not in TEXT_EXTENSIONS:
            raise DashboardError(
                HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "file type is not editable text"
            )
        try:
            with self._pinned_project(project_id) as (directory, _root):
                with _open_parent_directory(directory, pure) as (parent, name):
                    data, _ = self._read_regular(parent, name)
        except OSError as exc:
            raise DashboardError(
                HTTPStatus.FORBIDDEN, "text file cannot be opened safely"
            ) from exc
        try:
            content = data.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise DashboardError(
                HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "text file must be UTF-8"
            ) from exc
        return {
            "path": pure.as_posix(),
            "content": content,
            "version": _version(data),
            "writable": not self._is_protected(pure),
        }

    def _read_regular(self, parent: Directory, name: str) -> tuple[bytes, int]:
        descriptor = parent.open_regular(name, os.O_RDONLY)
        try:
            details = os.fstat(descriptor)
            if not stat.S_ISREG(details.st_mode):
                raise DashboardError(HTTPStatus.BAD_REQUEST, "path is not a file")
            if details.st_size > self.max_file_bytes:
                raise DashboardError(
                    HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
                    "file exceeds preview limit",
                )
            with os.fdopen(descriptor, "rb") as handle:
                descriptor = -1
                data = handle.read(self.max_file_bytes + 1)
            if len(data) > self.max_file_bytes:
                raise DashboardError(
                    HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
                    "file exceeds preview limit",
                )
            return data, stat.S_IMODE(details.st_mode)
        finally:
            if descriptor >= 0:
                os.close(descriptor)

    def _replace_text(
        self,
        parent: Directory,
        name: str,
        encoded: bytes,
        mode: int,
        expected_version: str,
    ) -> None:
        temporary_name = f".{name}.{uuid.uuid4().hex}.tmp"
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
        descriptor = parent.open_regular(temporary_name, flags, mode)
        replaced = False
        try:
            with os.fdopen(descriptor, "wb") as handle:
                descriptor = -1
                handle.write(encoded)
                handle.flush()
                os.fsync(handle.fileno())
                # Windows has no fchmod, and its chmod only moves the read-only
                # bit; the mode the file already carries is the right one there.
                if hasattr(os, "fchmod"):
                    os.fchmod(handle.fileno(), mode)
            latest, _ = self._read_regular(parent, name)
            if _version(latest) != expected_version:
                raise DashboardError(
                    HTTPStatus.CONFLICT, "file changed since it was opened"
                )
            try:
                parent.replace(temporary_name, name)
            except PermissionError as exc:
                # Windows refuses the rename while anything else holds the
                # target open; elsewhere the same errno means the file is
                # write-protected. Neither is an unsafe path, so neither may
                # report as one.
                raise DashboardError(
                    HTTPStatus.CONFLICT, "file is locked or not writable"
                ) from exc
            replaced = True
            parent.sync()
        finally:
            if descriptor >= 0:
                os.close(descriptor)
            if not replaced:
                # Best effort: the save already failed, and a cleanup that
                # cannot run must not replace that error with its own.
                with contextlib.suppress(OSError):
                    parent.unlink(temporary_name)

    @contextlib.contextmanager
    def _coordinated_edit(
        self, directory: Directory, relative: PurePosixPath, expected_version: str
    ) -> Iterator[None]:
        try:
            with directory.coordinated_edit(
                self.project_tool, relative.as_posix(), expected_version
            ):
                yield
        except Exception as exc:
            if exc.__class__.__name__ == "ProjectConflictError":
                raise DashboardError(HTTPStatus.CONFLICT, str(exc)) from exc
            if isinstance(exc, ValueError):
                raise DashboardError(
                    HTTPStatus.FORBIDDEN, "project path changed during the save"
                ) from exc
            if isinstance(exc, OSError):
                # Both backends refuse to take the project lock through a
                # redirected `.short-drama`. That is a refused path, not the
                # internal error it used to surface as.
                raise DashboardError(
                    HTTPStatus.FORBIDDEN, "text file cannot be replaced safely"
                ) from exc
            raise

    def write_text(
        self, project_id: str, relative: str, content: Any, expected_version: Any
    ) -> dict[str, Any]:
        pure = self._safe_relative(relative)
        if pure.suffix.casefold() not in TEXT_EXTENSIONS:
            raise DashboardError(
                HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "file type is not editable text"
            )
        if self._is_protected(pure):
            raise DashboardError(
                HTTPStatus.FORBIDDEN, "file is protected and read-only"
            )
        if not isinstance(content, str) or not isinstance(expected_version, str):
            raise DashboardError(
                HTTPStatus.BAD_REQUEST,
                "content and expectedVersion are required strings",
            )
        if re.fullmatch(r"[0-9a-f]{64}", expected_version) is None:
            raise DashboardError(
                HTTPStatus.BAD_REQUEST, "expectedVersion must be a SHA-256 digest"
            )
        encoded = content.encode("utf-8")
        if len(encoded) > self.max_file_bytes:
            raise DashboardError(
                HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "content exceeds file limit"
            )
        _validate_structured_text(pure, content)
        with (
            self._write_lock,
            self._pinned_project(project_id) as (directory, _root),
            self._coordinated_edit(directory, pure, expected_version),
        ):
            # The in-process lock and the project's file lock keep the version
            # check and atomic replace together. Pinned parents stop a
            # concurrent symlink swap from redirecting the write.
            try:
                with _open_parent_directory(directory, pure) as (parent, name):
                    current, mode = self._read_regular(parent, name)
                    if _version(current) != expected_version:
                        raise DashboardError(
                            HTTPStatus.CONFLICT,
                            "file changed since it was opened",
                        )
                    self._replace_text(
                        parent, name, encoded, mode, expected_version
                    )
            except OSError as exc:
                raise DashboardError(
                    HTTPStatus.FORBIDDEN, "text file cannot be replaced safely"
                ) from exc
        return {"path": pure.as_posix(), "version": _version(encoded), "saved": True}

    def media_info(self, project_id: str, relative: str) -> dict[str, Any]:
        pure = self._safe_relative(relative)
        with self._pinned_project(project_id) as (directory, _root):
            handle, content_type, size = self._open_media(directory, pure)
            handle.close()
            lifecycle = directory.lifecycle(self.project_tool, pure.as_posix())
        kind = "image"
        if content_type.startswith("video/"):
            kind = "video"
        elif content_type.startswith("audio/"):
            kind = "audio"
        result = {
            "path": pure.as_posix(),
            "kind": kind,
            "contentType": content_type,
            "size": size,
            "readOnly": True,
            "contentUrl": f"/api/media/content?{urlencode({'project': project_id, 'path': pure.as_posix()})}",
            "status": "ready",
        }
        if lifecycle is not None:
            result["lifecycle"] = lifecycle
        return result

    def open_media(
        self, project_id: str, relative: str
    ) -> tuple[BinaryIO, PurePosixPath, str, int]:
        pure = self._safe_relative(relative)
        with self._pinned_project(project_id) as (directory, _root):
            handle, content_type, size = self._open_media(directory, pure)
        return handle, pure, content_type, size

    def _open_media(
        self, directory: Directory, pure: PurePosixPath
    ) -> tuple[BinaryIO, str, int]:
        content_type = MEDIA_TYPES.get(pure.suffix.casefold())
        if content_type is None:
            raise DashboardError(
                HTTPStatus.UNSUPPORTED_MEDIA_TYPE, "unsupported preview media"
            )
        try:
            with _open_parent_directory(directory, pure) as (parent, name):
                descriptor = parent.open_regular(name, os.O_RDONLY)
        except OSError as exc:
            raise DashboardError(
                HTTPStatus.FORBIDDEN, "media file cannot be opened safely"
            ) from exc
        try:
            details = os.fstat(descriptor)
            if not stat.S_ISREG(details.st_mode):
                raise DashboardError(HTTPStatus.BAD_REQUEST, "media path is not a file")
            if details.st_size > self.max_media_bytes:
                raise DashboardError(
                    HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "media exceeds preview limit"
                )
            return os.fdopen(descriptor, "rb"), content_type, details.st_size
        except Exception:
            os.close(descriptor)
            raise


class DashboardHTTPServer(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, server_address: tuple[str, int], store: ProjectStore) -> None:
        self.store = store
        self.workspace_fingerprint = workspace_fingerprint(store.workspace)
        self.access_token = secrets.token_urlsafe(32)
        self.api_prefix = f"/_short_drama/{secrets.token_urlsafe(18)}"
        cookie_suffix = hashlib.sha256(self.api_prefix.encode("utf-8")).hexdigest()[:16]
        self.session_cookie = f"short_drama_{cookie_suffix}"
        super().__init__(server_address, DashboardHandler)

    def allowed_authority(self, authority: str) -> bool:
        parsed = urlsplit(f"//{authority}")
        if (
            not parsed.hostname
            or not _is_loopback(parsed.hostname)
            or parsed.username is not None
            or parsed.password is not None
            or bool(parsed.path or parsed.query or parsed.fragment)
        ):
            return False
        try:
            port = parsed.port
        except ValueError:
            return False
        return port == self.server_address[1]

    def server_close(self) -> None:
        try:
            super().server_close()
        finally:
            self.store.close()


class DashboardHandler(SimpleHTTPRequestHandler):
    server: DashboardHTTPServer

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, directory=str(STATIC_ROOT), **kwargs)

    def end_headers(self) -> None:
        self.send_header("Cache-Control", "no-store")
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("X-Frame-Options", "DENY")
        self.send_header(
            "Content-Security-Policy",
            "default-src 'self'; style-src 'self'; script-src 'self'; img-src 'self' data:; media-src 'self'",
        )
        super().end_headers()

    def _request_token(self) -> str:
        header = self.headers.get("X-Short-Drama-Token")
        if header:
            return header
        cookies = SimpleCookie()
        try:
            cookies.load(self.headers.get("Cookie", ""))
        except Exception:
            return ""
        morsel = cookies.get(self.server.session_cookie)
        return morsel.value if morsel is not None else ""

    def _security_ok(self, *, require_token: bool = False) -> bool:
        host = self.headers.get("Host", "")
        if not host or not self.server.allowed_authority(host):
            self._json(HTTPStatus.FORBIDDEN, {"error": "invalid Host header"})
            return False
        origin = self.headers.get("Origin")
        if origin:
            parsed = urlsplit(origin)
            if (
                parsed.scheme != "http"
                or not parsed.netloc
                or bool(parsed.path not in {"", "/"} or parsed.query or parsed.fragment)
                or not self.server.allowed_authority(parsed.netloc)
            ):
                self._json(HTTPStatus.FORBIDDEN, {"error": "invalid Origin header"})
                return False
        # Compare bytes, not str: `hmac.compare_digest` raises TypeError on a
        # str operand holding any codepoint above U+007F, and http.client
        # decodes header bytes as iso-8859-1, so a single 0x80+ byte in the
        # header would escape this pre-auth check and drop the connection with
        # no response at all.
        if require_token and not hmac.compare_digest(
            self._request_token().encode("utf-8", "surrogateescape"),
            self.server.access_token.encode("utf-8"),
        ):
            self._json(HTTPStatus.UNAUTHORIZED, {"error": "dashboard session required"})
            return False
        return True

    def _json(
        self, status: int, value: Any, *, headers: dict[str, str] | None = None
    ) -> None:
        body = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode(
            "utf-8"
        )
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        for name, content in (headers or {}).items():
            self.send_header(name, content)
        self.end_headers()
        self.wfile.write(body)

    def _query(self) -> tuple[str, dict[str, list[str]]]:
        parsed = urlsplit(self.path)
        path = parsed.path
        if path.startswith(f"{self.server.api_prefix}/api/"):
            path = path[len(self.server.api_prefix) :]
        return path, parse_qs(parsed.query, keep_blank_values=True)

    @staticmethod
    def _one(query: dict[str, list[str]], name: str) -> str:
        values = query.get(name)
        if not values or not values[0]:
            raise DashboardError(
                HTTPStatus.BAD_REQUEST, f"missing query parameter: {name}"
            )
        return values[0]

    @staticmethod
    def _byte_range(value: str | None, size: int) -> tuple[int, int] | None:
        if value is None:
            return None
        if not value.startswith("bytes=") or "," in value or size <= 0:
            raise DashboardError(
                HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "invalid byte range"
            )
        bounds = value[6:].strip()
        if "-" not in bounds:
            raise DashboardError(
                HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "invalid byte range"
            )
        first, last = bounds.split("-", 1)
        try:
            if first:
                start = int(first)
                end = int(last) if last else size - 1
                if start < 0 or start >= size or end < start:
                    raise ValueError
                end = min(end, size - 1)
            else:
                suffix = int(last)
                if suffix <= 0:
                    raise ValueError
                start = max(0, size - suffix)
                end = size - 1
        except ValueError as exc:
            raise DashboardError(
                HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "invalid byte range"
            ) from exc
        return start, end

    def _serve_media(self, query: dict[str, list[str]], *, send_body: bool) -> None:
        handle, _, content_type, size = self.server.store.open_media(
            self._one(query, "project"), self._one(query, "path")
        )
        try:
            try:
                selected = self._byte_range(self.headers.get("Range"), size)
            except DashboardError:
                self.send_response(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
                self.send_header("Content-Range", f"bytes */{size}")
                self.send_header("Content-Length", "0")
                self.send_header("Accept-Ranges", "bytes")
                self.end_headers()
                return
            start, end = selected if selected is not None else (0, size - 1)
            length = 0 if size == 0 else end - start + 1
            self.send_response(
                HTTPStatus.PARTIAL_CONTENT if selected is not None else HTTPStatus.OK
            )
            self.send_header("Content-Type", content_type)
            self.send_header("Content-Length", str(length))
            self.send_header("Accept-Ranges", "bytes")
            if selected is not None:
                self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
            self.end_headers()
            if not send_body or length == 0:
                return
            handle.seek(start)
            remaining = length
            while remaining:
                chunk = handle.read(min(64 * 1024, remaining))
                if not chunk:
                    break
                self.wfile.write(chunk)
                remaining -= len(chunk)
        except (BrokenPipeError, ConnectionResetError):
            pass
        finally:
            handle.close()

    def do_GET(self) -> None:
        path, query = self._query()
        if not self._security_ok(require_token=path.startswith("/api/")):
            return
        try:
            if path == "/api/projects":
                projects, warnings = self.server.store.discover()
                self._json(HTTPStatus.OK, {"projects": projects, "warnings": warnings})
                return
            if path == "/api/status":
                self._json(
                    HTTPStatus.OK, self.server.store.status(self._one(query, "project"))
                )
                return
            if path == "/api/tree":
                self._json(
                    HTTPStatus.OK, self.server.store.tree(self._one(query, "project"))
                )
                return
            if path == "/api/file":
                self._json(
                    HTTPStatus.OK,
                    self.server.store.read_text(
                        self._one(query, "project"), self._one(query, "path")
                    ),
                )
                return
            if path == "/api/media":
                info = self.server.store.media_info(
                    self._one(query, "project"), self._one(query, "path")
                )
                info["contentUrl"] = (
                    f"{self.server.api_prefix}{info['contentUrl']}"
                )
                self._json(
                    HTTPStatus.OK,
                    info,
                )
                return
            if path == "/api/media/content":
                self._serve_media(query, send_body=True)
                return
            if path.startswith("/api/"):
                self._json(HTTPStatus.NOT_FOUND, {"error": "API endpoint not found"})
                return
            if path == "/":
                self.path = "/index.html"
            super().do_GET()
        except DashboardError as exc:
            self._json(exc.status, {"error": exc.message})
        except (OSError, ValueError, json.JSONDecodeError) as exc:
            self._json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(exc)})
        except Exception:
            # A handler that lets anything escape closes the socket without a
            # status line, so the browser reports a network failure it cannot
            # act on. Every unexpected type still becomes an answered request.
            self._json(
                HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "internal dashboard error"}
            )

    def do_HEAD(self) -> None:
        path, query = self._query()
        if not self._security_ok(require_token=path.startswith("/api/")):
            return
        if path == "/api/media/content":
            try:
                self._serve_media(query, send_body=False)
            except DashboardError as exc:
                self._json(exc.status, {"error": exc.message})
            except OSError as exc:
                self._json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(exc)})
            except Exception:
                self._json(
                    HTTPStatus.INTERNAL_SERVER_ERROR,
                    {"error": "internal dashboard error"},
                )
            return
        if path.startswith("/api/"):
            self._json(
                HTTPStatus.METHOD_NOT_ALLOWED,
                {"error": "HEAD is not supported for API endpoints"},
            )
            return
        if path == "/":
            self.path = "/index.html"
        super().do_HEAD()

    def do_PUT(self) -> None:
        path, query = self._query()
        if not self._security_ok(require_token=path.startswith("/api/")):
            return
        if path != "/api/file":
            self._json(HTTPStatus.NOT_FOUND, {"error": "API endpoint not found"})
            return
        try:
            raw_length = self.headers.get("Content-Length")
            if raw_length is None:
                raise DashboardError(
                    HTTPStatus.LENGTH_REQUIRED, "Content-Length is required"
                )
            length = int(raw_length)
            if length < 0 or length > self.server.store.max_request_bytes:
                raise DashboardError(
                    HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "request body is too large"
                )
            if self.headers.get_content_type() != "application/json":
                raise DashboardError(
                    HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
                    "Content-Type must be application/json",
                )
            payload = json.loads(self.rfile.read(length).decode("utf-8"))
            if not isinstance(payload, dict):
                raise DashboardError(
                    HTTPStatus.BAD_REQUEST, "request body must be an object"
                )
            result = self.server.store.write_text(
                self._one(query, "project"),
                self._one(query, "path"),
                payload.get("content"),
                payload.get("expectedVersion"),
            )
            self._json(HTTPStatus.OK, result)
        except DashboardError as exc:
            self._json(exc.status, {"error": exc.message})
        except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
            self._json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON request"})
        except OSError as exc:
            self._json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(exc)})
        except Exception:
            self._json(
                HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "internal dashboard error"}
            )

    def do_POST(self) -> None:
        path, _ = self._query()
        if not self._security_ok(require_token=path.startswith("/api/")):
            return
        if path != "/api/session":
            self._json(HTTPStatus.NOT_FOUND, {"error": "API endpoint not found"})
            return
        self._json(
            HTTPStatus.OK,
            {"status": "ready", "apiBase": self.server.api_prefix},
            headers={
                "Set-Cookie": (
                    f"{self.server.session_cookie}={self.server.access_token}; "
                    f"HttpOnly; SameSite=Strict; Path={self.server.api_prefix}/"
                )
            },
        )


def workspace_fingerprint(workspace: Path) -> str:
    """Identify one workspace without exposing its path in a response header."""
    return hashlib.sha256(
        str(workspace).encode("utf-8", "surrogateescape")
    ).hexdigest()[:16]


def session_file_for(workspace: Path, override: Path | None = None) -> Path:
    if override is not None:
        return override.expanduser().resolve()
    return workspace / Path(str(SESSION_RELATIVE))


def read_session(path: Path) -> dict[str, Any] | None:
    """Return the recorded session, or ``None`` when there is nothing usable."""
    try:
        document = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(document, dict) or document.get("schema_version") != SESSION_SCHEMA:
        return None
    required = ("host", "port", "token", "fingerprint", "url", "pid")
    if any(key not in document for key in required):
        return None
    return document


def write_session(path: Path, document: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
    descriptor = os.open(
        str(temporary), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
    )
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(document, handle, ensure_ascii=False, indent=2, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    finally:
        with contextlib.suppress(OSError):
            temporary.unlink()


def session_lock_path(session_path: Path) -> Path:
    return session_path.with_name(session_path.name + SESSION_LOCK_SUFFIX)


def _try_exclusive_lock(handle: Any) -> bool:
    """Take the serving lock without waiting. False means someone else holds it."""
    if os.name == "nt":
        import msvcrt

        handle.seek(0, os.SEEK_END)
        if handle.tell() == 0:
            handle.write(b"0")
            handle.flush()
        handle.seek(0)
        # Reached through getattr for the same reason project_tool.py does: the
        # POSIX stubs mypy checks against do not declare these Windows names.
        locking = getattr(msvcrt, "locking")
        non_blocking = getattr(msvcrt, "LK_NBLCK")
        try:
            locking(handle.fileno(), non_blocking, 1)
        except OSError:
            return False
        return True
    import fcntl

    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        return False
    return True


class SessionUnavailable(RuntimeError):
    """This location cannot hold a session record (read-only workspace, ...).

    Serving does not depend on the record: it exists so --detach can report a
    URL and --status/--stop can find the server later. A workspace the creator
    can read but not write must still open, the way it did before sessions
    existed.
    """


@contextlib.contextmanager
def hold_session_lock(session_path: Path) -> Iterator[bool]:
    """Hold the serving lock, yielding whether WE hold it (vs. another process).

    Liveness is answered by this lock rather than by probing the recorded port.
    A PID can be reused and a port can be inherited by something unrelated; a
    lock is released by the kernel exactly when the serving process ends. It
    also keeps the dashboard free of any outbound network client.
    """
    lock_path = session_lock_path(session_path)
    try:
        lock_path.parent.mkdir(parents=True, exist_ok=True)
        handle = lock_path.open("a+b")
    except OSError as exc:
        raise SessionUnavailable(
            f"cannot use a dashboard session file at {lock_path}: {exc}"
        ) from exc
    if not _try_exclusive_lock(handle):
        handle.close()
        yield False
        return
    try:
        yield True
    finally:
        handle.close()


def session_is_live(session_path: Path) -> bool:
    """Answer whether a dashboard is still serving this workspace."""
    try:
        with hold_session_lock(session_path) as held:
            return not held
    except SessionUnavailable:
        return False


def session_matches(session: Mapping[str, Any], workspace: Path) -> bool:
    """Is this record about the workspace the caller asked for?

    A --session-file shared between two workspaces would otherwise hand the
    second creator a link that serves the first one's project.
    """
    return session.get("fingerprint") == workspace_fingerprint(workspace)


def watch_workspace(server: Any, workspace: Path, interval: float = 5.0) -> None:
    """Shut the server down if its workspace stops existing.

    A detached dashboard used to die with its shell. Now that it does not, a
    creator who deletes or moves the project would otherwise leave a server
    holding a port and answering on the old token URL for a tree that is gone --
    and its session record went with the directory, so nothing could stop it.
    """
    try:
        expected = os.stat(workspace)
    except OSError:
        return
    identity = (expected.st_dev, expected.st_ino)
    misses = 0

    def loop() -> None:
        nonlocal misses
        while True:
            time.sleep(interval)
            try:
                current = os.stat(workspace)
                gone = (current.st_dev, current.st_ino) != identity
            except OSError:
                gone = True
            misses = misses + 1 if gone else 0
            if misses >= 2:  # tolerate one transient stat failure
                threading.Thread(target=server.shutdown, daemon=True).start()
                return

    threading.Thread(target=loop, daemon=True).start()


def stop_session(path: Path) -> bool:
    """Stop the recorded dashboard and forget it. Returns whether one was live."""
    session = read_session(path)
    live = session is not None and session_is_live(path)
    if live:
        pid = session.get("pid") if session else None
        if isinstance(pid, int) and pid > 0:
            with contextlib.suppress(OSError):
                os.kill(pid, signal.SIGTERM)
            deadline = time.monotonic() + STOP_TIMEOUT_SECONDS
            while time.monotonic() < deadline and session_is_live(path):
                time.sleep(0.1)
    with contextlib.suppress(OSError):
        path.unlink()
    return live


def _detached_child(
    workspace: Path, *, host: str, port: int, session_path: Path
) -> subprocess.Popen[bytes]:
    command = [
        sys.executable,
        "-B",
        str(Path(__file__).resolve()),
        "--workspace",
        str(workspace),
        "--host",
        host,
        "--port",
        str(port),
        "--session-file",
        str(session_path),
    ]
    log_path = session_path.with_name(f"{session_path.stem}.log")
    log_path.parent.mkdir(parents=True, exist_ok=True)
    log = open(log_path, "ab")
    options: dict[str, Any] = {}
    if os.name == "nt":
        # DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP: no console to inherit and
        # no Ctrl-C from the launching shell.
        options["creationflags"] = 0x00000008 | 0x00000200
    else:
        options["start_new_session"] = True
    try:
        return subprocess.Popen(
            command,
            stdin=subprocess.DEVNULL,
            stdout=log,
            stderr=log,
            close_fds=True,
            cwd=str(workspace),
            **options,
        )
    finally:
        log.close()


def start_detached(
    workspace: Path, *, host: str, port: int, session_path: Path
) -> dict[str, Any]:
    """Start a dashboard that outlives the shell that asked for it."""
    process = _detached_child(workspace, host=host, port=port, session_path=session_path)
    deadline = time.monotonic() + DETACH_TIMEOUT_SECONDS
    while time.monotonic() < deadline:
        session = read_session(session_path)
        if session is not None and session.get("pid") == process.pid:
            return session
        if process.poll() is not None:
            break
        time.sleep(0.1)
    if process.poll() is None:
        with contextlib.suppress(OSError):
            process.kill()
    log_path = session_path.with_name(f"{session_path.stem}.log")
    raise RuntimeError(f"dashboard did not start; see {log_path}")


def create_server(
    workspace: Path,
    *,
    host: str = "127.0.0.1",
    port: int = 8765,
    suite_root: Path | None = None,
    max_depth: int = DEFAULT_MAX_DEPTH,
    max_nodes: int = DEFAULT_MAX_NODES,
    max_file_bytes: int = DEFAULT_MAX_FILE_BYTES,
    max_media_bytes: int = DEFAULT_MAX_MEDIA_BYTES,
) -> DashboardHTTPServer:
    try:
        address = ipaddress.i

#!/usr/bin/env python3
"""Small filesystem lifecycle for short-drama projects.

Creative work lives in project files and skills. This module keeps only the
mechanical boundaries that earn their cost: safe publication, exact creator
acceptance, a lightweight review verdict, direct-input freshness, and portable
text delivery. It does not call networks or media services.
"""

from __future__ import annotations

import argparse
import contextlib
import hashlib
import json
import math
import os
import re
import shutil
import stat
import sys
import unicodedata
import uuid
from collections.abc import Callable, Iterable, Mapping, Sequence
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Iterator


MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit(
        "short-drama needs Python {}.{} or newer; this interpreter is {}.{}".format(
            *MINIMUM_PYTHON, sys.version_info.major, sys.version_info.minor
        )
    )

PROJECT_FILE = "short-drama.json"
STATE_FILE = Path(".short-drama/state.json")
STATE_SCHEMA = "2.0"
DEFAULT_PROMPT_LANGUAGE = "en"
LANGUAGE_TAG_RE = re.compile(r"[A-Za-z]{2,8}(?:-[A-Za-z0-9]{1,8})*")
DELIVERY_SUFFIXES = {".md", ".json", ".jsonl"}
CREATOR_DOCUMENTS = (
    "剧本.md",
    "视觉设定.md",
    "分镜.md",
    "图片提示词.md",
    "视频提示词.md",
)
# The cut list is written after the media exists, so an episode that never
# reached editing is complete without one. It travels with a handover when it is
# there -- without it the next person has to re-cut the film from scratch -- but
# it is never reported as a missing creator document.
POST_PRODUCTION_DOCUMENTS = ("剪辑单.md",)
EXPORT_MEDIA_DIRECTORY = "制作成果"
EXPORT_SCHEMA = "1.0"
EPISODE_ID_RE = re.compile(r"EP(?:[0-9]{3}|[1-9][0-9]{3,})")
WINDOWS_FORBIDDEN_PATH_CHARACTERS = frozenset('<>:"|?*')
WINDOWS_RESERVED_PATH_STEMS = frozenset(
    {
        "con",
        "prn",
        "aux",
        "nul",
        *(f"com{number}" for number in range(1, 10)),
        *(f"lpt{number}" for number in range(1, 10)),
        "com¹",
        "com²",
        "com³",
        "lpt¹",
        "lpt²",
        "lpt³",
    }
)

CANONICAL_ROOTS = {
    "inputs": "输入",
    "development": "项目开发",
    "bible": "设定集",
    "episodes": "剧集",
    "delivery": "交付",
    "creator-decisions": "创作者决策",
    "reviews": "审查",
}
LEGACY_ROOTS = {role: role for role in CANONICAL_ROOTS}
ROOT_ROLE_ALIASES: dict[str, str] = {
    name.casefold(): role
    for roots in (CANONICAL_ROOTS, LEGACY_ROOTS)
    for role, name in roots.items()
}
LAYOUT_PINNING_ROLES = frozenset(
    {"development", "bible", "episodes", "delivery", "creator-decisions", "reviews"}
)
PUBLISHABLE_ROOT_ROLES = frozenset(
    {"development", "bible", "episodes", "creator-decisions", "reviews"}
)
PUBLISHABLE_ROOTS = tuple(
    roots[role]
    for roots in (CANONICAL_ROOTS, LEGACY_ROOTS)
    for role in CANONICAL_ROOTS
    if role in PUBLISHABLE_ROOT_ROLES
)
PROJECT_DIRS = (*CANONICAL_ROOTS.values(), ".short-drama")

PROTECTED_PUBLISH_ROLE_REASONS = {
    "inputs": "creator inputs are immutable publication sources",
    "delivery": "the delivery tree is written by the packaging gate, not by publication",
}
PROTECTED_PUBLISH_ROOTS = {
    name.casefold(): reason
    for role, reason in PROTECTED_PUBLISH_ROLE_REASONS.items()
    for name in (CANONICAL_ROOTS[role], LEGACY_ROOTS[role])
} | {".short-drama": "operational state cannot be a publication target"}

AUTHORITY_ROOT_TOKEN = "creator_authority"
EPISODE_LENGTH_POINTER = "/format/target_seconds_per_episode"
PACING_POINTER = "/format/pacing"
# Everything outside /creator_authority/* that a decision may still bind. Both
# are read by the write stage to turn a screenplay into seconds, so leaving them
# out of set-authority left hand-editing short-drama.json as the only way in.
FORMAT_POINTERS = (EPISODE_LENGTH_POINTER, PACING_POINTER)

class ProjectConflictError(RuntimeError):
    """A file changed while a guarded operation was in progress."""


class PackageBlockedError(RuntimeError):
    """A requested delivery contains an artifact that is not approved."""


class NonPortablePathError(ValueError):
    """A path would alias or fail on a supported filesystem."""


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def sha256_bytes(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _is_link_or_reparse(details: os.stat_result) -> bool:
    attributes = getattr(details, "st_file_attributes", 0)
    reparse_flag = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
    return stat.S_ISLNK(details.st_mode) or bool(attributes & reparse_flag)


def _fsync_directory(path: Path) -> None:
    if os.name == "nt":
        return
    descriptor = os.open(path, os.O_RDONLY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def _atomic_bytes(path: Path, content: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
    try:
        with temporary.open("xb") as handle:
            handle.write(content)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
        _fsync_directory(path.parent)
    finally:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass


def atomic_json(path: Path, document: Mapping[str, Any]) -> None:
    # allow_nan=False: Python reads and writes bare Infinity/NaN, but they are not
    # JSON. Writing one produces a manifest that every other reader rejects.
    encoded = (
        json.dumps(document, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False)
        + "\n"
    ).encode("utf-8")
    _atomic_bytes(path, encoded)


def normalize_language_tag(value: str, *, field: str) -> str:
    tag = value.strip()
    if not tag:
        raise ValueError(f"{field} must not be empty")
    if LANGUAGE_TAG_RE.fullmatch(tag) is None:
        raise ValueError(f"{field} is not a well-formed language tag: {value!r}")
    return tag


def project_languages(project: Mapping[str, Any]) -> dict[str, str]:
    format_block = project.get("format")
    prompt_language = (
        format_block.get("prompt_language")
        if isinstance(format_block, Mapping)
        else None
    )
    authority = project.get("creator_authority")
    production_profile = (
        authority.get("production_profile")
        if isinstance(authority, Mapping)
        else None
    )
    choices = (
        production_profile.get("choices")
        if isinstance(production_profile, Mapping)
        else None
    )
    video_prompt_language = (
        choices.get("video_prompt_language") if isinstance(choices, Mapping) else None
    )
    return {
        "language": str(project.get("language") or "zh-CN"),
        "prompt_language": str(prompt_language or DEFAULT_PROMPT_LANGUAGE),
        "video_prompt_language": str(
            video_prompt_language or prompt_language or DEFAULT_PROMPT_LANGUAGE
        ),
    }


def project_video_model_profile(project: Mapping[str, Any]) -> dict[str, Any]:
    """Expose the accepted prompt-facing model choices without guessing defaults."""
    authority = project.get("creator_authority")
    production_profile = (
        authority.get("production_profile")
        if isinstance(authority, Mapping)
        else None
    )
    if (
        not isinstance(production_profile, Mapping)
        or production_profile.get("status") != "accepted"
    ):
        return {}
    choices = production_profile.get("choices")
    if not isinstance(choices, Mapping):
        return {}
    fields = (
        "target_video_model",
        "video_prompt_dialect",
        "video_prompt_language",
        "native_duration_seconds",
        "supported_generation_modes",
        "audio_generation",
    )
    return {field: choices[field] for field in fields if field in choices}


def initialize_project(
    path: Path,
    *,
    title: str,
    language: str,
    aspect_ratio: str,
    prompt_language: str = DEFAULT_PROMPT_LANGUAGE,
    episode_count: int | None = None,
    target_seconds_per_episode: float | None = None,
    suite_root: Path | None = None,
) -> dict[str, Any]:
    language = normalize_language_tag(language, field="language")
    prompt_language = normalize_language_tag(prompt_language, field="prompt_language")
    if episode_count is not None and (
        isinstance(episode_count, bool)
        or not isinstance(episode_count, int)
        or episode_count <= 0
    ):
        raise ValueError("episode_count must be a positive integer")
    if target_seconds_per_episode is not None and (
        isinstance(target_seconds_per_episode, bool)
        or not isinstance(target_seconds_per_episode, (int, float))
        or not math.isfinite(target_seconds_per_episode)
        or target_seconds_per_episode <= 0
    ):
        raise ValueError("target_seconds_per_episode must be a positive number of seconds")
    root = path.expanduser().resolve()
    project_path = root / PROJECT_FILE
    if project_path.exists():
        raise FileExistsError(f"project already exists: {project_path}")

    root.mkdir(parents=True, exist_ok=True)
    for relative in PROJECT_DIRS:
        (root / relative).mkdir(parents=True, exist_ok=True)

    core = suite_root or Path(__file__).resolve().parents[1]
    project = json.loads(
        (core / "assets/project-template/short-drama.json").read_text(encoding="utf-8")
    )
    project.update(
        {
            "project_id": f"SD-{uuid.uuid4().hex[:12].upper()}",
            "title": title.strip() or "未命名短剧",
            "language": language,
            "created_at": utc_now(),
        }
    )
    project["format"]["aspect_ratio"] = aspect_ratio
    project["format"]["prompt_language"] = prompt_language
    if episode_count is not None:
        project["format"]["episode_count"] = episode_count
    if target_seconds_per_episode is not None:
        project["format"]["target_seconds_per_episode"] = target_seconds_per_episode
    state = {
        "schema_version": STATE_SCHEMA,
        "project_id": project["project_id"],
        "project_layout_mode": "auto",
        "updated_at": utc_now(),
        "last_action": "initialized",
        "artifacts": {},
    }
    atomic_json(root / STATE_FILE, state)
    atomic_json(project_path, project)
    return {"project_root": str(root), "project": project, "state": state}


def find_project(start: Path) -> Path:
    candidate = start.expanduser().resolve()
    if candidate.is_file():
        candidate = candidate.parent
    for directory in (candidate, *candidate.parents):
        if (directory / PROJECT_FILE).is_file():
            return directory
    raise FileNotFoundError(f"no {PROJECT_FILE} found from {start}")


def _has_nonportable_path_component(parts: tuple[str, ...]) -> bool:
    for part in parts:
        stem = part.split(".", 1)[0].casefold()
        if (
            part.endswith((" ", "."))
            or any(ord(character) < 32 or ord(character) == 127 for character in part)
            or any(character in WINDOWS_FORBIDDEN_PATH_CHARACTERS for character in part)
            or stem in WINDOWS_RESERVED_PATH_STEMS
        ):
            return True
    return False


def _relative_path(value: str | Path, *, allow_operations: bool = False) -> str:
    raw = str(value).replace("\\", "/")
    pure = PurePosixPath(raw)
    if not raw or pure.is_absolute() or any(part in ("", ".", "..") for part in pure.parts):
        raise ValueError(f"unsafe project-relative path: {value!s}")
    if _has_nonportable_path_component(pure.parts):
        raise NonPortablePathError(f"unsafe project-relative path: {value!s}")
    if not allow_operations and pure.parts[0].casefold() == ".short-drama":
        raise ValueError("operational state cannot be a publication target")
    return pure.as_posix()


def _portable_path_identity(value: str) -> str:
    return unicodedata.normalize("NFC", value.casefold())


def _validate_existing_path_spelling(root: Path, relative: str, *, label: str) -> None:
    current = root
    prefix: list[str] = []
    parts = PurePosixPath(relative).parts
    for index, part in enumerate(parts):
        try:
            entries = list(os.scandir(current))
        except (FileNotFoundError, NotADirectoryError):
            return
        identity = _portable_path_identity(part)
        matches = [entry for entry in entries if _portable_path_identity(entry.name) == identity]
        alias = next((entry.name for entry in matches if entry.name != part), None)
        if alias is not None:
            existing = PurePosixPath(*prefix, alias).as_posix()
            raise NonPortablePathError(
                f"{label} path spelling aliases an existing path: {relative} conflicts with {existing}"
            )
        exact = next((entry for entry in matches if entry.name == part), None)
        if exact is None or index == len(parts) - 1:
            return
        if not exact.is_dir(follow_symlinks=False):
            return
        current /= part
        prefix.append(part)


def _validate_path_set(root: Path, relatives: Iterable[str], *, label: str) -> list[str]:
    normalized: list[str] = []
    seen: dict[str, str] = {}
    for value in relatives:
        relative = _relative_path(value)
        identity = _portable_path_identity(relative)
        previous = seen.get(identity)
        if previous is not None and previous != relative:
            raise NonPortablePathError(
                f"{label} paths are not portable aliases: {previous} and {relative}"
            )
        if previous is None:
            seen[identity] = relative
            normalized.append(relative)
        _validate_existing_path_spelling(root, relative, label=label)
    return sorted(normalized)


def _root_role(name: str) -> str | None:
    return ROOT_ROLE_ALIASES.get(name.casefold())


def _root_layout_mode(name: str) -> str | None:
    role = _root_role(name)
    if role is None:
        return None
    if name == CANONICAL_ROOTS[role]:
        return "canonical"
    if name == LEGACY_ROOTS[role]:
        return "legacy"
    return None


def is_protected_project_text(value: str | Path) -> bool:
    raw = str(value).replace("\\", "/")
    pure = PurePosixPath(raw)
    if (
        not raw
        or pure.is_absolute()
        or any(part in ("", ".", "..") for part in pure.parts)
        or _has_nonportable_path_component(pure.parts)
    ):
        return True
    return (
        pure.name.casefold() == PROJECT_FILE
        or pure.parts[0].casefold() == ".short-drama"
        or _root_role(pure.parts[0]) == "delivery"
    )


def _directory_has_content(path: Path) -> bool:
    try:
        details = os.lstat(path)
        return stat.S_ISLNK(details.st_mode) or (
            stat.S_ISDIR(details.st_mode) and any(path.iterdir())
        )
    except FileNotFoundError:
        return False


def _read_state(root: Path) -> dict[str, Any]:
    state_path = root / STATE_FILE
    if not state_path.is_file() or state_path.is_symlink():
        raise ValueError("project state is missing or unsafe")
    document = json.loads(state_path.read_text(encoding="utf-8"))
    if not isinstance(document, dict):
        raise ValueError("project state must be an object")
    return _normalize_state(document)


def _legacy_artifact(record: Mapping[str, Any]) -> dict[str, Any]:
    candidate = record.get("candidate_targets")
    accepted_targets = record.get("accepted_targets")
    reviewed_targets = record.get("reviewed_targets")
    output_map = (
        candidate
        if isinstance(candidate, Mapping)
        else accepted_targets
        if isinstance(accepted_targets, Mapping)
        else {}
    )
    outputs = sorted(str(path) for path in output_map if isinstance(path, str))
    input_map = (
        record.get("candidate_inputs")
        if isinstance(candidate, Mapping)
        else record.get("accepted_inputs")
    )
    if not isinstance(input_map, Mapping):
        input_map = {}
    inputs = {
        str(path): str(value)
        for path, value in (input_map.items() if isinstance(input_map, Mapping) else [])
        if isinstance(path, str) and isinstance(value, str)
    }
    acceptance: dict[str, Any] | None = None
    if record.get("creator_acceptance") in {"accepted", "rejected"}:
        decision = str(record["creator_acceptance"])
        acceptance = {
            "decision": decision,
            "at": None,
            "outputs": {
                str(path): str(value)
                for path, value in (
                    accepted_targets.items()
                    if isinstance(accepted_targets, Mapping)
                    else []
                )
                if isinstance(path, str) and isinstance(value, str)
            },
            "note": "migrated from v0.3 state",
        }
    review: dict[str, Any] | None = None
    legacy_verdict = record.get("independent_review")
    if legacy_verdict in {"approve", "approve_with_notes", "revise", "provisional"}:
        review_outputs = (
            reviewed_targets
            if isinstance(reviewed_targets, Mapping)
            else accepted_targets
            if isinstance(accepted_targets, Mapping)
            else {}
        )
        review = {
            "verdict": legacy_verdict,
            "at": None,
            "outputs": {
                str(path): str(value)
                for path, value in review_outputs.items()
                if isinstance(path, str) and isinstance(value, str)
            },
            "reviewer": "migrated",
            "note": "migrated from v0.3 state",
        }
    return {
        "owner": str(record.get("owner") or "unknown"),
        "outputs": outputs,
        "inputs": inputs,
        "acceptance": acceptance,
        "review": review,
        "updated_at": None,
    }


def _normalize_state(document: Mapping[str, Any]) -> dict[str, Any]:
    artifacts_raw = document.get("artifacts", {})
    if not isinstance(artifacts_raw, Mapping):
        raise ValueError("state.artifacts must be an object")
    artifacts: dict[str, Any] = {}
    modern = document.get("schema_version") == STATE_SCHEMA
    for artifact_id, value in artifacts_raw.items():
        if not isinstance(artifact_id, str) or not isinstance(value, Mapping):
            raise ValueError("artifact state entries must be objects")
        if modern and isinstance(value.get("outputs"), list):
            outputs = [
                _relative_path(item)
                for item in value["outputs"]
                if isinstance(item, str)
            ]
            inputs_raw = value.get("inputs", {})
            if not isinstance(inputs_raw, Mapping):
                raise ValueError(f"{artifact_id}.inputs must be an object")
            inputs = {
                _relative_path(path, allow_operations=True): str(digest)
                for path, digest in inputs_raw.items()
                if isinstance(path, str) and isinstance(digest, str)
            }
            artifacts[artifact_id] = {
                "owner": str(value.get("owner") or "unknown"),
                "outputs": sorted(set(outputs)),
                "inputs": inputs,
                "acceptance": value.get("acceptance") if isinstance(value.get("acceptance"), Mapping) else None,
                "review": value.get("review") if isinstance(value.get("review"), Mapping) else None,
                "updated_at": value.get("updated_at"),
            }
        else:
            artifacts[artifact_id] = _legacy_artifact(value)
    mode = document.get("project_layout_mode", "auto")
    if mode not in {"auto", "canonical", "legacy"}:
        mode = "auto"
    bindings_raw = document.get("authority")
    bindings = {
        str(field): dict(binding)
        for field, binding in (bindings_raw.items() if isinstance(bindings_raw, Mapping) else [])
        if isinstance(field, str) and isinstance(binding, Mapping)
    }
    return {
        "schema_version": STATE_SCHEMA,
        "project_id": document.get("project_id"),
        "project_layout_mode": mode,
        "updated_at": document.get("updated_at"),
        "last_action": document.get("last_action") or "loaded",
        "artifacts": artifacts,
        "authority": bindings,
    }


def _save_state(root: Path, state: dict[str, Any], *, action: str) -> None:
    state["schema_version"] = STATE_SCHEMA
    state["updated_at"] = utc_now()
    state["last_action"] = action
    atomic_json(root / STATE_FILE, state)


def _project_layout_from_root(root: Path, state: Mapping[str, Any] | None = None) -> dict[str, Any]:
    current = state or _read_state(root)
    recorded = current.get("project_layout_mode", "auto")
    canonical_roles = sorted(
        role
        for role, name in CANONICAL_ROOTS.items()
        if role in LAYOUT_PINNING_ROLES and _directory_has_content(root / name)
    )
    legacy_roles = sorted(
        role
        for role, name in LEGACY_ROOTS.items()
        if role in LAYOUT_PINNING_ROLES and _directory_has_content(root / name)
    )
    nonstandard_roots = sorted(
        entry.name
        for entry in root.iterdir()
        if (entry.is_symlink() or entry.is_dir())
        if (role := _root_role(entry.name)) in LAYOUT_PINNING_ROLES
        if entry.name not in {CANONICAL_ROOTS[role], LEGACY_ROOTS[role]}
        if _directory_has_content(entry)
    )
    unsafe_roots = sorted(
        entry.name
        for entry in root.iterdir()
        if _root_role(entry.name) in LAYOUT_PINNING_ROLES and entry.is_symlink()
    )
    detected = {
        mode
        for mode, roles in (("canonical", canonical_roles), ("legacy", legacy_roles))
        if roles
    }
    conflict = bool(nonstandard_roots or unsafe_roots) or len(detected) > 1 or (
        recorded in {"canonical", "legacy"} and detected and detected != {recorded}
    )
    if conflict:
        mode = "mixed"
    elif recorded in {"canonical", "legacy"}:
        mode = str(recorded)
    elif detected:
        mode = next(iter(detected))
    else:
        mode = "canonical"
    roots = LEGACY_ROOTS if mode == "legacy" else CANONICAL_ROOTS
    return {
        "mode": mode,
        "pinned": recorded != "auto" or bool(detected),
        "roots": dict(roots),
        "nonstandardRoots": nonstandard_roots,
        "unsafeRoots": unsafe_roots,
    }


def project_layout(path: Path) -> dict[str, Any]:
    root = find_project(path)
    return _project_layout_from_root(root)


def _validate_project_output_layout(root: Path, relatives: Iterable[str]) -> str | None:
    families = {
        family
        for relative in relatives
        if (part := PurePosixPath(relative).parts[0])
        if _root_role(part) in LAYOUT_PINNING_ROLES
        if (family := _root_layout_mode(part)) is not None
    }
    if len(families) > 1:
        raise ValueError("不能在同一次发布中混用中文与旧版英文目录")
    family = next(iter(families), None)
    layout = _project_layout_from_root(root)
    if layout["mode"] == "mixed":
        raise ValueError("项目同时包含中文与旧版英文阶段目录，请先迁移并合并")
    if family is not None and layout["pinned"] and family != layout["mode"]:
        expected = "中文" if layout["mode"] == "canonical" else "旧版英文"
        raise ValueError(f"项目已使用{expected}目录布局，不能创建另一套平行目录")
    return family


def _validate_publication_layout(
    relative: str, *, allow_unregistered: bool
) -> None:
    pure = PurePosixPath(relative)
    first = pure.parts[0].casefold()
    role = _root_role(pure.parts[0])
    reason = PROTECTED_PUBLISH_ROOTS.get(first)
    if reason is not None:
        raise ValueError(reason)
    if pure.name.casefold() == PROJECT_FILE:
        raise ValueError("creator authority file cannot be a publication target")
    if role == "episodes":
        if len(pure.parts) < 3:
            raise ValueError("episode artifacts live in 剧集/<EP>/（兼容 episodes/<EP>/）")
        if EPISODE_ID_RE.fullmatch(pure.parts[1]) is None:
            raise ValueError(f"episode directory must use an EP001-style identifier: {pure.parts[1]}")
    if not allow_unregistered and role not in PUBLISHABLE_ROOT_ROLES:
        raise ValueError(
            f"{pure.parts[0]} is not a project stage directory; expected one of {', '.join(PUBLISHABLE_ROOTS)}"
        )
    if role is not None and pure.parts[0] not in {CANONICAL_ROOTS[role], LEGACY_ROOTS[role]}:
        raise ValueError(f"阶段目录大小写或拼写不规范：{pure.parts[0]}")


def _project_path(root: Path, relative: str, *, create_parent: bool = False) -> Path:
    root = root.resolve()
    target = root / relative
    current = root
    for part in PurePosixPath(relative).parts[:-1]:
        current /= part
        # ``is_symlink`` misses the reparse points that are not name
        # surrogates, and a Windows creator can make those without any extra
        # privilege. Ask about the attribute directly instead.
        try:
            details = os.lstat(current)
        except FileNotFoundError:
            details = None
        if details is not None and _is_link_or_reparse(details):
            raise ProjectConflictError(f"project parent cannot be a symlink: {part}")
        if details is not None and not stat.S_ISDIR(details.st_mode):
            raise ProjectConflictError(f"project parent is not a directory: {part}")
        if create_parent and details is None:
            current.mkdir()
    try:
        target_details = os.lstat(target)
    except (FileNotFoundError, NotADirectoryError):
        target_details = None
    if target_details is not None and (
        _is_link_or_reparse(target_details) or not stat.S_ISREG(target_details.st_mode)
    ):
        raise ProjectConflictError(f"project target is not a regular file: {relative}")
    if not target.parent.resolve().is_relative_to(root):
        raise ValueError(f"path escapes project root: {relative}")
    return target


def _live_hash(root: Path, relative: str) -> str | None:
    try:
        target = _project_path(root, relative)
    except (OSError, ProjectConflictError, ValueError):
        return None
    if not target.is_file():
        return None
    return sha256_file(target)


@contextlib.contextmanager
def _lock_handle(handle: Any) -> Iterator[None]:
    if os.name == "nt":
        import msvcrt

        handle.seek(0, os.SEEK_END)
        if handle.tell() == 0:
            handle.write(b"0")
            handle.flush()
        handle.seek(0)
        locking = getattr(msvcrt, "locking")
        lock = getattr(msvcrt, "LK_LOCK")
        unlock = getattr(msvcrt, "LK_UNLCK")
        locking(handle.fileno(), lock, 1)
        try:
            yield
        finally:
            handle.seek(0)
            locking(handle.fileno(), unlock, 1)
    else:
        import fcntl

        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)


@contextlib.contextmanager
def _project_lock(root: Path) -> Iterator[None]:
    lock_path = root / ".short-drama/project.lock"
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    with lock_path.open("a+b") as handle, _lock_handle(handle):
        yield


def _hash_mapping(root: Path, relatives: Iterable[str]) -> dict[str, str]:
    result: dict[str, str] = {}
    for relative in relatives:
        digest = _live_hash(root, relative)
        if digest is None:
            raise ProjectConflictError(f"project file is missing or unsafe: {relative}")
        result[relative] = digest
    return result


def _inputs_current(root: Path, record: Mapping[str, Any]) -> bool:
    inputs = record.get("inputs", {})
    return isinstance(inputs, Mapping) and all(
        isinstance(path, str)
        and isinstance(expected, str)
        and _live_hash(root, path) == expected
        for path, expected in inputs.items()
    )


def _artifact_state_from(
    record: Mapping[str, Any], live_hash: Callable[[str], str | None]
) -> str:
    """Decide one artifact's lifecycle state from the bytes on disk right now.

    The whole lifecycle rests on the two `!= live` comparisons below: an
    accepted artifact whose bytes were edited behind the tool's back has not
    been accepted in its current form, and must fall back to `update_needed`.

    This is written once and reached by two callers -- path-based and
    directory-fd-based -- which differ only in how they hash a file. It used to
    be two transcriptions of the same rules, with nothing asserting they agreed,
    and the dashboard renders the fd one.
    """

    outputs = record.get("outputs", [])
    if not isinstance(outputs, list) or not outputs:
        return "draft"
    live = {path: live_hash(path) for path in outputs if isinstance(path, str)}
    if len(live) != len(outputs) or any(value is None for value in live.values()):
        return "update_needed"
    inputs = record.get("inputs", {})
    if not isinstance(inputs, Mapping) or any(
        not isinstance(path, str)
        or not isinstance(expected, str)
        or live_hash(path) != expected
        for path, expected in inputs.items()
    ):
        return "update_needed"
    acceptance = record.get("acceptance")
    if not isinstance(acceptance, Mapping):
        return "needs_confirmation"
    if acceptance.get("decision") == "rejected":
        return "revise"
    if acceptance.get("decision") != "accepted" or acceptance.get("outputs") != live:
        return "update_needed"
    review = record.get("review")
    if not isinstance(review, Mapping):
        return "accepted"
    if review.get("outputs") != live:
        return "update_needed"
    verdict = review.get("verdict")
    if verdict in {"approve", "approve_with_notes"}:
        return "approved"
    if verdict == "revise":
        return "revise"
    return "accepted"


def _artifact_state(root: Path, record: Mapping[str, Any]) -> str:
    return _artifact_state_from(record, lambda path: _live_hash(root, path))


def _authority_report(
    project: Mapping[str, Any], state: Mapping[str, Any]
) -> dict[str, str]:
    """Say whether the manifest still holds what each bound decision wrote.

    `set-authority` is the only sanctioned way into `short-drama.json`, and it
    already records the digest of what it wrote. Comparing that against the file
    is what turns a hand edit from invisible into reported.
    """
    bindings = state.get("authority")
    if not isinstance(bindings, Mapping):
        return {}
    report: dict[str, str] = {}
    for field, binding in bindings.items():
        if not isinstance(field, str) or not isinstance(binding, Mapping):
            continue
        try:
            tokens = _authority_tokens(field)
        except ValueError:
            report[field] = "not_authority_field"
            continue
        missing = object()
        cursor: Any = project
        for token in tokens:
            cursor = cursor.get(token, missing) if isinstance(cursor, Mapping) else missing
            if cursor is missing:
                break
        if cursor is missing:
            report[field] = "missing"
            continue
        digest = hashlib.sha256(
            json.dumps(cursor, ensure_ascii=False, sort_keys=True).encode("utf-8")
        ).hexdigest()
        report[field] = "bound" if digest == binding.get("value_sha256") else "hand_edited"
    return report


def _build_status(
    *,
    project: Mapping[str, Any],
    state: Mapping[str, Any],
    layout: Mapping[str, Any],
    project_root: str,
    artifact_state: Callable[[Mapping[str, Any]], str],
) -> dict[str, Any]:
    counts: dict[str, int] = {}
    artifacts: dict[str, str] = {}
    # Which stage produced each file. The state already knows; without it every
    # reader has to guess from the path, and a stage added later reads as
    # "unrecognised" until someone updates that reader's own list of names.
    ownership: dict[str, str] = {}
    records = state.get("artifacts", {})
    if isinstance(records, Mapping):
        for artifact_id, record in records.items():
            if not isinstance(artifact_id, str) or not isinstance(record, Mapping):
                continue
            value = artifact_state(record)
            artifacts[artifact_id] = value
            counts[value] = counts.get(value, 0) + 1
            owner = record.get("owner")
            if not isinstance(owner, str) or not owner:
                continue
            for output in record.get("outputs", []) or []:
                if isinstance(output, str) and output:
                    ownership[output] = owner
    languages = project_languages(project)
    video_model_profile = project_video_model_profile(project)
    return {
        "project_id": project.get("project_id"),
        "title": project.get("title"),
        "language": languages["language"],
        "prompt_language": languages["prompt_language"],
        "video_prompt_language": languages["video_prompt_language"],
        "video_model_profile": video_model_profile,
        "project_root": project_root,
        "last_action": state.get("last_action"),
        "layout": dict(layout),
        "artifact_states": counts,
        "artifacts": artifacts,
        "ownership": ownership,
        "authority": _authority_report(project, state),
        "lifecycle": {"artifact_state": counts},
    }


def project_status(path: Path) -> dict[str, Any]:
    root = find_project(path)
    project = json.loads((root / PROJECT_FILE).read_text(encoding="utf-8"))
    if not isinstance(project, dict):
        raise ValueError("project manifest must be an object")
    state = _read_state(root)
    return _build_status(
        project=project,
        state=state,
        layout=_project_layout_from_root(root, state),
        project_root=str(root),
        artifact_state=lambda record: _artifact_state(root, record),
    )


def _open_directory_at(directory_fd: int, parts: Iterable[str]) -> int:
    descriptor = os.dup(directory_fd)
    try:
        for part in parts:
            child = os.open(
                part,
                os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
                dir_fd=descriptor,
            )
            os.close(descriptor)
            descriptor = child
        return descriptor
    except Exception:
        os.close(descriptor)
        raise


def _read_regular_at(directory_fd: int, relative: str) -> bytes:
    pure = PurePosixPath(relative)
    parent = _open_directory_at(directory_fd, pure.parts[:-1])
    descriptor = -1
    try:
        descriptor = os.open(
            pure.name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent
        )
        details = os.fstat(descriptor)
        if not stat.S_ISREG(details.st_mode):
            raise OSError("not a regular file")
        with os.fdopen(descriptor, "rb") as handle:
            descriptor = -1
            return handle.read()
    finally:
        if descriptor >= 0:
            os.close(descriptor)
        os.close(parent)


def _directory_has_content_at(directory_fd: int, name: str) -> tuple[bool, bool]:
    try:
        descriptor = os.open(
            name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory_fd
        )
    except FileNotFoundError:
        return False, False
    except OSError:
        return True, True
    try:
        with os.scandir(descriptor) as entries:
            return any(True for _ in entries), False
    finally:
        os.close(descriptor)


def _project_layout_at(directory_fd: int, state: Mapping[str, Any]) -> dict[str, Any]:
    recorded = state.get("project_layout_mode", "auto")
    canonical: list[str] = []
    legacy: list[str] = []
    unsafe: list[str] = []
    nonstandard: list[str] = []
    with os.scandir(directory_fd) as entries:
        root_entries = list(entries)
    for entry in root_entries:
        role = _root_role(entry.name)
        if role not in LAYOUT_PINNING_ROLES:
            continue
        has_content, unsafe_entry = _directory_has_content_at(directory_fd, entry.name)
        if not has_content:
            continue
        if unsafe_entry or entry.is_symlink():
            unsafe.append(entry.name)
            continue
        if entry.name == CANONICAL_ROOTS[role]:
            canonical.append(role)
        elif entry.name == LEGACY_ROOTS[role]:
            legacy.append(role)
        else:
            nonstandard.append(entry.name)
    detected = {mode for mode, values in (("canonical", canonical), ("legacy", legacy)) if values}
    conflict = bool(unsafe or nonstandard) or len(detected) > 1 or (
        recorded in {"canonical", "legacy"} and detected and detected != {recorded}
    )
    if conflict:
        mode = "mixed"
    elif recorded in {"canonical", "legacy"}:
        mode = str(recorded)
    elif detected:
        mode = next(iter(detected))
    else:
        mode = "canonical"
    roots = LEGACY_ROOTS if mode == "legacy" else CANONICAL_ROOTS
    return {
        "mode": mode,
        "pinned": recorded != "auto" or bool(detected),
        "roots": dict(roots),
        "nonstandardRoots": sorted(nonstandard),
        "unsafeRoots": sorted(unsafe),
    }


def project_status_at(directory_fd: int, *, project_root: str) -> dict[str, Any]:
    project = json.loads(_read_regular_at(directory_fd, PROJECT_FILE).decode("utf-8"))
    try:
        raw_state = json.loads(
            _read_regular_at(directory_fd, STATE_FILE.as_posix()).decode("utf-8")
        )
    except FileNotFoundError:
        raw_state = {
            "schema_version": STATE_SCHEMA,
            "project_id": project.get("project_id"),
            "project_layout_mode": "auto",
            "last_action": "untracked",
            "artifacts": {},
        }
    if not isinstance(project, dict) or not isinstance(raw_state, dict):
        raise ValueError("project files must contain objects")
    state = _normalize_state(raw_state)
    return _build_status(
        project=project,
        state=state,
        layout=_project_layout_at(directory_fd, state),
        project_root=project_root,
        artifact_state=lambda record: _artifact_state_at(directory_fd, record),
    )


def _live_hash_at(directory_fd: int, relative: str) -> str | None:
    try:
        return sha256_bytes(_read_regular_at(directory_fd, relative))
    except (OSError, ValueError):
        return None


def _artifact_state_at(directory_fd: int, record: Mapping[str, Any]) -> str:
    return _artifact_state_from(record, lambda path: _live_hash_at(directory_fd, path))


def project_path_lifecycle_at(
    directory_fd: int, relative: str
) -> dict[str, str] | None:
    """Return the one creator-facing state for a tracked path."""

    normalized = _relative_path(relative, allow_operations=True)
    try:
        raw_state = json.loads(
            _read_regular_at(directory_fd, STATE_FILE.as_posix()).decode("utf-8")
        )
    except (FileNotFoundError, UnicodeError, json.JSONDecodeError):
        return None
    if not isinstance(raw_state, Mapping):
        return None
    state = _normalize_state(raw_state)
    for record in state["artifacts"].values():
        if isinstance(record, Mapping) and normalized in record.get("outputs", []):
            return {"artifact_state": _artifact_state_at(directory_fd, record)}
    return None


# The three functions below are the path-based half of the dashboard contract.
# Windows has no ``openat``, so the dashboard pins a project root by verified
# path there instead of by directory descriptor and calls these. They compose
# the same rules the descriptor twins do -- ``_build_status``,
# ``_artifact_state_from``, ``_normalize_state`` -- so the two halves cannot
# drift on what a status or a lifecycle state means; they differ only in how a
# file is reached.


def _read_regular(root: Path, relative: str) -> bytes:
    """Read one project file, refusing every link and reparse point on the way.

    ``os.lstat`` never follows, so a component swapped for a symlink or a
    junction is rejected rather than traversed. This is the Windows stand-in
    for opening each component with ``O_NOFOLLOW``.
    """

    pure = PurePosixPath(relative)
    current = root
    for part in pure.parts[:-1]:
        current = current / part
        details = os.lstat(current)
        if _is_link_or_reparse(details) or not stat.S_ISDIR(details.st_mode):
            raise ProjectConflictError(f"project parent is unsafe: {part}")
    target = current / pure.name
    details = os.lstat(target)
    if _is_link_or_reparse(details) or not stat.S_ISREG(details.st_mode):
        raise ProjectConflictError(f"project file is unsafe: {relative}")
    with open(target, "rb") as handle:
        return handle.read()


def project_status_from_root(root: Path, *, project_root: str) -> dict[str, Any]:
    """Report status for an already-pinned project root.

    ``project_status`` locates the project first and requires recorded state;
    this twin takes the root the caller pinned and tolerates a project that has
    never been tracked, matching ``project_status_at``.
    """

    project = json.loads(_read_regular(root, PROJECT_FILE).decode("utf-8"))
    try:
        raw_state = json.loads(
            _read_regular(root, STATE_FILE.as_posix()).decode("utf-8")
        )
    except FileNotFoundError:
        raw_state = {
            "schema_version": STATE_SCHEMA,
            "project_id": project.get("project_id"),
            "project_layout_mode": "auto",
            "last_action": "untracked",
            "artifacts": {},
        }
    if not isinstance(project, dict) or not isinstance(raw_state, dict):
        raise ValueError("project files must contain objects")
    state = _normalize_state(raw_state)
    return _build_status(
        project=project,
        state=state,
        layout=_project_layout_from_root(root, state),
        project_root=project_root,
        artifact_state=lambda record: _artifact_state(root, record),
    )


def project_path_lifecycle(root: Path, relative: str) -> dict[str, str] | None:
    """Return the one creator-facing state for a tracked path."""

    normalized = _relative_path(relative, allow_operations=True)
    try:
        raw_state = json.loads(
            _read_regular(root, STATE_FILE.as_posix()).decode("utf-8")
        )
    except (OSError, ProjectConflictError, UnicodeError, json.JSONDecodeError):
        return None
    if not isinstance(raw_state, Mapping):
        return None
    state = _normalize_state(raw_state)
    for record in state["artifacts"].values():
        if isinstance(record, Mapping) and normalized in record.get("outputs", []):
            return {"artifact_state": _artifact_state(root, record)}
    return None


@contextlib.contextmanager
def coordinated_project_text_edit(
    root: Path, relative: str, expected_version: str
) -> Iterator[None]:
    normalized = _relative_path(relative)
    if not re.fullmatch(r"[0-9a-f]{64}", expected_version):
        raise ValueError("expected version must be a SHA-256 digest")
    # The descriptor twin opens the operations directory with O_NOFOLLOW. Match
    # it: a lock taken through a redirected `.short-drama` would leave two
    # dashboards each believing they hold the project.
    try:
        operations = os.lstat(root / ".short-drama")
    except FileNotFoundError:
        operations = None
    if operations is not None and _is_link_or_reparse(operations):
        raise OSError("project operations directory is unsafe")
    with _project_lock(root):
        current = sha256_bytes(_read_regular(root, normalized))
        if current != expected_version:
            raise ProjectConflictError("file changed since it was opened")
        yield


def _validate_structured_content(relative: str, content: bytes) -> None:
    try:
        text = content.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise ValueError(f"publication text must be UTF-8: {relative}") from exc
    suffix = PurePosixPath(relative).suffix.casefold()
    if suffix not in DELIVERY_SUFFIXES:
        raise ValueError(f"publication supports Markdown, JSON, and JSONL only: {relative}")
    if suffix == ".json":
        json.loads(text)
    elif suffix == ".jsonl":
        for number, line in enumerate(text.splitlines(), 1):
            if not line.strip():
                continue
            try:
                value = json.loads(line)
            except json.JSONDecodeError as exc:
                raise ValueError(f"invalid JSONL at {relative}:{number}") from exc
            if not isinstance(value, dict):
                raise ValueError(f"JSONL records must be objects: {relative}:{number}")


def _check_output_ownership(state: Mapping[str, Any], artifact_id: str, outputs: Iterable[str]) -> None:
    wanted = set(outputs)
    artifacts = state.get("artifacts", {})
    if not isinstance(artifacts, Mapping):
        return
    for other_id, other in artifacts.items():
        if other_id == artifact_id or not isinstance(other, Mapping):
            continue
        existing = other.get("outputs", [])
        overlap = wanted.intersection(existing if isinstance(existing, list) else [])
        if overlap:
            raise ValueError(
                f"project path already belongs to {other_id}: {', '.join(sorted(overlap))}"
            )


def publish_candidate(
    root: Path,
    *,
    owner: str,
    artifact_id: str,
    outputs: Mapping[str, str | bytes],
    inputs: Iterable[str] | None = None,
    allow_unregistered_path: bool = False,
) -> dict[str, Any]:
    root = find_project(root)
    if not owner.strip() or not artifact_id.strip():
        raise ValueError("owner and artifact_id are required")
    if not outputs:
        raise ValueError("publish needs at least one output")
    normalized_outputs = _validate_path_set(root, outputs, label="publication")
    prepared: dict[str, bytes] = {}
    for relative in normalized_outputs:
        _validate_publication_layout(
            relative, allow_unregistered=allow_unregistered_path
        )
        content = outputs[relative]
        encoded = content.encode("utf-8") if isinstance(content, str) else bytes(content)
        _validate_structured_content(relative, encoded)
        prepared[relative] = encoded

    input_paths = list(inputs or [])
    normalized_inputs = _validate_path_set(root, input_paths, label="input")
    if set(normalized_inputs).intersection(normalized_outputs):
        raise ValueError("an output cannot also be its own direct input")

    with _project_lock(root):
        state = _read_state(root)
        family = _validate_project_output_layout(root, normalized_outputs)
        _check_output_ownership(state, artifact_id, normalized_outputs)
        current_inputs = _hash_mapping(root, normalized_inputs)
        for relative, content in prepared.items():
            target = _project_path(root, relative, create_parent=True)
            _atomic_bytes(target, content)
        state["artifacts"][artifact_id] = {
            "owner": owner,
            "outputs": normalized_outputs,
            "inputs": current_inputs,
            "acceptance": None,
            "review": None,
            "updated_at": utc_now(),
        }
        if state.get("project_layout_mode") == "auto" and family is not None:
            state["project_layout_mode"] = family
        _save_state(root, state, action="published")
    return {
        "artifact_id": artifact_id,
        "owner": owner,
        "outputs": normalized_outputs,
        "state": "needs_confirmation",
    }


def record_creator_acceptance(
    root: Path,
    *,
    artifact_id: str,
    decision: str,
    note: str = "",
) -> dict[str, Any]:
    if decision not in {"accepted", "rejected"}:
        raise ValueError("decision must be accepted or rejected")
    root = find_project(root)
    with _project_lock(root):
        state = _read_state(root)
        record = state["artifacts"].get(artifact_id)
        if not isinstance(record, dict):
            raise KeyError(f"unknown artifact: {artifact_id}")
        outputs = _hash_mapping(root, record.get("outputs", []))
        if not _inputs_current(root, record):
            raise ProjectConflictError("direct input changed; republish before acceptance")
        record["acceptance"] = {
            "decision": decision,
            "at": utc_now(),
            "outputs": outputs,
            "note": note.strip(),
        }
        record["review"] = None
        record["updated_at"] = utc_now()
        _save_state(root, state, action="accepted" if decision == "accepted" else "rejected")
    return {"artifact_id": artifact_id, "decision": decision, "state": _artifact_state(root, record)}


def record_review(
    root: Path,
    *,
    artifact_id: str,
    verdict: str,
    reviewer: str = "",
    note: str = "",
) -> dict[str, Any]:
    if verdict not in {"approve", "approve_with_notes", "revise", "provisional"}:
        raise ValueError("unsupported review verdict")
    root = find_project(root)
    with _project_lock(root):
        state = _read_state(root)
        record = state["artifacts"].get(artifact_id)
        if not isinstance(record, dict):
            raise KeyError(f"unknown artifact: {artifact_id}")
        outputs = _hash_mapping(root, record.get("outputs", []))
        acceptance = record.get("acceptance")
        if (
            not isinstance(acceptance, Mapping)
            or acceptance.get("decision") != "accepted"
            or acceptance.get("outputs") != outputs
        ):
            raise ProjectConflictError("review requires current creator acceptance")
        if not _inputs_current(root, record):
            raise ProjectConflictError("direct input changed; republish before review")
        record["review"] = {
            "verdict": verdict,
            "at": utc_now(),
            "outputs": outputs,
            "reviewer": reviewer.strip(),
            "note": note.strip(),
        }
        record["updated_at"] = utc_now()
        _save_state(root, state, action="reviewed")
    return {"artifact_id": artifact_id, "verdict": verdict, "state": _artifact_state(root, record)}


def _authority_tokens(field: str) -> list[str]:
    if not field.startswith("/"):
        raise ValueError(f"--field must be a JSON pointer starting with /: {field}")
    tokens = [
        token.replace("~1", "/").replace("~0", "~") for token in field[1:].split("/")
    ]
    if any(not token for token in tokens):
        raise ValueError(f"--field has an empty pointer segment: {field}")
    if field not in FORMAT_POINTERS and (
        tokens[0] != AUTHORITY_ROOT_TOKEN or len(tokens) < 2
    ):
        raise ValueError(
            f"set-authority writes /{AUTHORITY_ROOT_TOKEN}/* and "
            f"{', '.join(FORMAT_POINTERS)} only"
        )
    if tokens[:2] == [AUTHORITY_ROOT_TOKEN, "decisions_artifact"]:
        # Where decisions are kept is project layout, not a creative choice; a
        # decision record must not move the place its own successors are read from.
        raise ValueError("decisions_artifact is project layout, not a creator choice")
    return tokens


def _accepted_decision_value(
    root: Path,
    state: Mapping[str, Any],
    *,
    decision_path: str,
    decision_id: str,
    field: str,
) -> Any:
    relative = _relative_path(decision_path)
    if _root_role(PurePosixPath(relative).parts[0]) != "creator-decisions":
        expected = CANONICAL_ROOTS["creator-decisions"]
        raise ValueError(f"creator decisions live in {expected}/: {relative}")
    try:
        _, record = _artifact_for_path(state, relative)
    except PackageBlockedError as exc:
        raise ValueError(f"creator decision file is not a published artifact: {relative}") from exc
    if _artifact_state(root, record) not in {"accepted", "approved"}:
        raise ValueError(f"creator decision file is not accepted and current: {relative}")
    text = _project_path(root, relative).read_text(encoding="utf-8")
    latest: tuple[int, Mapping[str, Any]] | None = None
    superseded_by: str | None = None
    for number, line in enumerate(text.splitlines(), 1):
        if not line.strip():
            continue
        try:
            decision = json.loads(line)
        except json.JSONDecodeError as exc:
            raise ValueError(f"{relative}:{number} is not a creator decision record: {exc}") from exc
        if not isinstance(decision, Mapping):
            continue
        # The file is append-only, so a revision arrives as a later line. Read the
        # whole file before deciding: the first match may already be retracted.
        if decision.get("decision_id") == decision_id:
            latest = (number, decision)
        elif (
            decision.get("supersedes_decision_id") == decision_id
            and decision.get("status") == "accepted"
        ):
            superseded_by = str(decision.get("decision_id"))
    if latest is None:
        raise KeyError(f"unknown creator decision: {decision_id}")
    if superseded_by is not None:
        raise ValueError(f"{decision_id} was superseded by {superseded_by}")
    number, decision = latest
    if decision.get("status") != "accepted":
        raise ValueError(f"{decision_id} is not an accepted creator decision")
    locators = decision.get("target_locators")
    if not isinstance(locators, list) or not any(
        isinstance(locator, Mapping)
        and locator.get("src") == "short-drama"
        and locator.get("field") == field
        for locator in locators
    ):
        raise ValueError(f"{decision_id} does not target {field}")
    if "accepted_value" not in decision:
        raise ValueError(f"{decision_id} carries no accepted_value at {relative}:{number}")
    return decision["accepted_value"]


def _json_kind(value: Any) -> str:
    """The JSON type of a value. 90 and 92.5 are both numbers."""
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, (int, float)):
        return "number"
    if isinstance(value, str):
        return "string"
    if isinstance(value, Mapping):
        return "object"
    if isinstance(value, list):
        return "array"
    return "null" if value is None else type(value).__name__


def _write_authority_value(project: dict[str, Any], tokens: list[str], value: Any) -> Any:
    cursor: Any = project
    blocks: list[dict[str, Any]] = []
    walked: list[str] = []
    for token in tokens[:-1]:
        cursor = cursor.get(token) if isinstance(cursor, dict) else None
        walked.append(token)
        if not isinstance(cursor, dict):
            raise ValueError(f"project manifest has no object at /{'/'.join(walked)}")
        if "status" in cursor:
            blocks.append(cursor)
    leaf = tokens[-1]
    if leaf not in cursor:
        # The manifest shape is declared by the project template. A decision may
        # fill a declared slot; inventing one would put a field downstream cannot
        # know to read.
        raise ValueError(f"project manifest declares no /{'/'.join([*walked, leaf])}")
    current = cursor.get(leaf)
    if isinstance(current, Mapping) and not isinstance(value, Mapping) and "status" not in current:
        # A choices map is merged, never replaced wholesale: replacing it would
        # silently drop the choices a previous decision already recorded.
        raise ValueError(f"/{'/'.join([*walked, leaf])} needs an object accepted_value")
    if isinstance(current, Mapping) and isinstance(value, Mapping) and "status" not in current:
        merged = {**current, **value}
        cursor[leaf] = merged
        for block in blocks:
            block["status"] = "accepted"
        return merged
    if not (isinstance(current, dict) and "status" in current):
        if current is not None and _json_kind(value) != _json_kind(current):
            raise ValueError(
                f"/{'/'.join([*walked, leaf])} is {_json_kind(current)}; "
                f"accepted_value is {_json_kind(value)}"
            )
        cursor[leaf] = value
        # Writing one choice inside an authority block accepts that block: a
        # downstream stage gates on the block's status before reading the choice.
        for block in blocks:
            block["status"] = "accepted"
        return value
    if not isinstance(value, Mapping) or not value:
        raise ValueError(f"/{'/'.join(tokens)} needs a non-empty object accepted_value")
    if "status" in value:
        raise ValueError("accepted_value must not carry its own status")
    block = dict(current)
    choices = block.get("choices")
    if isinstance(choices, Mapping):
        block["choices"] = {**choices, **value}
    else:
        block.update(value)
    block["status"] = "accepted"
    cursor[leaf] = block
    return block


def set_creator_authority(
    root: Path,
    *,
    field: str,
    decision_path: str,
    decision_id: str,
) -> dict[str, Any]:
    tokens = _authority_tokens(field)
    root = find_project(root)
    with _project_lock(root):
        state = _read_state(root)
        value = _accepted_decision_value(
            root,
            state,
            decision_path=decision_path,
            decision_id=decision_id,
            field=field,
        )
        if field == EPISODE_LENGTH_POINTER and not (
            isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0
        ):
            raise ValueError("target_seconds_per_episode must be a positive number of seconds")
        project_path = root / PROJECT_FILE
        project = json.loads(project_path.read_text(encoding="utf-8"))
        if not isinstance(project, dict):
            raise ValueError("project manifest must be an object")
        written = _write_authority_value(project, tokens, value)
        if field == PACING_POINTER:
            # Check what the manifest will hold, not what the decision said: an
            # object slot is merged, so a half decision leaves the other rate at
            # null and the estimate stays unusable while the write reports bound.
            # Type only — which rates make an estimate usable is the write
            # stage's call, and it stays in one place.
            if not isinstance(written, Mapping) or not written:
                raise ValueError("pacing must be an object of named rates")
            for name, rate in written.items():
                if not (
                    isinstance(rate, (int, float))
                    and not isinstance(rate, bool)
                    and rate > 0
                ):
                    raise ValueError(
                        f"pacing rate {name} must be a positive number; "
                        f"a decision that sets only some rates leaves the rest unset"
                    )
        atomic_json(project_path, project)
        # Record which decision produced the value, so a later reader can tell a
        # bound write from a hand edit and see what a re-bind replaced.
        bindings = state.get("authority")
        if not isinstance(bindings, dict):
            bindings = {}
            state["authority"] = bindings
        bindings[field] = {
            "decision": f"{_relative_path(decision_path)}#{decision_id}",
            # The written value, so a later reader can tell the manifest still
            # holds what the decision said rather than a hand edit made since.
            "value_sha256": 

#!/usr/bin/env python3
"""Offline self-test for the standalone project lifecycle."""

from __future__ import annotations

import json
import tempfile
import sys
from collections.abc import Iterator
from pathlib import Path
from typing import Any, NamedTuple

from project_tool import initialize_project, project_status, publish_candidate

MINIMUM_PYTHON = (3, 9)
if sys.version_info < MINIMUM_PYTHON:
    raise SystemExit("selftest.py requires Python 3.9 or newer")

SKILL_ROOT = Path(__file__).resolve().parent.parent
EXAMPLE_RECORDS = (
    "assets/creator-decision.example.jsonl",
    "assets/production-observation.example.jsonl",
    "assets/reference-observation.example.jsonl",
)

# ---------------------------------------------------------------------------
# REFERENCE RESOLVER -- reference implementation.
#
# Each skill carries its own copy of this block. The suite has no shared
# library on purpose: a skill must stay runnable after copying only its own
# directory, so duplicating these few lines across skills is the correct shape.
# Copy the block verbatim; do not import it.
# ---------------------------------------------------------------------------

SOURCES_RECORD_TYPE = "sources"
SOURCES_SCHEMA_VERSION = "1.0.0"


class ResolvedRef(NamedTuple):
    """An upstream reference with its snapshot resolved, whichever form it used."""

    owner: str
    artifact: str
    record_id: str | None
    field: str | None
    authority: str | None


class RefFinding(NamedTuple):
    """A structural defect in a reference object."""

    code: str
    location: str
    detail: str


def load_sources(document: Any) -> dict[str, dict[str, Any]]:
    """Return the ``sources`` declaration of a parsed file, or ``{}`` if absent.

    Accepts a parsed ``.json`` document (a dict) or the parsed record list of a
    ``.jsonl`` file, whose declaration lives on the first record.
    """
    if isinstance(document, list):
        document = document[0] if document else None
    if not isinstance(document, dict):
        return {}
    declared = document.get("sources")
    if not isinstance(declared, dict):
        return {}
    return {key: value for key, value in declared.items() if isinstance(value, dict)}


def resolve_ref(
    ref: Any, sources: dict[str, dict[str, Any]], location: str
) -> tuple[ResolvedRef | None, RefFinding | None]:
    """Resolve a reference object written in either the compact or expanded form."""
    if not isinstance(ref, dict):
        return None, RefFinding("REF_IS_NOT_AN_OBJECT", location, f"got {type(ref).__name__}")
    src = ref.get("src")
    if isinstance(src, str):
        entry = sources.get(src)
        if entry is None:
            return None, RefFinding(
                "REF_SRC_IS_NOT_DECLARED", location, f"src {src!r} has no sources entry"
            )
        owner, artifact = entry.get("owner"), entry.get("artifact")
        if not (isinstance(owner, str) and isinstance(artifact, str)):
            return None, RefFinding(
                "SOURCE_ENTRY_IS_INCOMPLETE",
                location,
                f"sources[{src!r}] needs owner/artifact",
            )
    elif all(isinstance(ref.get(key), str) for key in ("owner", "artifact")):
        owner, artifact = ref["owner"], ref["artifact"]
    else:
        return None, RefFinding(
            "REF_HAS_NO_UPSTREAM_BINDING", location, "needs src, or owner+artifact"
        )
    optional = {
        key: ref[key] for key in ("record_id", "field", "authority") if isinstance(ref.get(key), str)
    }
    return (
        ResolvedRef(
        owner,
        artifact,
            optional.get("record_id"),
            optional.get("field"),
            optional.get("authority"),
        ),
        None,
    )


# ---------------------------------------------------------------------------
# END REFERENCE RESOLVER
# ---------------------------------------------------------------------------

# A reference lives under a slot named by convention. `target_locators` is the
# creator-decision carrier that predates the suffix rule.
REF_CARRIER_SUFFIXES = ("_ref", "_refs")
EXTRA_REF_CARRIERS = ("target_locators",)


def is_ref_carrier(key: str) -> bool:
    return key.endswith(REF_CARRIER_SUFFIXES) or key in EXTRA_REF_CARRIERS


def iter_refs(node: Any, location: str) -> Iterator[tuple[Any, str]]:
    """Yield every reference object carried anywhere inside a record."""
    if isinstance(node, list):
        for index, item in enumerate(node):
            yield from iter_refs(item, f"{location}[{index}]")
        return
    if not isinstance(node, dict):
        return
    for key, value in node.items():
        where = f"{location}/{key}"
        if is_ref_carrier(key) and value is not None:
            if isinstance(value, list):
                for index, ref in enumerate(value):
                    yield ref, f"{where}[{index}]"
            else:
                yield value, where
            continue
        yield from iter_refs(value, where)


def read_records(path: Path) -> list[Any]:
    return [
        json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()
    ]


def check_records(records: list[Any], name: str) -> tuple[int, list[RefFinding]]:
    """Resolve every reference in a .jsonl file through its own declaration."""
    sources = load_sources(records)
    resolved = 0
    findings: list[RefFinding] = []
    for number, record in enumerate(records, 1):
        if isinstance(record, dict) and record.get("record_type") == SOURCES_RECORD_TYPE:
            continue
        for ref, where in iter_refs(record, f"{name}:{number}"):
            found, finding = resolve_ref(ref, sources, where)
            if finding is not None:
                findings.append(finding)
            if found is not None:
                resolved += 1
    return resolved, findings


CHECKS = 0


def require(condition: bool, message: str) -> None:
    global CHECKS
    if not condition:
        raise AssertionError(message)
    CHECKS += 1


def main() -> int:
    for relative in EXAMPLE_RECORDS:
        path = SKILL_ROOT / relative
        records = read_records(path)
        declared = load_sources(records)
        require(bool(declared), f"{relative} declares its upstream snapshots")
        resolved, findings = check_records(records, path.name)
        require(not findings, f"{relative} reference findings: {findings}")
        require(resolved > 0, f"{relative} carries at least one resolvable reference")

    sources = {"screenplay": {"owner": "short-drama-write", "artifact": "a.md", "hash": "0" * 64}}
    _, undeclared = resolve_ref({"src": "missing", "record_id": "BLK-1"}, sources, "synthetic")
    require(
        undeclared is not None and undeclared.code == "REF_SRC_IS_NOT_DECLARED",
        "a src without a sources entry is a structural finding",
    )
    _, unbound = resolve_ref({"record_id": "BLK-1"}, sources, "synthetic")
    require(
        unbound is not None and unbound.code == "REF_HAS_NO_UPSTREAM_BINDING",
        "a reference binding neither way is a structural finding",
    )
    expanded, finding = resolve_ref(
        {
            "owner": "short-drama-write",
            "artifact": "剧集/EP001/screenplay-index.jsonl",
            "hash": "a" * 64,
            "record_id": "BLK-EP001-SC001-A01",
        },
        {},
        "synthetic",
    )
    require(
        finding is None and expanded is not None and expanded.record_id == "BLK-EP001-SC001-A01",
        "a project written before the declaration still resolves",
    )

    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory) / "project"
        created = initialize_project(
            root,
            title="Self-test",
            language="zh-CN",
            prompt_language="en",
            aspect_ratio="9:16",
        )
        require(created["project"]["title"] == "Self-test", "project title")
        require(
            project_status(root)["project_id"] == created["project"]["project_id"],
            "project discovery",
        )

        published = publish_candidate(
            root,
            owner="selftest",
            artifact_id="SELFTEST-001",
            outputs={"剧集/EP001/screenplay.md": "# EP001\n"},
        )
        require(published["state"] == "needs_confirmation", "publication state")

        header = json.dumps(
            {
                "record_type": SOURCES_RECORD_TYPE,
                "schema_version": SOURCES_SCHEMA_VERSION,
                "sources": {
                    "screenplay": {
                        "owner": "short-drama-write",
                        "artifact": "剧集/EP001/screenplay.md",
                        "hash": "a" * 64,
                    }
                },
            },
            ensure_ascii=False,
        )
        record = json.dumps(
            {"shot_id": "SHOT-001", "source_ref": {"src": "screenplay", "record_id": "BLK-1"}},
            ensure_ascii=False,
        )
        declared_publication = publish_candidate(
            root,
            owner="selftest",
            artifact_id="SELFTEST-002",
            outputs={"剧集/EP001/storyboard/shots.jsonl": f"{header}\n{record}\n"},
        )
        require(
            declared_publication["state"] == "needs_confirmation",
            "a file declaring its upstream snapshots publishes",
        )

        try:
            publish_candidate(
                root,
                owner="selftest",
                artifact_id="SELFTEST-UNSAFE",
                outputs={"../escape.md": "unsafe"},
            )
        except ValueError:
            rejected = True
        else:
            rejected = False
        require(rejected, "unsafe publication path was accepted")

    print(f"{CHECKS} self-tests passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())