7.3 组合爆炸的控制

// file: core/src/inkwell/ne_07_3_control.ts
// tier: P
// verify: node --test 'core/test/ne_07_3_control.test.ts'
// 卷二 7.3 组合爆炸的控制：宽度扰动下的路径数（BigInt 精确 DP）、深度预算下的可达规模、
// 朴素乘积上界。与 sim/ne_07_3_perturb.py 同口径，数值逐项一致。
// 条件求值 / 效果应用复用公共内核 engine.ts（只读，不改）。
import type { DialogueNode, Spec } from "./engine.ts";

/** 结构路径数：DAG 上的精确 DP（BigInt，忽略 when）。 */
export function pathCount(nodes: DialogueNode[], start: string): bigint {
  const by = new Map(nodes.map((n) => [n.id, n]));
  const memo = new Map<string, bigint>();
  const dp = (id: string): bigint => {
    const hit = memo.get(id);
    if (hit !== undefined) return hit;
    const ch = by.get(id)!.choices ?? [];
    const v = ch.length === 0 ? 1n : ch.reduce((a, c) => a + dp(c.goto), 0n);
    memo.set(id, v);
    return v;
  };
  return dp(start);
}

/** 最长路径（边数）。 */
export function depthLongest(nodes: DialogueNode[], start: string): number {
  const by = new Map(nodes.map((n) => [n.id, n]));
  const memo = new Map<string, number>();
  const dp = (id: string): number => {
    const hit = memo.get(id);
    if (hit !== undefined) return hit;
    const ch = by.get(id)!.choices ?? [];
    const v = ch.length === 0 ? 0 : 1 + Math.max(...ch.map((c) => dp(c.goto)));
    memo.set(id, v);
    return v;
  };
  return dp(start);
}

/** 朴素乘积上界：dmax ^ 最长深度（每步都取最宽、且都走到最深）。 */
export function naiveProductBound(spec: Spec): bigint {
  const dmax = BigInt(Math.max(...spec.dialogue.map((n) => (n.choices ?? []).length)));
  return dmax ** BigInt(depthLongest(spec.dialogue, spec.meta.start));
}

/** 宽度扰动：把每个有选项节点的出度乘以 k（复制选项，目标不变，id 加后缀）。 */
export function widen(nodes: DialogueNode[], k: number): DialogueNode[] {
  return nodes.map((n) => {
    const ch = n.choices ?? [];
    if (ch.length === 0 || k <= 1) return n;
    const dup = ch.flatMap((c) =>
      Array.from({ length: k }, (_, j) => (j === 0 ? c : { ...c, id: `${c.id}#${j}` })));
    return { ...n, choices: dup };
  });
}

/** 深度预算：≤ cap 步内可达的节点数与跳转数。 */
export function depthReach(nodes: DialogueNode[], start: string, cap: number): { nodes: number; jumps: number } {
  const by = new Map(nodes.map((n) => [n.id, n]));
  const dist = new Map<string, number>([[start, 0]]);
  const q = [start];
  while (q.length) {
    const u = q.shift()!;
    if (dist.get(u)! >= cap) continue;
    for (const c of by.get(u)!.choices ?? []) {
      if (!dist.has(c.goto)) {
        dist.set(c.goto, dist.get(u)! + 1);
        q.push(c.goto);
      }
    }
  }
  const jumps = new Set<string>();
  for (const n of nodes) {
    if (!dist.has(n.id)) continue;
    for (const c of n.choices ?? []) if (dist.has(c.goto)) jumps.add(`${n.id}>${c.goto}`);
  }
  return { nodes: dist.size, jumps: jumps.size };
}

k=1:12位 k=2:21位 k=3:27位 k=4:31位
✔ 路径数的理论上界与真实结构路径数 (1.164959ms)
✔ 宽度扰动 ×k：路径数随 k 指数增长（BigInt 精确，与 Python 侧逐项一致） (0.812292ms)
✔ 深度预算：≤ L 步内可达的节点与跳转在最长深度处饱和 (0.713917ms)
ℹ tests 3
ℹ suites 0
ℹ pass 3
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 107.592333

