SW-05 p3 fragment 组成数据化：窗口/投影/Thread 注入进 profile.context（ADR-0017）

adr/0017-p3-context-profile-section.md

@@ -0,0 +1,50 @@
+# ADR-0017：p3 fragment 组成数据化（profile.context 段）
+
+- 状态：proposed
+- 日期：2026-08-22
+- 影响层：A 资产层（profiles/_schema.py + profile yaml）；B 层仅消费
+
+## 背景
+
+SW-05（上游依赖卡）：p3 的跨集上下文组成硬编码在 `pipeline.py`——
+`prev_episode_summary` 窗口恒为 1（只看上一集）、`known_facts` 投影恒为五字段、
+p2 规划的 `threads` 表永不注入 p3。弱模型需要更长的历史窗口与更窄的投影面，
+强模型可以相反；这些是 profile 级策略，不是代码常量。
+
+## 决定
+
+Profile 新增 `context` 段（缺省 = 原行为，零变化）：
+
+| 键 | 语义 | 缺省 | 消费点 |
+|---|---|---|---|
+| `context.prev_summary_window` | p3 `prev_episode_summary` 看近端 N 集（0=恒空） | 1 | `run_pipeline` p3 循环 + `recompile_episode`（`_window_join`：按时间序逐行拼接，远端在前、近端在后，与 compress_history 的【前情】→【上一集】布局一致） |
+| `context.known_fact_fields` | `known_facts` 投影字段（白名单子集） | `[id, content, episode_no, status, type]` | `_known_facts` |
+| `context.inject_threads` | 是否把 p2 的 Thread 表注入 p3（投影 `{id,title,status,state}`） | false | `run_pipeline`/`recompile_episode` 组装 fragment，p3 转发进 LLM 输入 |
+
+`profiles/_schema.py` 新增 `ContextSettings`（`extra="forbid"`；`known_fact_fields`
+经校验器限制在白名单内）。后续上下文预算旋钮（SW-06）并入同段。
+
+## 被否决的替代
+
+| 替代 | 为什么否决 |
+|---|---|
+| 环境变量 | 与 SW-04 同理：策略被运行时稀释，不进缓存键 |
+| 每集覆盖（brief 级） | 组成策略是 profile 资产，逐 brief 覆盖会让缓存键粒度爆炸 |
+| 直接注入全量 threads/facts | p3 输入预算失控；投影面必须显式声明 |
+
+## 对下游的约束
+
+- 改窗口/投影只动 profile yaml；`_KNOWN_FACT_FIELDS` 白名单新增字段需同步本 ADR 的表。
+- window=1 时 `prev_episode_summary` 与原实现逐字节一致（既有快照/桩测试守护）。
+- `known_fact_fields: []` 是合法的显式空投影（有意隐藏全部字段），不等同于未配置（review 修正）。
+- 窗口内多集按时间序排列（远端在前），与 compress_history 输出布局一致（review 澄清，初稿表述有误）。
+
+## 迁移
+
+非 breaking：三键全有缺省；在库 profile（short_drama_v1 / short_video_v1）显式
+写入缺省值以便发现。
+
+## 验证
+
+`tests/test_p3_context_config.py`：投影字段与缺省回退、窗口=1 回归、窗口=2 含
+祖父集摘要且近端在前、threads 开关注入/缺省不注入；全量 `pytest -m "not llm"` 绿。

profiles/_schema.py

@@ -4,7 +4,7 @@
 
 from typing import Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from spec.ir.nodes import Slug
 
@@ -36,6 +36,31 @@ class BeatTemplate(BaseModel):
     note: str = ""
 
 
