settled-reach/client/tests/test_step_canvas_disk_cache.gd

## T-1183 tests: step_canvas_disk_cache.gd — the disk-backed Tier 2/3 cache
## store layered beneath step_canvas_cache.gd's in-memory Tier 1 (D-255(d)).
## Covers: index round-trip (write/read/evict), the two sweeps' independence
## (a storage sweep never touches an in-TTL sim-state entry and vice versa),
## the Global retention floor, the per-body deep-rung cap, the version-tag
## mismatch path (miss + re-fetch, NEVER decode — asserted via "no canvas
## payload survives," not just "no crash"), persistence across a simulated
## restart (a fresh instance reads the same on-disk root), malformed index
## recovery (truncated/corrupt index -> empty cache, never a crash), orphan
## payload reconciliation (a stray .dat file with no index row is reclaimed
## by the background sweep; an indexed sibling survives untouched), corrupt-
## payload self-heal (garbage bytes in a payload file degrade to a miss and
## drop both the index row and the file, never a crash), the atomic index
## write (no .tmp file survives a completed put(), index.json always parses),
## and size_bytes accuracy (matches the payload file's actual on-disk length,
## captured at write time rather than re-read).
##
## Every test uses an INJECTED, disposable root path
## (user://test_step_canvas_disk_cache/<unique>/) — never the real
## user://atlas_cache/ — and removes it in after_test() so test runs never
## leak state into each other or into a real cache directory.
class_name TestStepCanvasDiskCache
extends GdUnitTestSuite

const StepCanvasDiskCache := preload(
	"res://ui/implant/apps/atlas/step_canvas/step_canvas_disk_cache.gd"
)

var _test_root: String = ""


func before_test() -> void:
	_test_root = "user://test_step_canvas_disk_cache/%d/" % Time.get_ticks_usec()


func after_test() -> void:
	var probe := StepCanvasDiskCache.new(_test_root)
	probe.clear_all()


## Minimal canvas-shaped stand-in for tests that only exercise cache
## mechanics (eviction, sweeps, round-tripping) and don't care about real
## canvas contents. Always carries width/height — get_canvas()'s shape guard
## (mirroring step_canvas_protocol.gd's own `_decode_encoded_canvas`
## discriminator) treats a payload missing either as corrupt, so every put()
## fixture in this suite must satisfy that contract, not just production
## payloads. `extra` merges in each test's own distinguishing marker
## key(s) — centralizing the shape here means a future shape-contract change
## touches one function, not every put() call site.
func _stub_canvas(extra: Dictionary = {}) -> Dictionary:
	var canvas: Dictionary = {"width": 1, "height": 1}
	canvas.merge(extra, true)
	return canvas


# =============================================================================
# Index round-trip: write / read / evict
# =============================================================================


