TheTree_mock/scripts/operations/operations_coordinator.gd

class_name OperationsCoordinator
extends RefCounted

## Standalone deterministic coordinator for the 14-day operating loop.
##
## This aggregate owns scheduling, demand, arrival carry, daily summaries and
## insolvency state.  It deliberately does not own Guests, Facilities or Gold.
## Those domains are reached only through capture/validate/apply/rollback
## callbacks so a future SimulationRuntime adapter can compose one transaction.

signal demand_updated(report: Dictionary)
signal arrivals_committed(report: Dictionary)
signal daily_close_completed(summary: Dictionary)
signal morning_boundary_completed(report: Dictionary)
signal operations_alerted(alert: Dictionary)
signal terminal_state_changed(old_state: StringName, new_state: StringName, reason: StringName)

const CommandResultType = preload("res://scripts/core/command_result.gd")

const SNAPSHOT_VERSION := 5
const MIN_SUPPORTED_SNAPSHOT_VERSION := 1
const MINUTES_PER_DAY := 1_440
const DEMAND_INTERVAL_MINUTES := 60
const ARRIVAL_INTERVAL_MINUTES := 30
const MORNING_MINUTE_OF_DAY := 360
const FIRST_DAILY_CLOSE_MINUTE := MINUTES_PER_DAY
const KEY_SEPARATOR := "\u001f"
const MAX_START_MINUTE := 360
const MAX_TRACKED_DAYS := 31
const MAX_TRACKED_MINUTES := MAX_TRACKED_DAYS * MINUTES_PER_DAY
const MAX_ADVANCE_SPAN_MINUTES := 31 * MINUTES_PER_DAY
const MAX_SAFE_COLLECTION_SIZE := 20_000
const MAX_VARIANT_DEPTH := 32
const MAX_VARIANT_NODES := 2_000_000
const MAX_TOTAL_STRING_LENGTH := 8_000_000
const MAX_SINGLE_STRING_LENGTH := 4_096
const MAX_CALLBACK_VARIANT_NODES := 20_000
const MAX_CALLBACK_STRING_LENGTH := 100_000
const MAX_DEMAND_PROFILES := 16
const MAX_WORLD_SEED := 2_147_483_647
const MAX_FINANCIAL_AMOUNT := 1_000_000_000_000_000
const MAX_SEQUENCE_VALUE := 99_999_999

const PHASE_CAPTURE: StringName = &"capture"
const PHASE_VALIDATE: StringName = &"validate"
const PHASE_APPLY: StringName = &"apply"
const PHASE_ROLLBACK: StringName = &"rollback"

const TERMINAL_RUNNING: StringName = &"RUNNING"
const TERMINAL_BANKRUPT: StringName = &"BANKRUPT"

const TIME_BANDS: Array[StringName] = [
	&"MORNING",
	&"DAY",
	&"EVENING",
	&"NIGHT",
	&"ANY",
]

const DEMAND_COMPONENTS: Array[StringName] = [
	&"base",
	&"reputation",
	&"unmet_need",
	&"event",
	&"recent_price_penalty",
	&"wait_penalty",
]

const DAILY_CATEGORIES: Array[StringName] = [
	&"revenue",
	&"refunds",
	&"cost_of_goods",
	&"wages",
	&"maintenance",
	&"utilities",
	&"taxes_and_interest",
	&"fines",
	&"construction_cash_flow",
]

const FINANCING_CATEGORIES: Array[StringName] = [&"financing_principal_inflows", &"financing_principal_outflows"]

const DEFAULT_CONFIG := {
	&"world_seed": 1,
	&"population_cap": 100,
	&"max_arrivals_per_interval": 10,
	# One arrival per 100 DemandIndex points per active profile and interval.
	# These are tuning values, not a final balance decision.
	&"arrival_rate_numerator": 1,
	&"arrival_rate_denominator": 100,
	# A profile may retain at most three full interval caps of fixed-point carry.
	&"arrival_carry_interval_limit": 3,
	&"max_component_magnitude": 10_000,
	&"bankruptcy_close_count": 3,
}

var current_minute := 0
var pause_requested := false
var last_error := ""

var _origin_minute := 0

# Old saves finish their open 06:00 period once, then use midnight.
# Historical insolvency evidence is retained without mixing it into new loss streaks.
var _legacy_calendar_last_day := 0
var _legacy_history_last_day := 0
# Closed records from versions 1-4 retain their original principal-excluded basis.
var _legacy_principal_profit_last_day := 0

var _config: Dictionary = DEFAULT_CONFIG.duplicate(true)
var _demand_profiles: Dictionary = {}
var _arrival_carry: Dictionary = {}
var _demand_history: Array[Dictionary] = []
var _arrival_reports: Array[Dictionary] = []
var _day_summaries: Dictionary = {}
var _morning_reports: Dictionary = {}
var _alerts: Array[Dictionary] = []
var _processed_demand_minutes: Dictionary = {}
var _processed_arrival_minutes: Dictionary = {}

var _population_count := 0
var _arrears_total := 0
# Serialized field name retained for compatibility; new closes count net losses.
var _consecutive_insolvency_count := 0
var _terminal_state: StringName = TERMINAL_RUNNING
var _next_arrival_plan_sequence := 1
var _next_guest_sequence := 1
var _arrival_plan_high_water := 0
var _guest_high_water := 0
var _fair_rotation_counter := 0
var _pending_boundary_minute := -1
var _deferred_boundary_minute := -1
var _pause_reason: StringName = &""
var _external_quarantined := false
var _quarantine_report: Dictionary = {}

var _is_advancing := false
var _notification_queue: Array[Dictionary] = []
var _notification_delivery_deferred := false
var _initialization_valid := true

var _midnight_boundary := Callable()
var _daily_close_boundary := Callable()
var _arrival_boundary := Callable()
var _morning_boundary := Callable()


func _init(config: Dictionary = {}) -> void:
	if not _canonical_key_collision(config).is_empty():
		last_error = "Operations constructor config has duplicate canonical keys."
		_initialization_valid = false
		_origin_minute = 0
		current_minute = 0
		return
	var start_value: Variant = _value(config, &"start_minute", 0)
	var start_minute := int(start_value) if _is_integral_value(start_value) else -1
	var candidate := config.duplicate(true)
	candidate.erase(&"start_minute")
	candidate.erase("start_minute")
	var normalized := _normalize_config(candidate)
	if bool(normalized[&"ok"]):
		_config = normalized[&"config"]
	else:
		last_error = String(normalized[&"error"])
		_initialization_valid = false
	if start_minute < 0 or start_minute > MAX_START_MINUTE:
		last_error = "Start minute must be within the explicit baseline range 0..360."
		_initialization_valid = false
		start_minute = 0
	_origin_minute = start_minute
	current_minute = start_minute


func configure_callbacks(
	daily_close_boundary: Callable = Callable(),
	arrival_boundary: Callable = Callable(),
	morning_boundary: Callable = Callable(),
	midnight_boundary: Callable = Callable()
) -> OperationsCoordinator:
	_daily_close_boundary = daily_close_boundary
	_arrival_boundary = arrival_boundary
	_morning_boundary = morning_boundary
	_midnight_boundary = midnight_boundary
	return self


func configure(config: Dictionary) -> CommandResult:
	if _is_advancing:
		return _failure("REENTRANT_MUTATION", "Operations state cannot mutate during advance notification or callbacks.")
	if not _initialization_valid:
		return _failure("INITIALIZATION_INVALID", last_error)
	if _external_quarantined:
		return _failure("EXTERNAL_ROLLBACK_FAILED", "External uncertainty quarantines configuration changes.")
	if _has_committed_operations():
		return _failure(
			"CONFIG_LOCKED",
			"Operating configuration cannot change after a boundary has committed."
		)
	var normalized := _normalize_config(config)
	if not bool(normalized[&"ok"]):
		return _failure("INVALID_CONFIG", String(normalized[&"error"]))
	var candidate_config: Dictionary = normalized[&"config"]
	if _population_count > int(candidate_config[&"population_cap"]):
		return _failure("INVALID_CONFIG", "Population cannot be silently truncated by a lower cap.")
	for profile_value: Variant in _demand_profiles.values():
		if not profile_value is Dictionary:
			return _failure("INVALID_CONFIG", "Existing demand state is malformed.")
		var components_value: Variant = _value(profile_value, &"components", null)
		if not components_value is Dictionary:
			return _failure("INVALID_CONFIG", "Existing demand components are malformed.")
		var candidate_components := _normalize_demand_components_for_limit(
			components_value,
			int(candidate_config[&"max_component_magnitude"])
		)
		if not bool(candidate_components[&"ok"]):
			return _failure(
				"INVALID_CONFIG",
				"New configuration would invalidate an existing demand profile."
			)
	_config = candidate_config
	last_error = ""
	return _success({&"config": get_config()})


func get_config() -> Dictionary:
	return _config.duplicate(true)


func set_demand_input(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName,
	components: Dictionary
) -> CommandResult:
	if _is_advancing:
		return _failure("REENTRANT_MUTATION", "Demand cannot mutate during advance notification or callbacks.")
	if not _initialization_valid:
		return _failure("INITIALIZATION_INVALID", last_error)
	if _external_quarantined:
		return _failure("EXTERNAL_ROLLBACK_FAILED", "External uncertainty quarantines demand changes.")
	if _terminal_state != TERMINAL_RUNNING:
		return _failure("TERMINAL_STATE", "Demand cannot change after bankruptcy.")
	var identity_error := _validate_demand_identity(facility_tag, guest_class, time_band)
	if not identity_error.is_empty():
		return _failure("INVALID_DEMAND_KEY", identity_error)
	var normalized := _normalize_demand_components(components)
	if not bool(normalized[&"ok"]):
		return _failure("INVALID_DEMAND_INPUT", String(normalized[&"error"]))
	var key := _demand_key(facility_tag, guest_class, time_band)
	if not _demand_profiles.has(key) and _demand_profiles.size() >= MAX_DEMAND_PROFILES:
		return _failure("DEMAND_PROFILE_LIMIT", "Demand profile limit has been reached.")
	var normalized_components: Dictionary = normalized[&"components"]
	var profile := {
		&"facility_tag": facility_tag,
		&"guest_class": guest_class,
		&"time_band": time_band,
		&"components": normalized_components,
		&"demand_index": _calculate_demand_index(normalized_components),
		&"last_evaluated_at": -1,
	}
	_demand_profiles[key] = profile
	if not _arrival_carry.has(key):
		_arrival_carry[key] = 0
	last_error = ""
	return _success({
		&"key": key,
		&"demand_index": profile[&"demand_index"],
	})


func remove_demand_input(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName
) -> CommandResult:
	if _is_advancing:
		return _failure("REENTRANT_MUTATION", "Demand cannot mutate during advance notification or callbacks.")
	if not _initialization_valid:
		return _failure("INITIALIZATION_INVALID", last_error)
	if _external_quarantined:
		return _failure("EXTERNAL_ROLLBACK_FAILED", "External uncertainty quarantines demand changes.")
	if _terminal_state != TERMINAL_RUNNING:
		return _failure("TERMINAL_STATE", "Demand cannot change after bankruptcy.")
	var key := _demand_key(facility_tag, guest_class, time_band)
	if not _demand_profiles.has(key):
		return _failure("DEMAND_NOT_FOUND", "Demand profile does not exist.")
	if int(_arrival_carry.get(key, 0)) > 0:
		return _failure(
			"ARRIVAL_CARRY_REMAINS",
			"A demand profile with pending arrival carry cannot be removed."
		)
	_demand_profiles.erase(key)
	_arrival_carry.erase(key)
	return _success({&"key": key})


func get_demand_profile(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName
) -> Dictionary:
	var key := _demand_key(facility_tag, guest_class, time_band)
	if not _demand_profiles.has(key):
		return {}
	return (_demand_profiles[key] as Dictionary).duplicate(true)


func get_demand_profiles() -> Array[Dictionary]:
	var result: Array[Dictionary] = []
	for key: String in _sorted_string_keys(_demand_profiles):
		result.append((_demand_profiles[key] as Dictionary).duplicate(true))
	return result


func get_demand_index(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName
) -> int:
	var profile := get_demand_profile(facility_tag, guest_class, time_band)
	return -1 if profile.is_empty() else int(profile[&"demand_index"])


func set_population_count(value: int) -> CommandResult:
	if _is_advancing:
		return _failure("REENTRANT_MUTATION", "Population cannot mutate during advance notification or callbacks.")
	if not _initialization_valid:
		return _failure("INITIALIZATION_INVALID", last_error)
	if _external_quarantined:
		return _failure("EXTERNAL_ROLLBACK_FAILED", "External uncertainty quarantines population changes.")
	if _terminal_state != TERMINAL_RUNNING:
		return _failure("TERMINAL_STATE", "Population cannot change after bankruptcy.")
	if value < 0 or value > int(_config[&"population_cap"]):
		return _failure(
			"INVALID_POPULATION",
			"Population must remain within zero and the configured cap."
		)
	_population_count = value
	return _success({
		&"population_count": _population_count,
		&"population_cap": int(_config[&"population_cap"]),
	})


func get_population_count() -> int:
	return _population_count


func get_origin_minute() -> int:
	return _origin_minute


func is_external_quarantined() -> bool:
	return _external_quarantined


func get_quarantine_report() -> Dictionary:
	return _quarantine_report.duplicate(true)


func advance_to(target_minute: int, defer_notifications := false) -> CommandResult:
	if _is_advancing:
		return _failure(
			"REENTRANT_ADVANCE",
			"Operations advance cannot re-enter from a callback or signal handler."
		)
	if _notification_delivery_deferred:
		return _failure(
			"NOTIFICATION_COMMIT_REQUIRED",
			"Deferred Operations notifications must be committed or discarded before advancing."
		)
	_is_advancing = true
	var result := _advance_to_internal(target_minute)
	# Domain signals are emitted only after every coordinator commit and clock
	# assignment for this call. Re-entrant advance remains guarded while they run.
	if defer_notifications:
		_notification_delivery_deferred = not _notification_queue.is_empty()
	else:
		_flush_notifications()
	_is_advancing = false
	return result


func has_deferred_notifications() -> bool:
	return _notification_delivery_deferred


func commit_deferred_notifications() -> bool:
	if _is_advancing:
		last_error = "Operations notification commit cannot re-enter an advance."
		return false
	if not _notification_delivery_deferred:
		return _notification_queue.is_empty()
	_is_advancing = true
	_notification_delivery_deferred = false
	_flush_notifications()
	_is_advancing = false
	last_error = ""
	return true


func discard_deferred_notifications() -> bool:
	if _is_advancing:
		last_error = "Operations notification discard cannot re-enter an advance."
		return false
	_notification_queue.clear()
	_notification_delivery_deferred = false
	last_error = ""
	return true


func _advance_to_internal(target_minute: int) -> CommandResult:
	if not _initialization_valid:
		return _failure("INITIALIZATION_INVALID", last_error)
	if target_minute < 0:
		return _failure("INVALID_TIME", "Target minute must be non-negative.")
	if target_minute < current_minute:
		return _failure(
			"TIME_REWIND_REJECTED",
			"Operations time cannot move backwards.",
			{&"current_minute": current_minute, &"target_minute": target_minute}
		)
	if target_minute > _origin_minute + MAX_TRACKED_MINUTES:
		return _failure("HISTORY_LIMIT_REACHED", "Uncompacted operations history is limited to 31 days.")
	if target_minute - current_minute > MAX_ADVANCE_SPAN_MINUTES:
		return _failure("ADVANCE_SPAN_TOO_LARGE", "One operations advance may cover at most 31 days.")
	if _external_quarantined:
		return _failure(
			"EXTERNAL_ROLLBACK_FAILED",
			"External rollback failed; normal retry is quarantined until verified recovery."
		)
	if _terminal_state != TERMINAL_RUNNING:
		return _failure("TERMINAL_STATE", "Operations are terminal after bankruptcy.")
	if pause_requested:
		return _failure("PAUSE_REQUESTED", "Operations require an explicit pause acknowledgement.")

	var started_at := current_minute
	var counts := {
		&"daily_closes": 0,
		&"morning_boundaries": 0,
		&"demand_updates": 0,
		&"arrival_boundaries": 0,
	}
	var completed_deferred := false
	if _deferred_boundary_minute != -1:
		if _deferred_boundary_minute != current_minute:
			return _failure("INVALID_DEFERRED_BOUNDARY", "Deferred boundary no longer matches the clock.")
		var deferred_result := _process_minute_boundaries(current_minute, counts)
		if not deferred_result.is_success():
			deferred_result.details[&"current_minute"] = current_minute
			deferred_result.details[&"failed_boundary_minute"] = current_minute
			return deferred_result
		if bool(deferred_result.details.get(&"deferred", false)):
			return _success({
				&"from_minute": started_at,
				&"requested_minute": target_minute,
				&"current_minute": current_minute,
				&"processed_minutes": 0,
				&"boundary_counts": counts,
				&"pause_requested": pause_requested,
				&"terminal_state": _terminal_state,
			})
		_deferred_boundary_minute = -1
		completed_deferred = true
	if target_minute == current_minute:
		if completed_deferred:
			return _success({
				&"current_minute": current_minute,
				&"processed_minutes": 0,
				&"deferred_completed": true,
				&"boundary_counts": counts,
			})
		var replay := _success({
			&"current_minute": current_minute,
			&"processed_minutes": 0,
			&"replay": true,
			&"boundary_counts": counts,
		})
		replay.is_replay = true
		return replay

	for minute in range(current_minute + 1, target_minute + 1):
		var boundary_result := _process_minute_boundaries(minute, counts)
		if not boundary_result.is_success():
			# The failing boundary made no coordinator commit. Earlier, lower-minute
			# boundaries stay committed, and retry begins at this exact minute.
			current_minute = minute - 1
			_pending_boundary_minute = minute
			boundary_result.details[&"current_minute"] = current_minute
			boundary_result.details[&"failed_boundary_minute"] = minute
			return boundary_result
		current_minute = minute
		if _pending_boundary_minute == minute:
			_pending_boundary_minute = -1
		if bool(boundary_result.details.get(&"deferred", false)):
			_deferred_boundary_minute = minute
		if _terminal_state != TERMINAL_RUNNING:
			break
		if pause_requested:
			break

	last_error = ""
	return _success({
		&"from_minute": started_at,
		&"requested_minute": target_minute,
		&"current_minute": current_minute,
		&"processed_minutes": current_minute - started_at,
		&"boundary_counts": counts,
		&"pause_requested": pause_requested,
		&"terminal_state": _terminal_state,
	})