#!/usr/bin/env python3
# file: sim/ne_07_3_perturb.py
# tier: B
# verify: /opt/homebrew/bin/python3 sim/ne_07_3_perturb.py
"""卷二 7.3 组合爆炸的控制（B 层，软告警不硬失败）。

对**分支宽度**与**深度**两条轴做参数扰动扫描，找覆盖率与路径数的相变点：

  ① 宽度因子 k：把每个有选项节点的出度乘以 k（复制选项，目标不变）→ 路径数爆炸，
     但**去重跳转覆盖率不变**（复制的是同一条边）。相变点 = 路径数越过 10^12 的 k。
  ② 深度预算 L：把探索限制在 ≤ L 步内 → 可达节点数与可达跳转数随 L 上升，到最长
     路径时饱和。相变点 = 覆盖率触顶的 L。

扰动只改**内存副本**，不写回 config/story_spec.json；越阈值只打印 WARN，**恒 exit 0**。
复用 sim/ne_07_2_sampling.py 的 eval_cond / apply_effects（只读）。产出 out/ne_07_3_perturb.json。
"""
from __future__ import annotations

import json
import math
import os
import sys
from collections import deque

import ne_07_2_sampling as s2

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")
KS = [1, 2, 3, 4]
DEPTHS = [5, 10, 15, 20, 22]
SAMPLE_N, SEED = 2000, 20261016
WARN_LOG10 = 12

def widen(nodes: list, k: int) -> list:
    """把每个有选项节点的出度乘以 k（复制选项，目标与 when 不变，id 加后缀）。"""
    out = []
    for n in nodes:
        m, ch = dict(n), n.get("choices") or []
        if ch and k > 1:
            m["choices"] = [dict(c, id=c["id"] if j == 0 else f"{c['id']}#{j}")
                            for c in ch for j in range(k)]
        out.append(m)
    return out

def path_count(nodes: list, start: str) -> int:
    """DAG 上的精确路径计数（BigInt 语义的 Python int，忽略 when）。"""
    by = {n["id"]: n for n in nodes}
    memo: dict = {}

    def dp(nid: str) -> int:
        if nid in memo:
            return memo[nid]
        ch = by[nid].get("choices") or []
        v = 1 if not ch else sum(dp(c["goto"]) for c in ch)
        memo[nid] = v
        return v

    return dp(start)

def jump_coverage(nodes: list, spec: dict, n_paths: int, total: int) -> int:
    """随机采样 n_paths 条路径，统计去重跳转覆盖率（千分比）。"""
    by = {n["id"]: n for n in nodes}
    start = spec["meta"]["start"]
    aff0 = {v: next(c["affinity"]["init"] for c in spec["characters"] if c["id"] == v)
            for v in sorted(spec["state"]["vars"])}
    init = {"aff": aff0, "flags": {f: 0 for f in sorted(spec["state"]["flags"])},
            "items": {i: 0 for i in sorted(spec["state"]["items"])}}
    rng, seen = SEED, set()

    def rnd(mod: int) -> int:
        nonlocal rng
        rng = (rng * 48271) % 2147483647
        return rng % mod

    for _ in range(n_paths):
        node, st = start, init
        for _step in range(64):
            if by[node].get("ending"):
                break
            opts = [c for c in (by[node].get("choices") or []) if s2.eval_cond(c.get("when"), st)]
            if not opts:
                break
            c = opts[rnd(len(opts))]
            seen.add((node, c["goto"]))
            st = s2.apply_effects(c.get("effects") or {}, st)
            node = c["goto"]
    return len(seen) * 1000 // total

def depth_reach(nodes: list, start: str, cap: int) -> tuple[int, int]:
    """≤ cap 步内可达的节点数与跳转数。"""
    by = {n["id"]: n for n in nodes}
    dist = {start: 0}
    q = deque([start])
    while q:
        u = q.popleft()
        if dist[u] >= cap:
            continue
        for c in (by[u].get("choices") or []):
            if c["goto"] not in dist:
                dist[c["goto"]] = dist[u] + 1
                q.append(c["goto"])
    reached = set(dist)
    jumps = {(n["id"], c["goto"]) for n in nodes if n["id"] in reached
             for c in (n.get("choices") or []) if c["goto"] in reached}
    return len(reached), len(jumps)