+class ContextSettings(BaseModel):
+    """SW-05 / ADR-0017：p3 fragment 组成旋钮。缺省值 = 原代码行为（零变化）。"""
+
+    model_config = ConfigDict(extra="forbid")
+    prev_summary_window: int = Field(
+        default=1, ge=0, description="p3 prev_episode_summary 的近端窗口（集数）"
+    )
+    known_fact_fields: list[str] = Field(
+        default_factory=lambda: ["id", "content", "episode_no", "status", "type"],
+        description="known_facts 投影字段（白名单子集）",
+    )
+    inject_threads: bool = Field(
+        default=False, description="是否把 p2 的 Thread 表注入 p3 fragment"
+    )
+
+    @field_validator("known_fact_fields")
+    @classmethod
+    def _within_whitelist(cls, v: list[str]) -> list[str]:
+        allowed = {"id", "content", "episode_no", "status", "type"}
+        bad = [x for x in v if x not in allowed]
+        if bad:
+            raise ValueError(f"known_fact_fields 白名单外字段：{bad}（允许：{sorted(allowed)}）")
+        return v
+
+
 class PipelineSettings(BaseModel):
     """SW-07 / ADR-0016：管线重试策略。缺省值 = 提案前的代码常量（零行为变化）。"""
 
@@ -87,6 +112,7 @@ class Profile(BaseModel):
 
     beat_templates: list[BeatTemplate] = Field(default_factory=list)
     novel: NovelSettings = Field(default_factory=NovelSettings)
+    context: ContextSettings = Field(default_factory=ContextSettings)
 
     #: SW-07 / ADR-0016：管线策略（重试与定向重生成次数）。缺省 = 原代码常量。
     pipeline: PipelineSettings = Field(default_factory=lambda: PipelineSettings())

profiles/short_drama_v1.yaml

@@ -52,6 +52,8 @@ model_tiers:
   p4_scene: tier_draft
   p5_dialogue: tier_draft
   p6_prose: tier_draft
+# SW-05 / ADR-0017：p3 跨集上下文组成（缺省即原行为）
+context: {prev_summary_window: 1, known_fact_fields: [id, content, episode_no, status, type], inject_threads: false}
 # SW-07 / ADR-0016：管线策略（缺省即原代码常量；弱模型/强模型 profile 可分道调参）
 pipeline: {pass_attempts: 2, phase_attempts: 3}
 retrieval: {top_k: 3}

profiles/short_video_v1.yaml

@@ -56,6 +56,8 @@ model_tiers:
   p4_scene: tier_draft
   p5_dialogue: tier_draft
   p6_prose: tier_draft
+# SW-05 / ADR-0017：p3 跨集上下文组成（缺省即原行为）
+context: {prev_summary_window: 1, known_fact_fields: [id, content, episode_no, status, type], inject_threads: false}
 # SW-07 / ADR-0016：管线策略（缺省即原代码常量；弱模型/强模型 profile 可分道调参）
 pipeline: {pass_attempts: 2, phase_attempts: 3}
 retrieval: {top_k: 3}

src/nsc/passes/p3_beatsheet.py

@@ -76,6 +76,10 @@ def run(ctx: PassContext, fragment: dict[str, Any]) -> dict[str, Any]:
     revivable = str(fragment.get("revivable_ideas", "") or "")
     if revivable:
         inputs["revivable_ideas"] = revivable
+    # SW-05：Thread 注入（profile.context.inject_threads 开关，pipeline 组装 JSON）
+    threads = str(fragment.get("threads", "") or "")
+    if threads:
+        inputs["threads"] = threads
     out = Module()(ctx, with_diag(inputs, fragment))
     raw_beats = inner_json(out["beats_json"], "p3_beatsheet", "beats_json")
     raw_sps = inner_json(out["setup_payoffs_json"], "p3_beatsheet", "setup_payoffs_json")

src/nsc/passes/pipeline.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 import re
 import sys
@@ -348,6 +349,12 @@ def track() -> None:
     check_stage(ctx, cur(), "after_p2", "after_p2")
 
     episodes = r2["episodes"]
