mtgodot-poc/project/testing/playable_flow.gd

class_name PlayableFlow
extends RefCounted

## NET-01 / CBT-01 联网闭环状态机（纯逻辑，可由假客户端驱动）。
##
## 只经生产入口操作：AppFlow 自动登录/选角、PlayerController.walk_to、
## NetPlay.select_target / set_attack_key / pick_ground_item、Quickbar.assign/activate。
## 成功证据只来自 PlayableProbe 的 source=server 观察或 M2Client 状态查询；
## 本地位置、施法开始、特效节点只作补充，不单独构成服务端结果。
##
## 所有时间来自注入的 clock_us，harness 可以不等待真实时间验证超时分支。
##
## STB-01 soak：同一套 MOVE…PICKUP 按轮重复，按墙钟期限（clock_us 经过的时长）结束，
## 不按轮数或帧数。每轮后依次：真实传输故障（需确认的本地代理）、合法传送门往返
## （需确认的路线）、三种逻辑分辨率切换（需窗口适配器）、≥30s 休息（内存低水位窗口）。
## 主动 reconnect() 与真实故障分开计数；未确认的条件保持 BLOCKED，不互相替代。

signal finished(status: String)

const Config = preload("res://testing/playable_config.gd")
const Metrics = preload("res://testing/playable_metrics.gd")

const LOGIN_TIMEOUT_US := 30_000_000
const SELECT_TIMEOUT_US := 45_000_000
const WORLD_TIMEOUT_US := 60_000_000
const MOVE_TIMEOUT_US := 20_000_000
const CONFIRM_TIMEOUT_US := 60_000_000
const TARGET_TIMEOUT_US := 30_000_000
const ATTACK_TIMEOUT_US := 90_000_000
const DROP_TIMEOUT_US := 30_000_000
const PICKUP_TIMEOUT_US := 15_000_000
const PICKUP_SETTLE_US := 2_000_000
const SKILL_CAST_TIMEOUT_US := 10_000_000
const SKILL_RETRY_US := 500_000
## Soak: the scene must stay ready this long after a re-login before rebind checks run,
## so a node still fading out from the old connection is not mistaken for a live duplicate.
const REBIND_SETTLE_US := 1_000_000
const FAULT_DISCONNECT_TIMEOUT_US := 30_000_000
## Recovery budget after the transport fault ends: ReconnectUI retry delay + login + map load.
const FAULT_RECOVERY_GRACE_US := 120_000_000
const WARP_TIMEOUT_US := 120_000_000
const RESIZE_SETTLE_US := 500_000
const RESIZE_TIMEOUT_US := 5_000_000
const RESIZE_TOLERANCE_PX := 2
const FRAME_HISTORY_CAP := 64
const RECONNECT_STAGES := ["MOVE_CONFIRM", "WARP", "FAULT"]
const SOAK_PLAY_STAGES := ["MOVE", "TARGET", "ATTACK", "DROP", "PICKUP", "RESIZE", "REST"]
const SOAK_STEPS := ["FAULT", "WARP", "RESIZE", "REST"]

const ARRIVE_M := 1.5
const CORRECTION_CM := 500.0
const CONFIRM_CM := 300.0
const DROP_RADIUS_CM := 600.0
const MAX_KILLS := 3
const MONSTER_CH_TYPE := 2
const SKILL_SLOT := 7
const DEFAULT_SKILL_REPEATS := 10
## One-shot effects must leave the tree within their MSE-defined lifetime + this grace.
const FX_GRACE_MS := 1000
## PlayerSkill codes that mean the fixture (not the client) is missing something.
const FIXTURE_SKILL_CODES := ["NOT_YET_LEARN", "NO_SKILL_DATA", "PASSIVE", "NOT_MATCHABLE_WEAPON", "EMPTY_ARROW", "HAVE_TO_RIDE", "NO_RIDING_SKILL"]

var report: RefCounted
var probe: Object
var flow: Object
var client: Object
var config: Dictionary = {}
var suite := "playable"
var allow_gameplay := false
var clock_us: Callable = func() -> int: return Time.get_ticks_usec()
## func(name: String) -> String; returns an evidence path relative to the report, or "".
var capture: Callable
## Soak only. PlayableMetrics fed once per frame by the live runner (null offline -> STB-FRAMES-01 BLOCKED).
var metrics: RefCounted
## Soak only. Window adapter (testing/playable_window.gd): availability / request_logical_size /
## logical_size / physical_size / viewport_size / screen_scale. null -> STB-RESIZE-01 BLOCKED.
var window: Object
## Unix wall-clock milliseconds for rest windows matched against the external RSS sampler.
var wall_ms: Callable = func() -> int: return int(Time.get_unix_time_from_system() * 1000.0)

var stage := "BOOT"
var done := false
var status := ""

var _stage_since := 0
var _main_vid := 0
var _bound_epoch := 0
var _entered_seen := false
var _waypoint := 0
var _expected_world := Vector3.ZERO
var _reconnect_epoch := 0
var _reconnect_entered := 0
var _rebind_since := -1
var _confirm_target_cm := Vector2.ZERO
var _confirm_blocked := false
var _kills := 0
var _abandoned := {}
var _target := {}
var _attack_down := false
var _damaged := false
var _dead := false
var _kill_pos_cm := Vector2.ZERO
var _pre_attack_iids := {}
var _drop := {}
var _pickup_before := 0
var _pickup_removed_at := -1
var _skill_index := 0
var _skill_successes := 0
var _skill_cast := {}
var _skill_next_try := 0
var _skill_draining := false
var _open_fx := {}   # fx_id -> {effect, lifetime_ms, spawned_us, case_id}
## §7.1 item 6: the first cast of each skill case is captured at its own event
## times (before / trigger / peak / end), not on a fixed one-second delay.
var _shots: Array = []        # evidence files attached to the skill case result
var _shot_phases := {}        # phase -> true once attempted for this case
var _shot_fx := {}            # fx_id -> peak due (us) for first-cast timed effects
var _shot_end_pending := false
var _start_us := 0
# --- soak ---
var _round := 0
var _rounds_done := 0
var _soak_start_us := -1
var _soak_wall_start_ms := 0
var _soak_blocked := ""
var _active_reconnects := 0
var _faults := {"close": 0, "unreachable": 0}
var _fault := {}
var _warps := 0
var _warp_leg := 0
var _warp_expected := ""
var _resize_index := -1
var _resize_since := 0
var _resize_before := Vector2i.ZERO
var _resize_counts := {}
var _resize_blocked := ""
var _severe_seen := {}
var _frame_history := {}      # interaction (stage) -> last FRAME_HISTORY_CAP bools
var _repeated_severe := {}    # interaction -> true once 2 of 3 consecutive repetitions had a >100ms frame

func start() -> void:
	_start_us = _now()
	report.set_required_cases(Config.required_cases(config, suite))
	var server: Dictionary = config.get("server", {})
	var address: Dictionary = {}
	if flow != null and flow.has_method("select_test_server"):
		address = flow.select_test_server(int(server.get("server_index", -1)), int(server.get("channel", 0)))
	var match_ok := not address.is_empty()
	for key in ["auth_host", "game_host"]:
		match_ok = match_ok and String(address.get(key, "")) == String(server.get(key, ""))
	for key in ["auth_port", "game_port"]:
		match_ok = match_ok and int(address.get(key, -1)) == int(server.get(key, -2))
	_event("address_selected", {"server_index": int(server.get("server_index", -1)),
		"channel": int(server.get("channel", 0)), "address_match": match_ok})
	if not match_ok:
		_finish_case("CONFIG-01", "FAIL", "client serverinfo address differs from the prechecked config address")
		_finish()
		return
	report.add_case("CONFIG-01", "PASS", "config validated and client address matches precheck")
	_enter("LOGIN")

