3.2 设定的一致性：冲突检测

#!/usr/bin/env python3
# file: sim/na_03_conflict_detect.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/na_03_conflict_detect.py
"""3.2 设定的一致性：冲突检测（S 层）。

设定冲突不是一个「感觉」，而是三种可以判定的结构问题，各给一个检测器：

  ① 立场失衡（signed cycle）：把 stance 的非零值当**带符号有向边**，枚举简单环，
     算符号积；积为 -1 的环即失衡环（「敌人的敌人是我的敌人」被破坏）。
  ② 解锁成环（unlock cycle）：地点 -> 解锁条件所依赖的 flag/item -> 能提供它的场景，
     接成有向图找环。边按「该条件的提供者数量」判软硬：唯一提供者=硬边，多提供者=软边；
     **全硬**的环是死锁，含软边的环是软环。
  ③ 规则反证（rule falsification）：每条 rule 由**读法表**映射成一条禁则，
     扫全部选项效果，数它能被多少个独立选项触发（反证源）；>= 2 源即**双源反证**。

每个检测器都配一个**注入式负例**：把一份人造的坏规格喂进去，证明检测器会报错，
而不是恒返回「无冲突」。真规格上三处都干净，负例上三处都触发。
"""
from __future__ import annotations

import copy
import json
import os
import sys


def load_spec() -> dict:
    here = os.path.dirname(os.path.abspath(__file__))
    with open(os.path.join(here, "../config/story_spec.json"), encoding="utf-8") as fh:
        return json.load(fh)


# ---------------------------------------------------------------- ① 立场失衡
def canon(path: list[str]) -> tuple[str, ...]:
    """环的规范形：旋转 + 反向取字典序最小，保证同一环只被记一次。"""
    n = len(path)
    rev = list(reversed(path))
    return min([tuple(path[i:] + path[:i]) for i in range(n)] +
               [tuple(rev[i:] + rev[:i]) for i in range(n)])


def cycles_of(edges: list[dict], min_len: int = 2) -> list[list[str]]:
    """有向图（由带 from/to 的边表给出）的全部简单环，含自环。

    min_len 是环的**最小节点数**：立场失衡只认长度 >= 3 的真环（长度 2 的往返是
    「互惠边」，不是环）；解锁依赖则要认长度 2 与自环。
    """
    adj: dict[str, list[dict]] = {}
    for e in edges:
        adj.setdefault(e["from"], []).append(e)
    out: list[list[str]] = [[e["from"], e["from"]] for e in edges if e["from"] == e["to"]]

    def dfs(s: str, path: list[str], seen: set[str]) -> None:
        for e in adj.get(path[-1], []):
            t = e["to"]
            if t == s and len(path) >= min_len:
                out.append(list(canon(path)))
            elif t not in seen and t > s:
                dfs(s, path + [t], seen | {t})

    for s in sorted(adj):
        dfs(s, [s], {s})
    uniq: list[list[str]] = []
    for c in out:
        if c not in uniq:
            uniq.append(c)
    return uniq


def detect_signed_cycles(spec: dict) -> dict:
    ids = [f["id"] for f in spec["world"]["factions"]]
    m = {f["id"]: dict(f["stance"]) for f in spec["world"]["factions"]}
    edges = [{"from": a, "to": b} for a in ids for b in ids if a != b and m[a][b] != 0]
    found = []
    for cyc in cycles_of(edges, min_len=3):
        prod = 1
        for i in range(len(cyc)):  # 首尾闭合，故取模
            prod *= m[cyc[i]][cyc[(i + 1) % len(cyc)]]
        found.append({"cycle": "->".join(cyc), "sign_product": prod, "frustrated": prod < 0})
    return {"edges": len(edges), "cycles": found, "frustrated": [c for c in found if c["frustrated"]]}


# ---------------------------------------------------------------- ② 解锁成环
def condition_providers(name: str, kind: str, spec: dict) -> list[str]:
    """能提供该 flag/item 的**场景**（去重保序）。"""
    out: list[str] = []
    for n in spec["dialogue"]:
        for c in (n.get("choices") or []):
            e = c.get("effects") or {}
            hit = (kind == "flag" and name in (e.get("set") or {})) or \
                  (kind == "item" and name in (e.get("grant") or {}))
            if hit and n["scene"] not in out:
                out.append(n["scene"])
    return out