func acknowledge_pause() -> CommandResult:
	if _is_advancing:
		return _failure("REENTRANT_MUTATION", "Pause acknowledgement cannot re-enter an advance.")
	if _terminal_state != TERMINAL_RUNNING:
		return _failure("TERMINAL_STATE", "Bankruptcy cannot be acknowledged away.")
	if _external_quarantined:
		return _failure(
			"EXTERNAL_ROLLBACK_FAILED",
			"External state is uncertain and requires recovery from a verified snapshot."
		)
	var had_request := pause_requested
	pause_requested = false
	_pause_reason = &""
	return _success({&"had_request": had_request})


func is_terminal() -> bool:
	return _terminal_state != TERMINAL_RUNNING


func get_terminal_state() -> StringName:
	return _terminal_state


func get_arrears_total() -> int:
	return _arrears_total


func get_consecutive_insolvency_count() -> int:
	return _consecutive_insolvency_count


func can_start_construction_or_investment() -> Dictionary:
	var allowed := _terminal_state == TERMINAL_RUNNING \
		and _arrears_total == 0 and not _external_quarantined
	var reason: StringName = &""
	if _terminal_state == TERMINAL_BANKRUPT:
		reason = &"BANKRUPT"
	elif _external_quarantined:
		reason = &"EXTERNAL_ROLLBACK_FAILED"
	elif _arrears_total > 0:
		reason = &"ARREARS_REMAIN"
	return {
		&"allowed": allowed,
		&"reason": reason,
		&"arrears_total": _arrears_total,
		&"consecutive_insolvency_count": _consecutive_insolvency_count,
	}


## Compact, defensive projection for the continuously visible operations HUD.
func get_dashboard_summary() -> Dictionary:
	var latest_day := -1
	for day: int in _day_summaries:
		latest_day = maxi(latest_day, day)
	var summaries: Array[Dictionary] = []
	if latest_day >= 0:
		var latest: Dictionary = _day_summaries[latest_day]
		summaries.append({&"day":latest[&"day"], &"net_profit":latest[&"net_profit"]})
	return {&"terminal_state":_terminal_state, &"consecutive_insolvency_count":_consecutive_insolvency_count,
		&"legacy_history_last_day":_legacy_history_last_day, &"day_summaries":summaries}


func get_day_summary(day: int) -> Dictionary:
	if not _day_summaries.has(day):
		return {}
	return (_day_summaries[day] as Dictionary).duplicate(true)


func get_day_summaries() -> Array[Dictionary]:
	var result: Array[Dictionary] = []
	for day: int in _sorted_int_keys(_day_summaries):
		result.append((_day_summaries[day] as Dictionary).duplicate(true))
	return result


func get_morning_report(day: int) -> Dictionary:
	if not _morning_reports.has(day):
		return {}
	return (_morning_reports[day] as Dictionary).duplicate(true)


func get_arrival_reports() -> Array[Dictionary]:
	return _arrival_reports.duplicate(true)


func get_demand_history() -> Array[Dictionary]:
	return _demand_history.duplicate(true)


func get_alerts() -> Array[Dictionary]:
	return _alerts.duplicate(true)


func get_arrival_carry(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName
) -> int:
	return int(_arrival_carry.get(_demand_key(facility_tag, guest_class, time_band), -1))


func deterministic_digest() -> String:
	return JSON.stringify(create_snapshot(), "", true, true).sha256_text()


func create_snapshot() -> Dictionary:
	if not _initialization_valid:
		last_error = "Operations coordinator initialization is invalid."
		return {}
	if _notification_delivery_deferred or not _notification_queue.is_empty():
		last_error = "Deferred Operations notifications must be resolved before snapshotting."
		return {}
	var field_cache: Dictionary = {}
	var internal_report := _snapshot_root_report({
		&"config": _config,
		&"demand_profiles": _demand_profiles,
		&"arrival_carry": _arrival_carry,
		&"demand_history": _demand_history,
		&"arrival_reports": _arrival_reports,
		&"day_summaries": _day_summaries,
		&"morning_reports": _morning_reports,
		&"alerts": _alerts,
		&"quarantine_report": _quarantine_report,
	}, field_cache, true)
	var shape_error := _internal_snapshot_shape_error()
	if not bool(internal_report[&"ok"]) or not shape_error.is_empty():
		last_error = "Operations state cannot be serialized: %s" % [
			internal_report[&"code"] if not bool(internal_report[&"ok"]) else shape_error
		]
		return {}
	var saved_profiles: Array[Dictionary] = []
	var saved_carry: Array[Dictionary] = []
	for key: String in _sorted_string_keys(_demand_profiles):
		var profile: Dictionary = _demand_profiles[key]
		saved_profiles.append({
			&"facility_tag": String(profile[&"facility_tag"]),
			&"guest_class": String(profile[&"guest_class"]),
			&"time_band": String(profile[&"time_band"]),
			&"components": _string_key_dictionary(profile[&"components"]),
			&"demand_index": int(profile[&"demand_index"]),
			&"last_evaluated_at": int(profile[&"last_evaluated_at"]),
		})
		saved_carry.append({
			&"facility_tag": String(profile[&"facility_tag"]),
			&"guest_class": String(profile[&"guest_class"]),
			&"time_band": String(profile[&"time_band"]),
			&"numerator": int(_arrival_carry.get(key, 0)),
		})

	var saved_mornings: Array[Dictionary] = []
	for day: int in _sorted_int_keys(_morning_reports):
		saved_mornings.append(_morning_reports[day])
	var saved_days: Array[Dictionary] = []
	for day: int in _sorted_int_keys(_day_summaries):
		saved_days.append(_day_summaries[day])

	# These borrowed containers never leave this synchronous method. The final
	# canonical conversion recursively creates owned dictionaries and arrays.
	var raw_snapshot := {
		&"schema_version": SNAPSHOT_VERSION,
		&"legacy_calendar_last_day": _legacy_calendar_last_day,
		&"legacy_history_last_day": _legacy_history_last_day,
		&"legacy_principal_profit_last_day": _legacy_principal_profit_last_day,
		&"origin_minute": _origin_minute,
		&"current_minute": current_minute,
		&"config": _string_key_dictionary(_config),
		&"demand_profiles": saved_profiles,
		&"arrival_carry": saved_carry,
		&"demand_history": _demand_history,
		&"arrival_reports": _arrival_reports,
		&"day_summaries": saved_days,
		&"morning_reports": saved_mornings,
		&"alerts": _alerts,
		&"processed_demand_minutes": _sorted_int_keys(_processed_demand_minutes),
		&"processed_arrival_minutes": _sorted_int_keys(_processed_arrival_minutes),
		&"population_count": _population_count,
		&"arrears_total": _arrears_total,
		&"consecutive_insolvency_count": _consecutive_insolvency_count,
		&"terminal_state": String(_terminal_state),
		&"pause_requested": pause_requested,
		&"next_arrival_plan_sequence": _next_arrival_plan_sequence,
		&"next_guest_sequence": _next_guest_sequence,
		&"arrival_plan_high_water": _arrival_plan_high_water,
		&"guest_high_water": _guest_high_water,
		&"fair_rotation_counter": _fair_rotation_counter,
		&"pending_boundary_minute": _pending_boundary_minute,
		&"deferred_boundary_minute": _deferred_boundary_minute,
		&"pause_reason": String(_pause_reason),
		&"external_quarantined": _external_quarantined,
		&"quarantine_report": _quarantine_report,
	}
	var tree_report := _snapshot_root_report(raw_snapshot, field_cache, false)
	if not bool(tree_report[&"ok"]):
		last_error = "Operations state cannot be serialized: %s" % tree_report[&"code"]
		return {}
	return tree_report[&"value"]


func restore_snapshot(snapshot: Dictionary) -> bool:
	if _is_advancing or _notification_delivery_deferred \
			or not _notification_queue.is_empty():
		last_error = "Operations snapshot restore cannot re-enter an advance."
		return false
	var parsed := _parse_snapshot(snapshot)
	if not bool(parsed.get(&"ok", false)):
		last_error = String(parsed.get(&"error", "Snapshot is invalid."))
		return false
	var state: Dictionary = parsed[&"state"]
	# Atomic swap begins only after the entire candidate passed validation.
	_legacy_calendar_last_day = int(state[&"legacy_calendar_last_day"])
	_legacy_history_last_day = int(state[&"legacy_history_last_day"])
	_legacy_principal_profit_last_day = int(state[&"legacy_principal_profit_last_day"])
	_origin_minute = int(state[&"origin_minute"])
	current_minute = int(state[&"current_minute"])
	_config = (state[&"config"] as Dictionary).duplicate(true)
	_demand_profiles = (state[&"demand_profiles"] as Dictionary).duplicate(true)
	_arrival_carry = (state[&"arrival_carry"] as Dictionary).duplicate(true)
	_demand_history = _copy_dictionary_array(state[&"demand_history"])
	_arrival_reports = _copy_dictionary_array(state[&"arrival_reports"])
	_day_summaries = (state[&"day_summaries"] as Dictionary).duplicate(true)
	_morning_reports = (state[&"morning_reports"] as Dictionary).duplicate(true)
	_alerts = _copy_dictionary_array(state[&"alerts"])
	_processed_demand_minutes = (state[&"processed_demand_minutes"] as Dictionary).duplicate(true)
	_processed_arrival_minutes = (state[&"processed_arrival_minutes"] as Dictionary).duplicate(true)
	_population_count = int(state[&"population_count"])
	_arrears_total = int(state[&"arrears_total"])
	_consecutive_insolvency_count = int(state[&"consecutive_insolvency_count"])
	_terminal_state = StringName(state[&"terminal_state"])
	pause_requested = bool(state[&"pause_requested"])
	_next_arrival_plan_sequence = int(state[&"next_arrival_plan_sequence"])
	_next_guest_sequence = int(state[&"next_guest_sequence"])
	_arrival_plan_high_water = int(state[&"arrival_plan_high_water"])
	_guest_high_water = int(state[&"guest_high_water"])
	_fair_rotation_counter = int(state[&"fair_rotation_counter"])
	_pending_boundary_minute = int(state[&"pending_boundary_minute"])
	_deferred_boundary_minute = int(state[&"deferred_boundary_minute"])
	_pause_reason = StringName(state[&"pause_reason"])
	_external_quarantined = bool(state[&"external_quarantined"])
	_quarantine_report = (state[&"quarantine_report"] as Dictionary).duplicate(true)
	_initialization_valid = true
	_notification_queue.clear()
	_notification_delivery_deferred = false
	last_error = ""
	return true


func validate_invariants() -> Dictionary:
	var errors: Array[Dictionary] = []
	if not _initialization_valid:
		errors.append({&"code": &"INITIALIZATION_INVALID"})
	if _origin_minute < 0 or _origin_minute > MAX_START_MINUTE:
		errors.append({&"code": &"INVALID_ORIGIN"})
	if current_minute < _origin_minute \
	or current_minute > _origin_minute + MAX_TRACKED_MINUTES:
		errors.append({&"code": &"NEGATIVE_TIME"})
	var normalized_config := _normalize_config(_config)
	if not bool(normalized_config[&"ok"]) or normalized_config[&"config"] != _config:
		errors.append({&"code": &"INVALID_CONFIG"})
	if _population_count < 0 or _population_count > int(_config[&"population_cap"]):
		errors.append({&"code": &"INVALID_POPULATION"})
	if _legacy_principal_profit_last_day < 0 or _legacy_principal_profit_last_day > _day_summaries.size():
		errors.append({&"code": &"INVALID_LEGACY_PRINCIPAL_PROFIT_MARKER"})
	if _arrears_total < 0 or _consecutive_insolvency_count < 0:
		errors.append({&"code": &"INVALID_INSOLVENCY_STATE"})
	if _terminal_state not in [TERMINAL_RUNNING, TERMINAL_BANKRUPT]:
		errors.append({&"code": &"INVALID_TERMINAL_STATE"})
	if _terminal_state == TERMINAL_BANKRUPT \
	and _consecutive_insolvency_count < int(_config[&"bankruptcy_close_count"]):
		errors.append({&"code": &"BANKRUPTCY_WITHOUT_THRESHOLD"})
	if _arrival_plan_high_water < 0 or _guest_high_water < 0 \
	or _arrival_plan_high_water > MAX_SEQUENCE_VALUE \
	or _guest_high_water > MAX_SEQUENCE_VALUE \
	or _next_arrival_plan_sequence != _arrival_plan_high_water + 1 \
	or _next_guest_sequence != _guest_high_water + 1:
		errors.append({&"code": &"INVALID_SEQUENCE"})
	if _fair_rotation_counter < 0 \
	or _fair_rotation_counter != _processed_arrival_minutes.size():
		errors.append({&"code": &"INVALID_FAIR_ROTATION_COUNTER"})
	if _pending_boundary_minute != -1 \
	and (_pending_boundary_minute != current_minute + 1 \
		or not _minute_has_scheduled_boundary(_pending_boundary_minute)):
		errors.append({&"code": &"INVALID_PENDING_BOUNDARY"})
	if _deferred_boundary_minute != -1 \
	and (_deferred_boundary_minute != current_minute \
		or not _minute_has_scheduled_boundary(_deferred_boundary_minute)):
		errors.append({&"code": &"INVALID_DEFERRED_BOUNDARY"})
	if _external_quarantined:
		if not pause_requested or _pause_reason != &"EXTERNAL_ROLLBACK_FAILED" \
		or _quarantine_report.is_empty():
			errors.append({&"code": &"INVALID_EXTERNAL_QUARANTINE"})
	elif not _quarantine_report.is_empty():
		errors.append({&"code": &"ORPHAN_QUARANTINE_REPORT"})
	if pause_requested and _pause_reason not in [&"LEDGER_IMBALANCE", &"EXTERNAL_ROLLBACK_FAILED"]:
		errors.append({&"code": &"INVALID_PAUSE_REASON"})
	if not pause_requested and not _pause_reason.is_empty():
		errors.append({&"code": &"ORPHAN_PAUSE_REASON"})

	for key: String in _sorted_string_keys(_demand_profiles):
		var profile_value: Variant = _demand_profiles[key]
		if not profile_value is Dictionary:
			errors.append({&"code": &"INVALID_DEMAND_PROFILE", &"key": key})
			continue
		var profile: Dictionary = profile_value
		var facility_tag := StringName(_value(profile, &"facility_tag", &""))
		var guest_class := StringName(_value(profile, &"guest_class", &""))
		var time_band := StringName(_value(profile, &"time_band", &""))
		if not _validate_demand_identity(facility_tag, guest_class, time_band).is_empty() \
		or key != _demand_key(facility_tag, guest_class, time_band):
			errors.append({&"code": &"INVALID_DEMAND_KEY", &"key": key})
			continue
		var components_value: Variant = _value(profile, &"components", null)
		if not components_value is Dictionary:
			errors.append({&"code": &"INVALID_DEMAND_COMPONENTS", &"key": key})
			continue
		var normalized_components := _normalize_demand_components_for_limit(
			components_value,
			int((normalized_config[&"config"] as Dictionary)[&"max_component_magnitude"])
		)
		if not bool(normalized_components[&"ok"]):
			errors.append({&"code": &"INVALID_DEMAND_COMPONENTS", &"key": key})
			continue
		if int(_value(profile, &"demand_index", -1)) \
		!= _calculate_demand_index(normalized_components[&"components"]):
			errors.append({&"code": &"DEMAND_INDEX_MISMATCH", &"key": key})
		var evaluated_at := int(_value(profile, &"last_evaluated_at", -2))
		var maximum_evaluated_minute := current_minute
		if _pending_boundary_minute == current_minute + 1:
			maximum_evaluated_minute = _pending_boundary_minute
		if evaluated_at < -1 or evaluated_at > maximum_evaluated_minute \
		or (evaluated_at >= 0 and evaluated_at % DEMAND_INTERVAL_MINUTES != 0):
			errors.append({&"code": &"INVALID_DEMAND_EVALUATION_TIME", &"key": key})
		if not _arrival_carry.has(key):
			errors.append({&"code": &"MISSING_ARRIVAL_CARRY", &"key": key})
		elif int(_arrival_carry[key]) < 0 or int(_arrival_carry[key]) > _max_carry_numerator():
			errors.append({&"code": &"INVALID_ARRIVAL_CARRY", &"key": key})
	for carry_key: Variant in _arrival_carry:
		if not _demand_profiles.has(carry_key):
			errors.append({&"code": &"ORPHAN_ARRIVAL_CARRY", &"key": carry_key})

	_validate_boundary_minute_set(
		_processed_demand_minutes,
		DEMAND_INTERVAL_MINUTES,
		&"INVALID_DEMAND_BOUNDARY",
		errors
	)
	_validate_boundary_minute_set(
		_processed_arrival_minutes,
		ARRIVAL_INTERVAL_MINUTES,
		&"INVALID_ARRIVAL_BOUNDARY",
		errors
	)
	_validate_histories(errors)
	_validate_financial_history(errors)
	_validate_boundary_coverage(errors)
	_validate_alerts_and_pause(errors)
	var candidate_snapshot := create_snapshot()
	if candidate_snapshot.is_empty() or not _variant_tree_report(candidate_snapshot)[&"ok"]:
		errors.append({&"code": &"UNSAVEABLE_VARIANT"})
	return {
		&"ok": errors.is_empty(),
		&"errors": errors,
		&"demand_profile_count": _demand_profiles.size(),
		&"arrival_boundary_count": _arrival_reports.size(),
		&"closed_day_count": _day_summaries.size(),
	}