## Probe observation entry. Every observation is persisted as a contract event
## before it is interpreted.
func on_observed(kind: String, data: Dictionary) -> void:
	var epoch := int(data.get("connection_epoch", 0))
	var stale := _bound_epoch > 0 and epoch != _bound_epoch and kind != "entered_game" and kind != "disconnected"
	var payload: Dictionary = data.duplicate()
	payload.erase("connection_epoch")
	payload.erase("actor_vid")
	payload.erase("target_vid")
	payload.erase("received_us")
	if stale:
		payload["stale_epoch"] = true
	report.record(kind, stage, _case_for_stage(), epoch, int(data.get("actor_vid", 0)),
		int(data.get("target_vid", 0)), payload)
	if done:
		return
	match kind:
		"login_failed":
			if stage == "LOGIN" or stage == "SELECT":
				_finish_case("NET-LOGIN-01", "FAIL", "login rejected by server")
				_finish()
		"char_list":
			if stage == "LOGIN":
				var slot := int(config.get("character_slot", -1))
				if slot in data.get("slots", []):
					report.add_case("NET-LOGIN-01", "PASS", "server char_list contains the configured slot")
					_enter("SELECT")
				else:
					_finish_case("NET-LOGIN-01", "FAIL", "configured character slot is absent")
					_finish()
		"entered_game":
			_entered_seen = true
			_on_entered_game(epoch)
		"disconnected":
			_on_disconnected()
		"entity_moved":
			if stage == "MOVE" and not stale and int(data.get("actor_vid", 0)) == _main_vid:
				_check_server_correction(data.get("pos_cm", null))
		"damage":
			if stage == "ATTACK" and not stale and _is_target(int(data.get("target_vid", 0))) and int(data.get("amount", 0)) > 0:
				_damaged = true
			if stage == "SKILL" and not stale:
				_skill_evidence("damage", int(data.get("target_vid", 0)))
		"vitals_changed":
			if stage == "ATTACK" and not stale and _is_target(int(data.get("actor_vid", 0))):
				var hp := int(data.get("hp", -1))
				if hp >= 0 and int(_target.get("hp", -1)) > 0 and hp < int(_target.hp):
					_damaged = true
				_target["hp"] = hp
		"target_info":
			if stage == "ATTACK" and not stale and _is_target(int(data.get("target_vid", 0))) and int(data.get("hp_percent", 100)) < 100:
				_damaged = true
		"entity_dead":
			var dead_vid := int(data.get("target_vid", 0))
			if stage == "ATTACK" and not stale and _is_target(dead_vid):
				_dead = true
			if stage == "SKILL" and not stale:
				_skill_evidence("target_dead", dead_vid)
		"entity_despawned":
			if stage == "ATTACK" and not stale and _is_target(int(data.get("actor_vid", 0))) and not _dead:
				_target_lost("target despawned before a server death result")
		"server_motion":
			if stage == "SKILL" and not stale and int(data.get("actor_vid", 0)) == _main_vid:
				_skill_evidence("server_motion", int(data.get("target_vid", 0)))
		"effect_cue", "fly_cue", "affect_added", "sp_changed":
			if stage == "SKILL" and not stale:
				var actor := int(data.get("actor_vid", 0))
				if kind in ["affect_added", "sp_changed"] or actor == _main_vid or actor == int(_skill_cast.get("target_vid", -1)):
					_skill_evidence(kind, int(data.get("target_vid", 0)))
		"cast_started":
			if stage == "SKILL" and String(data.get("source", "")) == "local":
				_on_cast_started(int(data.get("skill_id", 0)), int(data.get("target_vid", 0)))
		"skill_rejected":
			if stage == "SKILL":
				_on_skill_rejected(int(data.get("skill_id", 0)), String(data.get("code", "")))
		"fx_spawned":
			_on_fx_spawned(data)
			if stage == "SKILL":
				_skill_evidence(kind, 0)
				_sample_fx_spawned(data)
		"fx_finished":
			if not _on_fx_finished(data):
				return
			if stage == "SKILL":
				_skill_evidence(kind, 0)
				_sample_fx_finished(data)

func tick() -> void:
	if done:
		return
	var now := _now()
	var timeout_us := int(config.get("timeout_seconds", 120)) * 1_000_000
	if now - _start_us > timeout_us:
		_fail_stage("overall scenario timeout")
		return
	if not _check_open_fx(now):
		return
	if stage == "SKILL":
		_sample_due(now)
	match stage:
		"LOGIN":
			if now - _stage_since > LOGIN_TIMEOUT_US:
				_finish_case("NET-LOGIN-01", "FAIL", "no server char_list within 30s")
				_finish()
		"SELECT":
			if _entered_seen:
				var vid := int(client.get_main_vid()) if client.has_method("get_main_vid") else 0
				var main: Dictionary = client.get_entity(vid) if vid > 0 and client.has_method("get_entity") else {}
				if vid > 0 and not main.is_empty():
					_main_vid = vid
					_bound_epoch = _probe_epoch()
					report.add_case("NET-SELECT-01", "PASS", "entered_game received and main entity is present")
					_enter("WORLD")
					return
			if now - _stage_since > SELECT_TIMEOUT_US:
				_finish_case("NET-SELECT-01", "FAIL", "entered_game/main entity not observed within 45s")
				_finish()
		"WORLD":
			var ready := _world_ready()
			if ready.ready:
				if String(ready.map_key) != String(config.get("map_key", "")):
					_event("map_mismatch", {"map_key": ready.map_key, "expected_map_key": config.map_key})
					_finish_case("NET-WORLD-01", "FAIL", "loaded map_key differs from the configured map")
					_finish()
					return
				_pass("NET-WORLD-01", "scene, map, player and HUD ready")
				if not allow_gameplay:
					_block_remaining("gameplay_not_allowed")
					_finish()
					return
				if suite == "soak":
					_begin_soak(now)
					return
				_begin_move(0)
			elif now - _stage_since > WORLD_TIMEOUT_US:
				_event("world_not_ready", ready)
				_finish_case("NET-WORLD-01", "FAIL", "scene/map/player/HUD not ready within 60s")
				_finish()
		"MOVE":
			var player: Node3D = _context().get("player", null)
			if player != null and is_instance_valid(player) \
					and Vector2(player.position.x, player.position.z).distance_to(Vector2(_expected_world.x, _expected_world.z)) <= ARRIVE_M:
				_pass(_move_case(), "local arrival without server correction")
				if _waypoint + 1 < config.waypoints_cm.size():
					_begin_move(_waypoint + 1)
				else:
					_begin_confirm()
			elif now - _stage_since > MOVE_TIMEOUT_US:
				_fail_stage("waypoint not reached within 20s")
		"MOVE_CONFIRM":
			_tick_confirm(now)
		"TARGET":
			_tick_target(now)
		"ATTACK":
			_tick_attack(now)
		"DROP":
			_tick_drop(now)
		"PICKUP":
			_tick_pickup(now)
		"SKILL":
			_tick_skill(now)
		"FAULT":
			_tick_fault(now)
		"WARP":
			_tick_warp(now)
		"RESIZE":
			_tick_resize(now)
		"REST":
			_tick_rest(now)

## Cancel all pending input immediately. Safe to call repeatedly.
func cancel_inputs() -> void:
	if _attack_down:
		var net_play: Object = _context().get("net_play", null)
		if net_play != null and is_instance_valid(net_play) and net_play.has_method("set_attack_key"):
			net_play.set_attack_key(false)
		_attack_down = false
	_target = {}
	_skill_cast = {}

# --- stages -----------------------------------------------------------------

func _enter(next: String) -> void:
	_track_frames(stage)
	stage = next
	_stage_since = _now()
	var snap: Dictionary = probe.snapshot() if probe != null and probe.has_method("snapshot") else {}
	var flow_snap := _flow_snapshot()
	snap["state"] = next
	snap["scene_ready"] = bool(flow_snap.get("scene_ready", false))
	snap["reconnecting"] = bool(flow_snap.get("reconnecting", false))
	snap["character_count"] = int(flow_snap.get("character_count", 0))
	_event("state_enter", snap)

