fix(ui): harden step-canvas disk cache — PR #204 review round (T-1183)

Atomic index writes (tmp+rename), orphan-payload reconciliation sweep
folded into the background sweep, payload shape guard mirroring the
protocol's own width/height discriminator, size_bytes via get_position()
instead of a full payload re-read, and a 64-bit SHA-256 payload filename
(String.hash()'s 31-bit space made a silent wrong-map filename collision
a ~1-in-16k event per cap-full body; migration self-heals via the orphan
sweep). Tier-3 class doc rewritten to name the real D-253 seam — the
glaciation/flooded_q wire fields exist but carry static values and no
staleness signal crosses the wire; T-1190 tracks threading a wire TTL
into put(sim_ttl_sec) when the driving clock lands — replacing the false
'no sim-state wire field exists' premise. Four new tests plus a
_stub_canvas fixture helper centralizing the canvas shape contract.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

client/tests/test_step_canvas_disk_cache.gd

@@ -5,8 +5,15 @@
 ## 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), and malformed
-## index recovery (truncated/corrupt index -> empty cache, never a crash).
+## 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
@@ -31,6 +38,21 @@ func after_test() -> void:
 	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
 # =============================================================================
@@ -58,42 +80,44 @@ func test_put_then_get_round_trips_exact_canvas_including_png_bytes() -> void:
 
 func test_put_overwrites_existing_key() -> void:
 	var cache := StepCanvasDiskCache.new(_test_root)
-	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), {"v": 1})
-	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), {"v": 2})
+	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(
-		{"v": 2}
+		_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), {"rung": "District"})
-	cache.put("GJ1c", "Chunk", Vector2i(10, 20), Vector2i(64, 64), {"rung": "Chunk"})
+	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(
-		{"rung": "District"}
+		_stub_canvas({"rung": "District"})
 	)
 	assert_that(cache.get_canvas("GJ1c", "Chunk", Vector2i(10, 20), Vector2i(64, 64))).is_equal(
-		{"rung": "Chunk"}
+		_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), {"body": "c"})
-	cache.put("GJ1d", "District", Vector2i(10, 20), Vector2i(64, 64), {"body": "d"})
+	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(
-		{"body": "c"}
+		_stub_canvas({"body": "c"})
 	)
 	assert_that(cache.get_canvas("GJ1d", "District", Vector2i(10, 20), Vector2i(64, 64))).is_equal(
-		{"body": "d"}
+		_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), {"v": 1})
+	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)),
@@ -129,7 +153,7 @@ func test_global_rung_entry_is_flagged_retention_floor() -> void:
 
 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, {"width": 1})
+	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.
@@ -142,7 +166,7 @@ func test_global_entry_survives_visit_sweep_even_when_ancient() -> void:
 
 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, {"width": 1})
+	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
@@ -151,7 +175,7 @@ func test_global_entry_survives_capacity_sweep_even_when_it_alone_exceeds_budget
 	# 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), {"pad": PackedByteArray()}
+			"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()
@@ -164,7 +188,7 @@ func test_global_entry_survives_capacity_sweep_even_when_it_alone_exceeds_budget
 
 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), {"v": 1})
+	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)
@@ -179,14 +203,16 @@ func test_visit_sweep_never_touches_an_in_ttl_sim_state_entry() -> void:
 	# 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), {"flooded": true}, 0, 999_999)
+	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), {"v": 1})
+	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.
@@ -199,7 +225,9 @@ func test_staleness_sweep_never_touches_a_geometry_entry_regardless_of_age() ->
 
 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), {"flooded": true}, 0, 100)
+	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,
@@ -214,7 +242,7 @@ func test_staleness_sweep_evicts_sim_state_entry_past_its_own_ttl_even_if_recent
 
 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), {"v": 1})
+	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()
 
@@ -231,7 +259,7 @@ func test_deep_rung_cap_evicts_oldest_deep_rung_entry_when_exceeded() -> void:
 	# 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), {"i": i})
+		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"
 
@@ -251,7 +279,7 @@ 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), {"i": i})
+		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)
@@ -259,10 +287,10 @@ func test_deep_rung_cap_does_not_apply_to_shallow_rungs() -> void:
 
 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, {"width": 1})
