feat(迭代35卡①): 节点图参数化生成器核心批落地——五旋钮参数档 schema+.tres 三行 provisional+ConfigDB 注册与 level_gen_params_row 单点读;RngService 新增 pure_stream 纯流面(不推计数器不落日志,derive_at 派生面流化);NodeMapGenerator 产图 generate(params,seed)+产出侧自证 validate_output(产出约束五条/结构铁律/摆放约束三条/多终点深度互异/深度=自入口最长路径步数)+参数无解致命中止无兜底图;三套断言宿主 68 条(fv1–fv10)+run_all manifest 三行登记;生长读数 L1/L2/L3 = 11/19/29 节点、层高 6/9/13。卡② 停工呈报随批(§0 硬交付序列——覆盖批未立卡)

This commit is contained in:
DKChan committed 2026-10-24 05:40:04 +08:00
1 parent 2931b1f250
commit 63a6ea8055
15 files changed
+1372

No files matched your search

+537
View File
@@ -0,0 +1,537 @@
## 关卡节点图参数化长图生成器(节点图生成器核心批 卡① T2)。
## 纯函数产图面:generate(参数档, 种子) → 一张合法节点图({nodes, entries, terminals})——
## 图 = (参数档, 种子) 的函数,复现不依赖统一流计数器号位(契约 1),结构取数只经 RngService.pure_stream。
## 产前强制自证(契约 3 产出约束五条+结构铁律、契约 5 摆放约束三条、契约 2 深度赋值与多终点深度互异);
## 不合格产出=参数无解致命中止(契约 4:无兜底图、无软回退、无残图交付)。
## 生成器只吃参数档、不认关卡号(契约 6);节点深度 = 自入口最长路径步数(契约 2)。
class_name NodeMapGenerator
extends RefCounted
const NODE_TYPES := ["battle", "boss", "reward", "rest", "shop"]
const FILLER_TYPES := ["battle", "reward", "rest"]
const MAX_WIDTH := 4 # 产出约束·宽度=分叉生长(同深度节点数由岔口长出 1→2→3→4,禁摆平行层)
# 多终点图的支路互斥段必须各有独有类型(契约 5③),而商店与 Boss 是每条支路共有件,
# 独有类型只能来自 battle/reward/rest 三件 ⇒ 归属大岔口各区域的填充节点取互斥签名类型。
# 入口与主链段(大岔口上游,不属任何支路互斥段)恒为战斗节点——承载战斗保底与岔口下一跳互异基准形。
const REGION_SIGNATURES := ["rest", "reward", "battle"]
const SCOPE_STRUCTURE := 0 # 纯流面取数分域:结构流(代拍板 §10 ①——单流多次取数)
var _rng_service
var _params: Dictionary = {}
var _pool: Array = []
var _by_id: Dictionary = {}
var _next_id := 0
var _detour_heads: Array = [] # 岔口侧链首节点(下一跳类型互异的修型面)
var _detour_ids: Dictionary = {} # 侧链节点 id 集(配比填余不动侧链——战斗保底只由链上槽承载)
func _init(rng_service = null) -> void:
_rng_service = rng_service
## 产图主入口(§7 接口块定稿签名——不收 level_index,契约 6)
func generate(params: Dictionary, seed_val: int) -> Dictionary:
var errs := check_params(params)
if not errs.is_empty():
for e in errs:
push_error("[NodeMapGenerator] 参数无解: " + String(e["message"]))
_fatal("参数无解致命中止(契约 4——无兜底图、禁软回退)")
return {} # 致命中止后不可达;本行即「无残图交付」的显式证明
if _rng_service == null:
push_error("[NodeMapGenerator] 未注入 RngService(RNG 单点红线——生成器取数唯一源缺失)")
_fatal("未注入 RngService")
return {}
var built := _setup_and_build(params, seed_val)
var out_errs := validate_output(built, params)
if not out_errs.is_empty():
for e in out_errs:
push_error("[NodeMapGenerator] 产出不合格: " + String(e["message"]))
_fatal("产出自证不过=产出不合格(执行边界①——禁止进玩法)")
return {}
return built
## 调试面(卡① §8 测试方式两分法——能力验证可经调试接口直改参数档/种子驱动生成图):
## 只构型、不自证不致命,供断言逐条读产出规格违例;生产路径仍走 generate(不合格即致命中止)。
func generate_for_debug(params: Dictionary, seed_val: int) -> Dictionary:
if _rng_service == null:
return {}
return _setup_and_build(params, seed_val)
func _setup_and_build(params: Dictionary, seed_val: int) -> Dictionary:
_params = params
var rg: RandomNumberGenerator = _rng_service.pure_stream(seed_val, SCOPE_STRUCTURE)
_pool = []
_by_id = {}
_next_id = 0
_detour_heads = []
_detour_ids = {}
return _build(rg)
## 参数档可行性判定(执行边界③:参数无解=配置错误)——域校验外加深度预算与战斗保底可放性。
static func check_params(params: Dictionary) -> Array:
var errs: Array[Dictionary] = []
var L := int(params.get("layers", 0))
var e := int(params.get("exit_count", 0))
var en := int(params.get("entry_count", 0))
var bf := int(params.get("battle_floor", -1))
var bd := float(params.get("branch_density", -1.0))
if L < 6:
errs.append({"category": "domain", "message": "层深 %d 低于下限 6(层高已锁)" % L})
if e < 1 or e > 3:
errs.append({"category": "domain", "message": "出口个数 %d 越域 [1,3]" % e})
if en < 1:
errs.append({"category": "domain", "message": "入口个数 %d 越域(≥1)" % en})
if bd < 0.0 or bd > 1.0:
errs.append({"category": "domain", "message": "分支密度 %f 越域 [0,1]" % bd})
if bf < 0:
errs.append({"category": "domain", "message": "战斗保底 %d 越域(≥0)" % bf})
if int(params.get("fork_ways_min", 2)) < 2 or int(params.get("fork_ways_max", 3)) > 3:
errs.append({"category": "domain", "message": "岔口去路数域越界(锁 2/3)"})
var sum := 0.0
for rk in ["battle_ratio", "reward_ratio", "rest_ratio", "shop_ratio"]:
sum += float(params.get(rk, -1.0))
if not is_equal_approx(sum, 1.0):
errs.append({"category": "domain", "message": "配比四类合计 %f ≠ 1" % sum})
if not errs.is_empty():
return errs
var t := 0 if e == 1 else _trunk_len(L, e)
if e > 1 and t < 1:
errs.append({"category": "structural", "message": "层深 %d 与出口数 %d 无解(主链段放不下归属区域)" % [L, e]})
var battle_slots := en + t
if e == 1:
battle_slots += maxi(0, L - 2)
else:
for r in range(e):
if String(REGION_SIGNATURES[r]) == "battle":
battle_slots += maxi(0, (L - r) - t - 2)
if battle_slots < bf:
errs.append({"category": "structural", "message": "可放战斗节点 %d 低于战斗保底 %d(参数无解)" % [battle_slots, bf]})
return errs
## 产出侧自证(契约 3/契约 5/契约 2)——generate 产前强制;错误类别沿
## structural/dangling_next/dup_key/enum/domain 词表。params 为空时跳过与参数档对照的计数判据。
func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
var errs: Array[Dictionary] = []
var nodes: Array = graph.get("nodes", [])
if nodes.is_empty():
errs.append({"category": "structural", "message": "产出为空图(执行边界①——不合格产出禁止进玩法)"})
return errs
var by_id := {}
var indeg := {}
for n in nodes:
for k in ["id", "type", "depth", "next"]:
if not (n as Dictionary).has(k):
errs.append({"category": "structural", "message": "节点字段缺失: %s" % k})
return errs
var id: int = int(n["id"])
if by_id.has(id):
errs.append({"category": "dup_key", "message": "节点句柄重复: %d" % id})
by_id[id] = n
if not NODE_TYPES.has(String(n["type"])):
errs.append({"category": "enum", "message": "节点类型出域: %s" % n["type"]})
if int(n["depth"]) < 0:
errs.append({"category": "domain", "message": "节点深度为负: %d" % n["depth"]})
indeg[id] = 0
if not errs.is_empty():
return errs
for n in nodes:
for oid in (n["next"] as Array):
if not by_id.has(int(oid)):
errs.append({"category": "dangling_next", "message": "出边悬空: %d → %s" % [n["id"], oid]})
continue
if int(by_id[int(oid)]["depth"]) <= int(n["depth"]):
errs.append({"category": "structural", "message": "连线未朝前进方向(目标深度 %d 不严格大于源 %d)——环/回边" % [
int(by_id[int(oid)]["depth"]), int(n["depth"])]})
indeg[int(oid)] = int(indeg[int(oid)]) + 1
if not errs.is_empty():
return errs
var entries := []
var terminals := []
for n in nodes:
var id0: int = int(n["id"])
var outs: Array = n["next"]
if int(indeg[id0]) == 0:
entries.append(id0)
if outs.is_empty():
terminals.append(id0)
if String(n["type"]) != "boss":
errs.append({"category": "structural", "message": "终点(无出边)非 Boss: 节点 %d 类型 %s" % [id0, n["type"]]})
elif String(n["type"]) == "boss":
errs.append({"category": "structural", "message": "Boss 节点带出边: %d" % id0})
elif outs.size() != 1 and not (outs.size() >= 2 and outs.size() <= 3):
errs.append({"category": "domain", "message": "节点 %d 去路数 %d 非〔恰一条或岔口 2–3〕" % [id0, outs.size()]})
if not errs.is_empty():
return errs
# 深度=自入口最长路径步数(契约 2 赋值语义——出边严格前进的无环判据成立的前提)
var longest := _longest_paths(nodes, by_id, entries, indeg)
for n in nodes:
var id1: int = int(n["id"])
if int(n["depth"]) != int(longest.get(id1, -1)):
errs.append({"category": "domain", "message": "节点 %d 深度 %d ≠ 自入口最长路径步数 %d" % [
id1, n["depth"], int(longest.get(id1, -1))]})
var reaches := {}
for tid in terminals:
_mark_ancestors(int(tid), by_id, reaches)
for n in nodes:
if not reaches.has(int(n["id"])):
errs.append({"category": "structural", "message": "节点 %d 不可达任何终点(断头/孤儿)" % n["id"]})
if not errs.is_empty():
return errs
# 摆放约束①:每终点前一位恰一店、商店总数=终点数、终点前一层去路集合恰含一个终点
var shop_count := 0
for n in nodes:
if String(n["type"]) == "shop":
shop_count += 1
if shop_count != terminals.size():
errs.append({"category": "structural", "message": "商店总数 %d ≠ 终点数 %d(商店数=出口数已锁)" % [shop_count, terminals.size()]})
for tid in terminals:
var preds := []
for n in nodes:
if (n["next"] as Array).has(tid):
preds.append(n)
if preds.size() != 1:
errs.append({"category": "structural", "message": "终点 %d 直接前驱数 %d ≠ 1" % [tid, preds.size()]})
continue
var pred: Dictionary = preds[0]
if String(pred["type"]) != "shop":
errs.append({"category": "structural", "message": "终点 %d 前一位非商店: %s" % [tid, pred["type"]]})
continue
var t_hits := 0
for oid in (pred["next"] as Array):
if terminals.has(int(oid)):
t_hits += 1
if t_hits != 1:
errs.append({"category": "structural", "message": "终点 %d 前一层去路集合含终点 %d 个(归属锁定破)" % [tid, t_hits]})
if not _all_paths_one_shop(by_id, entries, terminals):
errs.append({"category": "structural", "message": "存在入口→终点路径穿店数 ≠ 1(必经恰一店破)"})
_check_fork_rules(nodes, by_id, errs)
var tdepths := {}
for tid in terminals:
var dd: int = int(by_id[tid]["depth"])
if tdepths.has(dd):
errs.append({"category": "structural", "message": "多终点同深度 %d(推论条款 A 破——定位信息只有两样)" % dd})
tdepths[dd] = true
var per_depth := {}
for n in nodes:
var d2: int = int(n["depth"])
per_depth[d2] = int(per_depth.get(d2, 0)) + 1
for d3 in per_depth:
if int(per_depth[d3]) > MAX_WIDTH:
errs.append({"category": "structural", "message": "深度 %d 节点数 %d 超分叉生长上限 4(平行层摆放)" % [d3, per_depth[d3]]})
var battles := 0
for n in nodes:
if String(n["type"]) == "battle":
battles += 1
if not params.is_empty():
if battles < int(params.get("battle_floor", 0)):
errs.append({"category": "domain", "message": "战斗节点数 %d 低于保底 %d" % [battles, params.get("battle_floor", 0)]})
if entries.size() != int(params.get("entry_count", entries.size())):
errs.append({"category": "structural", "message": "入口数 %d ≠ 参数档 entry_count %d" % [entries.size(), params.get("entry_count", 0)]})
if terminals.size() != int(params.get("exit_count", terminals.size())):
errs.append({"category": "structural", "message": "终点数 %d ≠ 参数档 exit_count %d" % [terminals.size(), params.get("exit_count", 0)]})
var maxd := 0
for tid in terminals:
maxd = maxi(maxd, int(by_id[tid]["depth"]))
if maxd != int(params.get("layers", maxd)):
errs.append({"category": "domain", "message": "层高(最长路径步数)%d ≠ 参数档 layers %d" % [maxd, params.get("layers", 0)]})
return errs
# ---- 构型(长图算法 HOW 归实现——D1:28;合法性由 validate_output 强制)----
func _build(rg: RandomNumberGenerator) -> Dictionary:
var L := int(_params["layers"])
var e := int(_params["exit_count"])
var en := int(_params["entry_count"])
var t := 0 if e == 1 else _trunk_len(L, e)
var entries: Array[int] = []
for i in range(en):
entries.append(_mk("battle", 0))
var prevs: Array[int] = entries.duplicate()
for d in range(1, t + 1):
var nid := _mk("battle", d)
for p in prevs:
(_by_id[p]["next"] as Array).append(nid)
prevs = [nid]
var terminals: Array[int] = []
for r in range(e):
var bdepth := L - r
var bid := _mk("boss", bdepth)
var sid := _mk("shop", bdepth - 1)
(_by_id[sid]["next"] as Array).append(bid)
var cur := sid
var sig := String(REGION_SIGNATURES[r]) if e > 1 else "battle"
for d in range(bdepth - 2, t, -1):
var fid := _mk(sig, d)
(_by_id[fid]["next"] as Array).append(cur)
cur = fid
terminals.append(bid)
for p in prevs:
(_by_id[p]["next"] as Array).append(cur)
# e>1 时 prevs 收束为主链段末节点一处 ⇒ 该节点即归属大岔口(各区域互不连达);
# e==1 时 prevs 为入口数组,直接接单区域头(无大岔口——L1 出口恒 1)。
_add_forks(rg, L, t, e)
_assign_types(rg, e)
var nodes: Array[Dictionary] = []
for n in _pool:
nodes.append(n)
return {"nodes": nodes, "entries": entries, "terminals": terminals}
## 岔口布置:主链段(深度 ≤ t-2)或单出口图链上(深度 ≤ L-3)的节点扩为 2–3 去路岔口;
## 侧支节点与主去路同深(保持「深度=自入口最长路径步数」不变量),两支各自汇入主去路的后继,
## 于是支路互斥段={主去路节点} 与 {侧支节点}——两支类型互异即同时满足「下一跳互异」与「各有独有类型」。
func _add_forks(rg: RandomNumberGenerator, L: int, t: int, e: int) -> void:
var wanted := clampi(int(round(float(_params["branch_density"]) * float(L))), 1, 24)
var anchors := []
for n in _pool:
if int(n["depth"]) > (t - 2 if e > 1 else L - 3):
continue
if (n["next"] as Array).size() != 1:
continue
anchors.append(n)
anchors = _shuffle(rg, anchors)
var selected := {} # 已扩岔节点 id
var succ_selected := {} # 已扩岔节点的主去路 id(禁再作锚点)
for a in anchors:
if wanted <= 0:
break
var node: Dictionary = a
var nid := int(node["id"])
if (node["next"] as Array).size() != 1:
continue # 同一节点不重复扩岔(去路数上限 3)
var v: Dictionary = _by_id[int((node["next"] as Array)[0])]
var vid := int(v["id"])
# 相邻两层不得同时扩岔:若主去路 v 也是岔口,v 的侧支会落进本岔口主支路的互斥段,
# 使侧支自己的类型不再独有(契约 5③)——两个方向都要拦。
if selected.has(vid) or succ_selected.has(nid):
continue
var v_outs: Array = v["next"]
if v_outs.is_empty():
continue # 主去路无后继(无可汇入点)——该锚点弃用
var extra := 2 if (int(_params["fork_ways_max"]) >= 3 and rg.randf() < 0.35) else 1
var made := 0
for k in range(extra):
if _depth_count(int(v["depth"])) >= MAX_WIDTH:
break
var wid := _mk("battle", int(v["depth"]))
(node["next"] as Array).append(wid)
(_by_id[wid]["next"] as Array).append(int(v_outs[0]))
_detour_heads.append(_by_id[wid])
_detour_ids[wid] = true
made += 1
if made > 0:
selected[nid] = true
succ_selected[vid] = true
wanted -= 1
# 确定性洗牌(Fisher-Yates,取数只经注入的纯流 RNG——Array.shuffle() 走引擎全局随机、
# 破「同(种子,参数档)⇒ 同图」的纯函数复现前提,禁用于生成器)
func _shuffle(rg: RandomNumberGenerator, arr: Array) -> Array:
var out := arr.duplicate()
for i in range(out.size() - 1, 0, -1):
var j := rg.randi_range(0, i)
var tmp = out[i]
out[i] = out[j]
out[j] = tmp
return out
func _depth_count(d: int) -> int:
var c := 0
for n in _pool:
if int(n["depth"]) == d:
c += 1
return c
## 类型装配:多出口图区域签名/主链段类型在构型期已定(支路互斥段的独有类型来源),此处只补
## 单出口图的配比填余(摆放优先、配比填余、战斗保底兜底——§8.1 次序),再统一修岔口下一跳互异。
func _assign_types(rg: RandomNumberGenerator, e: int) -> void:
if e == 1:
var free: Array = []
var chain_battles := 0
for n in _pool:
if String(n["type"]) == "shop" or String(n["type"]) == "boss":
continue
if _detour_ids.has(int(n["id"])):
continue # 侧链节点类型由岔口互异修型决定,不占配比填余面
if int(n["depth"]) == 1:
continue # 区域首节点留战斗——配比填余不破坏岔口下一跳互异的基准形
if String(n["type"]) == "battle":
chain_battles += 1
free.append(n)
free = _shuffle(rg, free)
var want := int(round(float(_pool.size()) * (float(_params["reward_ratio"]) + float(_params["rest_ratio"]))))
var floor_n := int(_params["battle_floor"])
var rest_first := rg.randf() < 0.5
for i in range(want):
if i >= free.size():
break
# 战斗保底优先:链上战斗数降到保底下限即停(侧链修型可能再让出战斗位,故保底只看链上)
if chain_battles - 1 < floor_n:
break
var node: Dictionary = free[i]
node["type"] = "rest" if ((i % 2 == 0) == rest_first) else "reward"
chain_battles -= 1
for head in _detour_heads:
var h: Dictionary = head
var fork = null
for n in _pool:
if (n["next"] as Array).has(int(h["id"])):
fork = n
break
if fork == null:
continue
var used := {}
for oid in (fork["next"] as Array):
if int(oid) != int(h["id"]):
used[String(_by_id[int(oid)]["type"])] = true
for cand in FILLER_TYPES:
if not used.has(cand):
h["type"] = cand
break
func _mk(type_: String, depth_: int) -> int:
var n := {"id": _next_id, "type": type_, "depth": depth_, "next": []}
_next_id += 1
_pool.append(n)
_by_id[int(n["id"])] = n
return int(n["id"])
static func _trunk_len(L: int, e: int) -> int:
return mini(int(L / 2), L - e - 1)
# ---- 自证辅助 ----
func _longest_paths(nodes: Array, by_id: Dictionary, entries: Array, indeg: Dictionary) -> Dictionary:
var out := {}
var din := {}
var q := []
for id in indeg:
din[int(id)] = int(indeg[int(id)])
for eid in entries:
out[int(eid)] = 0
q.append(int(eid))
while not q.is_empty():
var id: int = int(q.pop_front())
for oid in (by_id[id]["next"] as Array):
var nid: int = int(oid)
out[nid] = maxi(int(out.get(nid, 0)), int(out[id]) + 1)
din[nid] = int(din[nid]) - 1
if din[nid] == 0:
q.append(nid)
return out
func _mark_ancestors(t: int, by_id: Dictionary, reached: Dictionary) -> void:
var q := [t]
reached[t] = true
while not q.is_empty():
var id: int = int(q.pop_front())
for n in by_id.values():
if (n["next"] as Array).has(id) and not reached.has(int(n["id"])):
reached[int(n["id"])] = true
q.append(int(n["id"]))
func _all_paths_one_shop(by_id: Dictionary, entries: Array, terminals: Array) -> bool:
var stack := []
for eid in entries:
stack.append([int(eid), 1 if String(by_id[int(eid)]["type"]) == "shop" else 0])
while not stack.is_empty():
var frame: Array = stack.pop_back()
var id: int = int(frame[0])
var shops := int(frame[1])
var outs: Array = by_id[id]["next"]
if outs.is_empty():
if shops != 1 or not terminals.has(id):
return false
continue
for oid in outs:
stack.append([int(oid), shops + (1 if String(by_id[int(oid)]["type"]) == "shop" else 0)])
return true
func _check_fork_rules(nodes: Array, by_id: Dictionary, errs: Array) -> void:
for n in nodes:
var outs: Array = n["next"]
if outs.size() < 2:
continue
var head_types := {}
for oid in outs:
var ty := String(by_id[int(oid)]["type"])
if head_types.has(ty):
errs.append({"category": "structural", "message": "岔口 %d 各去路下一节点类型重复: %s" % [n["id"], ty]})
head_types[ty] = true
var reach_sets := []
for oid in outs:
reach_sets.append(_descendants(int(oid), by_id))
var comps := []
for i in range(reach_sets.size()):
var excl := {}
for rid in reach_sets[i]:
var shared := false
for j in range(reach_sets.size()):
if j != i and reach_sets[j].has(rid):
shared = true
break
if not shared:
excl[int(rid)] = String(by_id[int(rid)]["type"])
var arr := []
for v in excl.values():
arr.append(v)
arr.sort()
comps.append(arr)
for i in range(comps.size()):
for j in range(i + 1, comps.size()):
if comps[i] == comps[j]:
errs.append({"category": "structural", "message": "岔口 %d 支路 %d/%d 类型构成相同" % [n["id"], i, j]})
for i in range(comps.size()):
var uniq := false
for ty in comps[i]:
var in_other := false
for j in range(comps.size()):
if j != i and (comps[j] as Array).has(ty):
in_other = true
break
if not in_other:
uniq = true
break
if not uniq:
errs.append({"category": "structural", "message": "岔口 %d 支路 %d 无独有类型" % [n["id"], i]})
func _descendants(start: int, by_id: Dictionary) -> Dictionary:
var out := {start: true}
var q := [start]
while not q.is_empty():
var id: int = int(q.pop_front())
for oid in (by_id[id]["next"] as Array):
if not out.has(int(oid)):
out[int(oid)] = true
q.append(int(oid))
return out
func _fatal(msg: String) -> void:
assert(false, msg)
OS.crash("NodeMapGenerator: " + msg)