func _begin_move(index: int) -> void:
	_waypoint = index
	var point: Array = config.waypoints_cm[index]
	_expected_world = MapCoord.to_world(Vector3(float(point[0]) * 0.01, 0.0, -float(point[1]) * 0.01))
	_enter("MOVE")
	var controller: Object = _context().get("pc", null)
	if controller == null or not is_instance_valid(controller) or not controller.has_method("walk_to"):
		_fail_stage("PlayerController.walk_to is unavailable")
		return
	controller.walk_to(_expected_world)
	_event("move_requested", {"waypoint_index": index, "pos_cm": Vector2(float(point[0]), float(point[1]))})

func _check_server_correction(pos: Variant) -> void:
	var player: Node3D = _context().get("player", null)
	if not (pos is Vector3) or player == null or not is_instance_valid(player):
		return
	var local_cm := MapCoord.to_server_cm(player.position)
	var distance := local_cm.distance_to(Vector2(pos.x, pos.y))
	if distance > CORRECTION_CM:
		_event("server_correction", {"distance_cm": distance, "tolerance_cm": CORRECTION_CM, "waypoint_index": _waypoint})
		_fail_stage("server corrected the main character position by %.0fcm" % distance)

func _begin_confirm() -> void:
	var last: Array = config.waypoints_cm[config.waypoints_cm.size() - 1]
	_confirm_target_cm = Vector2(float(last[0]), float(last[1]))
	_arm_reconnect()
	_confirm_blocked = false
	_enter("MOVE_CONFIRM")
	cancel_inputs()
	if not client.has_method("reconnect") or not bool(client.reconnect()):
		_confirm_blocked = true
		_finish_case("NET-MOVE-CONFIRM-01", "BLOCKED", "reconnect is unavailable; no server position source")
		if suite == "soak":
			_finish_case("STB-RECONNECT-01", "BLOCKED", "reconnect is unavailable in round %d" % _round)
			_finish()
			return
		_enter_target()

func _tick_confirm(now: int) -> void:
	var bound := _rebound(now)
	if bound < 0:
		return
	if bound > 0:
		var pos: Variant = client.get_entity(_main_vid).get("pos_cm", null)
		var distance := Vector2(pos.x, pos.y).distance_to(_confirm_target_cm)
		_event("server_position", {"main_vid": _main_vid, "pos_cm": Vector2(pos.x, pos.y), "distance_cm": distance, "tolerance_cm": CONFIRM_CM})
		if distance <= CONFIRM_CM:
			_pass("NET-MOVE-CONFIRM-01", "server spawn position after re-login matches the last waypoint")
		else:
			_finish_case("NET-MOVE-CONFIRM-01", "FAIL", "server position after re-login is %.0fcm from the last waypoint" % distance)
			_finish()
			return
		if suite == "soak":
			_active_reconnects += 1
			_event("reconnect_verified", {"round": _round, "count": _active_reconnects})
		_enter_target()
		return
	if now - _stage_since > CONFIRM_TIMEOUT_US:
		_finish_case("NET-MOVE-CONFIRM-01", "FAIL", "re-login did not return a main entity/scene within 60s")
		_finish()

func _enter_target() -> void:
	_target = {}
	_damaged = false
	_dead = false
	_enter("TARGET")

func _tick_target(now: int) -> void:
	var candidate := _pick_monster()
	if not candidate.is_empty():
		var net_play: Object = _context().get("net_play", null)
		if net_play == null or not is_instance_valid(net_play) or not net_play.has_method("select_target"):
			_fail_stage("NetPlay.select_target is unavailable")
			return
		if bool(net_play.select_target(int(candidate.vid))):
			_target = candidate
			_target["epoch"] = _bound_epoch
			_event("target_selected", {"race": candidate.race, "pos_cm": candidate.pos_cm, "attempt": _kills + 1})
			if not report.has_case("NET-TARGET-01"):
				report.add_case("NET-TARGET-01", "PASS", "allowed live monster selected in the current connection epoch")
			if stage == "TARGET":
				_begin_attack()
			return
		_abandoned[int(candidate.vid)] = true
	if now - _stage_since > TARGET_TIMEOUT_US:
		if not report.has_case("NET-TARGET-01") or (suite == "soak" and _kills == 0):
			_finish_case("NET-TARGET-01", "BLOCKED", "no allowed live monster in range (fixture)")
		else:
			_finish_case("NET-DROP-01", "BLOCKED", "no further allowed monster after %d kills" % _kills)
		_finish()

func _pick_monster() -> Dictionary:
	if not client.has_method("get_entities"):
		return {}
	var allowed: Array = config.get("allowed_mob_vnums", []).map(func(v): return int(v))
	var main: Dictionary = client.get_entity(_main_vid) if client.has_method("get_entity") else {}
	var origin: Variant = main.get("pos_cm", Vector3.ZERO)
	var best := {}
	var best_distance := INF
	for entity in client.get_entities():
		var vid := int(entity.get("vid", 0))
		if vid <= 0 or _abandoned.has(vid) or bool(entity.get("dead", false)):
			continue
		if int(entity.get("ch_type", -1)) != MONSTER_CH_TYPE or not (int(entity.get("race", 0)) in allowed):
			continue
		var pos: Variant = entity.get("pos_cm", Vector3.ZERO)
		var distance := Vector2(pos.x, pos.y).distance_to(Vector2(origin.x, origin.y)) if pos is Vector3 and origin is Vector3 else 0.0
		if distance < best_distance:
			best_distance = distance
			best = {"vid": vid, "race": int(entity.get("race", 0)), "hp": int(entity.get("hp", -1)),
				"pos_cm": Vector2(pos.x, pos.y) if pos is Vector3 else Vector2.ZERO}
	return best

func _begin_attack() -> void:
	_damaged = false
	_dead = false
	_pre_attack_iids = _ground_iids()
	_enter("ATTACK")
	var net_play: Object = _context().get("net_play", null)
	if net_play == null or not net_play.has_method("set_attack_key"):
		_fail_stage("NetPlay.set_attack_key is unavailable")
		return
	net_play.set_attack_key(true)
	_attack_down = true

func _tick_attack(now: int) -> void:
	if _target.is_empty():
		return
	var entity: Dictionary = client.get_entity(int(_target.vid)) if client.has_method("get_entity") else {}
	var pos: Variant = entity.get("pos_cm", null)
	if pos is Vector3:
		_target["pos_cm"] = Vector2(pos.x, pos.y)
	if _dead:
		if not _damaged:
			# A death without any HP/damage evidence is not a verified combat result.
			cancel_inputs()
			_fail_stage("target died without a received damage/HP change")
			return
		_kills += 1
		_kill_pos_cm = _target.get("pos_cm", Vector2.ZERO)
		_abandoned[int(_target.vid)] = true
		cancel_inputs()
		_event("target_killed", {"attempt": _kills, "pos_cm": _kill_pos_cm})
		if not report.has_case("NET-ATTACK-01"):
			report.add_case("NET-ATTACK-01", "PASS", "received damage/HP change and death for the selected target")
		_drop = {}
		_enter("DROP")
		return
	if now - _stage_since > ATTACK_TIMEOUT_US:
		cancel_inputs()
		_fail_stage("no server damage and death result within 90s")

## NET-04: target vanished before death -> release input, never chase the old VID.
func _target_lost(reason: String) -> void:
	_event("target_lost", {"attempt": _kills + 1, "reason": reason})
	_abandoned[int(_target.get("vid", 0))] = true
	cancel_inputs()
	_kills += 1
	if _kills >= MAX_KILLS:
		_finish_case("NET-ATTACK-01", "BLOCKED", "target lost before death on %d attempts" % _kills)
		_finish()
		return
	_enter_target()