func test_miss_returns_null_and_has_reports_false() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	assert_that(cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).is_null()
	assert_bool(cache.has("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).is_false()


func test_put_then_get_round_trips_exact_canvas_including_png_bytes() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	var canvas := {
		"width": 64,
		"height": 64,
		"morphology": PackedByteArray([137, 80, 78, 71, 1, 2, 3]),  # PNG magic + junk, pre-decode
		"courses": [{"a": 1}],
	}
	cache.put("GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64), canvas)
	assert_bool(cache.has("GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64))).is_true()
	var got: Variant = cache.get_canvas("GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64))
	assert_that(got).is_equal(canvas)


func test_put_overwrites_existing_key() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 2}))
	assert_int(cache.entry_count("GJ1c")).is_equal(1)
	assert_that(cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).is_equal(
		_stub_canvas({"v": 2})
	)


func test_different_rungs_at_identical_center_extent_do_not_collide() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put(
		"GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64), _stub_canvas({"rung": "District"})
	)
	cache.put("GJ1c", "Chunk", Vector2i(10, 20), Vector2i(64, 64), _stub_canvas({"rung": "Chunk"}))
	assert_int(cache.entry_count("GJ1c")).is_equal(2)
	assert_that(cache.get_canvas("GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64))).is_equal(
		_stub_canvas({"rung": "District"})
	)
	assert_that(cache.get_canvas("GJ1c", "Chunk", Vector2i(10, 20), Vector2i(64, 64))).is_equal(
		_stub_canvas({"rung": "Chunk"})
	)


func test_different_bodies_do_not_collide() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64), _stub_canvas({"body": "c"}))
	cache.put("GJ1d", "District", Vector2i(10, 20), Vector2i(64, 64), _stub_canvas({"body": "d"}))
	assert_that(cache.get_canvas("GJ1c", "District", Vector2i(10, 20), Vector2i(64, 64))).is_equal(
		_stub_canvas({"body": "c"})
	)
	assert_that(cache.get_canvas("GJ1d", "District", Vector2i(10, 20), Vector2i(64, 64))).is_equal(
		_stub_canvas({"body": "d"})
	)


func test_get_touches_last_read_at() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	cache._debug_backdate_entry(
		"GJ1c",
		StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64)),
		1000,
		1000
	)
	var before_read: int = Time.get_unix_time_from_system()
	cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	# A fresh instance re-reads the persisted last_read_at from disk — proves
	# the touch was actually written through, not just held in memory.
	var reloaded := StepCanvasDiskCache.new(_test_root)
	reloaded.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	# The visit sweep, run immediately after, must NOT evict — last_read_at
	# was just updated to "now", far inside STORAGE_TTL_SEC.
	reloaded.run_visit_sweep("GJ1c")
	assert_bool(reloaded.has("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).override_failure_message(
		"a just-touched entry must survive an immediate visit sweep"
	).is_true()
	assert_int(before_read).is_greater_equal(1000)  # sanity: real clock, not the backdated stub


# =============================================================================
# Global retention floor
# =============================================================================


func test_global_rung_entry_is_flagged_retention_floor() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO, {"width": 1024, "height": 512})
	var key := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
	assert_bool(cache.has_entry_for_key("GJ1c", key)).is_true()


func test_global_entry_survives_visit_sweep_even_when_ancient() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO, _stub_canvas())
	var key := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
	# Backdate WAY past STORAGE_TTL_SEC (14 days) — a floored entry must be
	# immune regardless of how old last_read_at is.
	cache._debug_backdate_entry("GJ1c", key, 0, 0)
	cache.run_visit_sweep("GJ1c")
	assert_bool(cache.has("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)).override_failure_message(
		"the Global/rung-0 entry must never be evicted by the visit sweep"
	).is_true()


func test_global_entry_survives_capacity_sweep_even_when_it_alone_exceeds_budget() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO, _stub_canvas())
	var key := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
	# Force size_bytes artificially huge by backdating is not enough for
	# size — but the floored flag alone must exempt it from run_capacity_sweep
	# regardless of size accounting, so it is excluded from the sub-global
	# byte total entirely (verified indirectly: it survives even though this
	# test's other sub-global entries below would normally trigger eviction).
	for i in range(3):
		cache.put(
			"GJ1c", "Chunk", Vector2i(i, 0), Vector2i(64, 64), _stub_canvas({"pad": PackedByteArray()})
		)
	cache.run_capacity_sweep("GJ1c")
	assert_bool(cache.has("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)).is_true()


# =============================================================================
# The two Tier-2/3 sweeps are independent
# =============================================================================


func test_visit_sweep_evicts_old_geometry_entry_past_storage_ttl() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i(1, 1), Vector2i(64, 64), _stub_canvas({"v": 1}))
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i(1, 1), Vector2i(64, 64))
	var ancient: int = Time.get_unix_time_from_system() - StepCanvasDiskCache.STORAGE_TTL_SEC - 10
	cache._debug_backdate_entry("GJ1c", key, ancient, ancient)
	cache.run_visit_sweep("GJ1c")
	assert_bool(cache.has_entry_for_key("GJ1c", key)).override_failure_message(
		"an entry older than STORAGE_TTL_SEC must be evicted by the visit sweep"
	).is_false()


func test_visit_sweep_never_touches_an_in_ttl_sim_state_entry() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	# A recently-written sim-state entry with a LONG sim_ttl — well within
	# both its own TTL and STORAGE_TTL_SEC. The visit sweep (storage axis)
	# must not evict it purely because it's SimState-tagged.
	cache.put(
		"GJ1c", "Chunk", Vector2i(2, 2), Vector2i(64, 64), _stub_canvas({"flooded": true}), 0, 999_999
	)
	cache.run_visit_sweep("GJ1c")
	assert_bool(cache.has("GJ1c", "Chunk", Vector2i(2, 2), Vector2i(64, 64))).is_true()