+	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), {"i": i})
+		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()
@@ -272,10 +300,10 @@ 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), {"i": i})
+		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), {"body": "d"})
+	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)
 
@@ -287,13 +315,13 @@ func test_deep_rung_cap_is_per_body_not_global() -> void:
 
 func test_matching_schema_version_is_a_hit() -> void:
 	var cache := StepCanvasDiskCache.new(_test_root)
-	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), {"v": 1})
+	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), {"v": 1})
+	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
@@ -323,7 +351,7 @@ func test_mismatched_schema_version_is_a_miss_and_drops_the_entry() -> void:
 
 func test_mismatched_schema_version_has_reports_false() -> void:
 	var cache := StepCanvasDiskCache.new(_test_root)
-	cache.put("GJ1c", "District", Vector2i.ZERO, Vector2i(64, 64), {"v": 1})
+	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]
@@ -355,21 +383,19 @@ func test_fresh_instance_reads_entries_written_by_a_prior_instance() -> void:
 		"Quarter",
 		Vector2i(7, 8),
 		Vector2i(64, 64),
-		{"width": 64, "morphology": PackedByteArray([1, 2, 3])}
+		_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(
-		{"width": 64, "morphology": PackedByteArray([1, 2, 3])}
-	)
+	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, {"width": 1})
+	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
 
@@ -406,9 +432,9 @@ func test_truncated_index_json_is_treated_as_empty_cache_not_a_crash() -> void:
 
 	# 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), {"v": 1})
+	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(
-		{"v": 1}
+		_stub_canvas({"v": 1})
 	)
 
 
@@ -425,7 +451,7 @@ func test_index_with_wrong_top_level_json_type_is_treated_as_empty_cache() -> vo
 
 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), {"v": 1})
+	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]
@@ -439,6 +465,122 @@ func test_dangling_index_entry_with_missing_payload_file_is_a_miss() -> void:
 	).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())
 # =============================================================================
@@ -452,8 +594,8 @@ func test_make_key_global_ignores_center_and_extent() -> void:
 
 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), {"v": 1})
-	cache.put("GJ1d", "Chunk", Vector2i.ZERO, Vector2i(64, 64), {"v": 2})
+	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)

client/ui/implant/apps/atlas/step_canvas/step_canvas_disk_cache.gd

@@ -42,12 +42,20 @@ extends RefCounted
 ##            first, run on a coarse background timer.
 ##   Tier 3 (sim-state-tagged planes) — explicit TTL, staleness-motivated,
 ##     structurally separate from tiers 1/2 (`now > written_at + sim_ttl`),
-##     never touched by 2a/2b. NOT YET POPULATED: every EncodedStepCanvas
-##     field is geometry per D-227 today (frozen/flooded ride the existing
-##     glaciation/morphology-water-class fields, Araminta's round-1 schema —
-##     no distinct sim-state WIRE field exists yet for this ticket to tag).
-##     The `sim_ttl`/tier machinery below is WIRED and tested but has no
-##     production caller until a sim-state field lands on EncodedStepCanvas.
+##     never touched by 2a/2b. NOT YET POPULATED, but not for lack of a wire
+##     field: `glaciation`/`flooded_q` ARE distinct sim-state fields on
+##     EncodedStepCanvas today (step_canvas_protocol.gd, Araminta's round-2
+##     sim-state plane, T-1181). The real seam is that their VALUES are still
+##     static — `flooded_q` hardwired 0, `glaciation` the static
+##     DistrictProfile classification — byte-identical to an indefinitely-
+##     fresh derive (server/src/atlas/step_canvas.rs module doc, D-253
+##     wired-ahead), and no staleness signal crosses the wire at all yet (the
+##     server's SIM_STATE_TTL formula is tick-based and server-side only).
+##     TIER_GEOMETRY is therefore the CORRECT classification for every canvas
+##     today. The `sim_ttl`/tier machinery below is WIRED and tested but has
+##     no production caller until D-253's sim-state driving clock lands and
+##     the response gains a real TTL for on_response() to thread into the
+##     existing `put(sim_ttl_sec)` parameter — tracked in T-1190.
 ##
 ## **HARDENING (D-255(d), both mandatory):**
 ##   (i) Per-body deep-rung retention cap — DEEP_RUNGS (Block, Chunk; the