func _tick_drop(now: int) -> void:
	var allowed: Array = config.get("allowed_drop_vnums", []).map(func(v): return int(v))
	for item in _ground_items():
		var iid := int(item.get("vid", 0))
		if _pre_attack_iids.has(iid):
			continue
		var vnum := int(item.get("vnum", 0))
		var pos_cm := _ground_pos_cm(item)
		var distance := pos_cm.distance_to(_kill_pos_cm)
		if vnum == Config.GOLD_VNUM:
			_pre_attack_iids[iid] = true
			_event("gold_drop_seen", {"iid": iid, "distance_cm": distance})
			continue
		if not (vnum in allowed):
			continue
		var owner_ok := _owner_matches(String(item.get("owner", "")))
		if not owner_ok or distance > DROP_RADIUS_CM:
			_pre_attack_iids[iid] = true
			_event("drop_rejected", {"iid": iid, "vnum": vnum, "owner_match": owner_ok, "distance_cm": distance, "tolerance_cm": DROP_RADIUS_CM})
			continue
		_drop = {"iid": iid, "vnum": vnum}
		_event("drop_accepted", {"iid": iid, "vnum": vnum, "owner_match": true, "distance_cm": distance})
		_pass("NET-DROP-01", "new allowed ground item, pickable owner, near the killed target")
		_begin_pickup()
		return
	if now - _stage_since > DROP_TIMEOUT_US:
		if _kills < MAX_KILLS:
			_event("no_drop", {"attempt": _kills})
			_enter_target()
		else:
			_finish_case("NET-DROP-01", "BLOCKED", "no allowed pickable drop after %d kills (fixture)" % _kills)
			_finish()

func _begin_pickup() -> void:
	_pickup_before = _inventory_total(int(_drop.vnum))
	_pickup_removed_at = -1
	_enter("PICKUP")
	var net_play: Object = _context().get("net_play", null)
	if net_play == null or not net_play.has_method("pick_ground_item"):
		_fail_stage("NetPlay.pick_ground_item is unavailable")
		return
	net_play.pick_ground_item(int(_drop.iid))
	_event("pickup_requested", {"iid": _drop.iid, "vnum": _drop.vnum, "before": _pickup_before})

func _tick_pickup(now: int) -> void:
	var present := _ground_iids().has(int(_drop.iid))
	var after := _inventory_total(int(_drop.vnum))
	if not present and _pickup_removed_at < 0:
		_pickup_removed_at = now
	if not present and after > _pickup_before:
		_event("pickup_verified", {"iid": _drop.iid, "vnum": _drop.vnum, "before": _pickup_before, "after": after, "delta": after - _pickup_before})
		_pass("NET-PICKUP-01", "ground item removed and inventory vnum total increased")
		_after_closed_loop()
		return
	if _pickup_removed_at >= 0 and now - _pickup_removed_at > PICKUP_SETTLE_US:
		_event("pickup_unverified", {"iid": _drop.iid, "vnum": _drop.vnum, "before": _pickup_before, "after": after})
		_fail_stage("ground item vanished without an inventory increase (taken by another player or inventory full)")
		return
	if now - _stage_since > PICKUP_TIMEOUT_US:
		_fail_stage("pickup not accepted within 15s (inventory full, ownership or range)")

func _after_closed_loop() -> void:
	if suite == "soak":
		_soak_continue("")
		return
	if suite == "full" and not config.get("skill_cases", []).is_empty():
		_skill_index = 0
		_begin_skill_case()
		return
	_finish()

# --- CBT-01: skills ------------------------------------------------------------

func _begin_skill_case() -> void:
	_skill_successes = 0
	_skill_cast = {}
	_skill_next_try = 0
	_skill_draining = false
	_shots = []
	_shot_phases = {}
	_shot_fx = {}
	_shot_end_pending = false
	_enter("SKILL")
	var item: Dictionary = config.skill_cases[_skill_index]
	var quickbar: Object = _context().get("quickbar", null)
	if quickbar == null or not is_instance_valid(quickbar) or not quickbar.has_method("assign"):
		_fail_stage("Quickbar is unavailable")
		return
	# Local-only slot: no CG_QUICKSLOT_ADD persistence on the account.
	if not bool(quickbar.assign(SKILL_SLOT, "skill", int(item.skill_id), false)):
		_fail_stage("Quickbar.assign rejected the skill")

func _tick_skill(now: int) -> void:
	var item: Dictionary = config.skill_cases[_skill_index]
	if _skill_draining:
		# The case passes only after its one-shot effects actually left the tree;
		# _check_open_fx fails it once one outlives lifetime + grace.
		if _open_fx_for(_case_for_stage()).is_empty():
			_skill_draining = false
			report.add_case(Config.skill_case_id(item), "PASS", "server evidence observed for every cast; effects cleaned up",
				_shots.duplicate(), float(now - _stage_since) / 1000.0)
			_skill_index += 1
			if _skill_index >= config.skill_cases.size():
				_finish()
			else:
				_begin_skill_case()
		return
	if not _skill_cast.is_empty():
		if _skill_missing().is_empty():
			_skill_successes += 1
			_event("skill_verified", {"skill_id": item.skill_id, "count": _skill_successes})
			_skill_cast = {}
			_skill_next_try = now + SKILL_RETRY_US
			if _skill_successes >= int(item.get("repeats", DEFAULT_SKILL_REPEATS)):
				_skill_draining = true
				_tick_skill(now)
			return
		if now - int(_skill_cast.started_us) > SKILL_CAST_TIMEOUT_US:
			_fail_stage("cast %d missing evidence: %s" % [_skill_successes + 1, ",".join(_skill_missing())])
		return
	if now - _stage_since > SKILL_CAST_TIMEOUT_US * (int(item.get("repeats", DEFAULT_SKILL_REPEATS)) + 2):
		_fail_stage("skill case did not complete in its time budget")
		return
	if now < _skill_next_try:
		return
	if String(item.get("target", "enemy")) == "enemy":
		var target := _pick_monster()
		var net_play: Object = _context().get("net_play", null)
		if target.is_empty() or net_play == null or not bool(net_play.select_target(int(target.vid))):
			_skill_next_try = now + SKILL_RETRY_US
			return
	var quickbar: Object = _context().get("quickbar", null)
	if _skill_successes == 0:
		_shoot("before")
	_skill_cast = {"pending": true, "skill_id": int(item.skill_id), "started_us": now, "evidence": {}, "casts": 0}
	quickbar.activate(SKILL_SLOT)
	if _skill_cast.get("pending", false) and int(_skill_cast.casts) == 0:
		# No cast_started emitted synchronously: cooldown / gate silently held it.
		_skill_cast = {}
		_skill_next_try = now + SKILL_RETRY_US

func _on_cast_started(skill_id: int, target_vid: int) -> void:
	if _skill_cast.is_empty() and stage == "SKILL" and not _skill_draining \
			and skill_id == int(config.skill_cases[_skill_index].skill_id):
		# An earlier activation was out of range and reserved (__ReserveUseSkill);
		# NetPlay walked in and the reserved slot fired now. Track it as this case's cast.
		_skill_cast = {"pending": true, "skill_id": skill_id, "started_us": _now(), "evidence": {}, "casts": 0, "reserved": true}
	if _skill_cast.is_empty() or skill_id != int(_skill_cast.skill_id):
		return
	_skill_cast.casts = int(_skill_cast.casts) + 1
	if int(_skill_cast.casts) > 1:
		_fail_stage("duplicate cast request for one activation")
		return
	_skill_cast.pending = false
	if _skill_successes == 0:
		_shoot("trigger")
	# Bind the case to the target at cast time; later selection changes do not re-route evidence.
	_skill_cast.target_vid = target_vid
	_skill_cast.evidence["cast_started"] = true

func _on_skill_rejected(skill_id: int, code: String) -> void:
	var item: Dictionary = config.skill_cases[_skill_index]
	if skill_id != int(item.skill_id):
		return
	_skill_cast = {}
	if code in FIXTURE_SKILL_CODES:
		_finish_case(Config.skill_case_id(item), "BLOCKED", "skill fixture not ready: %s" % code)
		_finish()
		return
	_skill_next_try = _now() + SKILL_RETRY_US

func _skill_evidence(kind: String, target_vid: int) -> void:
	if _skill_cast.is_empty() or bool(_skill_cast.get("pending", false)):
		return
	if kind in ["damage", "target_dead"] and target_vid != int(_skill_cast.get("target_vid", 0)):
		return
	# Local kinds are recorded too, but validate() requires at least one server kind per case.
	_skill_cast.evidence[kind] = true