func test_staleness_sweep_never_touches_a_geometry_entry_regardless_of_age() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i(3, 3), Vector2i(64, 64), _stub_canvas({"v": 1}))
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i(3, 3), Vector2i(64, 64))
	# Backdate far into the past — if run_staleness_sweep incorrectly treated
	# geometry as staleness-bound, this would be evicted.
	cache._debug_backdate_entry("GJ1c", key, 0, 0)
	cache.run_staleness_sweep("GJ1c")
	assert_bool(cache.has_entry_for_key("GJ1c", key)).override_failure_message(
		"a Geometry-tier entry has no sim_ttl and must never be evicted by the staleness sweep"
	).is_true()


func test_staleness_sweep_evicts_sim_state_entry_past_its_own_ttl_even_if_recently_visited() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put(
		"GJ1c", "Chunk", Vector2i(4, 4), Vector2i(64, 64), _stub_canvas({"flooded": true}), 0, 100
	)
	var key := StepCanvasDiskCache.make_key("GJ1c", "Chunk", Vector2i(4, 4), Vector2i(64, 64))
	var now: int = Time.get_unix_time_from_system()
	# written_at far enough in the past that written_at + sim_ttl(100) < now,
	# but last_read_at is "just now" — proving capacity/visit recency does
	# NOT protect a sim-state entry from its own staleness clock.
	cache._debug_backdate_entry("GJ1c", key, now - 500, now)
	cache.run_staleness_sweep("GJ1c")
	assert_bool(cache.has_entry_for_key("GJ1c", key)).override_failure_message(
		"a sim-state entry past its own sim_ttl must be evicted regardless of last_read_at"
	).is_false()


func test_capacity_sweep_never_touches_an_entry_under_budget() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i(5, 5), Vector2i(64, 64), _stub_canvas({"v": 1}))
	cache.run_capacity_sweep("GJ1c")
	assert_bool(cache.has("GJ1c", "District", Vector2i(5, 5), Vector2i(64, 64))).is_true()


# =============================================================================
# Per-body deep-rung retention cap (D-255(d) hardening (i))
# =============================================================================


func test_deep_rung_cap_evicts_oldest_deep_rung_entry_when_exceeded() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	# Fill past the cap: MAX_DEEP_RUNG_ENTRIES_PER_BODY + 2 distinct Chunk
	# windows for one body, backdating each so eviction order is
	# deterministic (oldest last_read_at goes first).
	var cap: int = StepCanvasDiskCache.MAX_DEEP_RUNG_ENTRIES_PER_BODY
	for i in range(cap + 2):
		cache.put("GJ1c", "Chunk", Vector2i(i, 0), Vector2i(64, 64), _stub_canvas({"i": i}))
		var key := StepCanvasDiskCache.make_key("GJ1c", "Chunk", Vector2i(i, 0), Vector2i(64, 64))
		cache._debug_backdate_entry("GJ1c", key, i, i)  # strictly increasing "recency"

	assert_int(cache.entry_count("GJ1c")).override_failure_message(
		"deep-rung entries must never transiently exceed the cap, even mid-fill"
	).is_equal(cap)

	# The two OLDEST (i=0, i=1) must have been evicted; the newest must survive.
	assert_bool(cache.has("GJ1c", "Chunk", Vector2i(0, 0), Vector2i(64, 64))).is_false()
	assert_bool(cache.has("GJ1c", "Chunk", Vector2i(1, 0), Vector2i(64, 64))).is_false()
	assert_bool(
		cache.has("GJ1c", "Chunk", Vector2i(cap + 1, 0), Vector2i(64, 64))
	).override_failure_message("the most-recently-written deep-rung entry must survive").is_true()


func test_deep_rung_cap_does_not_apply_to_shallow_rungs() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	var cap: int = StepCanvasDiskCache.MAX_DEEP_RUNG_ENTRIES_PER_BODY
	for i in range(cap + 5):
		cache.put("GJ1c", "District", Vector2i(i, 0), Vector2i(64, 64), _stub_canvas({"i": i}))
	assert_int(cache.entry_count("GJ1c")).override_failure_message(
		"District is not a DEEP_RUNGS member — the deep-rung cap must not evict it"
	).is_equal(cap + 5)