func _process_minute_boundaries(minute: int, counts: Dictionary) -> CommandResult:
	# Migrated 06:00 financial periods retain their original payroll/history.
	# Calendar-midnight obligations still execute atomically at 00:00. Their
	# external receipt owns idempotency when a later boundary in this minute fails.
	if minute % MINUTES_PER_DAY == 0 and not _is_daily_close_boundary(minute) and _midnight_boundary.is_valid():
		var midnight := _run_boundary_transaction(_midnight_boundary, {&"boundary_minute": minute}, {&"ok": true})
		if not bool(midnight[&"ok"]):
			return _transaction_boundary_failure(midnight, &"MIDNIGHT_UPKEEP", minute, &"MIDNIGHT_UPKEEP_FAILED")
	# Finance precedes demand/arrivals at midnight. Morning work stays at 06:00.
	# During legacy migration, the final 06:00 close also precedes morning work.
	if _is_daily_close_boundary(minute):
		var close := _process_daily_close(minute)
		if not close.is_success():
			return close
		counts[&"daily_closes"] = int(counts[&"daily_closes"]) + int(not close.is_replay)
		if _terminal_state != TERMINAL_RUNNING:
			return _success({&"terminal": true})
		if pause_requested:
			# A ledger imbalance pauses before morning, demand and arrival work.
			return _success({&"deferred": true, &"after": &"DAILY_CLOSE"})

	if minute % MINUTES_PER_DAY == MORNING_MINUTE_OF_DAY:
		var morning := _process_morning_boundary(minute)
		if not morning.is_success():
			return morning
		counts[&"morning_boundaries"] = int(counts[&"morning_boundaries"]) + int(not morning.is_replay)
		if pause_requested:
			return _success({&"deferred": true, &"after": &"MORNING_BOUNDARY"})

	if minute % DEMAND_INTERVAL_MINUTES == 0:
		var demand := _process_demand_boundary(minute)
		if not demand.is_success():
			return demand
		counts[&"demand_updates"] = int(counts[&"demand_updates"]) + int(not demand.is_replay)
		if pause_requested:
			return _success({&"deferred": true, &"after": &"DEMAND_UPDATE"})

	if minute % ARRIVAL_INTERVAL_MINUTES == 0:
		var arrivals := _process_arrival_boundary(minute)
		if not arrivals.is_success():
			return arrivals
		counts[&"arrival_boundaries"] = int(counts[&"arrival_boundaries"]) + int(not arrivals.is_replay)
	return _success()


func _process_demand_boundary(minute: int) -> CommandResult:
	if _processed_demand_minutes.has(minute):
		var replay := _success({&"minute": minute})
		replay.is_replay = true
		return replay
	var values: Array[Dictionary] = []
	for key: String in _sorted_string_keys(_demand_profiles):
		var profile: Dictionary = _demand_profiles[key]
		var demand_index := _calculate_demand_index(profile[&"components"])
		profile[&"demand_index"] = demand_index
		profile[&"last_evaluated_at"] = minute
		values.append({
			&"facility_tag": String(profile[&"facility_tag"]),
			&"guest_class": String(profile[&"guest_class"]),
			&"time_band": String(profile[&"time_band"]),
			&"demand_index": demand_index,
		})
	var report := {&"minute": minute, &"values": values}
	_processed_demand_minutes[minute] = true
	_demand_history.append(report)
	_queue_notification(&"demand", [report.duplicate(true)])
	return _success(report)


func _process_arrival_boundary(minute: int) -> CommandResult:
	if _processed_arrival_minutes.has(minute):
		var replay := _success({&"minute": minute})
		replay.is_replay = true
		return replay
	var active_band := _time_band_for_minute(minute)
	var denominator := int(_config[&"arrival_rate_denominator"])
	var numerator_rate := int(_config[&"arrival_rate_numerator"])
	var carry_candidate := _arrival_carry.duplicate(true)
	var intent_rows: Array[Dictionary] = []
	for key: String in _sorted_string_keys(_demand_profiles):
		var profile: Dictionary = _demand_profiles[key]
		if profile[&"time_band"] not in [active_band, &"ANY"]:
			continue
		var raw := int(carry_candidate.get(key, 0)) \
		+ int(profile[&"demand_index"]) * numerator_rate
		var desired := raw / denominator
		intent_rows.append({
			&"key": key,
			&"facility_tag": profile[&"facility_tag"],
			&"guest_class": profile[&"guest_class"],
			&"time_band": profile[&"time_band"],
			&"demand_index": int(profile[&"demand_index"]),
			&"raw_numerator": raw,
			&"desired_count": desired,
		})
	var fair_start_index := 0
	if not intent_rows.is_empty():
		fair_start_index = posmod(
			int(_config[&"world_seed"]) + _fair_rotation_counter,
			intent_rows.size()
		)
		intent_rows = _rotated_dictionary_rows(intent_rows, fair_start_index)

	var local_capacity := int(_config[&"population_cap"]) - _population_count
	var capture_payload := {
		&"operation": &"ARRIVAL_PLAN",
		&"minute": minute,
		&"time_band": active_band,
		&"population_count": _population_count,
		&"population_cap": int(_config[&"population_cap"]),
		&"fair_rotation_counter": _fair_rotation_counter,
		&"fair_start_index": fair_start_index,
		&"intent_rows": intent_rows.duplicate(true),
	}
	var capture := _call_boundary(
		_arrival_boundary,
		capture_payload,
		PHASE_CAPTURE,
		null,
		{&"ok": true, &"token": null, &"capacity_remaining": local_capacity}
	)
	var token: Variant = capture.get(&"token", null)
	if not _boundary_result_succeeded(capture):
		return _failed_boundary_after_capture(
			_arrival_boundary, capture_payload, token,
			&"ARRIVAL_PLAN", minute, PHASE_CAPTURE, &"ARRIVAL_CAPTURE_FAILED"
		)
	var capacity_value: Variant = capture.get(&"capacity_remaining", null)
	if not _is_integral_value(capacity_value) or int(capacity_value) < 0:
		return _failed_boundary_after_capture(
			_arrival_boundary, capture_payload, token,
			&"ARRIVAL_PLAN", minute, PHASE_CAPTURE, &"ARRIVAL_CAPTURE_INVALID"
		)
	var external_capacity := int(capacity_value)
	# A13: the arrival boundary reports the weather share of planned arrivals
	# (rain 50, otherwise 100). Absent means 100 for callers without weather.
	var percent_value: Variant = capture.get(&"arrival_percent", 100)
	if not _is_integral_value(percent_value) or int(percent_value) < 0 or int(percent_value) > 100:
		return _failed_boundary_after_capture(
			_arrival_boundary, capture_payload, token,
			&"ARRIVAL_PLAN", minute, PHASE_CAPTURE, &"ARRIVAL_CAPTURE_INVALID"
		)
	var arrival_percent := int(percent_value)
	var remaining := mini(
		mini(local_capacity, external_capacity),
		int(_config[&"max_arrivals_per_interval"])
	)
	var plan_sequence := _next_arrival_plan_sequence
	var guest_sequence := _next_guest_sequence
	var plans: Array[Dictionary] = []
	var total_count := 0
	var capped_rows: Array[Dictionary] = []
	var accepted_by_row: Array[int] = []
	var planned_total := 0
	for row: Dictionary in intent_rows:
		var accepted := mini(int(row[&"desired_count"]), remaining)
		var key: String = row[&"key"]
		var retained := int(row[&"raw_numerator"]) - accepted * denominator
		var capped_retained := mini(retained, _max_carry_numerator())
		carry_candidate[key] = capped_retained
		if retained > capped_retained:
			capped_rows.append({
				&"facility_tag": String(row[&"facility_tag"]),
				&"guest_class": String(row[&"guest_class"]),
				&"discarded_numerator": retained - capped_retained,
			})
		accepted_by_row.append(accepted)
		planned_total += accepted
		remaining -= accepted
	# Rain keeps floor(planned * percent / 100) arrivals; the dropped arrivals
	# are lost to the weather and are not carried to the next boundary.
	var dropped := planned_total - planned_total * arrival_percent / 100
	var drop_index := accepted_by_row.size() - 1
	while dropped > 0 and drop_index >= 0:
		var cut := mini(dropped, accepted_by_row[drop_index])
		accepted_by_row[drop_index] -= cut
		dropped -= cut
		drop_index -= 1
	for row_index in intent_rows.size():
		var row: Dictionary = intent_rows[row_index]
		var accepted := accepted_by_row[row_index]
		if accepted <= 0:
			continue
		if plan_sequence > MAX_SEQUENCE_VALUE \
		or guest_sequence > MAX_SEQUENCE_VALUE - accepted + 1:
			return _failed_boundary_after_capture(
				_arrival_boundary, capture_payload, token,
				&"ARRIVAL_PLAN", minute, PHASE_CAPTURE, &"ID_SEQUENCE_EXHAUSTED"
			)
		var guest_ids: Array[String] = []
		for _index in accepted:
			guest_ids.append("guest-auto-%08d" % guest_sequence)
			guest_sequence += 1
		plans.append({
			&"plan_id": "arrival-plan-%08d" % plan_sequence,
			&"minute": minute,
			&"facility_tag": String(row[&"facility_tag"]),
			&"guest_class": String(row[&"guest_class"]),
			&"time_band": String(row[&"time_band"]),
			&"demand_index": int(row[&"demand_index"]),
			&"count": accepted,
			&"guest_ids": guest_ids,
		})
		plan_sequence += 1
		total_count += accepted

	var payload := capture_payload.duplicate(true)
	payload[&"plans"] = plans
	payload[&"total_count"] = total_count
	payload[&"idempotency_key"] = "arrival:%d" % minute
	var validated := _call_boundary(
		_arrival_boundary,
		payload,
		PHASE_VALIDATE,
		token,
		{&"ok": true, &"token": token}
	)
	if not _boundary_result_succeeded(validated):
		return _failed_boundary_after_capture(
			_arrival_boundary, payload, token,
			&"ARRIVAL_PLAN", minute, PHASE_VALIDATE, &"ARRIVAL_VALIDATE_FAILED"
		)
	token = validated.get(&"token", token)
	var applied := _call_boundary(
		_arrival_boundary,
		payload,
		PHASE_APPLY,
		token,
		{&"ok": true, &"accepted_count": total_count}
	)
	var accepted_value: Variant = applied.get(&"accepted_count", null)
	if not _boundary_result_succeeded(applied) \
	or not _is_integral_value(accepted_value) or int(accepted_value) != total_count:
		return _failed_boundary_after_capture(
			_arrival_boundary, payload, token,
			&"ARRIVAL_PLAN", minute, PHASE_APPLY, &"ARRIVAL_APPLY_FAILED"
		)

	_arrival_carry = carry_candidate
	_next_arrival_plan_sequence = plan_sequence
	_next_guest_sequence = guest_sequence
	_arrival_plan_high_water = plan_sequence - 1
	_guest_high_water = guest_sequence - 1
	_fair_rotation_counter += 1
	_population_count += total_count
	var report := {
		&"minute": minute,
		&"time_band": String(active_band),
		&"fair_rotation_counter": _fair_rotation_counter - 1,
		&"fair_start_index": fair_start_index,
		&"plans": plans,
		&"total_count": total_count,
		&"population_after": _population_count,
		&"capacity_remaining": int(_config[&"population_cap"]) - _population_count,
		&"carry_capped": capped_rows,
	}
	_processed_arrival_minutes[minute] = true
	_arrival_reports.append(report)
	if not capped_rows.is_empty():
		_append_alert({
			&"code": &"ARRIVAL_CARRY_CAPPED",
			&"minute": minute,
			&"rows": capped_rows,
		})
	_queue_notification(&"arrivals", [report.duplicate(true)])
	return _success(report)


func _process_daily_close(boundary_minute: int) -> CommandResult:
	var completed_day := _financial_day_for_boundary(boundary_minute)
	if _day_summaries.has(completed_day):
		var replay := _success({&"day": completed_day})
		replay.is_replay = true
		return replay
	var payload := {
		&"operation": &"DAILY_CLOSE",
		&"day": completed_day,
		&"boundary_minute": boundary_minute,
		&"period_start_minute": get_financial_period_start(completed_day),
		&"period_end_minute_exclusive": boundary_minute,
		&"idempotency_key": "daily-close:%d" % completed_day,
		&"prior_arrears_total": _arrears_total,
		&"steps": [
			&"FINALIZE_PENDING_SERVICES",
			&"AGGREGATE_LEDGER_CATEGORIES",
			&"RECONCILE_LEDGER",
			&"GENERATE_FACILITY_PROFIT_LOSS",
			&"DETECT_ANOMALIES",
			&"BUILD_DAY_SUMMARY",
			&"EVALUATE_LOSS_STREAK_AND_BANKRUPTCY",
		],
	}
	var transaction := _run_boundary_transaction(
		_daily_close_boundary,
		payload,
		_default_daily_apply_result()
	)
	if not bool(transaction[&"ok"]):
		return _transaction_boundary_failure(
			transaction, &"DAILY_CLOSE", boundary_minute, &"DAILY_CLOSE_FAILED"
		)
	var applied: Dictionary = transaction[&"result"]
	var normalized := _normalize_daily_result(applied)
	if not bool(normalized[&"ok"]):
		var rollback := _rollback_boundary_result(
			_daily_close_boundary,
			payload,
			transaction.get(&"token", null)
		)
		if not _boundary_result_succeeded(rollback):
			_enter_external_quarantine(
				&"DAILY_CLOSE", boundary_minute, &"RESULT_SCHEMA", &"DAILY_CLOSE_RESULT_INVALID"
			)
			return _boundary_failure(
				"EXTERNAL_ROLLBACK_FAILED",
				{&"operation": &"DAILY_CLOSE", &"phase": &"RESULT_SCHEMA"}
			)
		return _boundary_failure(
			"DAILY_CLOSE_RESULT_INVALID",
			{
				&"reason": normalized[&"error"],
				&"rollback_ok": true,
			}
		)

	var result: Dictionary = normalized[&"result"]
	var category_totals: Dictionary = result[&"category_totals"]
	var net_profit := _net_profit(category_totals)
	var previous_profit := net_profit
	if _day_summaries.has(completed_day - 1):
		previous_profit = int((_day_summaries[completed_day - 1] as Dictionary)[&"net_profit"])
	var rolling := _rolling_profit_with(net_profit)
	var arrears := int((result[&"settlement"] as Dictionary)[&"arrears_total"])
	var prior_streak := 0 if completed_day == _legacy_history_last_day + 1 else _consecutive_insolvency_count
	var consecutive := prior_streak + 1 if net_profit < 0 else 0
	var future_terminal := _terminal_state
	if consecutive >= int(_config[&"bankruptcy_close_count"]):
		future_terminal = TERMINAL_BANKRUPT
	var reconciliation_difference := int(result[&"reconciliation_difference"])
	var summary := {
		&"day": completed_day,
		&"period_start_minute": get_financial_period_start(completed_day),
		&"period_end_minute_exclusive": boundary_minute,
		&"closed_at": boundary_minute,
		&"category_totals": category_totals,
		&"net_profit": net_profit,
		&"previous_day_delta": net_profit - previous_profit,
		&"seven_day_average": rolling[&"quotient"],
		&"seven_day_average_detail": rolling,
		&"facility_profit_loss": result[&"facility_profit_loss"],
		&"anomalies": result[&"anomalies"],
		&"reconciliation_difference": reconciliation_difference,
		&"balanced": reconciliation_difference == 0,
		&"settlement": result[&"settlement"],
		&"arrears_total": arrears,
		&"consecutive_insolvency_count": consecutive,
		&"terminal_state": String(future_terminal),
	}

	# Commit coordinator state only after the external apply result and every
	# derived summary field have passed validation.
	_day_summaries[completed_day] = summary
	_arrears_total = arrears
	_consecutive_insolvency_count = consecutive
	if reconciliation_difference != 0:
		pause_requested = true
		_pause_reason = &"LEDGER_IMBALANCE"
		_append_alert({
			&"code": &"LEDGER_IMBALANCE",
			&"minute": boundary_minute,
			&"day": completed_day,
			&"difference": reconciliation_difference,
			&"automatic_correction": false,
		})
	if net_profit < 0:
		_append_alert({&"code": &"LOSS_STREAK", &"minute": boundary_minute,
			&"day": completed_day, &"net_profit": net_profit, &"consecutive_count": consecutive})
	if arrears > 0:
		_append_alert({
			&"code": &"ARREARS_REMAIN",
			&"minute": boundary_minute,
			&"day": completed_day,
			&"arrears_total": arrears,
			&"consecutive_count": consecutive,
		})
	if future_terminal != _terminal_state:
		var old_state := _terminal_state
		_terminal_state = future_terminal
		_append_alert({
			&"code": &"BANKRUPT",
			&"minute": boundary_minute,
			&"day": completed_day,
			&"arrears_total": arrears,
			&"consecutive_count": consecutive,
		})
		_queue_notification(
			&"terminal",
			[old_state, _terminal_state, &"CONSECUTIVE_LOSSES"]
		)
	_queue_notification(&"daily", [summary.duplicate(true)])
	return _success(summary)