func _skill_missing() -> Array:
	var missing: Array = []
	var item: Dictionary = config.skill_cases[_skill_index]
	for kind in item.required_evidence:
		if not _skill_cast.get("evidence", {}).has(String(kind)):
			missing.append(String(kind))
	return missing

# --- CBT-01: effect lifecycle ----------------------------------------------------
# Evidence comes from EffectRegistry spawn / tree-exit boundaries, never node counts.

func _on_fx_spawned(data: Dictionary) -> void:
	var lifetime_ms := int(data.get("lifetime_ms", -1))
	if lifetime_ms < 0 or String(data.get("source", "")) != "local":
		return # looping effects live as long as their owner
	_open_fx[int(data.get("fx_id", 0))] = {"effect": String(data.get("effect", "")),
		"lifetime_ms": lifetime_ms, "spawned_us": _now(), "case_id": _case_for_stage()}

## Returns false when the finish violated the lifetime contract (the run is over).
func _on_fx_finished(data: Dictionary) -> bool:
	var fx_id := int(data.get("fx_id", 0))
	var info: Dictionary = _open_fx.get(fx_id, {})
	_open_fx.erase(fx_id)
	var lifetime_ms := int(data.get("lifetime_ms", -1))
	var elapsed_ms := int(data.get("elapsed_ms", -1))
	if lifetime_ms >= 0 and elapsed_ms > lifetime_ms + FX_GRACE_MS:
		var case_id := String(info.get("case_id", _case_for_stage()))
		_fx_overdue(case_id, String(data.get("effect", "")), lifetime_ms, elapsed_ms)
		return false
	return true

func _check_open_fx(now: int) -> bool:
	for fx_id in _open_fx.keys():
		var info: Dictionary = _open_fx[fx_id]
		var elapsed_ms := (now - int(info.spawned_us)) / 1000
		if elapsed_ms > int(info.lifetime_ms) + FX_GRACE_MS:
			_open_fx.erase(fx_id)
			_fx_overdue(String(info.case_id), String(info.effect), int(info.lifetime_ms), elapsed_ms)
			return false
	return true

func _shoot(phase: String) -> void:
	if _shot_phases.has(phase) or not capture.is_valid() or _skill_index >= config.skill_cases.size():
		return
	_shot_phases[phase] = true
	var file := String(capture.call("%s-%s" % [Config.skill_case_id(config.skill_cases[_skill_index]), phase]))
	_event("skill_capture", {"phase": phase, "screenshot": file})
	if not file.is_empty():
		_shots.append(file)

## Peak = half of the effect's own defined lifetime, sampled on the tick it falls due.
func _sample_fx_spawned(data: Dictionary) -> void:
	var lifetime_ms := int(data.get("lifetime_ms", -1))
	if lifetime_ms < 0 or _skill_successes > 0 or _skill_cast.is_empty() or _shot_phases.has("end"):
		return
	_shot_fx[int(data.get("fx_id", 0))] = _now() + lifetime_ms * 500

func _sample_fx_finished(data: Dictionary) -> void:
	var fx_id := int(data.get("fx_id", 0))
	if not _shot_fx.has(fx_id):
		return
	if int(_shot_fx[fx_id]) > 0:
		_shoot("peak")   # finished before its half-life tick: take the last live frame
	_shot_fx.erase(fx_id)
	_shot_end_pending = _shot_fx.is_empty()

func _sample_due(now: int) -> void:
	for fx_id in _shot_fx:
		if int(_shot_fx[fx_id]) > 0 and now >= int(_shot_fx[fx_id]):
			_shoot("peak")
			_shot_fx[fx_id] = 0
	if _shot_end_pending:
		# One tick after the last first-cast effect left the tree.
		_shot_end_pending = false
		_shoot("end")

func _open_fx_for(case_id: String) -> Array:
	var out: Array = []
	for fx_id in _open_fx:
		if String(_open_fx[fx_id].case_id) == case_id:
			out.append(fx_id)
	return out

func _fx_overdue(case_id: String, effect: String, lifetime_ms: int, elapsed_ms: int) -> void:
	var message := "effect %s not cleaned within lifetime %dms + %dms (elapsed %dms)" % [effect, lifetime_ms, FX_GRACE_MS, elapsed_ms]
	_event("fx_overdue", {"effect": effect, "lifetime_ms": lifetime_ms, "elapsed_ms": elapsed_ms,
		"open_effects": _open_fx.size()})
	cancel_inputs()
	if report.has_case(case_id):
		report.add_failure("%s: %s" % [case_id, message])
	else:
		_finish_case(case_id, "FAIL", message)
	_finish()

# --- reconnect binding (NET-01 confirm, STB-01 soak) ------------------------------

func _arm_reconnect() -> void:
	_reconnect_epoch = _probe_epoch()
	_reconnect_entered = 0
	_rebind_since = -1

func _on_entered_game(epoch: int) -> void:
	if stage in RECONNECT_STAGES:
		if suite == "soak" and stage == "FAULT" and int(_fault.get("disconnected_us", -1)) < 0:
			_soak_fail("STB-DISCONNECT-01", "FAIL", "entered_game before the requested transport fault disconnected")
			return
		_reconnect_entered += 1
		if suite == "soak" and _reconnect_entered > 1:
			_soak_fail(_rebind_case(), "FAIL", "duplicate entered_game without a disconnect during %s" % stage)
			return
		if epoch > _reconnect_epoch:
			_bound_epoch = epoch
	elif suite == "soak" and stage in SOAK_PLAY_STAGES:
		_fail_stage("unexpected entered_game during %s (no reconnect was requested)" % stage)

## 0 = still waiting, 1 = main VID / scene / UI re-acquired in the new epoch, -1 = run finished.
## Soak additionally waits REBIND_SETTLE_US of continuous readiness, then checks for
## duplicate entity nodes and a target / attack key left over from the old connection.
func _rebound(now: int) -> int:
	if _bound_epoch <= _reconnect_epoch:
		_rebind_since = -1
		return 0
	var vid := int(client.get_main_vid()) if client.has_method("get_main_vid") else 0
	var main: Dictionary = client.get_entity(vid) if vid > 0 and client.has_method("get_entity") else {}
	var pos: Variant = main.get("pos_cm", null)
	var ready := _world_ready()
	if vid <= 0 or not (pos is Vector3) or (pos.x == 0.0 and pos.y == 0.0) or not ready.ready \
			or bool(_flow_snapshot().get("reconnecting", false)):
		_rebind_since = -1
		return 0
	_main_vid = vid
	if suite != "soak":
		return 1
	if _rebind_since < 0:
		_rebind_since = now
	if now - _rebind_since < REBIND_SETTLE_US:
		return 0
	var checks := _rebind_checks()
	_event("rebind_checked", {"main_vid": vid, "round": _round, "duplicate_vids": checks.duplicate_vids,
		"target_residue": checks.target_residue, "map_key": ready.map_key, "state": checks.status})
	if checks.status != "PASS":
		_soak_fail(_rebind_case(), checks.status, checks.reason)
		return -1
	return 1

func _rebind_checks() -> Dictionary:
	var out := {"status": "PASS", "reason": "", "duplicate_vids": [], "target_residue": []}
	var context := _context()
	var net_world: Variant = context.get("net_world", null)
	var parent: Variant = net_world.get("parent") if net_world != null and is_instance_valid(net_world) and "parent" in net_world else null
	if parent == null or not is_instance_valid(parent):
		out.status = "BLOCKED"
		out.reason = "NetWorld entity registry is unavailable; duplicate entity check not possible"
		return out
	var seen := {}
	for child in parent.get_children():
		if child.is_queued_for_deletion() or not child.has_meta("vid"):
			continue
		var vid := int(child.get_meta("vid"))
		if seen.has(vid) and not (vid in out.duplicate_vids):
			out.duplicate_vids.append(vid)
		seen[vid] = true
	var net_play: Variant = context.get("net_play", null)
	if net_play != null and is_instance_valid(net_play):
		var target_vid := int(net_play.get("_target_vid")) if "_target_vid" in net_play else 0
		if target_vid > 0 and client.get_entity(target_vid).is_empty():
			out.target_residue.append("target_vid")
		if "_attack_key_down" in net_play and bool(net_play.get("_attack_key_down")):
			out.target_residue.append("attack_key_down")
	if not out.duplicate_vids.is_empty():
		out.status = "FAIL"
		out.reason = "duplicate entity nodes after re-login: %s" % [out.duplicate_vids]
	elif not out.target_residue.is_empty():
		out.status = "FAIL"
		out.reason = "old target/attack state kept after re-login: %s" % [out.target_residue]
	return out