func test_deep_rung_cap_never_evicts_the_global_floor_entry() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO, _stub_canvas())
	var cap: int = StepCanvasDiskCache.MAX_DEEP_RUNG_ENTRIES_PER_BODY
	for i in range(cap + 3):
		cache.put("GJ1c", "Chunk", Vector2i(i, 0), Vector2i(64, 64), _stub_canvas({"i": i}))
	assert_bool(cache.has("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)).override_failure_message(
		"the Global floor entry must survive deep-rung cap eviction regardless of fill order"
	).is_true()


func test_deep_rung_cap_is_per_body_not_global() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	var cap: int = StepCanvasDiskCache.MAX_DEEP_RUNG_ENTRIES_PER_BODY
	for i in range(cap):
		cache.put("GJ1c", "Chunk", Vector2i(i, 0), Vector2i(64, 64), _stub_canvas({"i": i}))
	# A second body's Chunk entries must not be capped by the first body's
	# fill level — the cap is per-body.
	cache.put("GJ1d", "Chunk", Vector2i(0, 0), Vector2i(64, 64), _stub_canvas({"body": "d"}))
	assert_bool(cache.has("GJ1d", "Chunk", Vector2i(0, 0), Vector2i(64, 64))).is_true()
	assert_int(cache.entry_count("GJ1c")).is_equal(cap)


# =============================================================================
# Version-tag mismatch (D-255(d) hardening (ii)) — miss + re-fetch, NEVER decode
# =============================================================================


func test_matching_schema_version_is_a_hit() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	assert_that(cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).is_not_null()


func test_mismatched_schema_version_is_a_miss_and_drops_the_entry() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))

	# Simulate "written by an older/newer build": write a payload file
	# directly with a mismatched schema_version in the index, bypassing
	# put() (which always stamps the CURRENT version).
	var idx: Dictionary = cache._load_index("GJ1c")
	var entry: Dictionary = idx[key]
	entry["schema_version"] = "0.0.1-not-the-current-version"
	idx[key] = entry
	cache._save_index("GJ1c")

	# get_canvas() on the mismatched entry must return null — a MISS, not a
	# decode attempt (there is nothing here to "decode" beyond var_to_bytes
	# reconstruction of the stored Dictionary itself, so the stronger,
	# directly-assertable claim is: no canvas payload is ever handed back).
	var result: Variant = cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	assert_that(result).override_failure_message(
		"a schema-version-mismatched entry must be treated as a miss, never handed to the caller"
	).is_null()

	# The stale entry must also be DROPPED (re-fetch is the only path back
	# in — a lingering mismatched index row would just re-fail every read).
	assert_bool(cache.has_entry_for_key("GJ1c", key)).override_failure_message(
		"a version-mismatched entry must be dropped from the index on read, not left dangling"
	).is_false()


