diff --git a/adr/0018-context-wiring.md b/adr/0018-context-wiring.md new file mode 100644 index 0000000..5552252 --- /dev/null +++ b/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"` 绿。 diff --git a/profiles/_schema.py b/profiles/_schema.py index d2a7094..da94f53 100644 --- a/profiles/_schema.py +++ b/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 diff --git a/profiles/short_drama_v1.yaml b/profiles/short_drama_v1.yaml index 815db51..02909fa 100644 --- a/profiles/short_drama_v1.yaml +++ b/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} diff --git a/profiles/short_video_v1.yaml b/profiles/short_video_v1.yaml index 050ee7a..9372bd8 100644 --- a/profiles/short_video_v1.yaml +++ b/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} diff --git a/src/nsc/passes/__init__.py b/src/nsc/passes/__init__.py index fb8c10e..eb22d4a 100644 --- a/src/nsc/passes/__init__.py +++ b/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 供二分定位。""" diff --git a/src/nsc/passes/p3_beatsheet.py b/src/nsc/passes/p3_beatsheet.py index 16ddfcd..35693fb 100644 --- a/src/nsc/passes/p3_beatsheet.py +++ b/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") diff --git a/src/nsc/passes/p5_dialogue.py b/src/nsc/passes/p5_dialogue.py index 38bbf1f..bb2db23 100644 --- a/src/nsc/passes/p5_dialogue.py +++ b/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) diff --git a/src/nsc/passes/pipeline.py b/src/nsc/passes/pipeline.py index a8b5389..31a6854 100644 --- a/src/nsc/passes/pipeline.py +++ b/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 = [ diff --git a/tests/test_context_wiring.py b/tests/test_context_wiring.py new file mode 100644 index 0000000..a920bce --- /dev/null +++ b/tests/test_context_wiring.py @@ -0,0 +1,205 @@ +"""SW-06 接线休眠模块：assembler P2-P4 层进 p3/p5 输入装配；compress_history 进 p3 远端历史。 + +- nsc.context.assemble / compress_history 在 main 上已实现未接线（T-33 只落了模块）； +- 本卡接线：p3 的 prev_summary(P2)/known_facts(P3)/retrieved(P4) 与 p3/p5 的参考层(P5) + 过预算装配；history_compress 开时远端历史走 LLM 压缩； +- 缺省（budget=32768, history_compress=false）= 原行为逐字节不变。 +""" + +from __future__ import annotations + +from pathlib import Path + +import diskcache +import yaml + +import nsc.runtime.cache as cache_mod +from tests.test_pipeline_stub import FullStubRouter + +COMPRESS_MARK = "压缩成不超过" + + +class CompressRouter(FullStubRouter): + """回答历史压缩 summarize 调用（纯文本），其余走黄金桩。""" + + def __init__(self) -> None: + super().__init__() + self.summarize_calls: list[str] = [] + + def complete(self, tier, messages, *, json_mode=False, seed=None): + system = messages[0]["content"] + if COMPRESS_MARK in system: + self.summarize_calls.append(messages[-1]["content"]) + from nsc.runtime.models import LLMResult + + return LLMResult( + text=f"SUM<{len(messages[-1]['content'])}>", + model_id="stub/model", + tokens_in=1, + tokens_out=1, + cost_usd=0.0, + wall_ms=0, + ) + return super().complete(tier, messages, json_mode=json_mode, seed=seed) + + +class RecordingRouter(CompressRouter): + def __init__(self) -> None: + super().__init__() + self.p3_inputs: list[dict] = [] + + def _p3(self, inputs): + self.p3_inputs.append(inputs) + return super()._p3(inputs) + + +# ---------------------------------------------------------------- assemble_context 单元 +def _mini_ctx(tmp_path, context_cfg: dict | None): + from nsc.passes import PassContext + from nsc.runtime.provenance import RunsStore + + profile = {"id": "t", "version": "1", "context": context_cfg or {}} + return PassContext( + profile=profile, + brand={}, + router=None, + store=RunsStore(tmp_path / "runs.db"), + ruleset_ver="t", + spec_sha="t", + ) + + +def test_assemble_context_default_budget_keeps_everything(tmp_path): + from nsc.passes import assemble_context + + facts = ['{"id": "f1"}', '{"id": "f2"}'] + refs = [("bible_json", "BIBLE" * 100), ("profile_json", "PROFILE" * 100)] + prev, n_facts, rag, ref_keys = assemble_context( + _mini_ctx(tmp_path, {}), + p1_current="EPISODE" * 50, + prev_summary="PREV", + facts=facts, + rag=["RAG"], + refs=refs, + ) + assert prev == "PREV" and n_facts == 2 and rag == "RAG" + assert ref_keys == ["bible_json", "profile_json"] + + +def test_assemble_context_tight_budget_drops_in_order(tmp_path): + from nsc.passes import assemble_context + + # 预算：装下 P1 后剩 ~700：P3 可装、P4（1000 token）整层丢、P2 截尾、P5 低保 + prev, n_facts, rag, ref_keys = assemble_context( + _mini_ctx(tmp_path, {"budget": 800, "core_guarantee": 400}), + p1_current="E" * 100, # 50 token + prev_summary="P" * 4000, # 2000 token → 必截尾 + facts=['{"id": "f1"}', '{"id": "f2"}'], + rag=["R" * 2000], + refs=[("bible_json", "B" * 100)], + ) + assert rag == "", "P4 检索层超配额必须整层丢弃" + assert prev and len(prev) < 4000, "P2 超配额必须截尾保留末尾" + assert 0 <= n_facts <= 2 + assert isinstance(ref_keys, list) + + +# ---------------------------------------------------------------- p5 输入预算装配 +def test_p5_budgeted_inputs(tmp_path): + from nsc.passes import p5_dialogue + + inputs = { + "scene_json": '{"id": "s1"}', + "beats_json": "[]", + "characters_json": "C" * 200, + "profile_json": "P" * 200, + "retrieved_cases": "R" * 4000, + } + kept = p5_dialogue._budgeted_inputs(_mini_ctx(tmp_path, {}), inputs) + assert kept == inputs, "缺省预算下 p5 输入必须逐字段不变" + + tight = p5_dialogue._budgeted_inputs(_mini_ctx(tmp_path, {"budget": 500}), dict(inputs)) + assert tight["retrieved_cases"] == "", "预算吃紧时 P4 检索层丢弃" + assert "characters_json" in tight or "profile_json" in tight, "P5 参考层有低保" + # review 修正：降级不删键——signature 的必填 InputField 必须仍在（值可为空） + for key in ("characters_json", "profile_json"): + assert key in tight, f"预算降级不得删除 {key}（置空而非缺字段）" + + +def test_p3_degradation_keeps_keys(tmp_path): + """review 修正：p3 参考层降级保留键、置空值，不缺字段击穿 signature 契约。""" + from nsc.passes import p3_beatsheet + + out = p3_beatsheet._budgeted_inputs( + _mini_ctx(tmp_path, {"budget": 80, "core_guarantee": 40}), + { + "episode_json": "E" * 80, + "prev_episode_summary": "P" * 4000, + "retrieved_cases": "", + "bible_json": "B" * 4000, + "profile_json": "P" * 4000, + }, + [], + ) + assert {"episode_json", "bible_json", "profile_json"} <= set(out), "降级后必填键仍在" + assert out["bible_json"] == "" or out["profile_json"] == "", "超额参考层被置空" + assert out["prev_episode_summary"] == "", "P2 零配额整层丢弃后为空串（键仍在）" + + +# ---------------------------------------------------------------- compress_history 接线（全桩管线） +def _ctx(tmp_path, monkeypatch, router, context_cfg: dict): + monkeypatch.setenv("NSC_NO_CACHE", "1") + monkeypatch.setattr(cache_mod, "_cache", diskcache.Cache(str(tmp_path / "cache"))) + from nsc.passes import PassContext + from nsc.runtime.provenance import RunsStore + + profile = yaml.safe_load(Path("profiles/short_drama_v1.yaml").read_text("utf-8")) + profile["context"] = context_cfg + brand = yaml.safe_load(Path("brands/demo_tea/brand.yaml").read_text("utf-8")) + brief = yaml.safe_load(Path("examples/demo_tea/brief.yaml").read_text("utf-8")) + return PassContext( + profile=profile, + brand=brand, + brief=brief, + router=router, + store=RunsStore(tmp_path / "runs.db"), + ruleset_ver="t", + spec_sha="t", + out_dir=tmp_path / "out", + ) + + +def test_compress_wiring_far_and_recent(tmp_path, monkeypatch): + from nsc.passes.pipeline import run_pipeline + + router = RecordingRouter() + ctx = _ctx( + tmp_path, + monkeypatch, + router, + {"history_compress": True, "prev_summary_window": 3, "history_keep_recent": 1}, + ) + run_pipeline(ctx) + inputs = router.p3_inputs + # 第 2 集：历史只有 1 集 = 近端 → 原文窗口，无压缩调用参与该集 + assert "【前情】" not in inputs[1]["prev_episode_summary"] + # 第 4 集：窗口 3、近端 1 → 两集远端走压缩，一集近端保原文 + prev4 = inputs[3]["prev_episode_summary"] + assert prev4.startswith("【前情】") and "【上一集】" in prev4 + assert "SUM<" in prev4, "远端历史必须是 LLM 压缩摘要" + assert router.summarize_calls, "compress_history 必须经 make_llm_summarizer 调 LLM" + + +def test_compress_off_by_default(tmp_path, monkeypatch): + from nsc.passes.pipeline import run_pipeline + + router = RecordingRouter() + ctx = _ctx( + tmp_path, + monkeypatch, + router, + {"prev_summary_window": 3}, # 未开 history_compress：纯原文窗口（SW-05 行为） + ) + run_pipeline(ctx) + assert router.summarize_calls == [], "缺省不得产生压缩 LLM 调用" + assert "【前情】" not in router.p3_inputs[3]["prev_episode_summary"]