func _rebind_case() -> String:
	match stage:
		"FAULT": return "STB-DISCONNECT-01"
		"WARP": return "STB-WARP-01"
	return "STB-RECONNECT-01"

# --- STB-01: soak rounds ------------------------------------------------------------

func _soak_cfg() -> Dictionary:
	var soak: Variant = config.get("soak", {})
	return soak if soak is Dictionary else {}

func _begin_soak(now: int) -> void:
	var checked := Config.validate_soak(config)
	if not checked.ok:
		_block_remaining("soak config invalid: %s" % ", ".join(checked.errors))
		_finish()
		return
	_soak_start_us = now
	_soak_wall_start_ms = _wall_ms()
	_round = 0
	for size in _soak_cfg().resolutions.sizes:
		_resize_counts["%dx%d" % [int(size[0]), int(size[1])]] = 0
	_begin_round()

func _begin_round() -> void:
	_kills = 0
	_abandoned = {}
	_event("soak_round", {"round": _round, "wall_ms": _wall_ms()})
	_begin_move(0)

## Runs the post-pickup steps in SOAK_STEPS order, starting after `after`.
func _soak_continue(after: String) -> void:
	var soak := _soak_cfg()
	var start := SOAK_STEPS.find(after) + 1
	for i in range(start, SOAK_STEPS.size()):
		match SOAK_STEPS[i]:
			"FAULT":
				if String(soak.faults.get("status", "")) == "confirmed" and not _next_fault_type().is_empty():
					_begin_fault()
					return
			"WARP":
				var warp: Dictionary = soak.warp
				if String(warp.get("status", "")) == "confirmed" and _warps < int(warp.required):
					_begin_warp(0)
					return
			"RESIZE":
				var available: Dictionary = window.availability() if window != null and window.has_method("availability") else {}
				if window == null:
					_resize_blocked = "no window adapter: logical resolution switching needs a windowed display"
				elif not bool(available.get("ok", false)):
					_resize_blocked = "window resize unavailable: %s" % String(available.get("reason", "unknown"))
				else:
					_begin_resize()
					return
			"REST":
				_begin_rest()
				return

func _next_fault_type() -> String:
	var per := int(_soak_cfg().faults.get("per_type", 0))
	var best := ""
	for type in Config.SOAK_FAULT_TYPES:
		if int(_faults[type]) < per and (best.is_empty() or int(_faults[type]) < int(_faults[best])):
			best = type
	return best

## The fault itself is produced outside the client (script/playable_fault_proxy.mjs reads
## this event); the flow never calls reconnect() here — ReconnectUI's own retry recovers.
func _begin_fault() -> void:
	var faults: Dictionary = _soak_cfg().faults
	var type := _next_fault_type()
	_fault = {"type": type, "disconnected_us": -1}
	_arm_reconnect()
	cancel_inputs()
	_enter("FAULT")
	_event("fault_request", {"type": type, "attempt": int(_faults[type]) + 1, "round": _round,
		"timeout_ms": FAULT_DISCONNECT_TIMEOUT_US / 1000,
		"unreachable_seconds": int(faults.unreachable_seconds) if type == "unreachable" else 0})

func _tick_fault(now: int) -> void:
	var type := String(_fault.type)
	if int(_fault.disconnected_us) < 0:
		if now - _stage_since > FAULT_DISCONNECT_TIMEOUT_US:
			_soak_fail("STB-DISCONNECT-01", "FAIL", "requested %s fault produced no disconnect within 30s" % type)
		return
	var bound := _rebound(now)
	if bound < 0:
		return
	if bound > 0:
		_faults[type] = int(_faults[type]) + 1
		_event("fault_recovered", {"type": type, "round": _round, "count": _faults[type],
			"elapsed_ms": (now - int(_fault.disconnected_us)) / 1000})
		_fault = {}
		_soak_continue("FAULT")
		return
	var budget := int(_soak_cfg().faults.unreachable_seconds) * 1_000_000 + FAULT_RECOVERY_GRACE_US
	if now - int(_fault.disconnected_us) > budget:
		_soak_fail("STB-DISCONNECT-01", "FAIL", "%s fault not recovered by the reconnect path within %ds" % [type, budget / 1_000_000])

## Out-and-back through the configured portals. Only a new server connection epoch that
## lands on the expected map counts; the client never moves nodes to fake a map change.
func _begin_warp(leg: int) -> void:
	var warp: Dictionary = _soak_cfg().warp
	_warp_leg = leg
	var point: Array = warp.portal_cm if leg == 0 else warp.return_portal_cm
	_warp_expected = String(warp.destination_map_key) if leg == 0 else String(config.map_key)
	_arm_reconnect()
	cancel_inputs()
	_enter("WARP")
	var controller: Object = _context().get("pc", null)
	if controller == null or not is_instance_valid(controller) or not controller.has_method("walk_to"):
		_soak_fail("STB-WARP-01", "FAIL", "PlayerController.walk_to is unavailable")
		return
	controller.walk_to(MapCoord.to_world(Vector3(float(point[0]) * 0.01, 0.0, -float(point[1]) * 0.01)))
	_event("warp_requested", {"leg": leg, "round": _round, "portal_cm": Vector2(float(point[0]), float(point[1])),
		"expected_map_key": _warp_expected, "cross_server": bool(warp.cross_server)})

func _tick_warp(now: int) -> void:
	var bound := _rebound(now)
	if bound < 0:
		return
	var map_key := String(_flow_snapshot().get("map_key", ""))
	if bound > 0:
		if map_key != _warp_expected:
			_event("map_mismatch", {"map_key": map_key, "expected_map_key": _warp_expected, "leg": _warp_leg})
			_soak_fail("STB-WARP-01", "FAIL", "server reconnect after the portal loaded %s instead of %s" % [map_key, _warp_expected])
			return
		_warps += 1
		_event("warp_arrived", {"leg": _warp_leg, "round": _round, "count": _warps, "map_key": map_key,
			"elapsed_ms": (now - _stage_since) / 1000})
		if _warp_leg == 0:
			_begin_warp(1)
		else:
			_soak_continue("WARP")
		return
	if now - _stage_since > WARP_TIMEOUT_US:
		var why := " (map_key changed without a new server connection)" if map_key == _warp_expected else ""
		_soak_fail("STB-WARP-01", "FAIL", "no server reconnect into %s within 120s%s" % [_warp_expected, why])

func _begin_resize() -> void:
	_resize_index = -1
	_enter("RESIZE")
	_next_resize(_now())

func _next_resize(now: int) -> void:
	var sizes: Array = _soak_cfg().resolutions.sizes
	while true:
		_resize_index += 1
		if _resize_index >= sizes.size():
			_soak_continue("RESIZE")
			return
		var size := _resize_target()
		if metrics != null:
			metrics.mark("resize", {"size": "%dx%d" % [size.x, size.y]})
		_resize_before = window.logical_size()
		_resize_since = now
		if bool(window.request_logical_size(size)):
			return
		var key := "%dx%d" % [size.x, size.y]
		if _resize_blocked.is_empty():
			_resize_blocked = "logical size %s rejected by the window adapter (larger than the usable screen)" % key
		_event("window_size_rejected", {"requested_size": size, "round": _round, "screen_scale": float(window.screen_scale())})

func _resize_target() -> Vector2i:
	var item: Array = _soak_cfg().resolutions.sizes[_resize_index]
	return Vector2i(int(item[0]), int(item[1]))