def detect_unlock_cycles(spec: dict) -> dict:
    edges: list[dict] = []
    for loc in spec["world"]["locations"]:
        if loc["unlock_when"] is None:
            continue
        kind, name, _ = loc["unlock_when"]
        prov = condition_providers(name, kind, spec)
        for p in prov:
            edges.append({"from": p, "to": loc["id"], "cond": f"{kind}:{name}",
                          "providers": len(prov), "hard": len(prov) == 1})
    hard_of = {(e["from"], e["to"]): e["hard"] for e in edges}
    cycles = []
    for cyc in cycles_of(edges):
        pairs = [(cyc[i], cyc[i + 1]) for i in range(len(cyc) - 1)]
        hard = all(hard_of.get(p, False) for p in pairs)
        conds = sorted({e["cond"] for e in edges if (e["from"], e["to"]) in pairs})
        cycles.append({"cycle": "->".join(cyc), "conds": conds, "hard": hard})
    return {"edges": edges, "hard_edges": [e for e in edges if e["hard"]], "cycles": cycles,
            "hard_cycles": [c for c in cycles if c["hard"]]}


# ---------------------------------------------------------------- ③ 规则反证
def rule_violations(spec: dict) -> list[dict]:
    """读法表：每条 rule 映射成一条禁则，扫全部选项效果，返回违反它的选项。

    这张表是**本章的建模步骤**，不是 spec 自带的字段——换一种读法结论会变，
    所以它必须显式写出来被审阅，而不是藏在脚本里当常识。
    """
    nodes = {n["id"]: n for n in spec["scene_graph"]["nodes"]}
    out: list[dict] = []
    out.append({"rule": "rule_curfew", "reading": "灯塔区场景节点的 enter_when 必须非空",
                "violations": [nid for nid, n in nodes.items()
                               if nid.startswith("lighthouse_") and not n.get("enter_when")]})

    def scan(pred):
        return [f"{n['id']}/{c['id']}" for n in spec["dialogue"]
                for c in (n.get("choices") or []) if pred(n, c)]

    out.append({"rule": "rule_bell", "reading": "设置 bell_rung 的选项必须同时设置 dawn_choice_made",
                "violations": scan(lambda n, c: ((c.get("effects") or {}).get("set") or {}).get("bell_rung")
                                   and not ((c.get("effects") or {}).get("set") or {}).get("dawn_choice_made"))})
    out.append({"rule": "rule_guild", "reading": "授予 rope_knot 的选项必须带 when 前置",
                "violations": scan(lambda n, c: ((c.get("effects") or {}).get("grant") or {}).get("rope_knot")
                                   and c.get("when") is None)})
    out.append({"rule": "rule_church", "reading": "教会侧选项不得因 saw_lighthouse_light 给 shen 加分",
                "violations": scan(lambda n, c: n.get("speaker") == "shen"
                                   and "saw_lighthouse_light" in json.dumps(c.get("when"), ensure_ascii=False)
                                   and ((c.get("effects") or {}).get("add") or {}).get("shen", 0) > 0)})
    return out


# ---------------------------------------------------------------- 注入式负例
def inject_frustrated_cycle(spec: dict) -> dict:
    s = copy.deepcopy(spec)
    for f in s["world"]["factions"]:
        if f["id"] == "church":
            f["stance"]["guild"] = 1  # 教会改口称与船帮结盟 -> 环符号积翻负
    return s


def inject_hard_unlock_cycle(spec: dict) -> dict:
    s = copy.deepcopy(spec)
    for n in s["dialogue"]:  # 抽掉 know_tide_rule 的其它提供者
        for c in (n.get("choices") or []):
            (c.get("effects") or {}).get("set", {}).pop("knows_tide_rule", None)
    for n in s["dialogue"]:  # 让 tide_cave 自己提供自己要求的 flag
        if n["id"] == "d_tc_1":
            n["choices"][0]["effects"].setdefault("set", {})["knows_tide_rule"] = 1
    return s


def inject_double_source(spec: dict) -> dict:
    s = copy.deepcopy(spec)
    for n in s["dialogue"]:
        if n["id"] == "d_hub_a1":  # 再加一个无前置的绳结授予
            n["choices"].append({"id": "c_hub_a1c", "text": "再要一枚绳结", "when": None,
                                 "effects": {"grant": {"rope_knot": 1}}, "goto": "d_ch_1"})
    return s