func _process_morning_boundary(boundary_minute: int) -> CommandResult:
	var day := boundary_minute / MINUTES_PER_DAY + 1
	if _morning_reports.has(day):
		var replay := _success({&"day": day})
		replay.is_replay = true
		return replay
	var payload := {
		&"operation": &"MORNING_BOUNDARY",
		&"day": day,
		&"boundary_minute": boundary_minute,
		&"idempotency_key": "morning:%d" % day,
		&"steps": [
			&"START_DAY_SHIFT",
			&"RESTORE_PERSONAL_ACTION_POINTS",
			&"ROLL_INSPECTION",
			&"RESOLVE_MORNING_EVENTS",
		],
	}
	var transaction := _run_boundary_transaction(
		_morning_boundary,
		payload,
		{&"ok": true, &"event_payload": {}}
	)
	if not bool(transaction[&"ok"]):
		return _transaction_boundary_failure(
			transaction, &"MORNING_BOUNDARY", boundary_minute, &"MORNING_BOUNDARY_FAILED"
		)
	var applied: Dictionary = transaction[&"result"]
	var event_payload: Variant = applied.get(&"event_payload", {})
	if not event_payload is Dictionary or not _variant_tree_report(event_payload)[&"ok"]:
		var rollback := _rollback_boundary_result(
			_morning_boundary,
			payload,
			transaction.get(&"token", null)
		)
		if not _boundary_result_succeeded(rollback):
			_enter_external_quarantine(
				&"MORNING_BOUNDARY", boundary_minute, &"RESULT_SCHEMA", &"MORNING_RESULT_INVALID"
			)
			return _boundary_failure(
				"EXTERNAL_ROLLBACK_FAILED",
				{&"operation": &"MORNING_BOUNDARY", &"phase": &"RESULT_SCHEMA"}
			)
		return _boundary_failure(
			"MORNING_RESULT_INVALID",
			{&"rollback_ok": true}
		)
	var report := {
		&"day": day,
		&"minute": boundary_minute,
		&"event_payload": (event_payload as Dictionary).duplicate(true),
	}
	_morning_reports[day] = report
	_queue_notification(&"morning", [report.duplicate(true)])
	return _success(report)


func _run_boundary_transaction(
	callback: Callable,
	payload: Dictionary,
	default_apply_result: Dictionary
) -> Dictionary:
	var capture := _call_boundary(
		callback,
		payload,
		PHASE_CAPTURE,
		null,
		{&"ok": true, &"token": null}
	)
	var token: Variant = capture.get(&"token", null)
	if not _boundary_result_succeeded(capture):
		return _transaction_failure(callback, payload, token, PHASE_CAPTURE, capture)
	var validation := _call_boundary(
		callback,
		payload,
		PHASE_VALIDATE,
		token,
		{&"ok": true, &"token": token}
	)
	if not _boundary_result_succeeded(validation):
		return _transaction_failure(callback, payload, token, PHASE_VALIDATE, validation)
	token = validation.get(&"token", token)
	var applied := _call_boundary(
		callback,
		payload,
		PHASE_APPLY,
		token,
		default_apply_result
	)
	if not _boundary_result_succeeded(applied):
		return _transaction_failure(callback, payload, token, PHASE_APPLY, applied)
	return {&"ok": true, &"result": applied, &"token": token}


func _transaction_failure(
	callback: Callable,
	payload: Dictionary,
	token: Variant,
	phase: StringName,
	result: Dictionary
) -> Dictionary:
	var rollback := _rollback_boundary_result(callback, payload, token)
	return {
		&"ok": false,
		&"phase": phase,
		&"result": result,
		&"token": token,
		&"rollback_ok": _boundary_result_succeeded(rollback),
	}


func _rollback_boundary_result(callback: Callable, payload: Dictionary, token: Variant) -> Dictionary:
	return _call_boundary(
		callback,
		payload,
		PHASE_ROLLBACK,
		token,
		{&"ok": true}
	)


func _transaction_boundary_failure(
	transaction: Dictionary,
	operation: StringName,
	minute: int,
	failure_code: StringName
) -> CommandResult:
	if not bool(transaction.get(&"rollback_ok", false)):
		_enter_external_quarantine(
			operation,
			minute,
			StringName(transaction.get(&"phase", &"UNKNOWN")),
			failure_code
		)
		return _boundary_failure(
			"EXTERNAL_ROLLBACK_FAILED",
			{&"operation": operation, &"phase": transaction.get(&"phase", &"UNKNOWN")}
		)
	return _boundary_failure(failure_code, {
		&"operation": operation,
		&"phase": transaction.get(&"phase", &"UNKNOWN"),
		&"rollback_ok": true,
	})


func _failed_boundary_after_capture(
	callback: Callable,
	payload: Dictionary,
	token: Variant,
	operation: StringName,
	minute: int,
	phase: StringName,
	failure_code: StringName
) -> CommandResult:
	var rollback := _rollback_boundary_result(callback, payload, token)
	if not _boundary_result_succeeded(rollback):
		_enter_external_quarantine(operation, minute, phase, failure_code)
		return _boundary_failure(
			"EXTERNAL_ROLLBACK_FAILED",
			{&"operation": operation, &"phase": phase}
		)
	return _boundary_failure(failure_code, {
		&"operation": operation,
		&"phase": phase,
		&"rollback_ok": true,
	})


func _enter_external_quarantine(
	operation: StringName,
	minute: int,
	failed_phase: StringName,
	original_code: StringName
) -> void:
	if _external_quarantined:
		return
	_external_quarantined = true
	pause_requested = true
	_pause_reason = &"EXTERNAL_ROLLBACK_FAILED"
	_quarantine_report = {
		&"code": &"EXTERNAL_ROLLBACK_FAILED",
		&"operation": operation,
		&"minute": minute,
		&"failed_phase": failed_phase,
		&"original_code": original_code,
		&"recovery_required": true,
	}
	_append_alert(_quarantine_report)


static func _boundary_result_succeeded(result: Dictionary) -> bool:
	var ok_value: Variant = result.get(&"ok", result.get("ok", null))
	return typeof(ok_value) == TYPE_BOOL and bool(ok_value)


func _call_boundary(
	callback: Callable,
	payload: Dictionary,
	phase: StringName,
	token: Variant,
	default_result: Dictionary
) -> Dictionary:
	var payload_report := _variant_tree_report(payload)
	if not bool(payload_report[&"ok"]):
		return {&"ok": false, &"reason": payload_report[&"code"]}
	if not callback.is_valid():
		return default_result.duplicate(true)
	var value: Variant = callback.call(payload.duplicate(true), phase, token)
	if not value is Dictionary:
		return {&"ok": false, &"reason": &"NON_DICTIONARY_RESULT"}
	var result: Dictionary = value
	if not _canonical_key_collision(result).is_empty():
		return {&"ok": false, &"reason": &"DUPLICATE_CANONICAL_KEY"}
	# The rollback token is intentionally ephemeral and may be an external
	# snapshot handle. Every other returned value must obey payload budgets.
	var inspectable := result.duplicate(false)
	inspectable.erase(&"token")
	inspectable.erase("token")
	var report := _variant_tree_report(inspectable)
	if not bool(report[&"ok"]):
		return {&"ok": false, &"reason": report[&"code"]}
	if int(report[&"nodes"]) > MAX_CALLBACK_VARIANT_NODES \
	or int(report[&"string_length"]) > MAX_CALLBACK_STRING_LENGTH:
		return {&"ok": false, &"reason": &"CALLBACK_PAYLOAD_BUDGET"}
	return result.duplicate(true)


func _normalize_daily_result(source: Dictionary) -> Dictionary:
	if not _canonical_key_collision(source).is_empty():
		return {&"ok": false, &"error": "Daily result has duplicate canonical keys."}
	var categories_value: Variant = _value(source, &"category_totals", null)
	var facilities_value: Variant = _value(source, &"facility_profit_loss", null)
	var anomalies_value: Variant = _value(source, &"anomalies", null)
	var settlement_value: Variant = _value(source, &"settlement", null)
	if not categories_value is Dictionary \
	or not facilities_value is Dictionary \
	or not anomalies_value is Array \
	or not settlement_value is Dictionary:
		return {&"ok": false, &"error": "Daily close result schema is incomplete."}
	var categories := _normalize_categories(categories_value)
	if not bool(categories[&"ok"]):
		return categories
	var facilities := _normalize_facility_profit_loss(facilities_value)
	if not bool(facilities[&"ok"]):
		return facilities
	if not _variant_tree_report(anomalies_value)[&"ok"]:
		return {&"ok": false, &"error": "Daily anomalies contain an unsaveable value."}
	for anomaly: Variant in anomalies_value:
		if not anomaly is Dictionary:
			return {&"ok": false, &"error": "Daily anomaly rows must be dictionaries."}
	var difference_value: Variant = _value(source, &"reconciliation_difference", null)
	if not _is_integral_value(difference_value):
		return {&"ok": false, &"error": "Reconciliation difference must be an integer."}
	if int(difference_value) < -MAX_FINANCIAL_AMOUNT \
	or int(difference_value) > MAX_FINANCIAL_AMOUNT:
		return {&"ok": false, &"error": "Reconciliation difference exceeds checked limits."}
	var arrears_value: Variant = _value(settlement_value, &"arrears_total", null)
	if not _is_integral_value(arrears_value) or int(arrears_value) < 0 \
	or int(arrears_value) > MAX_FINANCIAL_AMOUNT:
		return {&"ok": false, &"error": "Settlement arrears must be a non-negative integer."}
	if not _variant_tree_report(settlement_value)[&"ok"]:
		return {&"ok": false, &"error": "Settlement contains an unsaveable value."}
	var settlement := (settlement_value as Dictionary).duplicate(true)
	settlement[&"arrears_total"] = int(arrears_value)
	return {
		&"ok": true,
		&"result": {
			&"category_totals": categories[&"categories"],
			&"facility_profit_loss": facilities[&"facilities"],
			&"anomalies": (anomalies_value as Array).duplicate(true),
			&"reconciliation_difference": int(difference_value),
			&"settlement": settlement,
		},
	}


func _normalize_categories(source: Dictionary) -> Dictionary:
	var collision := _canonical_key_collision(source)
	if not collision.is_empty():
		return {&"ok": false, &"error": "Daily categories contain duplicate canonical keys."}
	for raw_key: Variant in source:
		if typeof(raw_key) not in [TYPE_STRING, TYPE_STRING_NAME] \
		or (StringName(str(raw_key)) not in DAILY_CATEGORIES and StringName(str(raw_key)) not in FINANCING_CATEGORIES):
			return {
				&"ok": false,
				&"error": "Unknown financial category requires explicit classification: %s" % raw_key,
			}
	var result: Dictionary = {}
	for category: StringName in DAILY_CATEGORIES:
		var value: Variant = _value(source, category, 0)
		if not _is_integral_value(value) or int(value) < 0 \
		or int(value) > MAX_FINANCIAL_AMOUNT:
			return {
				&"ok": false,
				&"error": "Daily category '%s' must be a non-negative integer." % category,
			}
		result[category] = int(value)
	for category: StringName in FINANCING_CATEGORIES:
		if not source.has(category) and not source.has(String(category)):
			continue
		var value: Variant = _value(source, category, null)
		if not _is_integral_value(value) or int(value) < 0 or int(value) > MAX_FINANCIAL_AMOUNT:
			return {&"ok": false, &"error": "Principal financing totals must be non-negative checked integers."}
		result[category] = int(value)
	return {&"ok": true, &"categories": result}


func _normalize_facility_profit_loss(source: Dictionary) -> Dictionary:
	var result: Dictionary = {}
	var facility_ids: Array = source.keys()
	facility_ids.sort_custom(_variant_string_less)
	for raw_id: Variant in facility_ids:
		if typeof(raw_id) not in [TYPE_STRING, TYPE_STRING_NAME]:
			return {&"ok": false, &"error": "Facility P/L IDs must be strings."}
		var facility_id := StringName(str(raw_id))
		var row_value: Variant = source[raw_id]
		if facility_id.is_empty() or not row_value is Dictionary:
			return {&"ok": false, &"error": "Facility P/L row is malformed."}
		var row_source: Dictionary = row_value
		if not _canonical_key_collision(row_source).is_empty():
			return {&"ok": false, &"error": "Facility P/L row has duplicate canonical keys."}
		for row_key: Variant in row_source:
			if StringName(row_key) not in [&"revenue", &"expenses", &"net_profit"]:
				return {&"ok": false, &"error": "Facility P/L row contains an unknown field."}
		var revenue_value: Variant = _value(row_source, &"revenue", null)
		var expense_value: Variant = _value(row_source, &"expenses", null)
		if not _is_integral_value(revenue_value) or int(revenue_value) < 0 \
		or int(revenue_value) > MAX_FINANCIAL_AMOUNT \
		or not _is_integral_value(expense_value) or int(expense_value) < 0 \
		or int(expense_value) > MAX_FINANCIAL_AMOUNT:
			return {&"ok": false, &"error": "Facility P/L values must be non-negative integers."}
		var expected_net := int(revenue_value) - int(expense_value)
		var supplied_net: Variant = _value(row_source, &"net_profit", expected_net)
		if not _is_integral_value(supplied_net) or int(supplied_net) != expected_net:
			return {&"ok": false, &"error": "Facility P/L net does not reconcile."}
		result[facility_id] = {
			&"revenue": int(revenue_value),
			&"expenses": int(expense_value),
			&"net_profit": expected_net,
		}
	return {&"ok": true, &"facilities": result}


func _default_daily_apply_result() -> Dictionary:
	return {
		&"ok": true,
		&"category_totals": {},
		&"reconciliation_difference": 0,
		&"facility_profit_loss": {},
		&"anomalies": [],
		&"settlement": {&"arrears_total": _arrears_total},
	}


func _net_profit(categories: Dictionary) -> int:
	# Original daily expenses include debt repayment; borrowed principal is not sales.
	return _principal_excluded_profit(categories) - int(categories.get(&"financing_principal_outflows", 0))


func _principal_excluded_profit(categories: Dictionary) -> int:
	return int(categories.get(&"revenue", 0)) \
	- int(categories.get(&"refunds", 0)) \
	- int(categories.get(&"cost_of_goods", 0)) \
	- int(categories.get(&"wages", 0)) \
	- int(categories.get(&"maintenance", 0)) \
	- int(categories.get(&"utilities", 0)) \
	- int(categories.get(&"taxes_and_interest", 0)) \
	- int(categories.get(&"fines", 0))


func _rolling_profit_with(current_profit: int) -> Dictionary:
	var values: Array[int] = []
	var days := _sorted_int_keys(_day_summaries)
	var start := maxi(days.size() - 6, 0)
	for index in range(start, days.size()):
		values.append(int((_day_summaries[days[index]] as Dictionary)[&"net_profit"]))
	values.append(current_profit)
	var total := 0
	for value: int in values:
		total += value
	return {
		&"sum": total,
		&"count": values.size(),
		&"quotient": total / values.size(),
		&"remainder": total % values.size(),
	}


func _append_alert(alert: Dictionary) -> void:
	var safe := alert.duplicate(true)
	_alerts.append(safe)
	_queue_notification(&"alert", [safe.duplicate(true)])


func _queue_notification(kind: StringName, arguments: Array) -> void:
	_notification_queue.append({&"kind": kind, &"arguments": arguments.duplicate(true)})


func _flush_notifications() -> void:
	var queued := _notification_queue.duplicate(true)
	_notification_queue.clear()
	for item: Dictionary in queued:
		var arguments: Array = item[&"arguments"]
		match StringName(item[&"kind"]):
			&"demand":
				demand_updated.emit(arguments[0])
			&"arrivals":
				arrivals_committed.emit(arguments[0])
			&"daily":
				daily_close_completed.emit(arguments[0])
			&"morning":
				morning_boundary_completed.emit(arguments[0])
			&"alert":
				operations_alerted.emit(arguments[0])
			&"terminal":
				terminal_state_changed.emit(arguments[0], arguments[1], arguments[2])


func _normalize_config(overrides: Dictionary) -> Dictionary:
	if not _canonical_key_collision(overrides).is_empty():
		return {&"ok": false, &"error": "Operations config has duplicate canonical keys."}
	var result := DEFAULT_CONFIG.duplicate(true)
	for raw_key: Variant in overrides:
		if typeof(raw_key) not in [TYPE_STRING, TYPE_STRING_NAME]:
			return {&"ok": false, &"error": "Operations config keys must be strings."}
		var key := StringName(str(raw_key))
		if not result.has(key):
			return {&"ok": false, &"error": "Unknown operations config key: %s" % key}
		var value: Variant = overrides[raw_key]
		if not _is_integral_value(value):
			return {&"ok": false, &"error": "Operations config values must be integers."}
		result[key] = int(value)
	if int(result[&"world_seed"]) < 0:
		return {&"ok": false, &"error": "World seed must be non-negative."}
	if int(result[&"world_seed"]) > MAX_WORLD_SEED:
		return {&"ok": false, &"error": "World seed exceeds the supported deterministic range."}
	for positive_key: StringName in [
		&"population_cap",
		&"max_arrivals_per_interval",
		&"arrival_rate_numerator",
		&"arrival_rate_denominator",
		&"arrival_carry_interval_limit",
		&"max_component_magnitude",
		&"bankruptcy_close_count",
	]:
		if int(result[positive_key]) <= 0:
			return {&"ok": false, &"error": "Operations config requires positive limits."}
	if int(result[&"max_arrivals_per_interval"]) > int(result[&"population_cap"]):
		return {&"ok": false, &"error": "Interval arrival cap cannot exceed population cap."}
	if int(result[&"bankruptcy_close_count"]) > 100:
		return {&"ok": false, &"error": "Bankruptcy threshold is outside a safe range."}
	if int(result[&"population_cap"]) > 1_000 \
	or int(result[&"max_arrivals_per_interval"]) > 100 \
	or int(result[&"arrival_rate_numerator"]) > 1_000_000 \
	or int(result[&"arrival_rate_denominator"]) > 1_000_000 \
	or int(result[&"arrival_carry_interval_limit"]) > MAX_TRACKED_DAYS \
	or int(result[&"max_component_magnitude"]) > 1_000_000:
		return {&"ok": false, &"error": "Operations config exceeds checked-arithmetic limits."}
	return {&"ok": true, &"config": result}


