SW-06 接线休眠模块：assembler P2-P4 进 p3/p5、compress_history 进 p3 远端历史（ADR-0018）

adr/0018-context-wiring.md

@@ -0,0 +1,53 @@
+# ADR-0018：接线上下文预算装配与历史压缩（context.assembler / context.compress）
+
+- 状态：proposed
+- 日期：2026-08-22
+- 影响层：A 资产层（profiles/_schema.py + profile yaml）；B 层消费
+
+## 背景
+
+SW-06（上游依赖卡）：`nsc.context.assemble`（T-33）与 `compress_history` 在 main 上
+已实现但未接线——没有任何 Pass 的输入过预算装配，p3 的远端历史永远走原文窗口。
+
+## 决定
+
+1. **P2-P4 层接入 p3/p5**（`nsc.passes.assemble_context` 统一入口）：
+   - p3：P1=episode_json（不可裁剪锚）；P2=prev_episode_summary（SW-05 窗口文本）；
+     P3=known_facts 逐条；P4=retrieved_cases；P5=bible/profile 参考层（低保）。
+   - p5：P1=scene_json+beats_json；P4=retrieved_cases；P5=characters/profile 参考层。
+   - 预算读 `context.budget` / `context.core_guarantee`；降级顺序由 assembler 既定
+     语义决定（P4 整层丢 → P2 截尾 → P3 截断 → P5 低保）。
+2. **compress_history 接入 p3 远端历史**（`pipeline._history_text`）：
+   `context.history_compress: true` 且窗口宽于 `history_keep_recent` 时，窗口内远端集
+   经 `make_llm_summarizer`（LLM 出口走 models 路由）压缩成"【前情】"，近端集保
+   原文"【上一集】"；否则退回 SW-05 的原文窗口。
+
+## 缺省零变化（关键设计约束）
+
+- `budget=32768` 足够大 → 装配全存活，p3/p5 输入与接线前逐字节等价；
+- `history_compress=false` → 永不产生压缩 LLM 调用，前情文本 = SW-05 `_window_join`。
+- 压缩不设缺省开启的原因：compress_history 的输出带"【前情】/【上一集】"标记，
+  默认开启会改变既有 prompt 字节内容（缓存键漂移）；开关交给 profile 显式打开。
+
+## 被否决的替代
+
+| 替代 | 为什么否决 |
+|---|---|
+| 默认开启压缩 | 改变缺省 prompt 字节内容，违背"缺省零变化"约束 |
+| 在 pipeline 组装层做预算 | 装配是 Pass 输入语义（p6 先例在 Pass 内），pipeline 只管历史文本来源 |
+| p5 也接 P2/P3 | p5 输入无前情/事实层（场景级编译）；接了也是空层 |
+
+## 对下游的约束
+
+- 降级诊断（degraded/dropped）目前只进 assembler 返回值；若要进 runs 表需另卡。
+- `assemble_context` 的 P3/P5 存活重建是前缀式的：条目顺序即优先级，不得乱序。
+
+## 迁移
+
+非 breaking；在库 profile 写入缺省值。依赖 SW-05 的 `context` 段（本 ADR 与
+ADR-0017 同段扩容），本卡分支基于 sw/sw-05-p3-context-config。
+
+## 验证
+
+`tests/test_context_wiring.py`：缺省全存活/紧预算按序降级、p5 装配、压缩接线
+（远端 SUM / 近端原文 / 缺省零压缩调用）；全量 `pytest -m "not llm"` 绿。

profiles/_schema.py

@@ -50,6 +50,18 @@ class ContextSettings(BaseModel):
     inject_threads: bool = Field(
         default=False, description="是否把 p2 的 Thread 表注入 p3 fragment"
     )
+    #: SW-06 / ADR-0018：P0-P5 上下文预算与历史压缩（nsc.context.assembler/compress）。
+    budget: int = Field(
+        default=32768, gt=0, description="P0-P5 总预算（token）；P1 装不下即 PassFailure"
+    )
+    core_guarantee: int = Field(default=400, ge=1, description="P5 参考层低保额（token）")
+    history_compress: bool = Field(
+        default=False, description="远端历史是否走 LLM 压缩（compress_history 接线开关）"
+    )
+    history_keep_recent: int = Field(default=1, ge=0, description="历史压缩保留近端集数")
+    history_compress_ratio: float = Field(
+        default=0.1, gt=0, le=1, description="远端历史压缩目标长度比"
+    )
 
     @field_validator("known_fact_fields")
     @classmethod