def main() -> int:
    spec = load_spec()

    signed = detect_signed_cycles(spec)
    print(f"① 立场失衡：带符号有向边 {signed['edges']} 条，简单环 {len(signed['cycles'])} 个")
    for c in signed["cycles"]:
        print(f"   {c['cycle']:<44} 符号积 {c['sign_product']:>+d}  {'失衡' if c['frustrated'] else '平衡'}")
    print(f"   失衡环 {len(signed['frustrated'])} 个")

    unlock = detect_unlock_cycles(spec)
    soft = len(unlock["edges"]) - len(unlock["hard_edges"])
    print(f"\n② 解锁成环：依赖边 {len(unlock['edges'])} 条（硬边 {len(unlock['hard_edges'])} / 软边 {soft}）")
    for e in unlock["hard_edges"]:
        print(f"   硬边 {e['from']:<16}-> {e['to']:<18} {e['cond']:<24} 提供者 {e['providers']}")
    print(f"   环 {len(unlock['cycles'])} 个，死锁（全硬）{len(unlock['hard_cycles'])} 个")
    for c in unlock["cycles"]:
        print(f"   {c['cycle']:<40} 条件 {','.join(c['conds'])} {'死锁' if c['hard'] else '软环'}")

    rules = rule_violations(spec)
    print("\n③ 规则反证（反证源 >= 2 即双源反证）")
    for r in rules:
        print(f"   {r['rule']:<14} 反证源 {len(r['violations'])}  {r['violations']}")
        print(f"       读法：{r['reading']}")
    print(f"   双源反证 {sum(1 for r in rules if len(r['violations']) >= 2)} 条，"
          f"单源反证 {sum(1 for r in rules if len(r['violations']) == 1)} 条")

    n1 = detect_signed_cycles(inject_frustrated_cycle(spec))
    n2 = detect_unlock_cycles(inject_hard_unlock_cycle(spec))
    n3 = rule_violations(inject_double_source(spec))
    d3 = [r["rule"] for r in n3 if len(r["violations"]) >= 2]
    print("\n注入式负例")
    print(f"   ① church->guild 改 +1：失衡环 {len(n1['frustrated'])} 个 {[c['cycle'] for c in n1['frustrated']]}")
    print(f"   ② 抽掉 know_tide_rule 其它提供者：死锁环 {len(n2['hard_cycles'])} 个 {[c['cycle'] for c in n2['hard_cycles']]}")
    print(f"   ③ 再加一个无前置绳结授予：双源反证 {len(d3)} 条 {d3}")

    detected = bool(n1["frustrated"]) and bool(n2["hard_cycles"]) and bool(d3)
    here = os.path.dirname(os.path.abspath(__file__))
    out = os.path.join(here, "../out/na03_conflicts.json")
    os.makedirs(os.path.dirname(out), exist_ok=True)
    with open(out, "w", encoding="utf-8") as fh:
        json.dump({"signed": signed, "unlock": unlock, "rules": rules, "negative_detected": detected},
                  fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print(f"wrote out/na03_conflicts.json  negative_detected={detected}")
    if not detected:
        print("FAIL 负例未被检出，检测器恒返回「无冲突」")
        return 1
    print("OK na_03_conflict_detect：真规格 0 死锁 / 0 失衡环 / 0 双源反证，三个负例全部检出")
    return 0


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

① 立场失衡：带符号有向边 8 条，简单环 1 个
   church->guild->watch                         符号积 +1  平衡
   失衡环 0 个

② 解锁成环：依赖边 15 条（硬边 4 / 软边 11）
   硬边 tavern          -> counting_house     flag:met_hui             提供者 1
   硬边 pier            -> watch_post         flag:knows_yan_missing   提供者 1
   硬边 counting_house  -> harbor_office      item:ledger_page         提供者 1
   硬边 watch_post      -> lighthouse_top     item:lantern             提供者 1
   环 1 个，死锁（全硬）0 个
   lighthouse_base->lighthouse_base         条件 flag:saw_lighthouse_light 软环

③ 规则反证（反证源 >= 2 即双源反证）
   rule_curfew    反证源 0  []
       读法：灯塔区场景节点的 enter_when 必须非空
   rule_bell      反证源 1  ['d_fg_2/c_fg_2c']
       读法：设置 bell_rung 的选项必须同时设置 dawn_choice_made
   rule_guild     反证源 1  ['d_tavern_3/c_tavern_3a']
       读法：授予 rope_knot 的选项必须带 when 前置
   rule_church    反证源 0  []
       读法：教会侧选项不得因 saw_lighthouse_light 给 shen 加分
   双源反证 0 条，单源反证 2 条

注入式负例
   ① church->guild 改 +1：失衡环 1 个 ['church->guild->watch']
   ② 抽掉 know_tide_rule 其它提供者：死锁环 1 个 ['tide_cave->tide_cave']
   ③ 再加一个无前置绳结授予：双源反证 1 条 ['rule_guild']
wrote out/na03_conflicts.json  negative_detected=True
OK na_03_conflict_detect：真规格 0 死锁 / 0 失衡环 / 0 双源反证，三个负例全部检出