@@ -72,11 +80,13 @@ extends RefCounted
 ##
 ## **Sweep triggers (never per-frame — stig-round2.md §"Sweep triggers"):**
 ## callers invoke run_visit_sweep() once on body-open (cheap, index-only) and
-## run_background_sweep() on a coarse timer (LRU-capacity + Tier-3 TTL,
-## backgroundable). Neither is wired to _process()/a per-frame signal by this
-## file — the caller (step_canvas_viewer.gd) owns invoking these at the
-## right moments, matching D-227's "bookkeeping, not gameplay-adjacent work"
-## instruction.
+## run_background_sweep() on a coarse timer (LRU-capacity + Tier-3 TTL +
+## orphan-payload reconciliation, backgroundable — the orphan sweep does list
+## the body directory, unlike the other two, which is exactly why it lives on
+## the coarse timer and not body-open). Neither is wired to _process()/a
+## per-frame signal by this file — the caller (step_canvas_viewer.gd) owns
+## invoking these at the right moments, matching D-227's "bookkeeping, not
+## gameplay-adjacent work" instruction.
 ##
 ## D-227: every tier here is an evictable CACHE, never a source of truth.
 ## Deleting the whole cache root at any time changes client behavior only by
@@ -185,14 +195,18 @@ func _index_path(body_id: String) -> String:
 	return _body_dir(body_id) + INDEX_FILENAME
 
 
-## Stable non-negative filename for a composite key — the key string itself
-## isn't filesystem-safe on every target platform (':'/','), matching
-## stig-round2.md's own "named by a hash of key" directive. `key.hash() &
-## 0x7FFFFFFF` is the same non-negative-hash idiom reach_screen.gd already
-## uses elsewhere in this app for a stable derived value from a String.
+## Stable filename for a composite key — the key string itself isn't
+## filesystem-safe on every target platform (':'/','), matching
+## stig-round2.md's own "named by a hash of key" directive. 64 bits of
+## SHA-256 rather than String.hash(): the 31-bit space made a silent
+## filename collision (two live keys sharing one payload file — the wrong
+## map served as a valid hit) a ~1-in-16k event per cap-full body; at 64
+## bits it is negligible. The index stays keyed by the FULL key either way —
+## only the payload filename is derived. Changing this scheme later is safe:
+## old-scheme rows miss on their payload path and drop, old files are
+## reclaimed by run_orphan_sweep().
 static func _payload_filename(key: String) -> String:
-	var h: int = key.hash() & 0x7FFFFFFF
-	return "%08x.dat" % h
+	return key.sha256_text().substr(0, 16) + ".dat"
 
 
 func _payload_path(body_id: String, key: String) -> String:
@@ -227,7 +241,12 @@ static func _is_deep_rung(rung: String) -> bool:
 ## empty cache, don't crash, don't block first paint" instruction. The
 ## corrupt file is left on disk untouched (a caller that never puts() again
 ## for that body leaves it inert; the first successful save_index() call
-## overwrites it with a valid one).
+## overwrites it with a valid one). `_save_index()` writes atomically (below),
+## so a torn/truncated index can only arise from outside interference (manual
+## edit, platform-level file corruption) — never from a crash mid-write on
+## this file's own path. Any entries a discarded corrupt index loses are not
+## gone for good: their payload `.dat` files are reclaimed as orphans by
+## `run_orphan_sweep()` rather than leaking silently.
 func _load_index(body_id: String) -> Dictionary:
 	if _indexes.has(body_id):
 		return _indexes[body_id]
@@ -246,6 +265,14 @@ func _load_index(body_id: String) -> Dictionary:
 	return idx
 
 