profiles/short_drama_v1.yaml

@@ -53,7 +53,15 @@ model_tiers:
   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}
+context:
+  prev_summary_window: 1
+  known_fact_fields: [id, content, episode_no, status, type]
+  inject_threads: false
+  budget: 32768            # SW-06 / ADR-0018：P0-P5 总预算（token）
+  core_guarantee: 400      # P5 参考层低保额
+  history_compress: false  # 远端历史 LLM 压缩开关
+  history_keep_recent: 1
+  history_compress_ratio: 0.1
 # SW-07 / ADR-0016：管线策略（缺省即原代码常量；弱模型/强模型 profile 可分道调参）
 pipeline: {pass_attempts: 2, phase_attempts: 3}
 retrieval: {top_k: 3}

profiles/short_video_v1.yaml

@@ -57,7 +57,15 @@ model_tiers:
   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}
+context:
+  prev_summary_window: 1
+  known_fact_fields: [id, content, episode_no, status, type]
+  inject_threads: false
+  budget: 32768            # SW-06 / ADR-0018：P0-P5 总预算（token）
+  core_guarantee: 400      # P5 参考层低保额
+  history_compress: false  # 远端历史 LLM 压缩开关
+  history_keep_recent: 1
+  history_compress_ratio: 0.1
 # SW-07 / ADR-0016：管线策略（缺省即原代码常量；弱模型/强模型 profile 可分道调参）
 pipeline: {pass_attempts: 2, phase_attempts: 3}
 retrieval: {top_k: 3}

src/nsc/passes/__init__.py