+    # SW-05：p3 fragment 组成旋钮（profile.context.*，缺省 = 原行为）
+    p3_ctx = ctx.profile.get("context", {})
+    prev_window = max(0, int(p3_ctx.get("prev_summary_window", 1)))
+    fact_fields = _known_fact_fields_of(ctx.profile)
+    inject_threads = bool(p3_ctx.get("inject_threads", False))
+    ep_summaries: list[str] = []
     # p3（逐集）+ p4 + after_p3/p4 检查作为一个相位：L0 拦截（如 BM-002 植入间隔）时
     # 带诊断整体重试（D13 反馈驱动再生成，诊断累积）；重试前恢复相位前的状态。
     diag = ""
@@ -358,7 +365,6 @@ def track() -> None:
         }
         ep_state_snap = {e["id"]: list(e.get("state_changes", [])) for e in episodes}
         try:
-            prev_summary = ""
             for i, ep in enumerate(episodes):
                 placement = [
                     p for p in r2["placement_plan"] if int(p.get("episode_no", -1)) == ep["no"]
@@ -368,22 +374,24 @@ def track() -> None:
                     "bible": bible,
                     "placement": placement,
                     "required_brand_moment_beats": len(placement),
-                    "prev_episode_summary": prev_summary,
+                    "prev_episode_summary": _window_join(ep_summaries, prev_window),
                     "next_episode_promise": episodes[i + 1]["hook_promise"]
                     if i + 1 < len(episodes)
                     else "",
                     # ADR-0012：跨集回收上下文 + 状态变更声明域（facts/threads 等表见上）
-                    "known_facts": _known_facts(st["facts"]),
+                    "known_facts": _known_facts(st["facts"], fact_fields),
                     "declared_state": _declared_state(st),
                     "retrieved_cases": _retrieved(ctx, "beat_sequence", _ep_query(ep)),
                 }
                 if diag:
                     frag3["_previous_failure"] = diag
                 if revivable:
                     frag3["revivable_ideas"] = revivable
+                if inject_threads:
+                    frag3["threads"] = _threads_view(st["threads"])
                 r3 = _retry_pass(p3_beatsheet.run, ctx, frag3)
                 track()
-                prev_summary = "；".join(b["summary"] for b in r3["beats"])
+                ep_summaries.append("；".join(b["summary"] for b in r3["beats"]))
                 st["beats"] += r3["beats"]
                 st["setup_payoffs"] += r3["setup_payoffs"]
                 st["brand_moments"] += r3["brand_moments"]
@@ -503,23 +511,32 @@ def track() -> None:
     bank_db = _state_db(ctx, pid)
     # T-41：上次重编译存入 bank 的未复活素材作为可选注入层（空则不加）
     revivable = _revivable_layer(bank_db, pid)
+    # SW-05：p3 fragment 组成旋钮（与 run_pipeline 同一 profile.context.* 段）
+    _p3c = ctx.profile.get("context", {})
+    r_window = max(0, int(_p3c.get("prev_summary_window", 1)))
 
     frag3 = {
         "episode": ep,
         "bible": bible,
         "placement": _placement_of(raw, ep),
         "required_brand_moment_beats": len(_placement_of(raw, ep)),
-        "prev_episode_summary": _episode_digest(raw, ordered[idx - 1]["id"]) if idx > 0 else "",
+        "prev_episode_summary": _window_join(
+            [_episode_digest(raw, e["id"]) for e in ordered[max(0, idx - r_window) : idx]],
+            r_window,
+        ),
         "next_episode_promise": ordered[idx + 1]["hook_promise"] if idx + 1 < len(ordered) else "",
         # ADR-0012：其他集已成立的 facts 可被本集回收；状态声明域来自 IR 表
         "known_facts": _known_facts(
-            [f for f in raw.get("facts", []) if f.get("episode_no") != ep["no"]]
+            [f for f in raw.get("facts", []) if f.get("episode_no") != ep["no"]],
+            _known_fact_fields_of(ctx.profile),
         ),
         "declared_state": _declared_state(raw),
         "retrieved_cases": _retrieved(ctx, "beat_sequence", _ep_query(ep)),
     }
     if revivable:
         frag3["revivable_ideas"] = revivable