def main() -> int:
    spec = s2.load_spec()
    nodes, start = spec["dialogue"], spec["meta"]["start"]
    total_nodes = len(nodes)
    total_jumps = len({(n["id"], c["goto"]) for n in nodes for c in (n.get("choices") or [])})
    base_paths = path_count(nodes, start)
    base_log = math.log10(base_paths)
    warns: list[str] = []

    print(f"spec={spec['meta']['title']}  基线路径数 {base_paths}（10^{base_log:.2f}）"
          f"  跳转 {total_jumps} 条  节点 {total_nodes} 个")
    print(f"软告警阈值：路径数 > 10^{WARN_LOG10}。全部 exit 0（B 层口径）。")
    print("\n① 宽度因子 k（每个有选项节点的出度 ×k）→ 路径数与覆盖率")
    print(f"   {'k':>3}{'路径数':>26}{'log10':>8}{'跳转覆盖':>9}{'‰':>7}")
    width_rows = []
    for k in KS:
        wn = widen(nodes, k)
        pc = path_count(wn, start)
        lg = math.log10(pc)
        cov = jump_coverage(wn, spec, SAMPLE_N, total_jumps)
        width_rows.append({"k": k, "paths": pc, "log10": round(lg, 2), "jump_permille": cov})
        print(f"   {k:>3}{pc:>26}{lg:>8.2f}{total_jumps:>9}{cov:>7}")
        if lg > WARN_LOG10:
            warns.append(f"宽度 ×{k}：路径数 10^{lg:.2f} 越过 10^{WARN_LOG10}")

    print("\n② 深度预算 L（≤ L 步内可达）→ 节点覆盖与跳转覆盖")
    print(f"   {'L':>4}{'可达节点':>9}{'‰':>7}{'可达跳转':>9}{'‰':>7}")
    depth_rows = []
    for L in DEPTHS:
        rn, rj = depth_reach(nodes, start, L)
        depth_rows.append({"L": L, "nodes": rn, "node_permille": rn * 1000 // total_nodes,
                           "jumps": rj, "jump_permille": rj * 1000 // total_jumps})
        print(f"   {L:>4}{rn:>9}{rn * 1000 // total_nodes:>7}{rj:>9}{rj * 1000 // total_jumps:>7}")

    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ne_07_3_perturb.json"), "w", encoding="utf-8") as fh:
        json.dump({"base_paths": base_paths, "base_log10": round(base_log, 2),
                   "total_jumps": total_jumps, "total_nodes": total_nodes, "sample_n": SAMPLE_N,
                   "seed": SEED, "width_axis": width_rows, "depth_axis": depth_rows,
                   "warnings": warns}, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print()
    print("\n".join(f"WARN {w}" for w in warns) if warns
          else "无软告警：两条轴均未越过阈值")
    print("wrote out/ne_07_3_perturb.json")
    print(f"OK ne_07_3_perturb：扰动已跑完，{len(warns)} 条软告警，exit 0（B 层口径）")
    return 0

if __name__ == "__main__":
    sys.exit(main())

spec=雾港夜航  基线路径数 450208471680（10^11.65）  跳转 56 条  节点 48 个
软告警阈值：路径数 > 10^12。全部 exit 0（B 层口径）。

① 宽度因子 k（每个有选项节点的出度 ×k）→ 路径数与覆盖率
     k                       路径数   log10     跳转覆盖      ‰
     1              450208471680   11.65       56   1000
     2     739846127062351872000   20.87       56   1000
     3206730895807874991631881600   26.32       56   1000
     41532569005279951267429938626560   30.19       56   1000

② 深度预算 L（≤ L 步内可达）→ 节点覆盖与跳转覆盖
      L     可达节点      ‰     可达跳转      ‰
      5        9    187        9    160
     10       18    375       19    339
     15       31    645       34    607
     20       40    833       47    839
     22       48   1000       56   1000

WARN 宽度 ×2：路径数 10^20.87 越过 10^12
WARN 宽度 ×3：路径数 10^26.32 越过 10^12
WARN 宽度 ×4：路径数 10^30.19 越过 10^12
wrote out/ne_07_3_perturb.json
OK ne_07_3_perturb：扰动已跑完，3 条软告警，exit 0（B 层口径）