+## Atomic write: the full index is serialized to a `.tmp` sibling, closed,
+## then swapped into place with `DirAccess.rename_absolute()` — a single
+## filesystem rename, never a truncate-in-place. A crash between the tmp
+## write and the rename leaves the PREVIOUS index.json untouched (the tmp
+## file is simply orphaned garbage, ignored by `_load_index()`); a crash mid-
+## rename is not a Godot-visible state this store needs to reason about
+## (the OS makes rename atomic). This closes the truncate-in-place corruption
+## window `_load_index()`'s doc used to describe as the normal recovery case.
 func _save_index(body_id: String) -> void:
 	var idx: Dictionary = _indexes.get(body_id, {})
 	var dir_err := DirAccess.make_dir_recursive_absolute(_body_dir(body_id))
@@ -255,12 +282,22 @@ func _save_index(body_id: String) -> void:
 			% [body_id, error_string(dir_err)]
 		)
 		return
-	var file := FileAccess.open(_index_path(body_id), FileAccess.WRITE)
+	var index_path := _index_path(body_id)
+	var tmp_path := index_path + ".tmp"
+	var file := FileAccess.open(tmp_path, FileAccess.WRITE)
 	if file == null:
-		push_warning("StepCanvasDiskCache: cannot write index for '%s'" % body_id)
+		push_warning("StepCanvasDiskCache: cannot write index tmp file for '%s'" % body_id)
 		return
 	file.store_string(JSON.stringify(idx))
 	file.close()
+	var rename_err := DirAccess.rename_absolute(tmp_path, index_path)
+	if rename_err != OK:
+		push_warning(
+			"StepCanvasDiskCache: atomic index rename failed for '%s': %s"
+			% [body_id, error_string(rename_err)]
+		)
+		if FileAccess.file_exists(tmp_path):
+			DirAccess.remove_absolute(tmp_path)
 
 
 # =============================================================================
@@ -277,6 +314,14 @@ func _save_index(body_id: String) -> void:
 ## and the filesystem can disagree (manual deletion, platform storage
 ## pressure clearing files without updating the index), and a dangling index
 ## row must never be handed to a caller as a hit.
+##
+## **Corrupt/truncated payload contract:** a payload file that decodes to
+## something other than a well-formed canvas Dictionary degrades to a miss
+## with full self-heal, symmetric with the index side's own malformed-JSON
+## recovery — `file.get_var()` returns null on a failed/garbage decode, the
+## shape check below catches a decoded-but-wrong-shaped value, and either
+## way `_drop_entry()` removes BOTH the index row and the payload file before
+## returning null. No caller ever sees a partial or malformed canvas.
 func get_canvas(
 	body_id: String, rung: String, center: Vector2i, extent: Vector2i, min_wl_m: int = 0
 ) -> Variant:
@@ -301,7 +346,7 @@ func get_canvas(
 		return null
 	var canvas: Variant = file.get_var()
 	file.close()
-	if not canvas is Dictionary:
+	if not canvas is Dictionary or not canvas.has("width") or not canvas.has("height"):
 		_drop_entry(body_id, key)
 		return null
 
@@ -335,6 +380,15 @@ func has(
 ## from the rung it's writing, mirroring make_key()'s own Global handling) —
 ## a floored entry is exempt from BOTH sweeps unconditionally.
 ##
+## **Write order is deliberate: payload file, THEN index entry.** A crash
+## between the two steps leaves an orphaned payload `.dat` file with no
+## index row pointing at it — reclaimed later by `run_orphan_sweep()` — never
+## a dangling index row pointing at a payload that doesn't exist (the other
+## direction is already handled by `get_canvas()`'s missing-payload-file
+## check, but a crash can't actually produce it under this ordering). Both
+## directions of index/payload disagreement are covered: one by write order,
+## the other by the orphan sweep.
+##
 ## (i) HARDENING: for a deep-rung (Block/Chunk) write, enforces
 ## MAX_DEEP_RUNG_ENTRIES_PER_BODY SYNCHRONOUSLY before inserting — if this
 ## put() would exceed the cap, the oldest (by last_read_at) deep-rung entry
@@ -369,6 +423,12 @@ func put(
 		push_warning("StepCanvasDiskCache: cannot write payload for '%s'/'%s'" % [body_id, key])
 		return
 	file.store_var(canvas)
+	# Captured while the file is still open, right after the write — the
+	# cursor position IS the byte count just written, so this needs no
+	# separate re-read of the file to learn its size (get_file_as_bytes()
+	# would otherwise load the whole payload a second time just to call
+	# .size() on it).
+	var payload_size: int = file.get_position()
 	file.close()
 
 	var now: int = Time.get_unix_time_from_system()
@@ -377,7 +437,7 @@ func put(
 		"tier": TIER_SIM_STATE if sim_ttl_sec > 0 else TIER_GEOMETRY,
 		"written_at": now,
 		"last_read_at": now,
-		"size_bytes": FileAccess.get_file_as_bytes(_payload_path(body_id, key)).size(),
+		"size_bytes": payload_size,
 		"sim_ttl": sim_ttl_sec if sim_ttl_sec > 0 else null,
 		"retention_floor": _is_global_rung(rung),
 		"schema_version": current_schema_version(),
@@ -520,12 +580,48 @@ func run_staleness_sweep(body_id: String) -> void:
 		_drop_entry(body_id, key)
 
 
-## Convenience: the three sweeps a coarse background timer runs together —
+## Reconciliation sweep — reclaims payload `.dat` files on disk that no
+## longer have an index row pointing at them (an "orphan"). This is the
+## other half of `put()`'s deliberate payload-before-index write order: a
+## crash between the two writes leaves exactly this shape, and a discarded
+## corrupt index (`_load_index()`'s malformed-JSON recovery) orphans EVERY
+## payload for that body at once. Nothing else in this file ever scans the
+## body directory, so without this sweep orphaned files leak on disk forever
+## and are invisible to `run_capacity_sweep()`'s byte accounting (they aren't
+## indexed, so they're never counted, and never evicted by it either). Builds
+## the set of payload basenames the CURRENT index actually references (same
+## derivation `_payload_path()` uses — a hash of the key, not the stored
+## `file_path` string, so this is robust even if `file_path` was ever wrong),
+## lists the body directory once, and removes every `.dat` file not in that
+## set. Never touches `index.json`, `index.json.tmp`, or an indexed payload.
+func run_orphan_sweep(body_id: String) -> void:
+	var dir_path := _body_dir(body_id)
+	if not DirAccess.dir_exists_absolute(dir_path):
+		return
+	var idx := _load_index(body_id)
+	var referenced: Dictionary = {}  # basename String -> true
+	for key: String in idx.keys():
+		referenced[_payload_filename(key)] = true
+
+	var dir := DirAccess.open(dir_path)
+	if dir == null:
+		return
+	dir.list_dir_begin()
+	var entry := dir.get_next()
+	while entry != "":
+		if not dir.current_is_dir() and entry.ends_with(".dat") and not referenced.has(entry):
+			DirAccess.remove_absolute(dir_path + entry)
+		entry = dir.get_next()
+	dir.list_dir_end()
+
+
+## Convenience: the sweeps a coarse background timer runs together —
 ## run_visit_sweep() is deliberately NOT included here (it belongs on
 ## body-open only, per stig-round2.md's own sweep-trigger split).
 func run_background_sweep(body_id: String) -> void:
 	run_capacity_sweep(body_id)
 	run_staleness_sweep(body_id)
+	run_orphan_sweep(body_id)
 
 
 # =============================================================================

client/ui/implant/apps/atlas/step_canvas/step_canvas_request.gd

@@ -158,6 +158,9 @@ func on_response(response: Dictionary) -> void:
 	_retries = 0
 	_held_extent = _echoed_extent(canvas, _rung, response.get("extent", Vector2i.ZERO))
 	_cache.put(_body_id, _rung, _center, _extent, canvas, _min_wl_m)
+	# sim_ttl_sec deliberately unset — every wire field is static-valued
+	# today; threads from the wire when D-253's clock lands (T-1190; see
+	# step_canvas_disk_cache.gd's Tier-3 doc).
 	_disk_cache.put(_body_id, _rung, _center, _extent, canvas, _min_wl_m)
 	canvas_ready.emit(canvas)