func _normalize_demand_components(source: Dictionary) -> Dictionary:
	return _normalize_demand_components_for_limit(
		source,
		int(_config[&"max_component_magnitude"])
	)


func _normalize_demand_components_for_limit(source: Dictionary, magnitude_limit: int) -> Dictionary:
	if not _canonical_key_collision(source).is_empty():
		return {&"ok": false, &"error": "Demand input has duplicate canonical keys."}
	if source.size() != DEMAND_COMPONENTS.size():
		return {&"ok": false, &"error": "Demand input must contain exactly six components."}
	var result: Dictionary = {}
	for component: StringName in DEMAND_COMPONENTS:
		if not source.has(component) and not source.has(String(component)):
			return {&"ok": false, &"error": "Demand component is missing: %s" % component}
		var value: Variant = _value(source, component, null)
		if not _is_finite_numeric(value):
			return {&"ok": false, &"error": "Demand component is not finite: %s" % component}
		if float(value) < -float(magnitude_limit) or float(value) > float(magnitude_limit):
			return {&"ok": false, &"error": "Demand component exceeds the configured bound."}
		var normalized_value := int(round(float(value)))
		if normalized_value < -magnitude_limit or normalized_value > magnitude_limit:
			return {&"ok": false, &"error": "Demand component exceeds the configured bound."}
		if component in [&"base", &"unmet_need", &"event", &"recent_price_penalty", &"wait_penalty"] \
		and normalized_value < 0:
			return {&"ok": false, &"error": "Unsigned demand component cannot be negative."}
		result[component] = normalized_value
	return {&"ok": true, &"components": result}


func _calculate_demand_index(components: Dictionary) -> int:
	return clampi(
		int(components[&"base"])
		+ int(components[&"reputation"])
		+ int(components[&"unmet_need"])
		+ int(components[&"event"])
		- int(components[&"recent_price_penalty"])
		- int(components[&"wait_penalty"]),
		0,
		200
	)


func _validate_demand_identity(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName
) -> String:
	if facility_tag.is_empty() or guest_class.is_empty() or time_band.is_empty():
		return "Demand identity fields cannot be empty."
	if time_band not in TIME_BANDS:
		return "Demand time band is not recognized."
	if KEY_SEPARATOR in String(facility_tag) or KEY_SEPARATOR in String(guest_class):
		return "Demand identity contains a reserved separator."
	if String(facility_tag).length() > 128 or String(guest_class).length() > 128:
		return "Demand identity is too long."
	return ""


func _demand_key(
	facility_tag: StringName,
	guest_class: StringName,
	time_band: StringName
) -> String:
	return "%s%s%s%s%s" % [facility_tag, KEY_SEPARATOR, guest_class, KEY_SEPARATOR, time_band]


func _time_band_for_minute(absolute_minute: int) -> StringName:
	var minute_of_day := posmod(absolute_minute, MINUTES_PER_DAY)
	if minute_of_day >= 360 and minute_of_day < 600:
		return &"MORNING"
	if minute_of_day >= 600 and minute_of_day < 960:
		return &"DAY"
	if minute_of_day >= 960 and minute_of_day < 1_200:
		return &"EVENING"
	return &"NIGHT"


func _max_carry_numerator() -> int:
	return int(_config[&"arrival_rate_denominator"]) \
	* int(_config[&"max_arrivals_per_interval"]) \
	* int(_config[&"arrival_carry_interval_limit"])


func _dictionary_has_exact_string_keys(
	source: Dictionary,
	expected: Array[StringName]
) -> bool:
	if source.size() != expected.size():
		return false
	var seen: Dictionary = {}
	for raw_key: Variant in source:
		if typeof(raw_key) not in [TYPE_STRING, TYPE_STRING_NAME]:
			return false
		var key := StringName(raw_key)
		if not expected.has(key) or seen.has(key):
			return false
		seen[key] = true
	return seen.size() == expected.size()


func _current_snapshot_exact_shape_error(snapshot: Dictionary) -> String:
	var top_keys: Array[StringName] = [
		&"schema_version", &"origin_minute", &"current_minute", &"config",
		&"demand_profiles", &"arrival_carry", &"demand_history",
		&"arrival_reports", &"day_summaries", &"morning_reports", &"alerts",
		&"processed_demand_minutes", &"processed_arrival_minutes",
		&"population_count", &"arrears_total", &"consecutive_insolvency_count",
		&"terminal_state", &"pause_requested", &"next_arrival_plan_sequence",
		&"next_guest_sequence", &"arrival_plan_high_water", &"guest_high_water",
		&"fair_rotation_counter", &"pending_boundary_minute",
		&"deferred_boundary_minute", &"pause_reason", &"external_quarantined",
		&"quarantine_report",
	]
	if int(_value(snapshot, &"schema_version", 0)) >= 4:
		top_keys.append_array([&"legacy_calendar_last_day", &"legacy_history_last_day"])
	if int(_value(snapshot, &"schema_version", 0)) >= 5:
		top_keys.append(&"legacy_principal_profit_last_day")
	if not _dictionary_has_exact_string_keys(snapshot, top_keys):
		return "Operations snapshot has missing or unknown root fields."

	var profiles: Variant = _value(snapshot, &"demand_profiles", null)
	var carry: Variant = _value(snapshot, &"arrival_carry", null)
	var demand_history: Variant = _value(snapshot, &"demand_history", null)
	var arrivals: Variant = _value(snapshot, &"arrival_reports", null)
	var days: Variant = _value(snapshot, &"day_summaries", null)
	var mornings: Variant = _value(snapshot, &"morning_reports", null)
	if not profiles is Array or not carry is Array or not demand_history is Array \
			or not arrivals is Array or not days is Array or not mornings is Array:
		return "Operations snapshot history collections are malformed."
	for raw_profile: Variant in profiles:
		if not raw_profile is Dictionary or not _dictionary_has_exact_string_keys(
			raw_profile,
			[&"facility_tag", &"guest_class", &"time_band", &"components",
			 &"demand_index", &"last_evaluated_at"]
		):
			return "Operations demand profile has missing or unknown fields."
	for raw_carry: Variant in carry:
		if not raw_carry is Dictionary or not _dictionary_has_exact_string_keys(
			raw_carry, [&"facility_tag", &"guest_class", &"time_band", &"numerator"]
		):
			return "Operations arrival carry row has missing or unknown fields."
	for raw_report: Variant in demand_history:
		if not raw_report is Dictionary or not _dictionary_has_exact_string_keys(
			raw_report, [&"minute", &"values"]
		):
			return "Operations demand history row has missing or unknown fields."
		var values: Variant = (raw_report as Dictionary).get(&"values", null)
		if not values is Array:
			return "Operations demand history values are malformed."
		for raw_value: Variant in values:
			if not raw_value is Dictionary or not _dictionary_has_exact_string_keys(
				raw_value,
				[&"facility_tag", &"guest_class", &"time_band", &"demand_index"]
			):
				return "Operations demand history value has missing or unknown fields."
	for raw_report: Variant in arrivals:
		if not raw_report is Dictionary or not _dictionary_has_exact_string_keys(
			raw_report,
			[&"minute", &"time_band", &"fair_rotation_counter", &"fair_start_index",
			 &"plans", &"total_count", &"population_after", &"capacity_remaining",
			 &"carry_capped"]
		):
			return "Operations arrival report has missing or unknown fields."
		var plans: Variant = (raw_report as Dictionary).get(&"plans", null)
		var capped: Variant = (raw_report as Dictionary).get(&"carry_capped", null)
		if not plans is Array or not capped is Array:
			return "Operations arrival report collections are malformed."
		for raw_plan: Variant in plans:
			if not raw_plan is Dictionary or not _dictionary_has_exact_string_keys(
				raw_plan,
				[&"plan_id", &"minute", &"facility_tag", &"guest_class", &"time_band",
				 &"demand_index", &"count", &"guest_ids"]
			):
				return "Operations arrival plan has missing or unknown fields."
		for raw_capped: Variant in capped:
			if not raw_capped is Dictionary or not _dictionary_has_exact_string_keys(
				raw_capped, [&"facility_tag", &"guest_class", &"discarded_numerator"]
			):
				return "Operations v3 capped carry row has missing or unknown fields."
	for raw_day: Variant in days:
		if not raw_day is Dictionary or not _dictionary_has_exact_string_keys(
			raw_day,
			[&"day", &"period_start_minute", &"period_end_minute_exclusive",
			 &"closed_at", &"category_totals", &"net_profit", &"previous_day_delta",
			 &"seven_day_average", &"seven_day_average_detail", &"facility_profit_loss",
			 &"anomalies", &"reconciliation_difference", &"balanced", &"settlement",
			 &"arrears_total", &"consecutive_insolvency_count", &"terminal_state"]
		):
			return "Operations v3 DaySummary has missing or unknown fields."
		var detail: Variant = (raw_day as Dictionary).get(&"seven_day_average_detail", null)
		if not detail is Dictionary or not _dictionary_has_exact_string_keys(
			detail, [&"sum", &"count", &"quotient", &"remainder"]
		):
			return "Operations v3 rolling average evidence has missing or unknown fields."
		var anomalies: Variant = (raw_day as Dictionary).get(&"anomalies", null)
		if not anomalies is Array:
			return "Operations v3 DaySummary anomalies are malformed."
		for raw_anomaly: Variant in anomalies:
			if not raw_anomaly is Dictionary:
				return "Operations v3 DaySummary anomaly row is malformed."
			var anomaly: Dictionary = raw_anomaly
			var code_value: Variant = anomaly.get(&"code", null)
			if typeof(code_value) not in [TYPE_STRING, TYPE_STRING_NAME]:
				return "Operations v3 DaySummary anomaly code is malformed."
			match StringName(code_value):
				&"WAGE_ALLOCATION_EVIDENCE":
					if not _dictionary_has_exact_string_keys(
						anomaly,
						[&"code", &"version", &"new_wages_total", &"allocated_total",
						 &"unallocated_total", &"allocations", &"unallocated_by_staff",
						 &"facility_ids"]
					) or typeof(anomaly.get(&"version", null)) != TYPE_INT \
							or int(anomaly[&"version"]) != 1 \
							or typeof(anomaly.get(&"new_wages_total", null)) != TYPE_INT \
							or typeof(anomaly.get(&"allocated_total", null)) != TYPE_INT \
							or typeof(anomaly.get(&"unallocated_total", null)) != TYPE_INT \
							or not anomaly.get(&"allocations", null) is Array \
							or not anomaly.get(&"unallocated_by_staff", null) is Array \
							or not anomaly.get(&"facility_ids", null) is Array:
						return "Operations v3 wage allocation evidence is malformed."
					for raw_row: Variant in anomaly[&"allocations"]:
						if not raw_row is Dictionary or not _dictionary_has_exact_string_keys(
							raw_row, [&"facility_id", &"amount"]
						):
							return "Operations v3 wage allocation row is malformed."
					for raw_row: Variant in anomaly[&"unallocated_by_staff"]:
						if not raw_row is Dictionary or not _dictionary_has_exact_string_keys(
							raw_row, [&"staff_id", &"amount"]
						):
							return "Operations v3 unallocated wage evidence is malformed."
				&"UNALLOCATED_WAGES":
					if not _dictionary_has_exact_string_keys(
						anomaly, [&"code", &"staff_id", &"amount"]
					):
						return "Operations v3 unallocated wage anomaly is malformed."
				&"OPERATING_REFUND_EXCEEDS_PERIOD_COST":
					if not _dictionary_has_exact_string_keys(
						anomaly,
						[&"code", &"refund_total", &"cost_offset", &"excess_revenue"]
					):
						return "Operations v3 operating refund anomaly is malformed."
				_:
					return "Operations v3 DaySummary anomaly code is unknown."
	for raw_morning: Variant in mornings:
		if not raw_morning is Dictionary or not _dictionary_has_exact_string_keys(
			raw_morning, [&"day", &"minute", &"event_payload"]
		):
			return "Operations v3 morning report has missing or unknown fields."
	return ""