func test_mismatched_schema_version_has_reports_false() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	var idx: Dictionary = cache._load_index("GJ1c")
	var entry: Dictionary = idx[key]
	entry["schema_version"] = "9.9.9-future"
	idx[key] = entry
	cache._save_index("GJ1c")
	assert_bool(cache.has("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).is_false()


func test_current_schema_version_reads_project_yaml() -> void:
	# project.yaml's version field is "0.4.0" per CLAUDE.md at time of
	# writing — assert only that a non-placeholder value comes back (not the
	# exact string, so this test doesn't need updating every version bump).
	var v: String = StepCanvasDiskCache.current_schema_version()
	assert_str(v).override_failure_message(
		"current_schema_version() must read a real value from project.yaml, not the '?' fallback"
	).is_not_equal("?.?.?")


# =============================================================================
# Persistence across a simulated restart
# =============================================================================


func test_fresh_instance_reads_entries_written_by_a_prior_instance() -> void:
	var writer := StepCanvasDiskCache.new(_test_root)
	writer.put(
		"GJ1c",
		"Quarter",
		Vector2i(7, 8),
		Vector2i(64, 64),
		_stub_canvas({"morphology": PackedByteArray([1, 2, 3])})
	)

	# A brand-new instance, same root — simulates a game restart. No shared
	# in-memory state whatsoever between `writer` and `reader`.
	var reader := StepCanvasDiskCache.new(_test_root)
	var got: Variant = reader.get_canvas("GJ1c", "Quarter", Vector2i(7, 8), Vector2i(64, 64))
	assert_that(got).is_equal(_stub_canvas({"morphology": PackedByteArray([1, 2, 3])}))


func test_fresh_instance_sees_global_floor_flag_persisted_by_a_prior_instance() -> void:
	var writer := StepCanvasDiskCache.new(_test_root)
	writer.put("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO, _stub_canvas())
	var key := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
	writer._debug_backdate_entry("GJ1c", key, 0, 0)  # ancient, would evict if not floored

	var reader := StepCanvasDiskCache.new(_test_root)
	reader.run_visit_sweep("GJ1c")
	assert_bool(reader.has("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)).override_failure_message(
		"the retention-floor flag must persist across a simulated restart"
	).is_true()


# =============================================================================
# Malformed index recovery
# =============================================================================


func test_missing_index_file_is_treated_as_empty_cache() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	# No put() has ever happened for this body_id — no index file exists.
	assert_int(cache.entry_count("nonexistent_body")).is_equal(0)
	assert_that(cache.get_canvas("nonexistent_body", "District", Vector2i.ZERO, Vector2i(64, 64))).is_null()


func test_truncated_index_json_is_treated_as_empty_cache_not_a_crash() -> void:
	var body_id := "GJ1c"
	var dir_err := DirAccess.make_dir_recursive_absolute(_test_root + body_id + "/")
	assert_int(dir_err).is_equal(OK)
	var file := FileAccess.open(_test_root + body_id + "/index.json", FileAccess.WRITE)
	file.store_string('{"some_key": {"written_at": 123, "tru')  # deliberately truncated
	file.close()

	var cache := StepCanvasDiskCache.new(_test_root)
	assert_int(cache.entry_count(body_id)).is_equal(0)
	assert_that(cache.get_canvas(body_id, "District", Vector2i.ZERO, Vector2i(64, 64))).is_null()

	# The store must still be USABLE after a corrupt read — put()/get() work
	# normally and overwrite the bad file with a valid one on next save.
	cache.put(body_id, "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	assert_that(cache.get_canvas(body_id, "District", Vector2i.ZERO, Vector2i(64, 64))).is_equal(
		_stub_canvas({"v": 1})
	)


func test_index_with_wrong_top_level_json_type_is_treated_as_empty_cache() -> void:
	var body_id := "GJ1c"
	DirAccess.make_dir_recursive_absolute(_test_root + body_id + "/")
	var file := FileAccess.open(_test_root + body_id + "/index.json", FileAccess.WRITE)
	file.store_string("[1, 2, 3]")  # a JSON array, not the expected Dictionary
	file.close()

	var cache := StepCanvasDiskCache.new(_test_root)
	assert_int(cache.entry_count(body_id)).is_equal(0)


func test_dangling_index_entry_with_missing_payload_file_is_a_miss() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	var idx: Dictionary = cache._load_index("GJ1c")
	var entry: Dictionary = idx[key]
	# Delete the payload file directly (simulates platform storage pressure
	# clearing files without updating the index), leaving a dangling row.
	DirAccess.remove_absolute(str(entry["file_path"]))

	assert_that(cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).is_null()
	assert_bool(cache.has_entry_for_key("GJ1c", key)).override_failure_message(
		"a dangling index row (missing payload file) must be dropped on the miss, not left behind"
	).is_false()


# =============================================================================
# Orphan payload reconciliation (fix for the payload-before-index write
# order's other failure direction — a discarded/corrupt index, or a crash
# between the payload write and the index write, orphans a .dat file with
# no index row pointing at it)
# =============================================================================


func test_background_sweep_removes_an_orphaned_payload_with_no_index_row() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))

	# Drop a stray payload file directly into the body dir, bypassing put() —
	# simulates a crash between the payload write and the index write, or the
	# aftermath of a discarded corrupt index.
	var orphan_path := _test_root + "GJ1c/deadbeef.dat"
	var orphan_file := FileAccess.open(orphan_path, FileAccess.WRITE)
	orphan_file.store_var(_stub_canvas({"orphan": true}))
	orphan_file.close()
	assert_bool(FileAccess.file_exists(orphan_path)).override_failure_message(
		"test setup: the stray payload file must exist before the sweep runs"
	).is_true()

	cache.run_background_sweep("GJ1c")

	assert_bool(FileAccess.file_exists(orphan_path)).override_failure_message(
		"an orphaned .dat file (no index row referencing it) must be removed by the background sweep"
	).is_false()
	# The INDEXED entry sharing the same directory must be untouched — the
	# sweep must not treat "not the orphan" as "collateral damage".
	assert_that(cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))).override_failure_message(
		"an indexed entry's payload must survive the orphan sweep and remain a hit"
	).is_equal(_stub_canvas({"v": 1}))