+    if _p3c.get("inject_threads", False):
+        frag3["threads"] = _threads_view(raw.get("threads", []))
     r3 = _retry_pass(p3_beatsheet.run, ctx, frag3)
     track()
     r4 = _retry_pass(p4_scene.run, ctx, {"episode": ep, "beats": r3["beats"], "bible": bible})
@@ -647,21 +664,67 @@ def _episode_digest(raw: dict[str, Any], ep_id: str) -> str:
     return "；".join(b["summary"] for b in raw["beats"] if b["parent_id"] in scene_ids)
 
 
-def _known_facts(facts: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """p3 的跨集回收上下文（ADR-0012）：已成立 Fact 的最小可见面，供模型引用 id 做回收。"""
+# --- SW-05 / ADR-0017：p3 fragment 组成的 profile 旋钮（context.* 段，缺省 = 原行为） ---
+
+#: known_facts 投影字段全集（值 = 缺省填充，机制映射，非业务规则）。
+_KNOWN_FACT_FIELDS: dict[str, Any] = {
+    "id": "",
+    "content": "",
+    "episode_no": 1,
+    "status": "active",
+    "type": "plot_event",
+}
+
+
+def _known_fact_fields_of(profile: dict[str, Any]) -> tuple[str, ...]:
+    """profile.context.known_fact_fields → 投影字段（白名单交集，缺省 = 原五字段）。
+
+    显式空列表是合法配置（投影面为空，即有意隐藏全部字段），不与"未配置"混同。
+    """
+    raw = profile.get("context", {}).get("known_fact_fields")
+    wanted = list(_KNOWN_FACT_FIELDS) if raw is None else raw
+    return tuple(k for k in _KNOWN_FACT_FIELDS if k in wanted)
+
+
+def _known_facts(
+    facts: list[dict[str, Any]], fields: tuple[str, ...] = tuple(_KNOWN_FACT_FIELDS)
+) -> list[dict[str, Any]]:
+    """p3 的跨集回收上下文（ADR-0012）：已成立 Fact 的最小可见面，供模型引用 id 做回收。
+
+    SW-05：投影字段由 profile.context.known_fact_fields 决定（缺省 = 原五字段）。
+    """
     return [
-        {
-            "id": f["id"],
-            "content": f["content"],
-            "episode_no": f.get("episode_no", 1),
-            "status": f.get("status", "active"),
-            "type": f.get("type", "plot_event"),
-        }
+        {k: f.get(k, default) for k, default in _KNOWN_FACT_FIELDS.items() if k in fields}
         for f in facts
         if isinstance(f, dict)
     ]
 
 
+def _window_join(summaries: list[str], window: int) -> str:
+    """prev_episode_summary 窗口（SW-05）：按时间序拼接（远端在前、近端在后）。
+
+    与 compress_history 的【前情】→【上一集】布局一致（review 澄清：chronological）。
+    window=1 与原行为逐字节一致（单元素直接返回）；window=0 即恒空串。
+    """
+    n = max(0, int(window))
+    return "\n".join(summaries[-n:]) if n else ""
+
+
+def _threads_view(threads: list[Any]) -> str:
+    """Thread 注入面（SW-05）：p2 规划的叙事线索标题/状态，供 p3 做跨集呼应。"""
+    view = [
+        {
+            "id": t.get("id", ""),
+            "title": t.get("title", ""),
+            "status": t.get("status", "active"),
+            "state": t.get("state", ""),
+        }
+        for t in threads
+        if isinstance(t, dict)
+    ]
+    return json.dumps(view, ensure_ascii=False)
+
+
 def _declared_state(st: dict[str, Any]) -> dict[str, Any]:
     """p3 的状态变更声明域（ADR-0012）：只允许对已声明的 StateVariable/DarkThread key 变更。"""
     return {