卡①T3(核心数据面结构收敛批): 生成器族两文件拆分——board_generator generate收束(_board_params快照+_attempt_loop重试主循环+_accept_fallback三分出口)+_attempt布雷剃雷两域(_populate_mines Fisher-Yates取号段整体随迁+_shave_ring零RNG)+measure三算子域(_apply_chain_rule/_apply_single_rules/_derived_constraints/_apply_derived_rules)+_finalize删_db参5参化;node_map_generator validate_output六阶段域(_vo_node_fields/_vo_edges/_vo_flow_facts/_vo_depth_reach/_vo_shop_counts/_vo_terminal_shops/_vo_terminal_depths/_vo_width_rules/_vo_params_check)+check_params两分(_cp_domain/_cp_structural)+_build主链区域两段(_build_trunk/_build_region零RNG)+_add_forks锚收集段零RNG提取+_assign_types两分(_fill_ratios取号序整体搬迁+_fix_detour_heads零RNG)+_check_fork_rules签名两助手;RNG消费序冻结同种子逐位一致,17键全清,node_map三件+generator套件全绿

This commit is contained in:
DKChan committed 2026-10-28 07:24:56 +08:00
1 parent 9646ce545d
commit 331d6b039a
2 files changed
+346 -206

No files matched your search

+127 -82
View File
@@ -26,8 +26,15 @@ func generate(L: int, D: int, rng_stream: RandomNumberGenerator, db) -> BoardDat
if row.is_empty(): if row.is_empty():
_fatal("BoardGen 行缺失 (%d,%d)" % [L, D]) _fatal("BoardGen 行缺失 (%d,%d)" % [L, D])
return null return null
# 生成期表值单次读取快照(防热重载撕裂 attempt 语义) var params := _board_params(row, L, D)
var params := { var outcome := _attempt_loop(params, rng_stream, db)
_finalize(outcome["board"], params, outcome["attempts"], outcome["mode"], outcome["r0_rej"])
return outcome["board"]
## 生成期表值单次读取快照(防热重载撕裂 attempt 语义;零 RNG)
func _board_params(row: Dictionary, L: int, D: int) -> Dictionary:
return {
"L": L, "D": D, "L": L, "D": D,
"w": int(row.board_width), "h": int(row.board_height), "w": int(row.board_width), "h": int(row.board_height),
"density": float(row.mine_density), "density": float(row.mine_density),
@@ -41,10 +48,11 @@ func generate(L: int, D: int, rng_stream: RandomNumberGenerator, db) -> BoardDat
"shave_r": int(row.get("shave_ring_radius", 1)), "shave_r": int(row.get("shave_ring_radius", 1)),
"r0_cap": int(row.get("r0_max_cap", 0)), "r0_cap": int(row.get("r0_max_cap", 0)),
} }
var best_legal: BoardData = null
var best_legal_score := -1.0
var best_bad: BoardData = null ## 重试主循环:逐尝试评分择优,产出三分出口结果(board/mode/attempts/r0_rej)
var best_badness := [999999, 999999, 999999] func _attempt_loop(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> Dictionary: # type-hint: ignore(同 generate——测量面 stub 注入)
var best := {"legal": null, "legal_score": -1.0, "bad": null, "badness": [999999, 999999, 999999]}
var attempts_used := 0 var attempts_used := 0
var accepted: BoardData = null var accepted: BoardData = null
var accept_mode := "" var accept_mode := ""
@@ -53,35 +61,37 @@ func generate(L: int, D: int, rng_stream: RandomNumberGenerator, db) -> BoardDat
attempts_used = k attempts_used = k
var b := _attempt(params, rng_stream, db) var b := _attempt(params, rng_stream, db)
var badness := _badness(b, params) var badness := _badness(b, params)
if best_bad == null or _lex_less(badness, best_badness): if best["bad"] == null or _lex_less(badness, best["badness"]):
best_bad = b best["bad"] = b
best_badness = badness best["badness"] = badness
# R₀≤cap 重掷闸触发计数(迭代14① T3——尝试事实,与合法性判定独立记录;占比=T5 闸触发率读数) # R₀≤cap 重掷闸触发计数(迭代14① T3——尝试事实,与合法性判定独立记录;占比=T5 闸触发率读数)
if int(params.r0_cap) > 0 and b.compute_r0().size() > int(params.r0_cap): if int(params.r0_cap) > 0 and b.compute_r0().size() > int(params.r0_cap):
r0_cap_rejected += 1 r0_cap_rejected += 1
if legality(b, params): if legality(b, params):
var m: Dictionary[Variant, Variant] = measure_inferability(b) # 求解器每合法板恰一次(O3 收口——此前重复调用两次) var inf: float = measure_inferability(b)["inferability"] # 求解器每合法板恰一次(O3 收口——此前重复调用两次)
var inf: float = m["inferability"] if best["legal"] == null or inf > best["legal_score"]:
if best_legal == null or inf > best_legal_score: best["legal"] = b
best_legal = b best["legal_score"] = inf
best_legal_score = inf
if inf >= params.target: # 单一判据(棋盘生成:现行规则已回归"可推理率 ≥ 及格线") if inf >= params.target: # 单一判据(棋盘生成:现行规则已回归"可推理率 ≥ 及格线")
accepted = b accepted = b
accept_mode = "first_pass" accept_mode = "first_pass"
break break
if accepted == null: if accepted == null:
if best_legal != null: var fallback := _accept_fallback(params, best["legal"], best["legal_score"], best["bad"])
accepted = best_legal accepted = fallback[0]
accept_mode = "accept_best" accept_mode = fallback[1]
accepted.degraded_product = true # 降级标记(独立于 illegal_product——三分出口表要求标记随产物上抛) return {"board": accepted, "mode": accept_mode, "attempts": attempts_used, "r0_rej": r0_cap_rejected}
push_warning("[棋盘生成] 重试耗尽无达标——接受最优出图(强告警埋点,target=%.2f best=%.3f)" % [params.target, best_legal_score])
else:
accepted = best_bad ## 三分出口(重试耗尽):接受最优(降级标记)或接受最不坏(非法产物标记——玩家永不卡死)
accept_mode = "accept_worst" func _accept_fallback(params: Dictionary, best_legal: BoardData, best_legal_score: float, best_bad: BoardData) -> Array:
accepted.illegal_product = true if best_legal != null:
push_error("[棋盘生成] attempts 耗尽零合法——接受最不坏出图 + 非法产物标记(玩家永不卡死)") best_legal.degraded_product = true # 降级标记(独立于 illegal_product——三分出口表要求标记随产物上抛)
_finalize(accepted, params, attempts_used, accept_mode, db, r0_cap_rejected) push_warning("[棋盘生成] 重试耗尽无达标——接受最优出图(强告警埋点,target=%.2f best=%.3f)" % [params.target, best_legal_score])
return accepted return [best_legal, "accept_best"]
best_bad.illegal_product = true
push_error("[棋盘生成] attempts 耗尽零合法——接受最不坏出图 + 非法产物标记(玩家永不卡死)")
return [best_bad, "accept_worst"]
# ==================== 五阶段:单次尝试 ==================== # ==================== 五阶段:单次尝试 ====================
@@ -94,8 +104,20 @@ func _attempt(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> Boar
b.board_type = str(params.board_type) b.board_type = str(params.board_type)
b.level_index = params.L b.level_index = params.L
b.node_depth = params.D b.node_depth = params.D
# ② 布雷:N = round(d×(W×H−9)) 先定数;域 = 全格 − 出生 3×3;均匀布雷后"先布后删" # ② 布雷:N = round(d×(W×H−9)) 先定数;域 = 全格 − 出生 3×3;均匀布雷后"先布后删"(_populate_mines 承载——Fisher-Yates 取号段整体随迁)
# (−9 = 出生区格数,常量单点 BoardData.SPAWN_AREA_CELLS) var removed := _populate_mines(b, params, rng_stream)
# ③ 变异布置(地形变异 框架面):布置约束与算法随首批变异类型定稿(生成期布置接口结构位),
# 定稿前零布置——清单为空是合法态(P0),静默产出无变异图。
b.metrics = {
"metric_version": METRIC_VERSION,
"ring_removed": removed,
}
return b
## 布雷与剃雷域:域 = 全格 − 出生 3×3;均匀布雷(部分 Fisher-Yates,确定性,消费 board_gen 流)
## 后剃出生环雷,返回 ring_removed(指标语义不变——现值变小=收窄批预期)
func _populate_mines(b: BoardData, params: Dictionary, rng_stream: RandomNumberGenerator) -> int:
var spawn := {} var spawn := {}
for c in b.spawn_cells(): for c in b.spawn_cells():
spawn[c] = true spawn[c] = true
@@ -104,7 +126,7 @@ func _attempt(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> Boar
if not spawn.has(c): if not spawn.has(c):
domain.append(c) domain.append(c)
var n_mines: int = target_mine_count(params) var n_mines: int = target_mine_count(params)
# 部分 Fisher-Yates(确定性,消费 board_gen 流) # 部分 Fisher-Yates(确定性,消费 board_gen 流)——取号序冻结(契约:同种子逐位一致)
for i in range(0, mini(n_mines, domain.size())): for i in range(0, mini(n_mines, domain.size())):
var j := rng_stream.randi_range(i, domain.size() - 1) var j := rng_stream.randi_range(i, domain.size() - 1)
var tmp = domain[i] var tmp = domain[i]
@@ -116,9 +138,14 @@ func _attempt(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> Boar
# 先布后删:剃雷域 = 距 spawn_center 切比雪夫距离 ≤1+r 的格(迭代14①收窄批 T2——原「R₀ flood 域∪ # 先布后删:剃雷域 = 距 spawn_center 切比雪夫距离 ≤1+r 的格(迭代14①收窄批 T2——原「R₀ flood 域∪
# 外扩 1 环」全域剃雷=开局开阔地巨大根因〔10×10 板 R₀ 均值 68.6=板面 68.6%〕;收窄后开阔地= # 外扩 1 环」全域剃雷=开局开阔地巨大根因〔10×10 板 R₀ 均值 68.6=板面 68.6%〕;收窄后开阔地=
# 出生邻域外的自然 flood。r 读表 shave_ring_radius(r=1 → 5×5=25 格;r=0 即只剃出生 3×3)。 # 出生邻域外的自然 flood。r 读表 shave_ring_radius(r=1 → 5×5=25 格;r=0 即只剃出生 3×3)。
# 只减不增、无迭代语义不变;ring_removed 指标语义不变(现值变小=预期);零 RNG 消费变化—— # 只减不增、无迭代语义不变;零 RNG 消费变化。
# 同种子布雷序列逐位一致,仅移除域收窄。 var removed := _shave_ring(b, int(params.get("shave_r", 1)))
var shave_r: int = int(params.get("shave_r", 1)) _compute_digits(b)
return removed
## 剃雷环域(零 RNG):距出生心切比雪夫 ≤1+r 的环内剃雷,返回剃除数
func _shave_ring(b: BoardData, shave_r: int) -> int:
var ring := {} var ring := {}
for c in b.all_cells(): for c in b.all_cells():
if maxi(absi(c.x - b.spawn_center.x), absi(c.y - b.spawn_center.y)) <= 1 + shave_r: if maxi(absi(c.x - b.spawn_center.x), absi(c.y - b.spawn_center.y)) <= 1 + shave_r:
@@ -128,14 +155,7 @@ func _attempt(params: Dictionary, rng_stream: RandomNumberGenerator, db) -> Boar
if b.mines.has(c): if b.mines.has(c):
b.mines.erase(c) b.mines.erase(c)
removed += 1 removed += 1
_compute_digits(b) return removed
# ③ 变异布置(地形变异 框架面):布置约束与算法随首批变异类型定稿(生成期布置接口结构位),
# 定稿前零布置——清单为空是合法态(P0),静默产出无变异图。
b.metrics = {
"metric_version": METRIC_VERSION,
"ring_removed": removed,
}
return b
func _compute_digits(b: BoardData) -> void: func _compute_digits(b: BoardData) -> void:
@@ -164,49 +184,12 @@ func measure_inferability(b: BoardData) -> Dictionary:
while changed: while changed:
changed = false changed = false
# 算子① 连锁:0 格的 8 邻安全格全入 R # 算子① 连锁:0 格的 8 邻安全格全入 R
var frontier: Array[Variant] = r.keys() changed = _apply_chain_rule(b, r, safe_set) or changed
for x in frontier:
if safe_set.has(x) and b.digit_of(x) == 0:
for n in b.neighbors8(x):
if safe_set.has(n) and not r.has(n):
r[n] = true
changed = true
# 信息源:数字约束 single + subset(solver_rules=single_subset 冻结域) # 信息源:数字约束 single + subset(solver_rules=single_subset 冻结域)
var constraints := _build_constraints(b, r, known_mines, safe_set) var constraints := _build_constraints(b, r, known_mines, safe_set)
# single 规则 changed = _apply_single_rules(constraints, r, known_mines) or changed
for con in constraints: var derived := _derived_constraints(constraints)
if con.mines == 0: changed = _apply_derived_rules(derived, r, known_mines) or changed
for c in con.unknown:
if not r.has(c):
r[c] = true
changed = true
elif con.mines == con.unknown.size():
for c in con.unknown:
if not known_mines.has(c):
known_mines[c] = true
changed = true
# subset 规则:U1 ⊂ U2 ⇒ (d2−d1, U2∖U1)
var derived := []
for i in constraints.size():
for j in range(i + 1, constraints.size()):
var a: Dictionary[Variant, Variant] = constraints[i]
var bb: Dictionary[Variant, Variant] = constraints[j]
var sub := _subset_of(a.unknown, bb.unknown)
if sub == 1:
derived.append(_diff_constraint(a, bb))
elif sub == 2:
derived.append(_diff_constraint(bb, a))
for con in derived:
if con.mines == 0:
for c in con.unknown:
if not r.has(c): # 同 no-peek 口径:推导结论不再用 safe_set 复核(约束本身可靠)
r[c] = true
changed = true
elif con.mines == con.unknown.size():
for c in con.unknown:
if not known_mines.has(c):
known_mines[c] = true
changed = true
var r_inf := r.keys() var r_inf := r.keys()
var s_count := safe_set.size() var s_count := safe_set.size()
var inferability: float = float(r_inf.size()) / float(maxi(1, s_count)) var inferability: float = float(r_inf.size()) / float(maxi(1, s_count))
@@ -217,6 +200,68 @@ func measure_inferability(b: BoardData) -> Dictionary:
} }
## 算子① 连锁域:0 格的 8 邻安全格全入 R(返回本轮是否有新增)
func _apply_chain_rule(b: BoardData, r: Dictionary, safe_set: Dictionary) -> bool:
var changed := false
var frontier: Array[Variant] = r.keys()
for x in frontier:
if safe_set.has(x) and b.digit_of(x) == 0:
for n in b.neighbors8(x):
if safe_set.has(n) and not r.has(n):
r[n] = true
changed = true
return changed
## single 规则域:零雷约束开格/满雷约束标雷(返回本轮是否有新增)
func _apply_single_rules(constraints: Array, r: Dictionary, known_mines: Dictionary) -> bool:
var changed := false
for con in constraints:
if con.mines == 0:
for c in con.unknown:
if not r.has(c):
r[c] = true
changed = true
elif con.mines == con.unknown.size():
for c in con.unknown:
if not known_mines.has(c):
known_mines[c] = true
changed = true
return changed
## subset 派生域:U1 ⊂ U2 ⇒ (d2−d1, U2∖U1)(对约束对做包含判定,产出派生约束集)
func _derived_constraints(constraints: Array) -> Array:
var derived := []
for i in constraints.size():
for j in range(i + 1, constraints.size()):
var a: Dictionary[Variant, Variant] = constraints[i]
var bb: Dictionary[Variant, Variant] = constraints[j]
var sub := _subset_of(a.unknown, bb.unknown)
if sub == 1:
derived.append(_diff_constraint(a, bb))
elif sub == 2:
derived.append(_diff_constraint(bb, a))
return derived
## 派生规则域:派生约束的零雷开格/满雷标雷(同 no-peek 口径:推导结论不再用 safe_set 复核——约束本身可靠)
func _apply_derived_rules(derived: Array, r: Dictionary, known_mines: Dictionary) -> bool:
var changed := false
for con in derived:
if con.mines == 0:
for c in con.unknown:
if not r.has(c):
r[c] = true
changed = true
elif con.mines == con.unknown.size():
for c in con.unknown:
if not known_mines.has(c):
known_mines[c] = true
changed = true
return changed
func _build_constraints(b: BoardData, r: Dictionary, known_mines: Dictionary, safe_set: Dictionary) -> Array: func _build_constraints(b: BoardData, r: Dictionary, known_mines: Dictionary, safe_set: Dictionary) -> Array:
var out: Array[Dictionary] = [] var out: Array[Dictionary] = []
for c in r.keys(): for c in r.keys():
@@ -337,7 +382,7 @@ func _neighbor_mine_count(b: BoardData, c: Vector2i) -> int:
# ==================== 产出上报(生成期指标集) ==================== # ==================== 产出上报(生成期指标集) ====================
func _finalize(b: BoardData, params: Dictionary, attempts_used: int, accept_mode: String, _db, r0_cap_rejected: int = 0) -> void: # type-hint: ignore(同 generate——测量面 stub 注入) func _finalize(b: BoardData, params: Dictionary, attempts_used: int, accept_mode: String, r0_cap_rejected: int = 0) -> void:
var m := measure_inferability(b) var m := measure_inferability(b)
# 产物断言(棋盘生成 出生区安全与反退化布雷):实际雷数 ≤ 目标雷数 N——先布后删只减不增,失败即致命 # 产物断言(棋盘生成 出生区安全与反退化布雷):实际雷数 ≤ 目标雷数 N——先布后删只减不增,失败即致命
var n_target: int = target_mine_count(params) var n_target: int = target_mine_count(params)
+219 -124
View File
@@ -66,6 +66,14 @@ func _setup_and_build(params: Dictionary, seed_val: int) -> Dictionary:
## 参数档可行性判定(执行边界③:参数无解=配置错误)——域校验外加深度预算与战斗保底可放性。 ## 参数档可行性判定(执行边界③:参数无解=配置错误)——域校验外加深度预算与战斗保底可放性。
static func check_params(params: Dictionary) -> Array: static func check_params(params: Dictionary) -> Array:
var errs: Array[Dictionary] = _cp_domain(params)
if not errs.is_empty():
return errs
return _cp_structural(params)
## 域校验段:五旋钮值域/岔口去路数锁值/配比四类合计 1
static func _cp_domain(params: Dictionary) -> Array[Dictionary]:
var errs: Array[Dictionary] = [] var errs: Array[Dictionary] = []
var L := int(params.get("layers", 0)) var L := int(params.get("layers", 0))
var e := int(params.get("exit_count", 0)) var e := int(params.get("exit_count", 0))
@@ -89,8 +97,16 @@ static func check_params(params: Dictionary) -> Array:
sum += float(params.get(rk, -1.0)) sum += float(params.get(rk, -1.0))
if not is_equal_approx(sum, 1.0): if not is_equal_approx(sum, 1.0):
errs.append({"category": "domain", "message": "配比四类合计 %f ≠ 1" % sum}) errs.append({"category": "domain", "message": "配比四类合计 %f ≠ 1" % sum})
if not errs.is_empty(): return errs
return errs
## 结构可行性段:主链段容量与战斗保底可放性(多出口按区域签名累加可放槽)
static func _cp_structural(params: Dictionary) -> Array[Dictionary]:
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 t := 0 if e == 1 else _trunk_len(L, e) var t := 0 if e == 1 else _trunk_len(L, e)
if e > 1 and t < 1: if e > 1 and t < 1:
errs.append({"category": "structural", "message": "层深 %d 与出口数 %d 无解(主链段放不下归属区域)" % [L, e]}) errs.append({"category": "structural", "message": "层深 %d 与出口数 %d 无解(主链段放不下归属区域)" % [L, e]})
@@ -116,11 +132,33 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
return errs return errs
var by_id := {} var by_id := {}
var indeg := {} var indeg := {}
if not _vo_node_fields(nodes, errs, by_id, indeg):
return errs
if not _vo_edges(nodes, by_id, indeg, errs):
return errs
var flow := _vo_flow_facts(nodes, indeg, errs)
if not flow["ok"]:
return errs
if not _vo_depth_reach(nodes, by_id, flow["entries"], flow["terminals"], errs):
return errs
_vo_shop_counts(nodes, flow["terminals"], errs)
_vo_terminal_shops(nodes, flow["terminals"], errs)
if not _all_paths_one_shop(by_id, flow["entries"], flow["terminals"]):
errs.append({"category": "structural", "message": "存在入口→终点路径穿店数 ≠ 1(必经恰一店破)"})
_check_fork_rules(nodes, by_id, errs)
_vo_terminal_depths(flow["terminals"], by_id, errs)
_vo_width_rules(nodes, errs)
_vo_params_check(params, nodes, by_id, [flow["entries"], flow["terminals"]], errs)
return errs
## 阶段一·节点字段域:字段齐备/句柄唯一/类型词表/深度非负(by_id/indeg 引用填充)
func _vo_node_fields(nodes: Array, errs: Array, by_id: Dictionary, indeg: Dictionary) -> bool:
for n in nodes: for n in nodes:
for k in ["id", "type", "depth", "next"]: for k in ["id", "type", "depth", "next"]:
if not (n as Dictionary).has(k): if not (n as Dictionary).has(k):
errs.append({"category": "structural", "message": "节点字段缺失: %s" % k}) errs.append({"category": "structural", "message": "节点字段缺失: %s" % k})
return errs return false
var id: int = int(n["id"]) var id: int = int(n["id"])
if by_id.has(id): if by_id.has(id):
errs.append({"category": "dup_key", "message": "节点句柄重复: %d" % id}) errs.append({"category": "dup_key", "message": "节点句柄重复: %d" % id})
@@ -130,8 +168,11 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
if int(n["depth"]) < 0: if int(n["depth"]) < 0:
errs.append({"category": "domain", "message": "节点深度为负: %d" % n["depth"]}) errs.append({"category": "domain", "message": "节点深度为负: %d" % n["depth"]})
indeg[id] = 0 indeg[id] = 0
if not errs.is_empty(): return errs.is_empty()
return errs
## 阶段二·边域:出边悬空与环/回边判据(indeg 引用累计)
func _vo_edges(nodes: Array, by_id: Dictionary, indeg: Dictionary, errs: Array) -> bool:
for n in nodes: for n in nodes:
for oid in (n["next"] as Array): for oid in (n["next"] as Array):
if not by_id.has(int(oid)): if not by_id.has(int(oid)):
@@ -141,9 +182,11 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
errs.append({"category": "structural", "message": "连线未朝前进方向(目标深度 %d 不严格大于源 %d)——环/回边" % [ errs.append({"category": "structural", "message": "连线未朝前进方向(目标深度 %d 不严格大于源 %d)——环/回边" % [
int(by_id[int(oid)]["depth"]), int(n["depth"])]}) int(by_id[int(oid)]["depth"]), int(n["depth"])]})
indeg[int(oid)] = int(indeg[int(oid)]) + 1 indeg[int(oid)] = int(indeg[int(oid)]) + 1
if not errs.is_empty(): return errs.is_empty()
return errs
## 阶段三·流事实域:入口集(入度 0)/终点集(无出边)与 Boss 终点判据、去路数形态
func _vo_flow_facts(nodes: Array, indeg: Dictionary, errs: Array) -> Dictionary:
var entries := [] var entries := []
var terminals := [] var terminals := []
for n in nodes: for n in nodes:
@@ -159,11 +202,12 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
errs.append({"category": "structural", "message": "Boss 节点带出边: %d" % id0}) errs.append({"category": "structural", "message": "Boss 节点带出边: %d" % id0})
elif outs.size() != 1 and not (outs.size() >= 2 and outs.size() <= 3): 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()]}) errs.append({"category": "domain", "message": "节点 %d 去路数 %d 非〔恰一条或岔口 2–3〕" % [id0, outs.size()]})
if not errs.is_empty(): return {"entries": entries, "terminals": terminals, "ok": errs.is_empty()}
return errs
# 深度=自入口最长路径步数(契约 2 赋值语义——出边严格前进的无环判据成立的前提)
var longest := _longest_paths(nodes, by_id, entries, indeg) ## 阶段四·深度与可达域:深度=自入口最长路径步数(契约 2)+全节点可达某终点
func _vo_depth_reach(nodes: Array, by_id: Dictionary, entries: Array, terminals: Array, errs: Array) -> bool:
var longest := _longest_paths(nodes, by_id, entries)
for n in nodes: for n in nodes:
var id1: int = int(n["id"]) var id1: int = int(n["id"])
if int(n["depth"]) != int(longest.get(id1, -1)): if int(n["depth"]) != int(longest.get(id1, -1)):
@@ -175,16 +219,21 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
for n in nodes: for n in nodes:
if not reaches.has(int(n["id"])): if not reaches.has(int(n["id"])):
errs.append({"category": "structural", "message": "节点 %d 不可达任何终点(断头/孤儿)" % n["id"]}) errs.append({"category": "structural", "message": "节点 %d 不可达任何终点(断头/孤儿)" % n["id"]})
if not errs.is_empty(): return errs.is_empty()
return errs
# 摆放约束①:每终点前一位恰一店、商店总数=终点数、终点前一层去路集合恰含一个终点
## 摆放约束①前半·商店总数:商店总数=终点数(商店数=出口数已锁)
func _vo_shop_counts(nodes: Array, terminals: Array, errs: Array) -> void:
var shop_count := 0 var shop_count := 0
for n in nodes: for n in nodes:
if String(n["type"]) == "shop": if String(n["type"]) == "shop":
shop_count += 1 shop_count += 1
if shop_count != terminals.size(): if shop_count != terminals.size():
errs.append({"category": "structural", "message": "商店总数 %d ≠ 终点数 %d(商店数=出口数已锁)" % [shop_count, terminals.size()]}) errs.append({"category": "structural", "message": "商店总数 %d ≠ 终点数 %d(商店数=出口数已锁)" % [shop_count, terminals.size()]})
## 摆放约束①后半·终点前驱:每终点前一位恰一店且为唯一前驱、终点前一层去路集合恰含一个终点
func _vo_terminal_shops(nodes: Array, terminals: Array, errs: Array) -> void:
for tid in terminals: for tid in terminals:
var preds := [] var preds := []
for n in nodes: for n in nodes:
@@ -204,16 +253,19 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
if t_hits != 1: if t_hits != 1:
errs.append({"category": "structural", "message": "终点 %d 前一层去路集合含终点 %d 个(归属锁定破)" % [tid, t_hits]}) 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)
## 计数规则①·多终点深度互异(推论条款 A——定位信息只有两样)
func _vo_terminal_depths(terminals: Array, by_id: Dictionary, errs: Array) -> void:
var tdepths := {} var tdepths := {}
for tid in terminals: for tid in terminals:
var dd: int = int(by_id[tid]["depth"]) var dd: int = int(by_id[tid]["depth"])
if tdepths.has(dd): if tdepths.has(dd):
errs.append({"category": "structural", "message": "多终点同深度 %d(推论条款 A 破——定位信息只有两样)" % dd}) errs.append({"category": "structural", "message": "多终点同深度 %d(推论条款 A 破——定位信息只有两样)" % dd})
tdepths[dd] = true tdepths[dd] = true
## 计数规则②·宽度上限:同深度节点数 ≤ 分叉生长上限 4(禁平行层摆放)
func _vo_width_rules(nodes: Array, errs: Array) -> void:
var per_depth := {} var per_depth := {}
for n in nodes: for n in nodes:
var d2: int = int(n["depth"]) var d2: int = int(n["depth"])
@@ -222,23 +274,28 @@ func validate_output(graph: Dictionary, params: Dictionary = {}) -> Array:
if int(per_depth[d3]) > MAX_WIDTH: if int(per_depth[d3]) > MAX_WIDTH:
errs.append({"category": "structural", "message": "深度 %d 节点数 %d 超分叉生长上限 4(平行层摆放)" % [d3, per_depth[d3]]}) errs.append({"category": "structural", "message": "深度 %d 节点数 %d 超分叉生长上限 4(平行层摆放)" % [d3, per_depth[d3]]})
## 计数规则③·参数档对照:战斗保底/入口数/终点数/层高(params 为空时跳过)
func _vo_params_check(params: Dictionary, nodes: Array, by_id: Dictionary, flow_ends: Array, errs: Array) -> void:
var entries: Array = flow_ends[0]
var terminals: Array = flow_ends[1]
var battles := 0 var battles := 0
for n in nodes: for n in nodes:
if String(n["type"]) == "battle": if String(n["type"]) == "battle":
battles += 1 battles += 1
if not params.is_empty(): if params.is_empty():
if battles < int(params.get("battle_floor", 0)): return
errs.append({"category": "domain", "message": "战斗节点数 %d 低于保底 %d" % [battles, params.get("battle_floor", 0)]}) if battles < int(params.get("battle_floor", 0)):
if entries.size() != int(params.get("entry_count", entries.size())): errs.append({"category": "domain", "message": "战斗节点数 %d 低于保底 %d" % [battles, params.get("battle_floor", 0)]})
errs.append({"category": "structural", "message": "入口数 %d ≠ 参数档 entry_count %d" % [entries.size(), params.get("entry_count", 0)]}) if entries.size() != int(params.get("entry_count", entries.size())):
if terminals.size() != int(params.get("exit_count", terminals.size())): errs.append({"category": "structural", "message": "入口数 %d ≠ 参数档 entry_count %d" % [entries.size(), params.get("entry_count", 0)]})
errs.append({"category": "structural", "message": "终点数 %d ≠ 参数档 exit_count %d" % [terminals.size(), params.get("exit_count", 0)]}) if terminals.size() != int(params.get("exit_count", terminals.size())):
var maxd := 0 errs.append({"category": "structural", "message": "终点数 %d ≠ 参数档 exit_count %d" % [terminals.size(), params.get("exit_count", 0)]})
for tid in terminals: var maxd := 0
maxd = maxi(maxd, int(by_id[tid]["depth"])) for tid in terminals:
if maxd != int(params.get("layers", maxd)): maxd = maxi(maxd, int(by_id[tid]["depth"]))
errs.append({"category": "domain", "message": "层高(最长路径步数)%d ≠ 参数档 layers %d" % [maxd, params.get("layers", 0)]}) if maxd != int(params.get("layers", maxd)):
return errs errs.append({"category": "domain", "message": "层高(最长路径步数)%d ≠ 参数档 layers %d" % [maxd, params.get("layers", 0)]})
# ---- 构型(长图算法 HOW 归实现——D1:28;合法性由 validate_output 强制)---- # ---- 构型(长图算法 HOW 归实现——D1:28;合法性由 validate_output 强制)----
@@ -252,28 +309,11 @@ func _build(rg: RandomNumberGenerator) -> Dictionary:
var entries: Array[int] = [] var entries: Array[int] = []
for i in range(en): for i in range(en):
entries.append(_mk("battle", 0)) entries.append(_mk("battle", 0))
var prevs: Array[int] = entries.duplicate() var prevs: Array = _build_trunk(t, 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] = [] var terminals: Array[int] = []
for r in range(e): for r in range(e):
var bdepth := L - r _build_region(r, e, L, prevs, terminals)
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 收束为主链段末节点一处 ⇒ 该节点即归属大岔口(各区域互不连达);
# e==1 时 prevs 为入口数组,直接接单区域头(无大岔口——L1 出口恒 1)。 # e==1 时 prevs 为入口数组,直接接单区域头(无大岔口——L1 出口恒 1)。
@@ -281,41 +321,61 @@ func _build(rg: RandomNumberGenerator) -> Dictionary:
_assign_types(rg, e) _assign_types(rg, e)
var nodes: Array[Dictionary] = [] var nodes: Array[Dictionary] = []
for n in _pool: nodes.assign(_pool)
nodes.append(n)
return {"nodes": nodes, "entries": entries, "terminals": terminals} return {"nodes": nodes, "entries": entries, "terminals": terminals}
## 主链段(深度 1..t 逐层单链,主链段零 RNG):返回链末节点集(e>1 时收束为一处即归属大岔口)
func _build_trunk(t: int, prevs: Array) -> Array:
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]
return prevs
## 归属区域段(每出口一条支路:Boss 终点+店+签名填充链;区域段零 RNG):
## Boss 深度 = L−r、店在其前一位、签名链自店向主链段方向回铺,支路头接入 prevs(归属大岔口)
func _build_region(r: int, e: int, L: int, prevs: Array, terminals: Array) -> void:
var t := 0 if e == 1 else _trunk_len(L, 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)
## 岔口布置:主链段(深度 ≤ t-2)或单出口图链上(深度 ≤ L-3)的节点扩为 2–3 去路岔口; ## 岔口布置:主链段(深度 ≤ t-2)或单出口图链上(深度 ≤ L-3)的节点扩为 2–3 去路岔口;
## 侧支节点与主去路同深(保持「深度=自入口最长路径步数」不变量),两支各自汇入主去路的后继, ## 侧支节点与主去路同深(保持「深度=自入口最长路径步数」不变量),两支各自汇入主去路的后继,
## 于是支路互斥段={主去路节点} 与 {侧支节点}——两支类型互异即同时满足「下一跳互异」与「各有独有类型」。 ## 于是支路互斥段={主去路节点} 与 {侧支节点}——两支类型互异即同时满足「下一跳互异」与「各有独有类型」。
func _add_forks(rg: RandomNumberGenerator, L: int, t: int, e: int) -> void: 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 wanted := clampi(int(round(float(_params["branch_density"]) * float(L))), 1, 24)
var anchors := [] var anchors := _collect_fork_anchors(L, t, e)
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) anchors = _shuffle(rg, anchors)
var selected := {} # 已扩岔节点 id var selected := {} # 已扩岔节点 id
var succ_selected := {} # 已扩岔节点的主去路 id(禁再作锚点) var succ_selected := {} # 已扩岔节点的主去路 id(禁再作锚点)
for a in anchors: for a in anchors:
if wanted <= 0: if wanted <= 0:
break break
var node = a var nid := int(a["id"])
var nid := int(node["id"]) if (a["next"] as Array).size() != 1:
if (node["next"] as Array).size() != 1:
continue # 同一节点不重复扩岔(去路数上限 3) continue # 同一节点不重复扩岔(去路数上限 3)
var v = _by_id[int((node["next"] as Array)[0])] var v = _by_id[int((a["next"] as Array)[0])]
var vid := int(v["id"]) var vid := int(v["id"])
# 相邻两层不得同时扩岔:若主去路 v 也是岔口,v 的侧支会落进本岔口主支路的互斥段, # 相邻两层不得同时扩岔:若主去路 v 也是岔口,v 的侧支会落进本岔口主支路的互斥段,
# 使侧支自己的类型不再独有(契约 5③)——两个方向都要拦。 # 使侧支自己的类型不再独有(契约 5③)——两个方向都要拦。
if selected.has(vid) or succ_selected.has(nid): if selected.has(vid) or succ_selected.has(nid):
continue continue
var v_outs = v["next"] if (v["next"] as Array).is_empty():
if v_outs.is_empty():
continue # 主去路无后继(无可汇入点)——该锚点弃用 continue # 主去路无后继(无可汇入点)——该锚点弃用
var extra := 2 if (int(_params["fork_ways_max"]) >= 3 and rg.randf() < 0.35) else 1 var extra := 2 if (int(_params["fork_ways_max"]) >= 3 and rg.randf() < 0.35) else 1
var made := 0 var made := 0
@@ -323,8 +383,8 @@ func _add_forks(rg: RandomNumberGenerator, L: int, t: int, e: int) -> void:
if _depth_count(int(v["depth"])) >= MAX_WIDTH: if _depth_count(int(v["depth"])) >= MAX_WIDTH:
break break
var wid := _mk("battle", int(v["depth"])) var wid := _mk("battle", int(v["depth"]))
(node["next"] as Array).append(wid) (a["next"] as Array).append(wid)
(_by_id[wid]["next"] as Array).append(int(v_outs[0])) (_by_id[wid]["next"] as Array).append(int((v["next"] as Array)[0]))
_detour_heads.append(_by_id[wid]) _detour_heads.append(_by_id[wid])
_detour_ids[wid] = true _detour_ids[wid] = true
made += 1 made += 1
@@ -334,6 +394,18 @@ func _add_forks(rg: RandomNumberGenerator, L: int, t: int, e: int) -> void:
wanted -= 1 wanted -= 1
## 锚收集段(零 RNG):主链段(深度 ≤ t-2)或单出口图链上(深度 ≤ L-3)的去路恰一节点
func _collect_fork_anchors(L: int, t: int, e: int) -> Array:
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)
return anchors
# 确定性洗牌(Fisher-Yates,取数只经注入的纯流 RNG——Array.shuffle() 走引擎全局随机、 # 确定性洗牌(Fisher-Yates,取数只经注入的纯流 RNG——Array.shuffle() 走引擎全局随机、
# 破「同(种子,参数档)⇒ 同图」的纯函数复现前提,禁用于生成器) # 破「同(种子,参数档)⇒ 同图」的纯函数复现前提,禁用于生成器)
func _shuffle(rg: RandomNumberGenerator, arr: Array) -> Array: func _shuffle(rg: RandomNumberGenerator, arr: Array) -> Array:
@@ -358,31 +430,42 @@ func _depth_count(d: int) -> int:
## 单出口图的配比填余(摆放优先、配比填余、战斗保底兜底——§8.1 次序),再统一修岔口下一跳互异。 ## 单出口图的配比填余(摆放优先、配比填余、战斗保底兜底——§8.1 次序),再统一修岔口下一跳互异。
func _assign_types(rg: RandomNumberGenerator, e: int) -> void: func _assign_types(rg: RandomNumberGenerator, e: int) -> void:
if e == 1: if e == 1:
var free: Array[Dictionary] = [] _fill_ratios(rg)
var chain_battles := 0 _fix_detour_heads()
for n in _pool:
if String(n["type"]) == "shop" or String(n["type"]) == "boss":
continue ## 配比填余段(单出口图;shuffle+randf 取号序整体搬迁——战斗保底优先、侧链不占填余面):
if _detour_ids.has(int(n["id"])): ## 摆放优先、配比填余、战斗保底兜底——§8.1 次序
continue # 侧链节点类型由岔口互异修型决定,不占配比填余面 func _fill_ratios(rg: RandomNumberGenerator) -> void:
if int(n["depth"]) == 1: var free: Array[Dictionary] = []
continue # 区域首节点留战斗——配比填余不破坏岔口下一跳互异的基准形 var chain_battles := 0
if String(n["type"]) == "battle": for n in _pool:
chain_battles += 1 if String(n["type"]) == "shop" or String(n["type"]) == "boss":
free.append(n) continue
free = _shuffle(rg, free) if _detour_ids.has(int(n["id"])):
var want := int(round(float(_pool.size()) * (float(_params["reward_ratio"]) + float(_params["rest_ratio"])))) continue # 侧链节点类型由岔口互异修型决定,不占配比填余面
var floor_n := int(_params["battle_floor"]) if int(n["depth"]) == 1:
var rest_first := rg.randf() < 0.5 continue # 区域首节点留战斗——配比填余不破坏岔口下一跳互异的基准形
for i in range(want): if String(n["type"]) == "battle":
if i >= free.size(): chain_battles += 1
break free.append(n)
# 战斗保底优先:链上战斗数降到保底下限即停(侧链修型可能再让出战斗位,故保底只看链上) free = _shuffle(rg, free)
if chain_battles - 1 < floor_n: var want := int(round(float(_pool.size()) * (float(_params["reward_ratio"]) + float(_params["rest_ratio"]))))
break var floor_n := int(_params["battle_floor"])
var node = free[i] var rest_first := rg.randf() < 0.5
node["type"] = "rest" if ((i % 2 == 0) == rest_first) else "reward" for i in range(want):
chain_battles -= 1 if i >= free.size():
break
# 战斗保底优先:链上战斗数降到保底下限即停(侧链修型可能再让出战斗位,故保底只看链上)
if chain_battles - 1 < floor_n:
break
var node = free[i]
node["type"] = "rest" if ((i % 2 == 0) == rest_first) else "reward"
chain_battles -= 1
## 岔口修型段(零 RNG):侧链首节点类型取岔口各去路下一节点类型的补集首件——下一跳互异承面
func _fix_detour_heads() -> void:
for head in _detour_heads: for head in _detour_heads:
var h = head var h = head
var fork = null var fork = null
@@ -416,12 +499,15 @@ static func _trunk_len(L: int, e: int) -> int:
# ---- 自证辅助 ---- # ---- 自证辅助 ----
func _longest_paths(nodes: Array, by_id: Dictionary, entries: Array, indeg: Dictionary) -> Dictionary: func _longest_paths(nodes: Array, by_id: Dictionary, entries: Array) -> Dictionary:
var out := {} var out := {}
var din := {} var din := {}
var q := [] var q := []
for id in indeg: for n in nodes:
din[int(id)] = int(indeg[int(id)]) din[int(n["id"])] = 0
for n in nodes:
for oid in (n["next"] as Array):
din[int(oid)] = int(din[int(oid)]) + 1
for eid in entries: for eid in entries:
out[int(eid)] = 0 out[int(eid)] = 0
q.append(int(eid)) q.append(int(eid))
@@ -476,44 +562,53 @@ func _check_fork_rules(nodes: Array, by_id: Dictionary, errs: Array) -> void:
if head_types.has(ty): if head_types.has(ty):
errs.append({"category": "structural", "message": "岔口 %d 各去路下一节点类型重复: %s" % [n["id"], ty]}) errs.append({"category": "structural", "message": "岔口 %d 各去路下一节点类型重复: %s" % [n["id"], ty]})
head_types[ty] = true head_types[ty] = true
var reach_sets := [] var comps := _fork_signatures(outs, by_id)
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 i in range(comps.size()):
for j in range(i + 1, comps.size()): for j in range(i + 1, comps.size()):
if comps[i] == comps[j]: if comps[i] == comps[j]:
errs.append({"category": "structural", "message": "岔口 %d 支路 %d/%d 类型构成相同" % [n["id"], i, j]}) errs.append({"category": "structural", "message": "岔口 %d 支路 %d/%d 类型构成相同" % [n["id"], i, j]})
for i in range(comps.size()): for i in range(comps.size()):
var uniq := false if not _fork_has_unique(comps, i):
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]}) errs.append({"category": "structural", "message": "岔口 %d 支路 %d 无独有类型" % [n["id"], i]})
## 支路签名构造:各去路的独占后代类型集(他人不可达的后代类型,排序数组——支路互斥段签名)
func _fork_signatures(outs: Array, by_id: Dictionary) -> Array:
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)
return comps
## 独有类型判定:支路 i 的签名中存在他支路不含的类型
func _fork_has_unique(comps: Array, i: int) -> bool:
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:
return true
return false
func _descendants(start: int, by_id: Dictionary) -> Dictionary: func _descendants(start: int, by_id: Dictionary) -> Dictionary:
var out := {start: true} var out := {start: true}
var q := [start] var q := [start]