func _tick_resize(now: int) -> void:
	if now - _resize_since < RESIZE_SETTLE_US:
		return
	var size := _resize_target()
	var logical: Vector2i = window.logical_size()
	var viewport: Vector2i = window.viewport_size()
	if absi(logical.x - size.x) <= RESIZE_TOLERANCE_PX and absi(logical.y - size.y) <= RESIZE_TOLERANCE_PX \
			and viewport.x > 0 and viewport.y > 0 and _world_ready().ready:
		# Physical pixels differ from logical points on Retina; only the logical size is judged.
		var switched := absi(_resize_before.x - size.x) > RESIZE_TOLERANCE_PX or absi(_resize_before.y - size.y) > RESIZE_TOLERANCE_PX
		var key := "%dx%d" % [size.x, size.y]
		if switched:
			_resize_counts[key] = int(_resize_counts.get(key, 0)) + 1
		_event("window_resized", {"requested_size": size, "logical_size": logical, "physical_size": window.physical_size(),
			"viewport_size": viewport, "screen_scale": float(window.screen_scale()), "round": _round,
			"count": int(_resize_counts.get(key, 0)), "accepted": switched})
		_next_resize(now)
		return
	if now - _resize_since > RESIZE_TIMEOUT_US:
		_event("window_resize_timeout", {"requested_size": size, "logical_size": logical, "viewport_size": viewport,
			"physical_size": window.physical_size(), "screen_scale": float(window.screen_scale()), "round": _round})
		_soak_fail("STB-RESIZE-01", "FAIL", "window stayed at logical %dx%d after requesting %dx%d" % [logical.x, logical.y, size.x, size.y])

func _begin_rest() -> void:
	cancel_inputs()
	_enter("REST")
	_event("soak_rest", {"round": _round, "state": "start", "wall_ms": _wall_ms()})

func _tick_rest(now: int) -> void:
	if now - _stage_since < int(_soak_cfg().rest_seconds) * 1_000_000:
		return
	_event("soak_rest", {"round": _round, "state": "end", "wall_ms": _wall_ms()})
	_rounds_done = _round + 1
	if now - _soak_start_us >= int(_soak_cfg().duration_seconds) * 1_000_000:
		_finish()
		return
	_round += 1
	_begin_round()

func _soak_fail(case_id: String, case_status: String, reason: String) -> void:
	cancel_inputs()
	_finish_case(case_id, case_status, reason)
	_finish()

## §9.3: one repetition of an interaction = one stay in a stage; it is "severe" when
## a >100ms frame tagged with that phase arrived during the stay.
func _track_frames(previous: String) -> void:
	if suite != "soak" or metrics == null or _soak_start_us < 0 or previous.is_empty():
		return
	var phase := previous.to_lower()
	var total := int(metrics.severe_by_phase().get(phase, 0))
	var delta := total - int(_severe_seen.get(phase, 0))
	_severe_seen[phase] = total
	var history: Array = _frame_history.get(phase, [])
	history.append(delta > 0)
	if history.size() > FRAME_HISTORY_CAP:
		history.pop_front()
	_frame_history[phase] = history
	if delta > 0:
		_event("frame_severe", {"interaction": phase, "severe": delta, "round": _round})
		if history.size() >= 3 and not Metrics.repeated_severe({phase: history.slice(-3)}).is_empty():
			_repeated_severe[phase] = true

func _close_soak() -> void:
	var now := _now()
	var soak := _soak_cfg()
	var elapsed := now - _soak_start_us
	var duration_s := int(soak.duration_seconds)
	var failures: Array = report.failures()
	if not report.has_case("STB-DURATION-01"):
		if not failures.is_empty():
			report.add_case("STB-DURATION-01", "FAIL", "soak ended by a failure after %d rounds (%.0fs of %ds)" % [_rounds_done, elapsed / 1e6, duration_s])
		elif not _soak_blocked.is_empty():
			report.add_case("STB-DURATION-01", "BLOCKED", "soak ended early: %s" % _soak_blocked)
		elif elapsed < duration_s * 1_000_000:
			report.add_case("STB-DURATION-01", "BLOCKED", "wall-clock %.0fs is below the %ds soak" % [elapsed / 1e6, duration_s])
		elif _rounds_done < int(soak.min_rounds):
			report.add_case("STB-DURATION-01", "BLOCKED", "%d rounds completed, need at least %d" % [_rounds_done, int(soak.min_rounds)])
		else:
			report.add_case("STB-DURATION-01", "PASS", "%d move/combat/pickup rounds over %.0fs wall clock" % [_rounds_done, elapsed / 1e6])
	if not report.has_case("STB-FRAMES-01"):
		if metrics == null:
			report.add_case("STB-FRAMES-01", "BLOCKED", "no frame metrics attached to this run")
		elif not _repeated_severe.is_empty():
			report.add_case("STB-FRAMES-01", "FAIL", ">100ms frames in 2 of 3 consecutive repetitions of: %s" % ", ".join(_repeated_severe.keys()))
		else:
			var unattributed: Dictionary = metrics.long_unattributed_by_phase()
			unattributed.erase("loading")
			var count := 0
			for phase in unattributed:
				count += int(unattributed[phase])
			if count > 0:
				report.add_case("STB-FRAMES-01", "BLOCKED", "%d gameplay frames >50ms need attribution: %s" % [count, unattributed])
			else:
				report.add_case("STB-FRAMES-01", "PASS", "no repeated >100ms interaction; every gameplay frame >50ms has a marker")
	if not report.has_case("STB-RECONNECT-01"):
		var required := int(soak.reconnects)
		if _active_reconnects >= required:
			report.add_case("STB-RECONNECT-01", "PASS", "%d active reconnects re-acquired main VID, scene and UI without residue" % _active_reconnects)
		else:
			report.add_case("STB-RECONNECT-01", "BLOCKED", "%d of %d active reconnects completed" % [_active_reconnects, required])
	if not report.has_case("STB-WARP-01"):
		var warp: Dictionary = soak.warp
		if String(warp.get("status", "")) != "confirmed":
			report.add_case("STB-WARP-01", "BLOCKED", "warp route not confirmed by the environment owner (portal, destination, cross-server support)")
		elif _warps >= int(warp.required):
			report.add_case("STB-WARP-01", "PASS", "%d map changes through configured portals, each on a new server connection" % _warps)
		else:
			report.add_case("STB-WARP-01", "BLOCKED", "%d of %d portal map changes completed" % [_warps, int(warp.required)])
	if not report.has_case("STB-DISCONNECT-01"):
		var faults: Dictionary = soak.faults
		if String(faults.get("status", "")) != "confirmed":
			report.add_case("STB-DISCONNECT-01", "BLOCKED", "real transport faults (connection closed, short unreachability) need a confirmed local test proxy; active reconnect is not a substitute")
		elif _next_fault_type().is_empty():
			report.add_case("STB-DISCONNECT-01", "PASS", "recovered from real faults %s" % [_faults])
		else:
			report.add_case("STB-DISCONNECT-01", "BLOCKED", "real faults recovered %s, need %d per type" % [_faults, int(faults.per_type)])
	if not report.has_case("STB-RESIZE-01"):
		var short: Array = []
		for key in _resize_counts:
			if int(_resize_counts[key]) < int(soak.resolutions.switches_per_size):
				short.append(key)
		if not _resize_blocked.is_empty():
			report.add_case("STB-RESIZE-01", "BLOCKED", _resize_blocked)
		elif short.is_empty():
			report.add_case("STB-RESIZE-01", "PASS", "switched to each logical size at least %d times: %s" % [int(soak.resolutions.switches_per_size), _resize_counts])
		else:
			report.add_case("STB-RESIZE-01", "BLOCKED", "logical size switches %s below %d for %s" % [_resize_counts, int(soak.resolutions.switches_per_size), short])
	var interactions := {}
	for phase in _frame_history:
		var hits := 0
		for hit in _frame_history[phase]:
			hits += int(bool(hit))
		interactions[phase] = {"recent": _frame_history[phase].size(), "severe": hits}
	report.set_section("soak", {"duration_seconds": duration_s, "elapsed_seconds": elapsed / 1e6, "rounds": _rounds_done,
		"rest_seconds": int(soak.rest_seconds), "warmup_rounds": int(soak.warmup_rounds),
		"active_reconnects": _active_reconnects, "real_faults": _faults.duplicate(), "fault_mode": String(soak.faults.get("status", "")),
		"warps": _warps, "warp_status": String(soak.warp.get("status", "")), "cross_server": soak.warp.get("cross_server", null),
		"resizes": _resize_counts.duplicate(), "wall_start_ms": _soak_wall_start_ms, "wall_end_ms": _wall_ms(),
		"interactions": interactions, "repeated_severe": _repeated_severe.keys(),
		"exits": "counted by the runner (script/playable_soak.sh)",
		"cold_start": "进程冷启动/系统缓存状态未知；未执行系统缓存清空"})