func _parse_snapshot(snapshot: Dictionary) -> Dictionary:
	var tree_report := _variant_tree_report(snapshot)
	if not bool(tree_report[&"ok"]):
		return {
			&"ok": false,
			&"error": "Operations snapshot Variant tree is invalid: %s" % tree_report[&"code"],
		}
	var version_value: Variant = _value(snapshot, &"schema_version", null)
	if not _is_integral_value(version_value):
		return {&"ok": false, &"error": "Operations snapshot version must be an integer."}
	var version := int(version_value)
	if version < MIN_SUPPORTED_SNAPSHOT_VERSION or version > SNAPSHOT_VERSION:
		return {&"ok": false, &"error": "Unsupported operations snapshot version."}
	if version >= 3:
		var exact_shape_error := _current_snapshot_exact_shape_error(snapshot)
		if not exact_shape_error.is_empty():
			return {&"ok": false, &"error": exact_shape_error}
	var config_value: Variant = _value(snapshot, &"config", null)
	if not config_value is Dictionary:
		return {&"ok": false, &"error": "Operations snapshot has no config."}
	var normalized_config := _normalize_config(config_value)
	if not bool(normalized_config[&"ok"]):
		return normalized_config
	var current_value: Variant = _value(snapshot, &"current_minute", null)
	if not _is_integral_value(current_value):
		return {&"ok": false, &"error": "Operations snapshot time must be an integer."}
	var restored_minute := int(current_value)
	# v1/v2 never had an authoritative origin or partial-boundary markers.
	# Ignore look-alike future fields on a downgraded payload instead of letting
	# them weaken legacy coverage validation.
	var origin_value: Variant = _value(snapshot, &"origin_minute", null) if version >= 3 else 0
	if not _is_integral_value(origin_value):
		return {&"ok": false, &"error": "Operations snapshot origin must be an integer."}
	var restored_origin := int(origin_value)
	if restored_origin < 0 or restored_origin > MAX_START_MINUTE \
	or restored_minute < restored_origin \
	or restored_minute > restored_origin + MAX_TRACKED_MINUTES:
		return {&"ok": false, &"error": "Operations snapshot time is invalid."}
	var pending_value: Variant = _value(snapshot, &"pending_boundary_minute", -1) \
		if version >= 3 else -1
	var deferred_value: Variant = _value(snapshot, &"deferred_boundary_minute", -1) \
		if version >= 3 else -1
	if not _is_integral_value(pending_value) or not _is_integral_value(deferred_value):
		return {&"ok": false, &"error": "Pending boundary markers must be integers."}
	var pending_minute := int(pending_value)
	var deferred_minute := int(deferred_value)
	if pending_minute != -1 \
	and (pending_minute != restored_minute + 1 \
		or not _minute_has_scheduled_boundary(pending_minute)):
		return {&"ok": false, &"error": "Pending operations boundary is invalid."}
	if deferred_minute != -1 \
	and (deferred_minute != restored_minute \
		or not _minute_has_scheduled_boundary(deferred_minute)):
		return {&"ok": false, &"error": "Deferred operations boundary is invalid."}
	var maximum_committed_minute := pending_minute if pending_minute >= 0 else restored_minute

	var profiles_value: Variant = _value(snapshot, &"demand_profiles", null)
	var carry_value: Variant = _value(snapshot, &"arrival_carry", null)
	if not profiles_value is Array or not carry_value is Array:
		return {&"ok": false, &"error": "Demand snapshot collections are malformed."}
	if profiles_value.size() > MAX_SAFE_COLLECTION_SIZE or carry_value.size() > MAX_SAFE_COLLECTION_SIZE:
		return {&"ok": false, &"error": "Demand snapshot collections are too large."}
	if profiles_value.size() > MAX_DEMAND_PROFILES:
		return {&"ok": false, &"error": "Demand snapshot exceeds the profile limit."}
	var restored_profiles: Dictionary = {}
	for profile_value: Variant in profiles_value:
		if not profile_value is Dictionary:
			return {&"ok": false, &"error": "Demand profile snapshot row is malformed."}
		var facility_tag := StringName(_value(profile_value, &"facility_tag", &""))
		var guest_class := StringName(_value(profile_value, &"guest_class", &""))
		var time_band := StringName(_value(profile_value, &"time_band", &""))
		var identity_error := _validate_demand_identity(facility_tag, guest_class, time_band)
		if not identity_error.is_empty():
			return {&"ok": false, &"error": identity_error}
		var components_value: Variant = _value(profile_value, &"components", null)
		if not components_value is Dictionary:
			return {&"ok": false, &"error": "Demand components snapshot is malformed."}
		var normalized_components := _normalize_demand_components_for_limit(
			components_value,
			int((normalized_config[&"config"] as Dictionary)[&"max_component_magnitude"])
		)
		if not bool(normalized_components[&"ok"]):
			return normalized_components
		var demand_index_value: Variant = _value(profile_value, &"demand_index", null)
		var evaluated_value: Variant = _value(profile_value, &"last_evaluated_at", null)
		if not _is_integral_value(demand_index_value) or not _is_integral_value(evaluated_value):
			return {&"ok": false, &"error": "Demand derived fields must be integers."}
		var demand_index := int(demand_index_value)
		if demand_index != _calculate_demand_index(normalized_components[&"components"]):
			return {&"ok": false, &"error": "Saved DemandIndex does not match its inputs."}
		var evaluated_at := int(evaluated_value)
		if evaluated_at < -1 or evaluated_at > maximum_committed_minute \
		or (evaluated_at >= 0 and evaluated_at % DEMAND_INTERVAL_MINUTES != 0):
			return {&"ok": false, &"error": "Demand evaluation time is invalid."}
		var key := _demand_key(facility_tag, guest_class, time_band)
		if restored_profiles.has(key):
			return {&"ok": false, &"error": "Demand profile IDs are duplicated."}
		restored_profiles[key] = {
			&"facility_tag": facility_tag,
			&"guest_class": guest_class,
			&"time_band": time_band,
			&"components": normalized_components[&"components"],
			&"demand_index": demand_index,
			&"last_evaluated_at": evaluated_at,
		}
	var restored_carry: Dictionary = {}
	for row_value: Variant in carry_value:
		if not row_value is Dictionary:
			return {&"ok": false, &"error": "Arrival carry row is malformed."}
		var key := _demand_key(
			StringName(_value(row_value, &"facility_tag", &"")),
			StringName(_value(row_value, &"guest_class", &"")),
			StringName(_value(row_value, &"time_band", &""))
		)
		var numerator_value: Variant = _value(row_value, &"numerator", null)
		if not _is_integral_value(numerator_value):
			return {&"ok": false, &"error": "Arrival carry numerator must be an integer."}
		var numerator := int(numerator_value)
		if not restored_profiles.has(key) or restored_carry.has(key) \
		or numerator < 0 or numerator > _max_carry_for_config(normalized_config[&"config"]):
			return {&"ok": false, &"error": "Arrival carry is invalid."}
		restored_carry[key] = numerator
	if restored_carry.size() != restored_profiles.size():
		return {&"ok": false, &"error": "Every demand profile requires one carry row."}

	var histories := _parse_snapshot_histories(
		snapshot,
		version,
		maximum_committed_minute
	)
	if not bool(histories[&"ok"]):
		return histories
	var population_value: Variant = _value(snapshot, &"population_count", null)
	if population_value == null:
		if version == 1 and (histories[&"arrival_reports"] as Array).is_empty():
			population_value = 0
		else:
			return {
				&"ok": false,
				&"error": "Legacy population is ambiguous when arrival history exists.",
			}
	var arrears_value: Variant = _value(snapshot, &"arrears_total", null)
	var consecutive_value: Variant = _value(snapshot, &"consecutive_insolvency_count", null)
	var next_plan_value: Variant = _value(snapshot, &"next_arrival_plan_sequence", null)
	var next_guest_value: Variant = _value(snapshot, &"next_guest_sequence", null)
	if not _is_integral_value(population_value) \
	or not _is_integral_value(arrears_value) \
	or not _is_integral_value(consecutive_value) \
	or not _is_integral_value(next_plan_value) \
	or not _is_integral_value(next_guest_value):
		return {&"ok": false, &"error": "Operations scalar snapshot types are invalid."}
	var population := int(population_value)
	var arrears := int(arrears_value)
	var consecutive := int(consecutive_value)
	var terminal_value: Variant = _value(snapshot, &"terminal_state", null)
	if typeof(terminal_value) not in [TYPE_STRING, TYPE_STRING_NAME]:
		return {&"ok": false, &"error": "Terminal state must be a string."}
	var terminal := StringName(terminal_value)
	var next_plan := int(next_plan_value)
	var next_guest := int(next_guest_value)
	if population < 0 or population > int((normalized_config[&"config"] as Dictionary)[&"population_cap"]) \
	or arrears < 0 or consecutive < 0 or next_plan < 1 or next_guest < 1:
		return {&"ok": false, &"error": "Operations scalar state is invalid."}
	if terminal not in [TERMINAL_RUNNING, TERMINAL_BANKRUPT]:
		return {&"ok": false, &"error": "Operations terminal state is invalid."}
	if terminal == TERMINAL_BANKRUPT \
	and consecutive < int((normalized_config[&"config"] as Dictionary)[&"bankruptcy_close_count"]):
		return {&"ok": false, &"error": "Bankruptcy threshold was not reached."}
	var pause_value: Variant = _value(snapshot, &"pause_requested", false)
	if typeof(pause_value) != TYPE_BOOL:
		return {&"ok": false, &"error": "Pause flag must be boolean."}
	var plan_high_water := next_plan - 1
	var guest_high_water := next_guest - 1
	var fair_counter := (histories[&"arrival_reports"] as Array).size()
	if version >= 3:
		var plan_high_value: Variant = _value(snapshot, &"arrival_plan_high_water", null)
		var guest_high_value: Variant = _value(snapshot, &"guest_high_water", null)
		var fair_value: Variant = _value(snapshot, &"fair_rotation_counter", null)
		if not _is_integral_value(plan_high_value) \
		or not _is_integral_value(guest_high_value) \
		or not _is_integral_value(fair_value):
			return {&"ok": false, &"error": "ID high-water and fairness state must be integers."}
		plan_high_water = int(plan_high_value)
		guest_high_water = int(guest_high_value)
		fair_counter = int(fair_value)
	if plan_high_water < 0 or plan_high_water > MAX_SEQUENCE_VALUE \
	or guest_high_water < 0 or guest_high_water > MAX_SEQUENCE_VALUE \
	or next_plan != plan_high_water + 1 or next_guest != guest_high_water + 1:
		return {&"ok": false, &"error": "ID high-water state is invalid."}
	var pause_reason_value: Variant = _value(snapshot, &"pause_reason", &"") \
		if version >= 3 else (&"LEDGER_IMBALANCE" if bool(pause_value) else &"")
	if typeof(pause_reason_value) not in [TYPE_STRING, TYPE_STRING_NAME]:
		return {&"ok": false, &"error": "Pause reason must be a string."}
	var pause_reason := StringName(pause_reason_value)
	var quarantine_value: Variant = _value(snapshot, &"external_quarantined", false) \
		if version >= 3 else false
	var quarantine_report_value: Variant = _value(snapshot, &"quarantine_report", {}) \
		if version >= 3 else {}
	if typeof(quarantine_value) != TYPE_BOOL or not quarantine_report_value is Dictionary:
		return {&"ok": false, &"error": "External quarantine snapshot state is malformed."}

	var legacy_history := 0
	var legacy_calendar := 0
	if version < 4:
		for day: int in histories[&"day_summaries"]:
			legacy_history = maxi(legacy_history, day)
		legacy_calendar = legacy_history + 1
	else:
		var calendar_value: Variant = _value(snapshot, &"legacy_calendar_last_day", null)
		var history_value: Variant = _value(snapshot, &"legacy_history_last_day", null)
		if not _is_integral_value(calendar_value) or not _is_integral_value(history_value):
			return {&"ok": false, &"error": "Invalid legacy calendar markers."}
		legacy_calendar = int(calendar_value)
		legacy_history = int(history_value)
		if legacy_history < 0 or legacy_calendar < 0 or legacy_calendar > MAX_TRACKED_DAYS + 1 \
			or (legacy_calendar != 0 and legacy_calendar != legacy_history + 1) \
			or (legacy_calendar == 0 and legacy_history != 0):
			return {&"ok": false, &"error": "Invalid legacy calendar transition."}
		for day in range(1, legacy_history + 1):
			if not (histories[&"day_summaries"] as Dictionary).has(day):
				return {&"ok": false, &"error": "Missing legacy financial history."}
	var last_closed_day := 0
	for day: int in histories[&"day_summaries"]:
		last_closed_day = maxi(last_closed_day, day)
	var legacy_principal := last_closed_day
	if version >= 5:
		var principal_value: Variant = _value(snapshot, &"legacy_principal_profit_last_day", null)
		if not _is_integral_value(principal_value):
			return {&"ok": false, &"error": "Legacy principal profit marker must be an integer."}
		legacy_principal = int(principal_value)
	if legacy_principal < 0 or legacy_principal > last_closed_day:
		return {&"ok": false, &"error": "Legacy principal profit marker exceeds closed history."}
	var state := {
		&"legacy_principal_profit_last_day": legacy_principal,
		&"legacy_calendar_last_day": legacy_calendar,
		&"legacy_history_last_day": legacy_history,
		&"origin_minute": restored_origin,
		&"current_minute": restored_minute,
		&"config": normalized_config[&"config"],
		&"demand_profiles": restored_profiles,
		&"arrival_carry": restored_carry,
		&"demand_history": histories[&"demand_history"],
		&"arrival_reports": histories[&"arrival_reports"],
		&"day_summaries": histories[&"day_summaries"],
		&"morning_reports": histories[&"morning_reports"],
		&"alerts": histories[&"alerts"],
		&"processed_demand_minutes": histories[&"processed_demand_minutes"],
		&"processed_arrival_minutes": histories[&"processed_arrival_minutes"],
		&"population_count": population,
		&"arrears_total": arrears,
		&"consecutive_insolvency_count": consecutive,
		&"terminal_state": terminal,
		&"pause_requested": bool(pause_value),
		&"next_arrival_plan_sequence": next_plan,
		&"next_guest_sequence": next_guest,
		&"arrival_plan_high_water": plan_high_water,
		&"guest_high_water": guest_high_water,
		&"fair_rotation_counter": fair_counter,
		&"pending_boundary_minute": pending_minute,
		&"deferred_boundary_minute": deferred_minute,
		&"pause_reason": pause_reason,
		&"external_quarantined": bool(quarantine_value),
		&"quarantine_report": (quarantine_report_value as Dictionary).duplicate(true),
	}
	# Validate the fully assembled state using a temporary instance without
	# copying any callback into save data.
	var candidate := OperationsCoordinator.new(state[&"config"])
	candidate._legacy_calendar_last_day = legacy_calendar
	candidate._legacy_history_last_day = legacy_history
	candidate._legacy_principal_profit_last_day = legacy_principal
	candidate._origin_minute = restored_origin
	candidate.current_minute = restored_minute
	candidate._config = (state[&"config"] as Dictionary).duplicate(true)
	candidate._demand_profiles = restored_profiles.duplicate(true)
	candidate._arrival_carry = restored_carry.duplicate(true)
	candidate._demand_history = _copy_dictionary_array(state[&"demand_history"])
	candidate._arrival_reports = _copy_dictionary_array(state[&"arrival_reports"])
	candidate._day_summaries = (state[&"day_summaries"] as Dictionary).duplicate(true)
	candidate._morning_reports = (state[&"morning_reports"] as Dictionary).duplicate(true)
	candidate._alerts = _copy_dictionary_array(state[&"alerts"])
	candidate._processed_demand_minutes = (state[&"processed_demand_minutes"] as Dictionary).duplicate(true)
	candidate._processed_arrival_minutes = (state[&"processed_arrival_minutes"] as Dictionary).duplicate(true)
	candidate._population_count = population
	candidate._arrears_total = arrears
	candidate._consecutive_insolvency_count = consecutive
	candidate._terminal_state = terminal
	candidate.pause_requested = bool(state[&"pause_requested"])
	candidate._next_arrival_plan_sequence = next_plan
	candidate._next_guest_sequence = next_guest
	candidate._arrival_plan_high_water = plan_high_water
	candidate._guest_high_water = guest_high_water
	candidate._fair_rotation_counter = fair_counter
	candidate._pending_boundary_minute = pending_minute
	candidate._deferred_boundary_minute = deferred_minute
	candidate._pause_reason = pause_reason
	candidate._external_quarantined = bool(quarantine_value)
	candidate._quarantine_report = (quarantine_report_value as Dictionary).duplicate(true)
	var report := candidate.validate_invariants()
	if not bool(report[&"ok"]):
		return {&"ok": false, &"error": "Operations snapshot invariants failed.", &"report": report}
	return {&"ok": true, &"state": state}


func _parse_snapshot_histories(
	snapshot: Dictionary,
	version: int,
	maximum_committed_minute: int
) -> Dictionary:
	var demand_history_value: Variant = _value(snapshot, &"demand_history", [])
	var arrivals_value: Variant = _value(snapshot, &"arrival_reports", null)
	var days_value: Variant = _value(snapshot, &"day_summaries", null)
	var mornings_value: Variant = _value(snapshot, &"morning_reports", null)
	var alerts_value: Variant = _value(snapshot, &"alerts", [])
	var demand_minutes_value: Variant = _value(snapshot, &"processed_demand_minutes", [])
	var arrival_minutes_value: Variant = _value(snapshot, &"processed_arrival_minutes", [])
	if not demand_history_value is Array or not arrivals_value is Array \
	or not days_value is Array or not mornings_value is Array \
	or not alerts_value is Array or not demand_minutes_value is Array \
	or not arrival_minutes_value is Array:
		return {&"ok": false, &"error": "Operations history collections are malformed."}
	for collection: Array in [
		demand_history_value,
		arrivals_value,
		days_value,
		mornings_value,
		alerts_value,
		demand_minutes_value,
		arrival_minutes_value,
	]:
		if collection.size() > MAX_SAFE_COLLECTION_SIZE:
			return {&"ok": false, &"error": "Operations history collection is too large."}
	if not _variant_tree_report(demand_history_value)[&"ok"] \
	or not _variant_tree_report(arrivals_value)[&"ok"] \
	or not _variant_tree_report(days_value)[&"ok"] \
	or not _variant_tree_report(mornings_value)[&"ok"] \
	or not _variant_tree_report(alerts_value)[&"ok"]:
		return {&"ok": false, &"error": "Operations history contains unsaveable data."}
	for row_value: Variant in demand_history_value:
		if not row_value is Dictionary:
			return {&"ok": false, &"error": "Demand history row is malformed."}
	var migrated_arrivals: Array = []
	var arrival_index := 0
	for row_value: Variant in arrivals_value:
		if not row_value is Dictionary:
			return {&"ok": false, &"error": "Arrival history row is malformed."}
		var migrated_row := (row_value as Dictionary).duplicate(true)
		if version < 3:
			migrated_row[&"fair_rotation_counter"] = arrival_index
			migrated_row[&"fair_start_index"] = 0
		migrated_arrivals.append(migrated_row)
		arrival_index += 1
	for row_value: Variant in alerts_value:
		if not row_value is Dictionary:
			return {&"ok": false, &"error": "Operations alert row is malformed."}
	var demand_minutes := _minute_array_to_set(
		demand_minutes_value,
		DEMAND_INTERVAL_MINUTES,
		maximum_committed_minute
	)
	var arrival_minutes := _minute_array_to_set(
		arrival_minutes_value,
		ARRIVAL_INTERVAL_MINUTES,
		maximum_committed_minute
	)
	if not bool(demand_minutes[&"ok"]) or not bool(arrival_minutes[&"ok"]):
		return {&"ok": false, &"error": "Processed boundary minute list is invalid."}
	var migrated_demand_history: Array = (demand_history_value as Array).duplicate(true)
	if version == 1 and migrated_demand_history.is_empty() \
	and not (demand_minutes[&"set"] as Dictionary).is_empty():
		for minute: int in _sorted_int_keys(demand_minutes[&"set"]):
			migrated_demand_history.append({&"minute": minute, &"values": []})
	var days: Dictionary = {}
	for row_value: Variant in days_value:
		if not row_value is Dictionary:
			return {&"ok": false, &"error": "DaySummary row is malformed."}
		var day_value: Variant = _value(row_value, &"day", null)
		if not _is_integral_value(day_value):
			return {&"ok": false, &"error": "DaySummary day must be an integer."}
		var day := int(day_value)
		if day < 1 or days.has(day):
			return {&"ok": false, &"error": "DaySummary day is invalid or duplicated."}
		days[day] = (row_value as Dictionary).duplicate(true)
	var mornings: Dictionary = {}
	for row_value: Variant in mornings_value:
		if not row_value is Dictionary:
			return {&"ok": false, &"error": "Morning report row is malformed."}
		var day_value: Variant = _value(row_value, &"day", null)
		if not _is_integral_value(day_value):
			return {&"ok": false, &"error": "Morning report day must be an integer."}
		var day := int(day_value)
		if day < 1 or mornings.has(day):
			return {&"ok": false, &"error": "Morning report day is invalid or duplicated."}
		mornings[day] = (row_value as Dictionary).duplicate(true)
	return {
		&"ok": true,
		&"demand_history": migrated_demand_history,
		&"arrival_reports": migrated_arrivals,
		&"day_summaries": days,
		&"morning_reports": mornings,
		&"alerts": (alerts_value as Array).duplicate(true) if version >= 2 else [],
		&"processed_demand_minutes": demand_minutes[&"set"],
		&"processed_arrival_minutes": arrival_minutes[&"set"],
	}