# =============================================================================
# Corrupt payload degrades to a self-healing miss (payload-shape guard)
# =============================================================================


func test_corrupt_payload_bytes_degrade_to_miss_with_self_heal() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	var idx: Dictionary = cache._load_index("GJ1c")
	var payload_path := str(idx[key]["file_path"])

	# Overwrite the payload file with garbage bytes that are not a valid
	# store_var() encoding at all — a stronger corruption than a merely
	# wrong-shaped Dictionary, exercising the get_var() -> null path.
	var garbage_file := FileAccess.open(payload_path, FileAccess.WRITE)
	garbage_file.store_buffer(PackedByteArray([0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x01]))
	garbage_file.close()

	var result: Variant = cache.get_canvas("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	assert_that(result).override_failure_message(
		"a corrupt payload file must degrade to a miss, never crash or return a partial canvas"
	).is_null()
	assert_bool(cache.has_entry_for_key("GJ1c", key)).override_failure_message(
		"a corrupt payload's index row must be dropped as part of the self-heal"
	).is_false()
	assert_bool(FileAccess.file_exists(payload_path)).override_failure_message(
		"a corrupt payload file must be deleted as part of the self-heal"
	).is_false()


# =============================================================================
# Atomic index write
# =============================================================================


func test_index_write_leaves_no_tmp_file_and_index_json_is_valid() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))

	var index_path := _test_root + "GJ1c/index.json"
	var tmp_path := index_path + ".tmp"
	assert_bool(FileAccess.file_exists(tmp_path)).override_failure_message(
		"no .tmp file should remain next to index.json after a completed put()"
	).is_false()

	var file := FileAccess.open(index_path, FileAccess.READ)
	var content := file.get_as_text()
	file.close()
	var json := JSON.new()
	var err := json.parse(content)
	assert_int(err).override_failure_message(
		"index.json must parse as valid JSON after an atomic write"
	).is_equal(OK)
	assert_bool(json.data is Dictionary).is_true()


# =============================================================================
# size_bytes accuracy (captured at write time, no re-read)
# =============================================================================


func test_size_bytes_matches_actual_payload_file_length_on_disk() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put(
		"GJ1c",
		"District",
		Vector2i.ZERO,
		Vector2i(64, 64),
		_stub_canvas({"morphology": PackedByteArray([1, 2, 3, 4, 5])})
	)
	var key := StepCanvasDiskCache.make_key("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64))
	var idx: Dictionary = cache._load_index("GJ1c")
	var entry: Dictionary = idx[key]
	var payload_path := str(entry["file_path"])
	var actual_size: int = FileAccess.get_file_as_bytes(payload_path).size()
	assert_int(int(entry["size_bytes"])).override_failure_message(
		"size_bytes must equal the payload file's actual on-disk length"
	).is_equal(actual_size)


# =============================================================================
# Key discipline — mirrors StepCanvasCache exactly (shared make_key())
# =============================================================================


func test_make_key_global_ignores_center_and_extent() -> void:
	var k1 := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i(10, 20), Vector2i(64, 64))
	var k2 := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i(999, -999), Vector2i(1, 1))
	assert_str(k1).is_equal(k2)


func test_clear_all_removes_every_body_and_the_root_directory() -> void:
	var cache := StepCanvasDiskCache.new(_test_root)
	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 1}))
	cache.put("GJ1d", "Chunk", Vector2i.ZERO, Vector2i(64, 64), _stub_canvas({"v": 2}))
	cache.clear_all()
	var reader := StepCanvasDiskCache.new(_test_root)
	assert_int(reader.entry_count("GJ1c")).is_equal(0)
	assert_int(reader.entry_count("GJ1d")).is_equal(0)
	assert_bool(DirAccess.dir_exists_absolute(_test_root)).is_false()