@@ -21,6 +21,7 @@
 __all__ = [
     "PassContext",
     "PassFailure",
+    "assemble_context",
     "cached_pass",
     "contract_text",
     "generate_json",
@@ -66,6 +67,60 @@ def with_diag(inputs: dict[str, Any], fragment: dict[str, Any]) -> dict[str, Any
     return {**inputs, "_previous_failure": diag} if diag else inputs
 
 
+def assemble_context(
+    ctx: Any,
+    *,
+    p1_current: str,
+    prev_summary: str,
+    facts: list[str],
+    rag: list[str],
+    refs: list[tuple[str, str]],
+) -> tuple[str, int, str, list[str]]:
+    """SW-06 / ADR-0018：把 P2-P4 层与 P5 参考层过 nsc.context.assemble 预算装配。
+
+    - p1_current：不可裁剪的"当前内容"锚（P1）；
+    - facts：逐条序列化后的 fact 串（P3），返回存活条数（前缀式）；
+    - rag：检索参考（P4 整层一次判定），返回存活文本（丢弃则空串）；
+    - refs：(输入键, 文本) 参考层（P5 低保），返回存活键列表（前缀式）。
+    预算缺省 32768 足够大 → 全存活，输出与输入逐字节等价（原行为）。
+    """
+    from nsc.context import assemble
+
+    cfg = (ctx.profile.get("context") or {}) if isinstance(ctx.profile, dict) else {}
+    res = assemble(
+        p0_system="",
+        p1_current=p1_current,
+        p2_prev_summary=prev_summary,
+        p3_facts=facts,
+        p4_rag=rag,
+        p5_bible=[text for _key, text in refs],
+        budget=int(cfg.get("budget", 32768)),
+        core_guarantee=int(cfg.get("core_guarantee", 400)),
+    )
+    layers = {lay.name: lay.text for lay in res.layers}
+
+    n_facts = 0
+    acc = ""
+    p3_text = layers.get("P3", "")
+    for fs in facts:
+        cand = fs if not acc else acc + "\n" + fs
+        if p3_text.startswith(cand):
+            acc, n_facts = cand, n_facts + 1
+        else:
+            break
+
+    kept_keys: list[str] = []
+    tail = layers.get("P5", "")
+    for key, text in refs:
+        if text and tail.startswith(text):
+            kept_keys.append(key)
+            tail = tail[len(text) + 1 :]  # 跳过层内 join 分隔符 "\n"
+        else:
+            break
+
+    return layers.get("P2", ""), n_facts, layers.get("P4", ""), kept_keys
+
+
 class PassFailure(Exception):  # noqa: N818  名字由 docs/HANDOFF_STRONG_MODEL.md 约定
     """结构性失败：禁止静默降级（AGENTS.md §7）。携带 node_id 供二分定位。"""
 

src/nsc/passes/p3_beatsheet.py

@@ -25,6 +25,7 @@
     DSPyPass,
     PassContext,
     PassFailure,
+    assemble_context,
     cached_pass,
     contract_text,
     inner_json,
@@ -51,6 +52,35 @@ class Module(DSPyPass):
     optional_outputs = ("facts_json", "state_changes_json")
 
 
+def _budgeted_inputs(
+    ctx: PassContext, inputs: dict[str, Any], facts_list: list[Any]
+) -> dict[str, Any]:
+    """SW-06 / ADR-0018：P2(前情)/P3(known_facts)/P4(检索) + P5 参考层过预算装配。
+
+    预算缺省足够大 → 全存活，输入与组装时逐字节等价（原行为）。
+    降级保留键、置空值（review 修正）：signature 的 InputField 是必填契约，
+    预算降级体现在内容为空，而不是缺字段击穿调用。
+    """
+    _prev, n_facts, rag, ref_keys = assemble_context(
+        ctx,
+        p1_current=inputs["episode_json"],
+        prev_summary=inputs["prev_episode_summary"],
+        facts=[json.dumps(f, ensure_ascii=False) for f in facts_list],
+        rag=[inputs["retrieved_cases"]] if inputs["retrieved_cases"] else [],
+        refs=[("bible_json", inputs["bible_json"]), ("profile_json", inputs["profile_json"])],
+    )
+    out = {
+        **inputs,
+        "prev_episode_summary": _prev,
+        "known_facts": json.dumps(facts_list[:n_facts], ensure_ascii=False),
+        "retrieved_cases": rag,
+    }
+    for _k in ("bible_json", "profile_json"):
+        if _k not in ref_keys:
+            out[_k] = ""
+    return out
+
+
 @cached_pass("p3_beatsheet")
 def run(ctx: PassContext, fragment: dict[str, Any]) -> dict[str, Any]:
     ep = fragment["episode"]
@@ -80,6 +110,7 @@ def run(ctx: PassContext, fragment: dict[str, Any]) -> dict[str, Any]:
     threads = str(fragment.get("threads", "") or "")
     if threads:
         inputs["threads"] = threads
+    inputs = _budgeted_inputs(ctx, inputs, list(fragment.get("known_facts", [])))
     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/p5_dialogue.py

@@ -17,6 +17,7 @@
     DSPyPass,
     PassContext,
     PassFailure,
+    assemble_context,
     cached_pass,
     contract_text,
     inner_json,
@@ -73,6 +74,30 @@ def _dialogue_length_target(chars_lo: int, chars_hi: int, scene_secs: float, cps
     )
 
 
+def _budgeted_inputs(ctx: PassContext, inputs: dict[str, Any]) -> dict[str, Any]:
+    """SW-06 / ADR-0018：p5 的 P4(检索) 与 P5 参考层过预算装配（p1=当前场+Beat，不可裁剪）。
+
+    预算缺省足够大 → 全存活，返回与输入逐字段相等（原行为）。
+    """
+    _prev, _n_facts, rag, ref_keys = assemble_context(
+        ctx,
+        p1_current=inputs["scene_json"] + "\n" + inputs["beats_json"],
+        prev_summary="",
+        facts=[],
+        rag=[inputs["retrieved_cases"]] if inputs.get("retrieved_cases") else [],
+        refs=[
+            ("characters_json", inputs["characters_json"]),
+            ("profile_json", inputs["profile_json"]),
+        ],
+    )
+    out = {**inputs, "retrieved_cases": rag}
+    # 降级保留键、置空值（review 修正）：同 p3，不缺字段击穿 signature 契约。
+    for _k in ("characters_json", "profile_json"):
+        if _k not in ref_keys:
+            out[_k] = ""
+    return out
+
+
 @cached_pass("p5_dialogue")
 def run(ctx: PassContext, fragment: dict[str, Any]) -> dict[str, Any]:
     scene = fragment["scene"]
@@ -113,7 +138,7 @@ def run(ctx: PassContext, fragment: dict[str, Any]) -> dict[str, Any]:
         },
         fragment,
     )