func _wall_ms() -> int:
	return int(wall_ms.call()) if wall_ms.is_valid() else int(Time.get_unix_time_from_system() * 1000.0)

# --- failure / teardown ---------------------------------------------------------

func _on_disconnected() -> void:
	if stage == "MOVE_CONFIRM" or (suite == "soak" and stage in ["WARP", "FAULT"]):
		# Expected during a requested reconnect / portal / transport fault; the next
		# entered_game after a disconnect is not a duplicate signal.
		_reconnect_entered = 0
		if stage == "FAULT" and not _fault.is_empty() and int(_fault.disconnected_us) < 0:
			_fault.disconnected_us = _now()
			_event("fault_disconnected", {"type": _fault.type, "round": _round,
				"elapsed_ms": (int(_fault.disconnected_us) - _stage_since) / 1000})
		return
	cancel_inputs()
	match stage:
		"LOGIN":
			_finish_case("NET-LOGIN-01", "FAIL", "disconnected before char_list")
		"SELECT", "WORLD":
			# NET-03: a disconnect while entering must end the run without half-loaded operations.
			_finish_case(_case_for_stage(), "FAIL", "disconnected while entering the game")
		_:
			_finish_case(_case_for_stage(), "FAIL", "unexpected disconnect during %s" % stage)
	_finish()

func _fail_stage(reason: String) -> void:
	cancel_inputs()
	_finish_case(_case_for_stage(), "FAIL", reason)
	_finish()

func _finish_case(case_id: String, case_status: String, reason: String) -> void:
	var soak_active := suite == "soak" and _soak_start_us >= 0
	if soak_active and case_status == "BLOCKED" and _soak_blocked.is_empty():
		_soak_blocked = "soak round %d %s: %s" % [_round, case_id, reason]
	if report.has_case(case_id):
		# A later soak round repeats cases that already have a result: record the round
		# failure against the run instead of a duplicate case result.
		if soak_active and case_status == "FAIL":
			report.add_failure("soak round %d %s: %s" % [_round, case_id, reason])
		return
	var evidence: Array = []
	if stage == "SKILL" and case_id == _case_for_stage():
		evidence.append_array(_shots)
	if capture.is_valid():
		var path := String(capture.call(case_id))
		if not path.is_empty():
			evidence.append(path)
	report.add_case(case_id, case_status, reason, evidence, float(_now() - _stage_since) / 1000.0)

## PASS result; a soak round repeating an already-passed case records nothing new.
func _pass(case_id: String, reason: String) -> void:
	if suite == "soak" and report.has_case(case_id):
		return
	report.add_case(case_id, "PASS", reason)

func _block_remaining(reason: String) -> void:
	for id in report.required_cases():
		if not report.has_case(String(id)):
			report.add_case(String(id), "BLOCKED", reason)

func _finish() -> void:
	if done:
		return
	cancel_inputs()
	if suite == "soak" and _soak_start_us >= 0:
		_close_soak()
	done = true
	_event("finished", {"state": stage})
	status = String(report.finish().status)
	finished.emit(status)

# --- helpers ---------------------------------------------------------------------

func _case_for_stage() -> String:
	match stage:
		"BOOT": return "CONFIG-01"
		"LOGIN": return "NET-LOGIN-01"
		"SELECT": return "NET-SELECT-01"
		"WORLD": return "NET-WORLD-01"
		"MOVE": return _move_case()
		"MOVE_CONFIRM": return "NET-MOVE-CONFIRM-01"
		"TARGET": return "NET-TARGET-01" if not report.has_case("NET-TARGET-01") else "NET-DROP-01"
		"ATTACK": return "NET-ATTACK-01" if not report.has_case("NET-ATTACK-01") else "NET-DROP-01"
		"DROP": return "NET-DROP-01"
		"PICKUP": return "NET-PICKUP-01"
		"SKILL":
			var skills: Array = config.get("skill_cases", [])
			return Config.skill_case_id(skills[_skill_index]) if _skill_index < skills.size() else "CBT"
		"FAULT": return "STB-DISCONNECT-01"
		"WARP": return "STB-WARP-01"
		"RESIZE": return "STB-RESIZE-01"
		"REST": return "STB-DURATION-01"
	return stage

func _move_case() -> String:
	return "NET-MOVE-%02d" % (_waypoint + 1)

func _event(kind: String, payload: Dictionary) -> void:
	report.record(kind, stage, _case_for_stage(), _probe_epoch(), _main_vid, int(_target.get("vid", 0)), payload)

func _now() -> int:
	return int(clock_us.call())

func _probe_epoch() -> int:
	return int(probe.connection_epoch) if probe != null else 0

## §9.1 segment tags for the live frame sampler; frames before the world is ready are "loading".
func frame_tags() -> Dictionary:
	var ready := _world_ready()
	var net_world: Variant = _context().get("net_world", null)
	var parent: Variant = net_world.get("parent") if net_world != null and is_instance_valid(net_world) and "parent" in net_world else null
	var actors: int = parent.get_child_count() if parent != null and is_instance_valid(parent) else 0
	return {"map": ready.map_key, "phase": stage.to_lower() if ready.ready else "loading",
		"actor_count": Metrics.actor_bucket(actors)}

func _is_target(vid: int) -> bool:
	return not _target.is_empty() and vid == int(_target.vid) and int(_target.get("epoch", -1)) == _bound_epoch

func _flow_snapshot() -> Dictionary:
	if flow != null and is_instance_valid(flow) and flow.has_method("get_playable_snapshot"):
		return flow.get_playable_snapshot()
	return {}

## Borrowed references: re-read every use, never cached across reconnect/scene reload.
func _context() -> Dictionary:
	if flow != null and is_instance_valid(flow) and flow.has_method("get_playable_context"):
		return flow.get_playable_context()
	return {}

func _world_ready() -> Dictionary:
	var snap := _flow_snapshot()
	var context := _context()
	var player: Variant = context.get("player", null)
	var hud: Variant = context.get("hud", null)
	var player_ready: bool = player != null and is_instance_valid(player)
	var hud_ready: bool = hud != null and is_instance_valid(hud)
	var scene_ready := bool(snap.get("scene_ready", false))
	return {"ready": scene_ready and player_ready and hud_ready, "scene_ready": scene_ready,
		"player_ready": player_ready, "hud_ready": hud_ready, "map_key": String(snap.get("map_key", ""))}

func _ground_items() -> Array:
	return client.get_ground_items() if client.has_method("get_ground_items") else []

func _ground_iids() -> Dictionary:
	var out := {}
	for item in _ground_items():
		out[int(item.get("vid", 0))] = true
	return out

func _ground_pos_cm(item: Dictionary) -> Vector2:
	var pos: Variant = item.get("pos", null)
	if pos is Vector3:
		return Vector2(pos.x * 100.0, -pos.z * 100.0)
	return Vector2(INF, INF)

func _owner_matches(owner: String) -> bool:
	if owner.is_empty():
		return true
	var main: Dictionary = client.get_entity(_main_vid) if client.has_method("get_entity") else {}
	return String(main.get("name", "")) == owner

## Stacks are summed over every inventory cell, so a merge into an existing stack counts.
func _inventory_total(vnum: int) -> int:
	if not client.has_method("get_inventory"):
		return -1
	var total := 0
	for item in client.get_inventory():
		if int(item.get("vnum", 0)) == vnum:
			total += maxi(1, int(item.get("count", 1)))
	return total