func _validate_histories(errors: Array[Dictionary]) -> void:
	var last_demand := -1
	for value: Variant in _demand_history:
		if not value is Dictionary:
			errors.append({&"code": &"INVALID_DEMAND_HISTORY"})
			continue
		var minute_value: Variant = _value(value, &"minute", null)
		if not _is_integral_value(minute_value):
			errors.append({&"code": &"INVALID_DEMAND_HISTORY_TIME"})
			continue
		var minute := int(minute_value)
		if minute <= last_demand or not _processed_demand_minutes.has(minute):
			errors.append({&"code": &"INVALID_DEMAND_HISTORY_ORDER"})
		last_demand = minute
		var values_value: Variant = _value(value, &"values", null)
		if not values_value is Array:
			errors.append({&"code": &"INVALID_DEMAND_HISTORY_VALUES", &"minute": minute})
			continue
		var identities: Dictionary = {}
		for row_value: Variant in values_value:
			if not row_value is Dictionary:
				errors.append({&"code": &"INVALID_DEMAND_HISTORY_ROW", &"minute": minute})
				continue
			var row: Dictionary = row_value
			var facility_value: Variant = _value(row, &"facility_tag", null)
			var guest_value: Variant = _value(row, &"guest_class", null)
			var band_value: Variant = _value(row, &"time_band", null)
			var index_value: Variant = _value(row, &"demand_index", null)
			if typeof(facility_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or typeof(guest_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or typeof(band_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or not _is_integral_value(index_value) \
			or int(index_value) < 0 or int(index_value) > 200:
				errors.append({&"code": &"INVALID_DEMAND_HISTORY_ROW", &"minute": minute})
				continue
			var identity_error := _validate_demand_identity(
				StringName(facility_value), StringName(guest_value), StringName(band_value)
			)
			var identity := _demand_key(
				StringName(facility_value), StringName(guest_value), StringName(band_value)
			)
			if not identity_error.is_empty() or identities.has(identity):
				errors.append({&"code": &"INVALID_DEMAND_HISTORY_IDENTITY", &"minute": minute})
			else:
				identities[identity] = true
	var last_arrival := -1
	var plan_ids: Dictionary = {}
	var guest_ids: Dictionary = {}
	var maximum_plan_sequence := 0
	var maximum_guest_sequence := 0
	var last_plan_sequence := 0
	var last_guest_sequence := 0
	var expected_fair_counter := 0
	for value: Variant in _arrival_reports:
		if not value is Dictionary:
			errors.append({&"code": &"INVALID_ARRIVAL_HISTORY"})
			continue
		var report: Dictionary = value
		var minute_value: Variant = _value(report, &"minute", null)
		var fair_value: Variant = _value(report, &"fair_rotation_counter", null)
		var fair_start_value: Variant = _value(report, &"fair_start_index", null)
		var population_value: Variant = _value(report, &"population_after", null)
		var capacity_value: Variant = _value(report, &"capacity_remaining", null)
		var report_band_value: Variant = _value(report, &"time_band", null)
		if not _is_integral_value(minute_value) or not _is_integral_value(fair_value) \
		or not _is_integral_value(fair_start_value) \
		or not _is_integral_value(population_value) \
		or not _is_integral_value(capacity_value) \
		or typeof(report_band_value) not in [TYPE_STRING, TYPE_STRING_NAME]:
			errors.append({&"code": &"INVALID_ARRIVAL_HISTORY_SCALAR"})
			continue
		var minute := int(minute_value)
		if minute <= last_arrival or not _processed_arrival_minutes.has(minute):
			errors.append({&"code": &"INVALID_ARRIVAL_HISTORY_ORDER"})
		last_arrival = minute
		if int(fair_value) != expected_fair_counter:
			errors.append({&"code": &"FAIR_ROTATION_HISTORY_MISMATCH"})
		if int(fair_start_value) < 0 or int(fair_start_value) >= MAX_DEMAND_PROFILES:
			errors.append({&"code": &"INVALID_FAIR_START_INDEX"})
		if StringName(report_band_value) != _time_band_for_minute(minute):
			errors.append({&"code": &"ARRIVAL_TIME_BAND_MISMATCH"})
		var population_after := int(population_value)
		if population_after < 0 or population_after > int(_config[&"population_cap"]) \
		or int(capacity_value) != int(_config[&"population_cap"]) - population_after:
			errors.append({&"code": &"ARRIVAL_CAPACITY_DERIVATION_MISMATCH"})
		expected_fair_counter += 1
		var plans_value: Variant = _value(report, &"plans", null)
		if not plans_value is Array:
			errors.append({&"code": &"INVALID_ARRIVAL_PLANS"})
			continue
		var total := 0
		for plan_value: Variant in plans_value:
			if not plan_value is Dictionary:
				errors.append({&"code": &"INVALID_ARRIVAL_PLAN"})
				continue
			var plan: Dictionary = plan_value
			var plan_id_value: Variant = _value(plan, &"plan_id", null)
			var plan_minute_value: Variant = _value(plan, &"minute", null)
			var plan_id := String(plan_id_value)
			var plan_sequence := _sequence_from_id(plan_id, "arrival-plan-")
			var count_value: Variant = _value(plan, &"count", null)
			var ids_value: Variant = _value(plan, &"guest_ids", null)
			var facility_value: Variant = _value(plan, &"facility_tag", null)
			var guest_value: Variant = _value(plan, &"guest_class", null)
			var band_value: Variant = _value(plan, &"time_band", null)
			var demand_value: Variant = _value(plan, &"demand_index", null)
			if typeof(plan_id_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or not _is_integral_value(plan_minute_value) or int(plan_minute_value) != minute \
			or not _is_integral_value(count_value) \
			or typeof(facility_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or typeof(guest_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or typeof(band_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
			or not _is_integral_value(demand_value) \
			or int(demand_value) < 0 or int(demand_value) > 200:
				errors.append({&"code": &"INVALID_ARRIVAL_PLAN_COUNT"})
				continue
			var count := int(count_value)
			if plan_sequence <= 0 or plan_ids.has(plan_id) or count <= 0 \
			or plan_sequence <= last_plan_sequence \
			or not ids_value is Array or ids_value.size() != count \
			or not _validate_demand_identity(
				StringName(facility_value), StringName(guest_value), StringName(band_value)
			).is_empty():
				errors.append({&"code": &"INVALID_ARRIVAL_PLAN"})
				continue
			plan_ids[plan_id] = true
			last_plan_sequence = plan_sequence
			maximum_plan_sequence = maxi(maximum_plan_sequence, plan_sequence)
			for guest_id_value: Variant in ids_value:
				var guest_id := String(guest_id_value)
				var guest_sequence := _sequence_from_id(guest_id, "guest-auto-")
				if typeof(guest_id_value) not in [TYPE_STRING, TYPE_STRING_NAME] \
				or guest_sequence <= 0 or guest_sequence <= last_guest_sequence \
				or guest_ids.has(guest_id):
					errors.append({&"code": &"DUPLICATE_ARRIVAL_GUEST"})
				guest_ids[guest_id] = true
				last_guest_sequence = guest_sequence
				maximum_guest_sequence = maxi(maximum_guest_sequence, guest_sequence)
			total += count
		var total_value: Variant = _value(report, &"total_count", null)
		if not _is_integral_value(total_value) \
		or total != int(total_value) \
		or total > int(_config[&"max_arrivals_per_interval"]) \
		or total > population_after:
			errors.append({&"code": &"ARRIVAL_TOTAL_MISMATCH"})
		var capped_value: Variant = _value(report, &"carry_capped", null)
		if not capped_value is Array:
			errors.append({&"code": &"INVALID_CARRY_CAPPED_HISTORY"})
		else:
			for capped_row: Variant in capped_value:
				if not capped_row is Dictionary:
					errors.append({&"code": &"INVALID_CARRY_CAPPED_HISTORY"})
	if _demand_history.size() != _processed_demand_minutes.size():
		errors.append({&"code": &"DEMAND_BOUNDARY_COUNT_MISMATCH"})
	if _arrival_reports.size() != _processed_arrival_minutes.size():
		errors.append({&"code": &"ARRIVAL_BOUNDARY_COUNT_MISMATCH"})
	if maximum_plan_sequence > _arrival_plan_high_water \
	or maximum_guest_sequence > _guest_high_water:
		errors.append({&"code": &"ID_HIGH_WATER_BELOW_HISTORY"})


func _validate_financial_history(errors: Array[Dictionary]) -> void:
	var maximum_committed_minute := current_minute
	if _pending_boundary_minute == current_minute + 1:
		maximum_committed_minute = _pending_boundary_minute
	var expected_consecutive := 0
	var expected_arrears := 0
	var expected_terminal: StringName = TERMINAL_RUNNING
	var prior_net := 0
	var profits: Array[int] = []
	for day: int in _sorted_int_keys(_day_summaries):
		var summary: Dictionary = _day_summaries[day]
		var scalar_fields_valid := true
		for scalar_key: StringName in [
			&"day", &"closed_at", &"period_start_minute", &"period_end_minute_exclusive",
			&"net_profit", &"previous_day_delta", &"seven_day_average",
			&"reconciliation_difference", &"arrears_total", &"consecutive_insolvency_count",
		]:
			if not _is_integral_value(_value(summary, scalar_key, null)):
				scalar_fields_valid = false
		if not scalar_fields_valid:
			errors.append({&"code": &"INVALID_DAY_SUMMARY_SCALAR_TYPE", &"day": day})
			continue
		var expected_close := _daily_close_minute_for_day(day)
		if int(_value(summary, &"day", 0)) != day \
		or int(_value(summary, &"closed_at", -1)) != expected_close \
		or int(_value(summary, &"period_start_minute", -1)) != get_financial_period_start(day) \
		or int(_value(summary, &"period_end_minute_exclusive", -1)) != expected_close \
		or expected_close <= _origin_minute \
		or expected_close > maximum_committed_minute:
			errors.append({&"code": &"INVALID_DAY_SUMMARY_BOUNDARY", &"day": day})
			continue
		var categories_value: Variant = _value(summary, &"category_totals", null)
		if not categories_value is Dictionary:
			errors.append({&"code": &"INVALID_DAY_CATEGORIES", &"day": day})
			continue
		var categories := _normalize_categories(categories_value)
		if not bool(categories[&"ok"]):
			errors.append({&"code": &"INVALID_DAY_CATEGORIES", &"day": day})
			continue
		var net := _principal_excluded_profit(categories[&"categories"]) \
			if day <= _legacy_principal_profit_last_day else _net_profit(categories[&"categories"])
		if net != int(_value(summary, &"net_profit", 0)):
			errors.append({&"code": &"NET_PROFIT_MISMATCH", &"day": day})
		var difference_value: Variant = _value(summary, &"reconciliation_difference", null)
		var balanced_value: Variant = _value(summary, &"balanced", null)
		if not _is_integral_value(difference_value) \
		or typeof(balanced_value) != TYPE_BOOL \
		or bool(balanced_value) != (int(difference_value) == 0):
			errors.append({&"code": &"RECONCILIATION_DERIVATION_MISMATCH", &"day": day})
		var facilities_value: Variant = _value(summary, &"facility_profit_loss", null)
		if not facilities_value is Dictionary:
			errors.append({&"code": &"INVALID_FACILITY_PROFIT_LOSS", &"day": day})
		else:
			var facilities := _normalize_facility_profit_loss(facilities_value)
			if not bool(facilities[&"ok"]) \
			or _to_snapshot_variant(facilities[&"facilities"]) != _to_snapshot_variant(facilities_value):
				errors.append({&"code": &"FACILITY_PROFIT_LOSS_MISMATCH", &"day": day})
		var anomalies_value: Variant = _value(summary, &"anomalies", null)
		if not anomalies_value is Array or not _variant_tree_report(anomalies_value)[&"ok"]:
			errors.append({&"code": &"INVALID_DAY_ANOMALIES", &"day": day})
		if int(_value(summary, &"previous_day_delta", 0)) != (0 if profits.is_empty() else net - prior_net):
			errors.append({&"code": &"PREVIOUS_DAY_DELTA_MISMATCH", &"day": day})
		profits.append(net)
		prior_net = net
		var start := maxi(profits.size() - 7, 0)
		var sum := 0
		for index in range(start, profits.size()):
			sum += profits[index]
		var count := profits.size() - start
		if int(_value(summary, &"seven_day_average", 0)) != sum / count:
			errors.append({&"code": &"SEVEN_DAY_AVERAGE_MISMATCH", &"day": day})
		var average_detail_value: Variant = _value(summary, &"seven_day_average_detail", null)
		if not average_detail_value is Dictionary \
		or int(_value(average_detail_value, &"sum", 0)) != sum \
		or int(_value(average_detail_value, &"count", 0)) != count \
		or int(_value(average_detail_value, &"quotient", 0)) != sum / count \
		or int(_value(average_detail_value, &"remainder", 0)) != sum % count:
			errors.append({&"code": &"SEVEN_DAY_DETAIL_MISMATCH", &"day": day})
		expected_arrears = int(_value(summary, &"arrears_total", -1))
		if expected_arrears < 0:
			errors.append({&"code": &"INVALID_DAY_ARREARS", &"day": day})
		var settlement_value: Variant = _value(summary, &"settlement", null)
		if not settlement_value is Dictionary \
		or not _is_integral_value(_value(settlement_value, &"arrears_total", null)) \
		or int(_value(settlement_value, &"arrears_total", -2)) != expected_arrears \
		or not _variant_tree_report(settlement_value)[&"ok"]:
			errors.append({&"code": &"SETTLEMENT_ARREARS_MISMATCH", &"day": day})
		if day == _legacy_history_last_day + 1:
			expected_consecutive = 0
		var failed_close := expected_arrears > 0 if day <= _legacy_history_last_day else net < 0
		expected_consecutive = expected_consecutive + 1 if failed_close else 0
		if int(_value(summary, &"consecutive_insolvency_count", -1)) != expected_consecutive:
			errors.append({&"code": &"INSOLVENCY_COUNT_MISMATCH", &"day": day})
		if expected_consecutive >= int(_config[&"bankruptcy_close_count"]):
			expected_terminal = TERMINAL_BANKRUPT
		if StringName(_value(summary, &"terminal_state", &"")) != expected_terminal:
			errors.append({&"code": &"DAY_TERMINAL_MISMATCH", &"day": day})
	if not _day_summaries.is_empty():
		if expected_arrears != _arrears_total or expected_consecutive != _consecutive_insolvency_count:
			errors.append({&"code": &"CURRENT_INSOLVENCY_MISMATCH"})
		if expected_terminal != _terminal_state:
			errors.append({&"code": &"CURRENT_TERMINAL_MISMATCH"})
	for day: int in _sorted_int_keys(_morning_reports):
		var report: Dictionary = _morning_reports[day]
		var expected_minute := (day - 1) * MINUTES_PER_DAY + MORNING_MINUTE_OF_DAY
		if not _is_integral_value(_value(report, &"day", null)) \
		or not _is_integral_value(_value(report, &"minute", null)) \
		or int(_value(report, &"day", 0)) != day \
		or int(_value(report, &"minute", -1)) != expected_minute \
		or expected_minute <= _origin_minute \
		or expected_minute > maximum_committed_minute or not _is_save_safe(report):
			errors.append({&"code": &"INVALID_MORNING_REPORT", &"day": day})


func _validate_boundary_minute_set(
	values: Dictionary,
	interval: int,
	code: StringName,
	errors: Array[Dictionary]
) -> void:
	var maximum_committed_minute := current_minute
	if _pending_boundary_minute == current_minute + 1:
		maximum_committed_minute = _pending_boundary_minute
	for minute_value: Variant in values:
		var minute := int(minute_value)
		if minute <= _origin_minute or minute > maximum_committed_minute or minute % interval != 0 \
		or not bool(values[minute_value]):
			errors.append({&"code": code, &"minute": minute})


func _validate_boundary_coverage(errors: Array[Dictionary]) -> void:
	var partial_minute := -1
	var complete_through := current_minute
	if _pending_boundary_minute == current_minute + 1:
		partial_minute = _pending_boundary_minute
	elif _deferred_boundary_minute == current_minute:
		partial_minute = _deferred_boundary_minute
		complete_through = current_minute - 1
	elif _terminal_state == TERMINAL_BANKRUPT and _is_daily_close_boundary(current_minute):
		partial_minute = current_minute
		complete_through = current_minute - 1

	for minute in range(_origin_minute + 1, complete_through + 1):
		if _is_daily_close_boundary(minute):
			var completed_day := _financial_day_for_boundary(minute)
			if not _day_summaries.has(completed_day):
				errors.append({&"code": &"MISSING_DAILY_CLOSE", &"minute": minute})
		if minute % MINUTES_PER_DAY == MORNING_MINUTE_OF_DAY:
			var morning_day := minute / MINUTES_PER_DAY + 1
			if not _morning_reports.has(morning_day):
				errors.append({&"code": &"MISSING_MORNING_BOUNDARY", &"minute": minute})
		if minute % DEMAND_INTERVAL_MINUTES == 0 \
		and not _processed_demand_minutes.has(minute):
			errors.append({&"code": &"MISSING_DEMAND_BOUNDARY", &"minute": minute})
		if minute % ARRIVAL_INTERVAL_MINUTES == 0 \
		and not _processed_arrival_minutes.has(minute):
			errors.append({&"code": &"MISSING_ARRIVAL_BOUNDARY", &"minute": minute})

	if partial_minute >= 0:
		_validate_partial_boundary_prefix(partial_minute, errors)


func _validate_partial_boundary_prefix(minute: int, errors: Array[Dictionary]) -> void:
	var states: Array[bool] = []
	if _is_daily_close_boundary(minute):
		states.append(_day_summaries.has(_financial_day_for_boundary(minute)))
	if minute % MINUTES_PER_DAY == MORNING_MINUTE_OF_DAY:
		states.append(_morning_reports.has(minute / MINUTES_PER_DAY + 1))
	if minute % DEMAND_INTERVAL_MINUTES == 0:
		states.append(_processed_demand_minutes.has(minute))
	if minute % ARRIVAL_INTERVAL_MINUTES == 0:
		states.append(_processed_arrival_minutes.has(minute))
	var found_gap := false
	for processed: bool in states:
		if not processed:
			found_gap = true
		elif found_gap:
			errors.append({&"code": &"NON_PREFIX_PARTIAL_BOUNDARY", &"minute": minute})
			return


func _validate_alerts_and_pause(errors: Array[Dictionary]) -> void:
	var imbalance_by_day: Dictionary = {}
	var arrears_by_day: Dictionary = {}
	var losses_by_day: Dictionary = {}
	var bankrupt_count := 0
	var quarantine_count := 0
	for alert: Dictionary in _alerts:
		if not _variant_tree_report(alert)[&"ok"]:
			errors.append({&"code": &"INVALID_ALERT_PAYLOAD"})
			continue
		var code := StringName(_value(alert, &"code", &""))
		match code:
			&"LEDGER_IMBALANCE":
				var day := int(_value(alert, &"day", 0))
				imbalance_by_day[day] = int(imbalance_by_day.get(day, 0)) + 1
			&"LOSS_STREAK":
				var day := int(_value(alert, &"day", 0))
				losses_by_day[day] = int(losses_by_day.get(day, 0)) + 1
				var summary: Dictionary = _day_summaries.get(day, {})
				if summary.is_empty() or day <= _legacy_history_last_day \
					or int(_value(alert, &"net_profit", 0)) >= 0 \
					or _value(alert, &"minute", null) != summary.get(&"closed_at") \
					or _value(alert, &"net_profit", null) != summary.get(&"net_profit") \
					or _value(alert, &"consecutive_count", null) != summary.get(&"consecutive_insolvency_count"):
					errors.append({&"code": &"INVALID_LOSS_ALERT", &"day": day})
			&"ARREARS_REMAIN":
				var day := int(_value(alert, &"day", 0))
				arrears_by_day[day] = int(arrears_by_day.get(day, 0)) + 1
			&"BANKRUPT":
				bankrupt_count += 1
			&"EXTERNAL_ROLLBACK_FAILED":
				quarantine_count += 1
			&"ARRIVAL_CARRY_CAPPED":
				pass
			_:
				errors.append({&"code": &"UNKNOWN_ALERT", &"alert_code": code})
	for day: int in _sorted_int_keys(_day_summaries):
		var summary: Dictionary = _day_summaries[day]
		var imbalanced := int(_value(summary, &"reconciliation_difference", 0)) != 0
		if int(imbalance_by_day.get(day, 0)) != int(imbalanced):
			errors.append({&"code": &"IMBALANCE_ALERT_MISMATCH", &"day": day})
		var new_loss := day > _legacy_history_last_day and int(_value(summary, &"net_profit", 0)) < 0
		if int(losses_by_day.get(day, 0)) != int(new_loss):
			errors.append({&"code": &"LOSS_ALERT_MISMATCH", &"day": day})
		var has_arrears := int(_value(summary, &"arrears_total", 0)) > 0
		if int(arrears_by_day.get(day, 0)) != int(has_arrears):
			errors.append({&"code": &"ARREARS_ALERT_MISMATCH", &"day": day})
	if bankrupt_count != int(_terminal_state == TERMINAL_BANKRUPT):
		errors.append({&"code": &"BANKRUPT_ALERT_MISMATCH"})
	if quarantine_count != int(_external_quarantined):
		errors.append({&"code": &"QUARANTINE_ALERT_MISMATCH"})
	if pause_requested and _pause_reason == &"LEDGER_IMBALANCE" and imbalance_by_day.is_empty():
		errors.append({&"code": &"PAUSE_WITHOUT_IMBALANCE_ALERT"})
	if pause_requested and _pause_reason == &"LEDGER_IMBALANCE":
		var paused_day := _financial_day_for_boundary(current_minute)
		var paused_summary: Dictionary = _day_summaries.get(paused_day, {})
		if not _is_daily_close_boundary(current_minute) \
		or paused_summary.is_empty() \
		or int(_value(paused_summary, &"reconciliation_difference", 0)) == 0 \
		or (_terminal_state == TERMINAL_RUNNING \
			and _deferred_boundary_minute != current_minute):
			errors.append({&"code": &"STALE_LEDGER_PAUSE"})
	if _external_quarantined:
		var quarantine_report_found := false
		var expected_report: Variant = _to_snapshot_variant(_quarantine_report)
		for alert: Dictionary in _alerts:
			if _to_snapshot_variant(alert) == expected_report:
				quarantine_report_found = true
				break
		if not quarantine_report_found:
			errors.append({&"code": &"QUARANTINE_REPORT_NOT_ALERTED"})


func _minute_array_to_set(values: Array, interval: int, maximum: int) -> Dictionary:
	var result: Dictionary = {}
	var prior := -1
	for value: Variant in values:
		if not _is_integral_value(value):
			return {&"ok": false}
		var minute := int(value)
		if minute <= prior or minute < 0 or minute > maximum or minute % interval != 0:
			return {&"ok": false}
		result[minute] = true
		prior = minute
	return {&"ok": true, &"set": result}


func _internal_snapshot_shape_error() -> StringName:
	if _demand_profiles.size() > MAX_DEMAND_PROFILES:
		return &"DEMAND_PROFILE_LIMIT"
	for profile_value: Variant in _demand_profiles.values():
		if not profile_value is Dictionary:
			return &"INVALID_DEMAND_PROFILE"
	for row_value: Variant in _demand_history:
		if not row_value is Dictionary:
			return &"INVALID_DEMAND_HISTORY"
	for row_value: Variant in _arrival_reports:
		if not row_value is Dictionary:
			return &"INVALID_ARRIVAL_HISTORY"
	for row_value: Variant in _alerts:
		if not row_value is Dictionary:
			return &"INVALID_ALERT_HISTORY"
	for row_value: Variant in _day_summaries.values():
		if not row_value is Dictionary:
			return &"INVALID_DAY_SUMMARY"
	for row_value: Variant in _morning_reports.values():
		if not row_value is Dictionary:
			return &"INVALID_MORNING_REPORT"
	return &""


func _has_committed_operations() -> bool:
	return not _processed_demand_minutes.is_empty() \
	or not _processed_arrival_minutes.is_empty() \
	or not _day_summaries.is_empty() \
	or not _morning_reports.is_empty()


func _minute_has_scheduled_boundary(minute: int) -> bool:
	if minute < 0:
		return false
	return minute % ARRIVAL_INTERVAL_MINUTES == 0 \
		or minute % DEMAND_INTERVAL_MINUTES == 0 \
		or minute % MINUTES_PER_DAY == MORNING_MINUTE_OF_DAY \
		or _is_daily_close_boundary(minute)


func _is_daily_close_boundary(minute: int) -> bool:
	var day := minute / MINUTES_PER_DAY
	return day > 0 and minute == get_daily_close_minute(day)


func _financial_day_for_boundary(boundary_minute: int) -> int:
	return boundary_minute / MINUTES_PER_DAY


func _daily_close_minute_for_day(day: int) -> int:
	return get_daily_close_minute(day)


func get_daily_close_minute(day: int) -> int:
	return day * MINUTES_PER_DAY + (MORNING_MINUTE_OF_DAY if day <= _legacy_calendar_last_day else 0)


func is_legacy_financial_history(day: int) -> bool:
	return day <= _legacy_history_last_day


func get_open_financial_day() -> int:
	return _day_summaries.size() + 1


func get_financial_period_start(day: int) -> int:
	return get_daily_close_minute(day - 1) if day > 1 else (MORNING_MINUTE_OF_DAY if _legacy_calendar_last_day > 0 else 0)


func _max_carry_for_config(config: Dictionary) -> int:
	return int(config[&"arrival_rate_denominator"]) \
	* int(config[&"max_arrivals_per_interval"]) \
	* int(config[&"arrival_carry_interval_limit"])


static func _is_integral_value(value: Variant) -> bool:
	return typeof(value) == TYPE_INT


static func _is_finite_numeric(value: Variant) -> bool:
	if typeof(value) == TYPE_INT:
		return true
	if typeof(value) != TYPE_FLOAT:
		return false
	return is_finite(float(value))


static func _is_save_safe(value: Variant, depth := 0) -> bool:
	if depth != 0:
		# Kept for compatibility with older internal callers; whole-tree budget
		# validation always begins at depth zero.
		return false
	return bool(_variant_tree_report(value)[&"ok"])


static func _variant_tree_report(value: Variant) -> Dictionary:
	# Shared per-traversal counters: [visited nodes, total string length].
	var budget: Array[int] = [0, 0]
	var issue := _inspect_variant_tree(value, budget, 0)
	if not issue.is_empty():
		return {&"ok": false, &"code": issue}
	return {
		&"ok": true,
		&"code": &"OK",
		&"nodes": budget[0],
		&"string_length": budget[1],
	}


static func _inspect_variant_tree(
	value: Variant,
	budget: Array[int],
	depth: int
) -> StringName:
	var issues: Array[StringName] = []
	_visit_snapshot_tree(value, budget, depth, issues, false)
	return issues[0] if not issues.is_empty() else &""


static func _snapshot_tree_report(value: Variant) -> Dictionary:
	var budget: Array[int] = [0, 0]
	var issues: Array[StringName] = []
	var encoded: Variant = _visit_snapshot_tree(value, budget, 0, issues, true)
	if not issues.is_empty():
		return {&"ok": false, &"code": issues[0]}
	return {&"ok": true, &"code": &"OK", &"nodes": budget[0],
		&"string_length": budget[1], &"value": encoded}


# Capture validated, canonical copies for unchanged root fields.
# The cache never survives create_snapshot and only root-depth subtrees reuse it.
# Use only for that synchronous internal-to-output conversion: these four fields
# have identical source values and depths in both root dictionaries. No callbacks
# or state mutation may occur between capture and reuse. Aggregate-limit failure
# still uses the original visitor to retain its exact error precedence.
static func _snapshot_root_report(source: Dictionary, cache: Dictionary, capture: bool) -> Dictionary:
	var budget: Array[int] = [1, 0]
	var issues: Array[StringName] = []
	var converted: Dictionary = {}
	if source.size() > MAX_SAFE_COLLECTION_SIZE:
		return {&"ok": false, &"code": &"COLLECTION_TOO_LARGE"}
	if not _canonical_key_collision(source).is_empty():
		return {&"ok": false, &"code": &"DUPLICATE_CANONICAL_KEY"}
	for key: Variant in source:
		if typeof(key) not in [TYPE_STRING, TYPE_STRING_NAME, TYPE_INT]:
			return {&"ok": false, &"code": &"INVALID_DICTIONARY_KEY"}
		budget[0] += 1
		if budget[0] > MAX_VARIANT_NODES:
			return {&"ok": false, &"code": &"MAX_VARIANT_NODES"}
		var name := str(key)
		var length := name.length()
		if length > MAX_SINGLE_STRING_LENGTH:
			return {&"ok": false, &"code": &"STRING_TOO_LONG"}
		budget[1] += length
		if budget[1] > MAX_TOTAL_STRING_LENGTH:
			return {&"ok": false, &"code": &"TOTAL_STRING_BUDGET"}
		var node_start := budget[0]
		var string_start := budget[1]
		var reusable := name in ["demand_history", "arrival_reports", "alerts", "quarantine_report"]
		var value: Variant
		var cached: Dictionary = cache.get(name, {}) if not capture else {}
		if not cached.is_empty() and budget[0] + cached[&"nodes"] <= MAX_VARIANT_NODES and budget[1] + cached[&"strings"] <= MAX_TOTAL_STRING_LENGTH:
			budget[0] += cached[&"nodes"]
			budget[1] += cached[&"strings"]
			value = cached[&"value"]
		else:
			# Fall back to the original source-order visitor when an aggregate
			# limit can fail, preserving which budget error occurs first.
			value = _visit_snapshot_tree(source[key], budget, 1, issues, not capture or reusable)
			if not issues.is_empty():
				return {&"ok": false, &"code": issues[0]}
		if capture and reusable:
			cache[name] = {&"nodes": budget[0] - node_start, &"strings": budget[1] - string_start, &"value": value}
		elif not capture:
			converted[name] = value
	converted.sort()
	return {&"ok": true, &"code": &"OK", &"nodes": budget[0], &"string_length": budget[1], &"value": converted}


static func _snapshot_tree_issue(issues: Array[StringName], code: StringName) -> Variant:
	issues.append(code)
	return null


static func _visit_snapshot_tree(
	value: Variant,
	budget: Array[int],
	depth: int,
	issues: Array[StringName],
	encode: bool
) -> Variant:
	# The validator and encoder share one traversal so their depth, key and
	# aggregate budgets cannot drift. Validation-only callers allocate no copies.
	if depth > MAX_VARIANT_DEPTH:
		return _snapshot_tree_issue(issues, &"MAX_VARIANT_DEPTH")
	budget[0] += 1
	if budget[0] > MAX_VARIANT_NODES:
		return _snapshot_tree_issue(issues, &"MAX_VARIANT_NODES")
	match typeof(value):
		TYPE_NIL, TYPE_BOOL, TYPE_INT:
			return value
		TYPE_FLOAT:
			return value if is_finite(float(value)) else _snapshot_tree_issue(issues, &"NON_FINITE_NUMBER")
		TYPE_STRING, TYPE_STRING_NAME:
			var text := String(value)
			if text.length() > MAX_SINGLE_STRING_LENGTH:
				return _snapshot_tree_issue(issues, &"STRING_TOO_LONG")
			budget[1] += text.length()
			if budget[1] > MAX_TOTAL_STRING_LENGTH:
				return _snapshot_tree_issue(issues, &"TOTAL_STRING_BUDGET")
			return text
		TYPE_ARRAY:
			var values: Array = value
			if values.size() > MAX_SAFE_COLLECTION_SIZE:
				return _snapshot_tree_issue(issues, &"COLLECTION_TOO_LARGE")
			var converted: Variant = [] if encode else null
			for child: Variant in values:
				var encoded: Variant = _visit_snapshot_tree(child, budget, depth + 1, issues, encode)
				if not issues.is_empty():
					return null
				if encode:
					converted.append(encoded)
			return converted
		TYPE_DICTIONARY:
			var source: Dictionary = value
			if source.size() > MAX_SAFE_COLLECTION_SIZE:
				return _snapshot_tree_issue(issues, &"COLLECTION_TOO_LARGE")
			if not _canonical_key_collision(source).is_empty():
				return _snapshot_tree_issue(issues, &"DUPLICATE_CANONICAL_KEY")
			var converted: Variant = {} if encode else null
			# Validate in source order, retaining the historical error precedence.
			for key: Variant in source:
				if typeof(key) not in [TYPE_STRING, TYPE_STRING_NAME, TYPE_INT]:
					return _snapshot_tree_issue(issues, &"INVALID_DICTIONARY_KEY")
				if depth + 1 > MAX_VARIANT_DEPTH:
					return _snapshot_tree_issue(issues, &"MAX_VARIANT_DEPTH")
				budget[0] += 1
				if budget[0] > MAX_VARIANT_NODES:
					return _snapshot_tree_issue(issues, &"MAX_VARIANT_NODES")
				var string_key := str(key)
				var key_length := string_key.length()
				if key_length > MAX_SINGLE_STRING_LENGTH:
					return _snapshot_tree_issue(issues, &"STRING_TOO_LONG")
				budget[1] += key_length
				if budget[1] > MAX_TOTAL_STRING_LENGTH:
					return _snapshot_tree_issue(issues, &"TOTAL_STRING_BUDGET")
				var encoded: Variant = _visit_snapshot_tree(source[key], budget, depth + 1, issues, encode)
				if not issues.is_empty():
					return null
				if encode:
					converted[string_key] = encoded
			if not encode:
				return null
			# Every key is now a unique String. Native string sorting preserves
			# canonical byte order without repeated Variant-to-string callbacks.
			converted.sort()
			return converted
		_:
			return _snapshot_tree_issue(issues, &"UNSAVEABLE_VARIANT")


static func _canonical_key_collision(source: Dictionary) -> Dictionary:
	var seen: Dictionary = {}
	for key: Variant in source:
		var canonical := str(key)
		if seen.has(canonical):
			return {
				&"canonical_key": canonical,
				&"first_type": typeof(seen[canonical]),
				&"second_type": typeof(key),
			}
		seen[canonical] = key
	return {}


static func _to_snapshot_variant(value: Variant) -> Variant:
	match typeof(value):
		TYPE_STRING_NAME:
			return String(value)
		TYPE_ARRAY:
			var result: Array = []
			for child: Variant in value:
				result.append(_to_snapshot_variant(child))
			return result
		TYPE_DICTIONARY:
			var source: Dictionary = value
			var result: Dictionary = {}
			var keys: Array = source.keys()
			keys.sort_custom(_variant_string_less)
			for key: Variant in keys:
				result[str(key)] = _to_snapshot_variant(source[key])
			return result
		_:
			return value


static func _value(source: Dictionary, key: StringName, default_value: Variant) -> Variant:
	if source.has(key):
		return source[key]
	var string_key := String(key)
	if source.has(string_key):
		return source[string_key]
	return default_value


static func _sorted_string_keys(source: Dictionary) -> Array[String]:
	var result: Array[String] = []
	for key: Variant in source:
		result.append(str(key))
	result.sort()
	return result


static func _sorted_int_keys(source: Dictionary) -> Array[int]:
	var result: Array[int] = []
	for key: Variant in source:
		result.append(int(key))
	result.sort()
	return result


static func _rotated_dictionary_rows(
	rows: Array[Dictionary],
	start_index: int
) -> Array[Dictionary]:
	var result: Array[Dictionary] = []
	if rows.is_empty():
		return result
	for offset in rows.size():
		result.append(rows[posmod(start_index + offset, rows.size())])
	return result


static func _sequence_from_id(identifier: String, prefix: String) -> int:
	if not identifier.begins_with(prefix):
		return -1
	var suffix := identifier.substr(prefix.length())
	if suffix.length() != 8 or not suffix.is_valid_int():
		return -1
	var value := int(suffix)
	return value if value > 0 and identifier == "%s%08d" % [prefix, value] else -1


static func _string_key_dictionary(source: Dictionary) -> Dictionary:
	var result: Dictionary = {}
	var keys: Array = source.keys()
	keys.sort_custom(_variant_string_less)
	for key: Variant in keys:
		var value: Variant = source[key]
		result[str(key)] = value.duplicate(true) if value is Array or value is Dictionary else value
	return result


static func _copy_dictionary_array(source: Array) -> Array[Dictionary]:
	var result: Array[Dictionary] = []
	for value: Variant in source:
		assert(value is Dictionary, "Dictionary-array parser must reject non-dictionary rows before copy.")
		result.append((value as Dictionary).duplicate(true))
	return result


static func _variant_string_less(left: Variant, right: Variant) -> bool:
	return str(left) < str(right)


static func _success(details: Dictionary = {}) -> CommandResult:
	return CommandResultType.succeeded(0, details)


func _failure(code: StringName, message: String, details: Dictionary = {}) -> CommandResult:
	last_error = message
	var enriched := details.duplicate(true)
	enriched[&"code"] = code
	return CommandResultType.failed(
		CommandResultType.Code.VALIDATION_FAILED,
		0,
		message,
		enriched
	)


func _boundary_failure(code: StringName, report: Dictionary) -> CommandResult:
	return _failure(
		code,
		"An operations boundary failed without committing coordinator state.",
		{&"boundary": report.duplicate(true)}
	)


func is_legacy_principal_profit_history(day: int) -> bool:
	return day > 0 and day <= _legacy_principal_profit_last_day