-    out = cast(dict[str, Any], Module()(ctx, inputs))
+    out = cast(dict[str, Any], Module()(ctx, _budgeted_inputs(ctx, inputs)))
     lines = _parse_lines(ctx, scene, beats, out, fragment["characters"])
     # T-31 自检子步（默认开）：本场 L0 findings → revision_brief 五节 → 一次自我修订
     lines, out = _self_check(ctx, inputs, scene, beats, lines, fragment["characters"], out)

src/nsc/passes/pipeline.py

@@ -354,7 +354,7 @@ def track() -> None:
     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] = []
+    ep_summaries: list[tuple[int, Any]] = []  # (episode.no, beat 摘要串)
     # p3（逐集）+ p4 + after_p3/p4 检查作为一个相位：L0 拦截（如 BM-002 植入间隔）时
     # 带诊断整体重试（D13 反馈驱动再生成，诊断累积）；重试前恢复相位前的状态。
     diag = ""
@@ -374,7 +374,7 @@ def track() -> None:
                     "bible": bible,
                     "placement": placement,
                     "required_brand_moment_beats": len(placement),
-                    "prev_episode_summary": _window_join(ep_summaries, prev_window),
+                    "prev_episode_summary": _history_text(ctx, ep_summaries, ep["no"], prev_window),
                     "next_episode_promise": episodes[i + 1]["hook_promise"]
                     if i + 1 < len(episodes)
                     else "",
@@ -391,7 +391,7 @@ def track() -> None:
                     frag3["threads"] = _threads_view(st["threads"])
                 r3 = _retry_pass(p3_beatsheet.run, ctx, frag3)
                 track()
-                ep_summaries.append("；".join(b["summary"] for b in r3["beats"]))
+                ep_summaries.append((ep["no"], "；".join(b["summary"] for b in r3["beats"])))
                 st["beats"] += r3["beats"]
                 st["setup_payoffs"] += r3["setup_payoffs"]
                 st["brand_moments"] += r3["brand_moments"]
@@ -520,8 +520,10 @@ def track() -> None:
         "bible": bible,
         "placement": _placement_of(raw, ep),
         "required_brand_moment_beats": len(_placement_of(raw, ep)),
-        "prev_episode_summary": _window_join(
-            [_episode_digest(raw, e["id"]) for e in ordered[max(0, idx - r_window) : idx]],
+        "prev_episode_summary": _history_text(
+            ctx,
+            [(e["no"], _episode_digest(raw, e["id"])) for e in ordered[:idx]],
+            ep["no"],
             r_window,
         ),
         "next_episode_promise": ordered[idx + 1]["hook_promise"] if idx + 1 < len(ordered) else "",
@@ -710,6 +712,32 @@ def _window_join(summaries: list[str], window: int) -> str:
     return "\n".join(summaries[-n:]) if n else ""
 
 
+def _history_text(
+    ctx: PassContext, hist: list[tuple[int, str]], current_no: int, window: int
+) -> str:
+    """p3 前情文本（SW-06 / ADR-0018）。
+
+    history_compress 开且窗口宽于 keep_recent 时，远端集走 compress_history
+    （LLM 压缩，经 make_llm_summarizer 路由），近端集保原文；否则退回 SW-05 的
+    原文窗口 _window_join（缺省路径，逐字节同原实现）。
+    """
+    cfg = ctx.profile.get("context", {}) or {}
+    keep_recent = max(0, int(cfg.get("history_keep_recent", 1)))
+    n = max(0, int(window))
+    visible = hist[-n:] if n else []
+    if bool(cfg.get("history_compress")) and n > keep_recent and len(visible) > keep_recent:
+        from nsc.context import compress_history, make_llm_summarizer
+
+        return compress_history(
+            [{"no": no, "text": text} for no, text in visible],
+            current_no,
+            make_llm_summarizer(ctx.router),
+            keep_recent=keep_recent,
+            ratio=float(cfg.get("history_compress_ratio", 0.1)),
+        )
+    return _window_join([text for _no, text in hist], n)
+
+
 def _threads_view(threads: list[Any]) -> str:
     """Thread 注入面（SW-05）：p2 规划的叙事线索标题/状态，供 p3 做跨集呼应。"""
     view = [