Merge remote-tracking branch 'origin/t1183-client-cache'

CHANGELOG.md

@@ -7,6 +7,7 @@ Format based on [Keep a Changelog](https://keepachangelog.com/).
 ## [Unreleased]
 
 ### Added
+- **The Atlas remembers across restarts** (D-255, T-1183) — every map view you visit is now kept on disk as well as in memory: relaunch the game, reopen a body, and previously-visited zoom steps draw instantly from the local cache with no server round-trip — only genuinely new ground fetches. Storage stays tidy on its own: fine-grained views of a body you haven't visited in ~two weeks quietly reclaim their space, each body's deep-zoom footprint is capped, and the whole-body overview of every visited body is kept forever. The cache is update-safe by construction — entries from an older game version are silently refetched, never misread
 - **The stepped Atlas map** (D-255, T-1182) — the body map is rebuilt on the step-canvas pipe: opening a body lands on the whole-body Global view, and the mouse wheel walks six discrete zoom steps (Global → Region → District → Quarter → Block → 64 m Chunk), cursor-anchored, with edge-scroll and WASD panning and a hard full-zoom-out snap back to Global. Terrain draws texel-exact from the server's data canvas (no client-side scaling — the whole family of zoom-compensation bugs is structurally gone, ~9,400 lines lighter), while river courses and settlement markers stay crisp at constant pixel sizes on a separate annotation layer. While a new step loads, the previous view holds with a pending wash — no blank frames. Within one session, revisited views come from an in-memory cache; the disk cache that survives restarts is the next ticket (T-1183)
 - **Lakes now show whether they drain** (D-227, T-1185) — every overflow lake's exit river is wired into the served course network: where a basin fills past its spill point, the water's escape path appears as a real river continuation (most commonly a short spill onto adjacent open ground — the honest dominant pattern; a long dramatic exit river to the sea is the exception). A closed (endorheic) basin gains nothing — the absence of any outgoing course IS the signal, exactly as ruled: no extra data bit, just hydrology you can read off the map
 - **Step-canvas serving — the map's new data pipe** (D-255, T-1181) — the server now answers "what is at this world coordinate at this zoom step" as a single tagged message per step: all six Atlas ladder rungs (the whole-body Global opener down to 64 m Chunk) served through one StepCanvasRequest/StepCanvasResponse envelope, with dense terrain fields shipped as compact per-field PNGs and river courses/cliffs as native lists (the encoding the workshop measured smallest-and-fastest). Backed by a keep-always whole-body tier (~9 MB for all ~267 bodies) plus a per-rung cache with honest freshness rules (terrain never goes stale — only storage evicts it; seasonal/tidal planes carry real clocks). Byte-identical cache-hit-vs-miss is enforced by a dedicated acceptance test on every rung — the guarantee that a player walking somewhere without ever opening the map gets the identical world. River detail at deep zoom is capped at the density that already reads well one rung up, eliminating a measured 38-87% derive-cost penalty with no visible loss. Server side only — the client map component that draws these canvases is the next ticket (T-1182)

client/tests/test_step_canvas_disk_cache.gd

@@ -0,0 +1,603 @@
+## 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()

client/tests/test_step_canvas_request.gd

@@ -1,20 +1,45 @@
-## T-1182 tests: step_canvas_request.gd — request lifecycle (cache hit/miss,
-## staleness gate, the extent ECHO rule). Live mode is required for
-## SimBridge.request_step_canvas() to actually send (test_mode short-circuits
-## it), so these tests exercise on_response()/request_now() against a
-## directly-constructed StepCanvasRequest node without a live bridge
-## connection — request_now() on a cache MISS calls into SimBridge, which is
-## a silent no-op in test_mode (SimBridge.test_mode defaults true outside
-## SR_LIVE=1), so no live server is needed for these assertions.
+## T-1182/T-1183 tests: step_canvas_request.gd — request lifecycle (cache
+## hit/miss, staleness gate, the extent ECHO rule) plus the T-1183 disk-tier
+## integration (Tier-2-before-wire miss path, write-through to both tiers).
+## Live mode is required for SimBridge.request_step_canvas() to actually send
+## (test_mode short-circuits it), so these tests exercise
+## on_response()/request_now() against a directly-constructed
+## StepCanvasRequest node without a live bridge connection — request_now()
+## on a cache MISS calls into SimBridge, which is a silent no-op in
+## test_mode (SimBridge.test_mode defaults true outside SR_LIVE=1), so no
+## live server is needed for these assertions.
+##
+## Every test constructs its StepCanvasRequest with an INJECTED disk-cache
+## root (never the real user://atlas_cache/ — see step_canvas_request.gd's
+## `_init()` doc) and removes it in after_test(), matching
+## test_step_canvas_disk_cache.gd's own isolation discipline.
 class_name TestStepCanvasRequest
 extends GdUnitTestSuite
 
 const StepCanvasRequestScript := preload("res://ui/implant/apps/atlas/step_canvas/step_canvas_request.gd")
+const StepCanvasDiskCache := preload(
+	"res://ui/implant/apps/atlas/step_canvas/step_canvas_disk_cache.gd"
+)
+
+var _test_disk_root: String = ""
+
+
+func before_test() -> void:
+	_test_disk_root = "user://test_step_canvas_request/%d/" % Time.get_ticks_usec()
+
+
+func after_test() -> void:
+	StepCanvasDiskCache.new(_test_disk_root).clear_all()
+
+
+func _make_request() -> Variant:
+	var req = auto_free(StepCanvasRequestScript.new(null, _test_disk_root))
+	add_child(req)
+	return req
 
 
 func test_cache_hit_emits_canvas_ready_synchronously_with_no_pending_state() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	var canvas := {"width": 64, "height": 64}
 	req.get_cache().put("GJ1c", "District", Vector2i(0, 0), Vector2i(64, 64), canvas)
 
@@ -28,15 +53,13 @@ func test_cache_hit_emits_canvas_ready_synchronously_with_no_pending_state() ->
 
 
 func test_cache_miss_sets_pending_true() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "Chunk", Vector2i(0, 0), Vector2i(64, 64))
 	assert_bool(req.is_pending()).is_true()
 
 
 func test_on_response_ignores_a_response_for_a_different_body() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "District", Vector2i(0, 0), Vector2i(64, 64))
 
 	var received: Array = []
@@ -55,8 +78,7 @@ func test_on_response_ignores_a_response_for_a_different_body() -> void:
 
 
 func test_on_response_ignores_a_response_for_a_different_rung() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "District", Vector2i(0, 0), Vector2i(64, 64))
 
 	var received: Array = []
@@ -75,8 +97,7 @@ func test_on_response_ignores_a_response_for_a_different_rung() -> void:
 
 
 func test_on_response_ignores_a_stale_center_for_a_fixed_rung() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "District", Vector2i(0, 0), Vector2i(64, 64))
 
 	var received: Array = []
@@ -98,8 +119,7 @@ func test_on_response_ignores_a_stale_center_for_a_fixed_rung() -> void:
 ## check must NOT compare center at all (an echoed (0,0) sentinel must not
 ## be rejected as "stale" against whatever was requested).
 func test_on_response_global_rung_ignores_center_in_staleness_check() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
 
 	var received: Array = []
@@ -122,8 +142,7 @@ func test_on_response_global_rung_ignores_center_in_staleness_check() -> void:
 ## from the RESPONSE's own echoed extent, never the requested one — a
 ## server-side clamp can shrink the actual canvas below what was asked for.
 func test_on_response_holds_the_echoed_extent_not_the_requested_one() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "Chunk", Vector2i(0, 0), Vector2i(9_999, 9_999))
 
 	req.on_response(
@@ -143,8 +162,7 @@ func test_on_response_holds_the_echoed_extent_not_the_requested_one() -> void:
 ## Global's held extent comes from the canvas's own width/height (the wire
 ## extent echo is a fixed (0,0) sentinel for that rung — nothing to read).
 func test_on_response_global_held_extent_derives_from_canvas_dimensions() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
 
 	req.on_response(
@@ -162,8 +180,7 @@ func test_on_response_global_held_extent_derives_from_canvas_dimensions() -> voi
 
 
 func test_on_response_ready_with_null_canvas_retries_rather_than_adopting() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "District", Vector2i(0, 0), Vector2i(64, 64))
 
 	var received: Array = []
@@ -183,16 +200,14 @@ func test_on_response_ready_with_null_canvas_retries_rather_than_adopting() -> v
 
 
 func test_on_response_not_found_gives_up_immediately() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "Chunk", Vector2i(0, 0), Vector2i(64, 64))
 	req.on_response({"body_id": "GJ1c", "rung": "Chunk", "status": "NotFound"})
 	assert_bool(req.is_pending()).is_false()
 
 
 func test_on_response_stores_a_fresh_ready_canvas_in_the_cache() -> void:
-	var req = auto_free(StepCanvasRequestScript.new())
-	add_child(req)
+	var req = _make_request()
 	req.request_now("GJ1c", "District", Vector2i(5, 5), Vector2i(64, 64))
 	req.on_response(
 		{
@@ -206,3 +221,103 @@ func test_on_response_stores_a_fresh_ready_canvas_in_the_cache() -> void:
 		}
 	)
 	assert_bool(req.get_cache().has("GJ1c", "District", Vector2i(5, 5), Vector2i(64, 64))).is_true()
+
+
+# =============================================================================
+# T-1183: Tier-2 (disk) integration
+# =============================================================================
+
+
+## A fresh Ready response must write through to BOTH tiers, not just Tier 1
+## — the whole point of the disk tier is that a LATER, separate
+## StepCanvasRequest instance (e.g. after a Tier-1 LRU eviction in this
+## session, or a full app restart) can still serve this key without a wire
+## round-trip.
+func test_on_response_writes_through_to_both_the_memory_and_disk_tiers() -> void:
+	var req = _make_request()
+	req.request_now("GJ1c", "Quarter", Vector2i(3, 3), Vector2i(64, 64))
+	req.on_response(
+		{
+			"body_id": "GJ1c",
+			"rung": "Quarter",
+			"center": Vector2i(3, 3),
+			"extent": Vector2i(64, 64),
+			"min_wl_m": 0,
+			"status": "Ready",
+			"canvas": {"width": 64, "height": 64},
+		}
+	)
+	assert_bool(req.get_cache().has("GJ1c", "Quarter", Vector2i(3, 3), Vector2i(64, 64))).is_true()
+	assert_bool(
+		req.get_disk_cache().has("GJ1c", "Quarter", Vector2i(3, 3), Vector2i(64, 64))
+	).override_failure_message("a fresh Ready response must ALSO land in the disk tier").is_true()
+
+
+## A Tier-1 miss that IS present on disk must be served from disk —
+## synchronously, no Pending state — and promoted into Tier 1 so the next
+## identical request is a pure in-memory hit.
+func test_disk_hit_on_tier1_miss_is_served_without_going_pending() -> void:
+	var req = _make_request()
+	var canvas := {"width": 64, "height": 64, "morphology": PackedByteArray([1, 2, 3])}
+	req.get_disk_cache().put("GJ1c", "Block", Vector2i(9, 9), Vector2i(64, 64), canvas)
+	# Confirm this is genuinely a Tier-1 miss before the request.
+	assert_bool(req.get_cache().has("GJ1c", "Block", Vector2i(9, 9), Vector2i(64, 64))).is_false()
+
+	var received: Array = []
+	req.canvas_ready.connect(func(c: Dictionary) -> void: received.append(c))
+	req.request_now("GJ1c", "Block", Vector2i(9, 9), Vector2i(64, 64))
+
+	assert_int(received.size()).is_equal(1)
+	assert_that(received[0]).is_equal(canvas)
+	assert_bool(req.is_pending()).override_failure_message(
+		"a disk-tier hit must be served synchronously, never leave the request Pending"
+	).is_false()
+
+
+## A disk hit promotes into Tier 1 — the NEXT identical request must be
+## servable purely from memory (this is what makes the disk tier "cache-
+## accelerated": the second lookup for the same key never touches disk
+## again this session).
+func test_disk_hit_promotes_into_tier1_for_the_next_lookup() -> void:
+	var req = _make_request()
+	var canvas := {"width": 32, "height": 32}
+	req.get_disk_cache().put("GJ1c", "Block", Vector2i(1, 1), Vector2i(64, 64), canvas)
+	req.request_now("GJ1c", "Block", Vector2i(1, 1), Vector2i(64, 64))
+	assert_bool(req.get_cache().has("GJ1c", "Block", Vector2i(1, 1), Vector2i(64, 64))).override_failure_message(
+		"a disk-tier hit must be promoted into Tier 1 on read"
+	).is_true()
+
+
+## A Global-rung write-through must set the disk tier's retention-floor flag
+## — the client-side mechanism for "never evicted by either sweep axis"
+## (D-255(d)).
+func test_global_rung_write_through_sets_the_disk_retention_floor() -> void:
+	var req = _make_request()
+	req.request_now("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
+	req.on_response(
+		{
+			"body_id": "GJ1c",
+			"rung": "Global",
+			"center": Vector2i.ZERO,
+			"extent": Vector2i.ZERO,
+			"min_wl_m": 0,
+			"status": "Ready",
+			"canvas": {"width": 19_139, "height": 9_569},
+		}
+	)
+	var key := StepCanvasDiskCache.make_key("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)
+	# Backdate to ancient, then run the visit sweep — a floored entry must
+	# survive regardless of age, proving the write-through path (not just
+	# put() called directly, as the disk-cache unit tests already cover)
+	# correctly threads the Global rung into the floor flag.
+	req.get_disk_cache()._debug_backdate_entry("GJ1c", key, 0, 0)
+	req.get_disk_cache().run_visit_sweep("GJ1c")
+	assert_bool(req.get_disk_cache().has("GJ1c", "Global", Vector2i.ZERO, Vector2i.ZERO)).is_true()
+
+
+## Neither tier is touched by a miss that never resolves (still Pending) —
+## write-through only happens on an adopted Ready response.
+func test_disk_tier_is_not_written_while_still_pending() -> void:
+	var req = _make_request()
+	req.request_now("GJ1c", "Chunk", Vector2i(0, 0), Vector2i(64, 64))
+	assert_bool(req.get_disk_cache().has("GJ1c", "Chunk", Vector2i(0, 0), Vector2i(64, 64))).is_false()

client/tests/test_step_canvas_viewer.gd

@@ -325,3 +325,63 @@ func test_edge_scroll_inactive_well_inside_the_viewport() -> void:
 	v._app_has_focus = true
 	v._last_mouse_pos = Vector2(400.0, 300.0)  # dead center — far from any edge
 	assert_bool(v._is_cursor_edge_scrolling()).is_false()
+
+
+# =============================================================================
+# T-1183: disk-cache sweep-trigger wiring
+# =============================================================================
+
+
+## enter() must invoke the disk cache's (2a) visit sweep for the entered
+## body — a smoke test that the wiring exists and doesn't crash; the sweep
+## LOGIC itself (what gets evicted and why) is covered exhaustively by
+## test_step_canvas_disk_cache.gd. Uses a distinctive body_id with nothing
+## ever cached under it, so the sweep is a true no-op read (no writes to the
+## real user://atlas_cache/ directory this test could leak).
+func test_enter_runs_the_disk_cache_visit_sweep_without_crashing() -> void:
+	var v: StepCanvasViewer = auto_free(StepCanvasViewer.new())
+	add_child(v)
+	v.enter({"body_id": "T1183_sweep_smoke_test_body", "body_radius_km": 6238.4}, {})
+	# If the wiring is broken (e.g. calling a method that doesn't exist), the
+	# enter() call itself would already have failed above — reaching here
+	# with the expected held rung is the assertion.
+	assert_str(v.get_held_rung()).is_equal(StepCanvasTransport.RUNG_GLOBAL)
+
+
+## The coarse background sweep timer exists, is not per-frame (a real
+## Timer node, not a _process()-driven counter), autostarts, and is set to
+## the documented coarse interval — never a sub-frame or per-frame value.
+func test_disk_sweep_timer_is_coarse_and_autostarts() -> void:
+	var v: StepCanvasViewer = auto_free(StepCanvasViewer.new())
+	add_child(v)
+	var timer: Timer = v.get_node("DiskSweepTimer")
+	assert_object(timer).is_not_null()
+	# Godot resets Timer.autostart to false once the timer has actually
+	# started after entering the tree (documented engine behavior — the flag
+	# is a one-shot "start me on _ready()" instruction, not a persistent
+	# state mirror). The real behavioral guarantee is "the timer is running,
+	# unpaused, without anyone having to call start() explicitly" —
+	# is_stopped() == false is the correct read of that.
+	assert_bool(timer.is_stopped()).override_failure_message(
+		"the disk sweep timer must autostart running, no explicit start() call needed"
+	).is_false()
+	assert_float(timer.wait_time).override_failure_message(
+		"the disk sweep timer must be coarse (minutes), never a per-frame interval"
+	).is_greater_equal(60.0)
+
+
+## The timer's timeout must actually route to the disk cache's
+## run_background_sweep() for the currently-entered body — verified by
+## invoking the private handler directly (the same "call the handler, don't
+## wait on a real Timer" pattern used elsewhere in this cluster for
+## non-blocking test speed) against an injected-root request so this test
+## touches no real cache files.
+func test_disk_sweep_timeout_handler_runs_background_sweep_for_the_current_body() -> void:
+	var v: StepCanvasViewer = auto_free(StepCanvasViewer.new())
+	add_child(v)
+	v.enter({"body_id": "T1183_sweep_smoke_test_body", "body_radius_km": 6238.4}, {})
+	# No live server, no cached entries for this body — the assertion is
+	# that calling the handler does not crash and leaves the (empty) cache
+	# consistent, mirroring the enter()-sweep smoke test above.
+	v._on_disk_sweep_timeout()
+	assert_int(v.get_request().get_disk_cache().entry_count("T1183_sweep_smoke_test_body")).is_equal(0)

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

@@ -0,0 +1,664 @@
+extends RefCounted
+
+## Client-side DISK-BACKED cache store for decoded StepCanvasResponse canvas
+## payloads (T-1183, D-255(d) Tier 2/3: "client disk-backed FileAccess store").
+## Layers BENEATH step_canvas_cache.gd's in-memory LRU (Tier 1, T-1182) — a
+## Tier-1 miss checks this store before going to the wire; a fresh wire
+## response is written through to BOTH tiers (step_canvas_request.gd is the
+## integration point). Never SQLite in any shape (workshop-rejected
+## independently by both Stig and Dudley — stig-round2.md §(c), plain
+## `FileAccess` + a JSON index won on both agents' analyses).
+##
+## **What is persisted: THE WIRE PAYLOAD, PRE-DECODE.** The `canvas`
+## Dictionary this store writes/reads is byte-identical to what
+## step_canvas_protocol.gd's `_decode_encoded_canvas()` produces — dense
+## fields (`morphology`/`elev_q`/`moisture_q`/`vegetation`/`glaciation`/
+## `flooded_q`) are still PNG-encoded `PackedByteArray`s, `temp_dc`/
+## `settlement_id` are raw msgpack-decoded arrays, `courses`/`cliffs` are raw
+## sparse lists. `Image.load_png_from_buffer()` — the actual pixel decode —
+## never runs here; it happens only in step_canvas_terrain_layer.gd at draw
+## time, exactly as it does for a Tier-1 hit today. "The disk file IS the
+## wire payload, no re-encoding for storage" (stig-round1.md §4).
+##
+## **Cache key — extends the exact Tier-1 tuple** (StepCanvasCache.make_key,
+## which itself mirrors server/src/atlas/step_canvas.rs's own
+## StepCanvasCache key): (body_id, rung, center, extent, min_wl_m). Global
+## (rung 0) collapses center/extent to the (0,0)/(0,0) sentinel identically
+## (make_key() delegates to StepCanvasCache.make_key() directly — one
+## composite-key discipline, not two).
+##
+## **Three-tier eviction (stig-round2.md §(c), the concrete spec this file
+## implements verbatim):**
+##   Tier 1 (Global/rung-0 geometry) — RETENTION FLOOR, no sweep at all. The
+##     only tier this file exempts unconditionally from both sweeps below.
+##   Tier 2 (sub-global geometry) — TWO independent, separately-triggered
+##     sweeps, kept apart per Jeroen's storage-eviction ruling (staleness and
+##     storage are distinct axes; geometry never goes stale, D-227):
+##       (2a) time-since-last-visit — evict any entry whose last_read_at is
+##            older than STORAGE_TTL_SEC, run on body-open (cheap: an index
+##            scan, no bulk file I/O).
+##       (2b) LRU-capacity — if total sub-global disk usage for a body
+##            exceeds SUB_GLOBAL_BYTE_BUDGET, evict oldest-touched entries
+##            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, 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
+##       two finest, matching troblum-round2.md S3's specific pan-assembly
+##       concern) are capped at MAX_DEEP_RUNG_ENTRIES_PER_BODY resident
+##       entries, independent of the Global floor and independent of the
+##       general Tier-2 byte budget — closes the D-226(d) accumulation gap
+##       against a systematic exhaustive pan (most plausibly the
+##       AtlasAgentInterface QA channel, D-226 item 4) structurally rather
+##       than by practical improbability. Enforced synchronously in put():
+##       inserting past the cap evicts the oldest deep-rung entry for that
+##       body FIRST (LRU order), before the general 2b sweep ever runs.
+##   (ii) Schema/version tag — every persistent entry is stamped with
+##        CACHE_SCHEMA_VERSION at write time (see that constant's own doc
+##        for the version-tag source decision + rationale). A read-time
+##        mismatch is treated as a cache miss: the stale-schema file is
+##        discarded (never handed to a caller, never decoded) and the index
+##        entry is dropped. This is the one place D-192's co-ship guarantee
+##        does not reach (D-192 2026-07-23 amendment) — a disk cache
+##        survives a game update; the live wire does not need this because
+##        client+server always launch in lockstep.
+##
+## **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 +
+## 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
+## causing re-fetches — verified structurally by test_step_canvas_disk_cache.
+## gd's malformed-index-recovery + missing-payload-file cases, not just by
+## comment.
+
+const StepCanvasCache := preload("res://ui/implant/apps/atlas/step_canvas/step_canvas_cache.gd")
+
+## Tier tags, mirroring stig-round2.md's IndexEntry.tier field exactly.
+const TIER_GEOMETRY: String = "Geometry"
+const TIER_SIM_STATE: String = "SimState"
+
+const DEFAULT_ROOT: String = "user://atlas_cache/"
+const INDEX_FILENAME: String = "index.json"
+
+## (2a) Time-since-last-visit threshold — stig-round2.md's "days-to-weeks of
+## real wall-clock time, tunable — not a round-2 architecture call, a tuning
+## pass once this ships" starting point. 14 days as the initial tunable.
+const STORAGE_TTL_SEC: int = 14 * 24 * 60 * 60
+
+## (2b) Sub-global per-body disk budget — stig-round2.md's "same order of
+## magnitude as tier 1... a tuning-pass number once real play-pattern data
+## exists" starting point. 256 MiB/body.
+const SUB_GLOBAL_BYTE_BUDGET: int = 256 * 1024 * 1024
+
+## (i) Deep-rung hardening cap — the two finest rungs (Block 128m, Chunk
+## 64m), independent of SUB_GLOBAL_BYTE_BUDGET and the Global floor. A
+## per-body ENTRY COUNT (not bytes) because the accumulation risk S3 names is
+## about DISTINCT WINDOWS assembling whole-body coverage, not aggregate
+## bytes — capping entry count directly bounds how much ground a client can
+## have simultaneously resident at fine spacing, which is the D-226(d)
+## purpose (information content), not a byte-budget proxy for it.
+const DEEP_RUNGS: Array = ["Block", "Chunk"]
+const MAX_DEEP_RUNG_ENTRIES_PER_BODY: int = 512
+
+## (ii) Schema/version tag source: the game's `project.yaml` `version:`
+## field (e.g. "0.4.0"), read via the identical technique
+## loading_screen.gd's `_read_client_version()` already uses (line-scan for
+## "version:", no YAML parser dependency). CHOSEN over a `generator_sha`-
+## style stamp because:
+##   - project.yaml's version is ALREADY the project's single source of
+##     truth for "what build is this" (CLAUDE.md: "Version source of truth:
+##     project.yaml") — reusing it needs no new stamp-generation machinery
+##     anywhere, client or server.
+##   - The D-255(c) wire shape (EncodedStepCanvas's field set) changes in
+##     lockstep with client releases in this single-repo, subprocess-co-ship
+##     project (D-192) — there is no independent server-only wire-schema
+##     versioning surface a generator_sha would need to track separately;
+##     the client's own version IS the wire-schema version for this
+##     project's deployment model.
+##   - It is legible for a human debugging a stale-cache report ("this file
+##     was written by 0.4.0, I am running 0.5.0") in a way a SHA is not.
+## A mismatch — ANY string difference, not just an older/newer comparison
+## (CLAUDE.md's semver note is for systems.db's schema_version LINEAGE,
+## which needs ordering for migration; this cache has no migration path at
+## all, so exact-match-or-miss is the correct, simpler rule) — is a cache
+## miss: re-fetch, never decode. See read_entry()'s version check.
+## (Declared as a const-adjacent static func, ahead of the instance vars
+## below, per gdlint's class-definitions-order: statics precede prvvars.)
+
+var _root: String = DEFAULT_ROOT
+
+## Per-body loaded index, held in memory for the session — matching
+## step_canvas_cache.gd's own "erase+reinsert = move-to-MRU" idiom, now
+## keyed one level up (body_id -> key -> IndexEntry Dictionary). Loaded
+## lazily on first touch per body, not all at once (a player who never
+## revisits most bodies never pays the parse cost for their index files).
+var _indexes: Dictionary = {}  # body_id String -> (key String -> IndexEntry Dictionary)
+
+
+static func current_schema_version() -> String:
+	var yaml_path := ProjectSettings.globalize_path("res://") + "/../project.yaml"
+	if not FileAccess.file_exists(yaml_path):
+		return "?.?.?"
+	var file := FileAccess.open(yaml_path, FileAccess.READ)
+	if file == null:
+		return "?.?.?"
+	var content := file.get_as_text()
+	file.close()
+	for line: String in content.split("\n"):
+		if line.begins_with("version:"):
+			var parts := line.split(":", false, 1)
+			if parts.size() >= 2:
+				return parts[1].strip_edges()
+	return "?.?.?"
+
+
+## `root` is injectable so tests never touch the real
+## user://atlas_cache/ directory (test_step_canvas_disk_cache.gd passes a
+## disposable per-test-run subdirectory and removes it in after_test()).
+func _init(root: String = DEFAULT_ROOT) -> void:
+	_root = root if root.ends_with("/") else root + "/"
+
+
+# =============================================================================
+# Path / key helpers
+# =============================================================================
+
+
+func _body_dir(body_id: String) -> String:
+	return _root + body_id.validate_filename() + "/"
+
+
+func _index_path(body_id: String) -> String:
+	return _body_dir(body_id) + INDEX_FILENAME
+
+
+## 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:
+	return key.sha256_text().substr(0, 16) + ".dat"
+
+
+func _payload_path(body_id: String, key: String) -> String:
+	return _body_dir(body_id) + _payload_filename(key)
+
+
+## Delegates to StepCanvasCache.make_key() directly — ONE composite-key
+## discipline shared by both tiers, never a second parallel key scheme.
+static func make_key(
+	body_id: String, rung: String, center: Vector2i, extent: Vector2i, min_wl_m: int = 0
+) -> String:
+	return StepCanvasCache.make_key(body_id, rung, center, extent, min_wl_m)
+
+
+static func _is_global_rung(rung: String) -> bool:
+	return rung == "Global"
+
+
+static func _is_deep_rung(rung: String) -> bool:
+	return DEEP_RUNGS.has(rung)
+
+
+# =============================================================================
+# Index load / save (malformed-recovery hardened)
+# =============================================================================
+
+
+## Returns the in-memory index Dictionary for `body_id`, loading it from disk
+## on first touch. A missing OR malformed (truncated/corrupt JSON, wrong top-
+## level type) index file is treated as an EMPTY cache for that body — never
+## a crash, matching stig-round1.md's "missing/corrupt index -> treat as
+## 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). `_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]
+	var idx: Dictionary = {}
+	var path := _index_path(body_id)
+	if FileAccess.file_exists(path):
+		var file := FileAccess.open(path, FileAccess.READ)
+		if file != null:
+			var content := file.get_as_text()
+			file.close()
+			var json := JSON.new()
+			var err := json.parse(content)
+			if err == OK and json.data is Dictionary:
+				idx = json.data
+	_indexes[body_id] = idx
+	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))
+	if dir_err != OK:
+		push_warning(
+			"StepCanvasDiskCache: cannot create cache dir for '%s': %s"
+			% [body_id, error_string(dir_err)]
+		)
+		return
+	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 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)
+
+
+# =============================================================================
+# Read / write entries
+# =============================================================================
+
+
+## Fetch a cached canvas, touching last_read_at (drives 2a/2b). Returns null
+## on a miss — including a SCHEMA-VERSION-MISMATCH miss (ii): a stale-schema
+## entry is dropped from the index and its payload file deleted WITHOUT ever
+## being decoded, exactly the "mismatch = cache miss, re-fetch, never
+## decode" contract. Also returns null (and drops the stale index row) if
+## the payload file is missing on disk despite an index entry — the index
+## 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:
+	var key := make_key(body_id, rung, center, extent, min_wl_m)
+	var idx := _load_index(body_id)
+	if not idx.has(key):
+		return null
+	var entry: Dictionary = idx[key]
+
+	if str(entry.get("schema_version", "")) != current_schema_version():
+		_drop_entry(body_id, key)
+		return null
+
+	var payload_path := _payload_path(body_id, key)
+	if not FileAccess.file_exists(payload_path):
+		_drop_entry(body_id, key)
+		return null
+
+	var file := FileAccess.open(payload_path, FileAccess.READ)
+	if file == null:
+		_drop_entry(body_id, key)
+		return null
+	var canvas: Variant = file.get_var()
+	file.close()
+	if not canvas is Dictionary or not canvas.has("width") or not canvas.has("height"):
+		_drop_entry(body_id, key)
+		return null
+
+	entry["last_read_at"] = Time.get_unix_time_from_system()
+	idx[key] = entry
+	_save_index(body_id)
+	return canvas
+
+
+## True if a (currently valid — schema-matched, payload present) entry
+## exists, without touching last_read_at (a pure existence check, mirroring
+## step_canvas_cache.gd's own has()/get_canvas() split).
+func has(
+	body_id: String, rung: String, center: Vector2i, extent: Vector2i, min_wl_m: int = 0
+) -> bool:
+	var key := make_key(body_id, rung, center, extent, min_wl_m)
+	var idx := _load_index(body_id)
+	if not idx.has(key):
+		return false
+	var entry: Dictionary = idx[key]
+	if str(entry.get("schema_version", "")) != current_schema_version():
+		return false
+	return FileAccess.file_exists(_payload_path(body_id, key))
+
+
+## Persist a canvas Dictionary — the exact pre-decode wire payload shape
+## step_canvas_protocol.gd produces (PNG bytes untouched). `sim_ttl_sec`, if
+## > 0, tags the entry Tier 3 (SimState, real staleness TTL); omitted/0 tags
+## Tier 2 (Geometry, D-227 never-stale). `retention_floor` should be true
+## ONLY for rung == "Global" callers (step_canvas_request.gd computes this
+## 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
+## for this body is evicted first. This runs on every deep-rung put(), not
+## just during the periodic background sweep — the cap must never be
+## transiently exceeded even by one entry, since the accumulation risk it
+## closes is about a sustained systematic pan, not a single burst.
+func put(
+	body_id: String,
+	rung: String,
+	center: Vector2i,
+	extent: Vector2i,
+	canvas: Dictionary,
+	min_wl_m: int = 0,
+	sim_ttl_sec: int = 0
+) -> void:
+	var key := make_key(body_id, rung, center, extent, min_wl_m)
+	var idx := _load_index(body_id)
+
+	if _is_deep_rung(rung) and not idx.has(key):
+		_enforce_deep_rung_cap(body_id, idx)
+
+	var dir_err := DirAccess.make_dir_recursive_absolute(_body_dir(body_id))
+	if dir_err != OK:
+		push_warning(
+			"StepCanvasDiskCache: cannot create cache dir for '%s': %s"
+			% [body_id, error_string(dir_err)]
+		)
+		return
+	var file := FileAccess.open(_payload_path(body_id, key), FileAccess.WRITE)
+	if file == null:
+		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()
+	var entry: Dictionary = {
+		"file_path": _payload_path(body_id, key),
+		"tier": TIER_SIM_STATE if sim_ttl_sec > 0 else TIER_GEOMETRY,
+		"written_at": now,
+		"last_read_at": now,
+		"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(),
+	}
+	idx[key] = entry
+	_save_index(body_id)
+
+
+## TEST-SUPPORT ONLY: overwrite an existing entry's `written_at`/
+## `last_read_at` so sweep tests can exercise STORAGE_TTL_SEC/sim_ttl
+## expiry without waiting on real wall-clock time. No production caller —
+## step_canvas_request.gd never calls this. A no-op if `key` isn't cached.
+func _debug_backdate_entry(body_id: String, key: String, written_at: int, last_read_at: int) -> void:
+	var idx := _load_index(body_id)
+	if not idx.has(key):
+		return
+	var entry: Dictionary = idx[key]
+	entry["written_at"] = written_at
+	entry["last_read_at"] = last_read_at
+	idx[key] = entry
+	_save_index(body_id)
+
+
+## Evict the single oldest (lowest last_read_at) deep-rung entry for
+## `body_id`, if adding one more would push the deep-rung count for this
+## body over MAX_DEEP_RUNG_ENTRIES_PER_BODY. A no-op while under the cap.
+func _enforce_deep_rung_cap(body_id: String, idx: Dictionary) -> void:
+	var deep_keys: Array = []
+	for key: String in idx.keys():
+		var entry: Dictionary = idx[key]
+		if not bool(entry.get("retention_floor", false)) and _entry_is_deep_rung(key):
+			deep_keys.append(key)
+	if deep_keys.size() < MAX_DEEP_RUNG_ENTRIES_PER_BODY:
+		return
+	deep_keys.sort_custom(
+		func(a: String, b: String) -> bool:
+			return int(idx[a].get("last_read_at", 0)) < int(idx[b].get("last_read_at", 0))
+	)
+	var evict_count: int = deep_keys.size() - MAX_DEEP_RUNG_ENTRIES_PER_BODY + 1
+	for i in range(evict_count):
+		_drop_entry(body_id, deep_keys[i])
+
+
+## A key's rung is embedded as its second colon-delimited field
+## (StepCanvasCache.make_key()'s own "%s:%s:..." shape) — parsed back out
+## rather than carried as a separate index column, since the key already
+## encodes it and a second copy would be a redundant field to keep in sync.
+static func _entry_is_deep_rung(key: String) -> bool:
+	var parts := key.split(":")
+	if parts.size() < 2:
+		return false
+	return _is_deep_rung(parts[1])
+
+
+func _drop_entry(body_id: String, key: String) -> void:
+	var idx := _load_index(body_id)
+	if not idx.has(key):
+		return
+	var entry: Dictionary = idx[key]
+	var path := str(entry.get("file_path", _payload_path(body_id, key)))
+	if FileAccess.file_exists(path):
+		DirAccess.remove_absolute(path)
+	idx.erase(key)
+	_save_index(body_id)
+
+
+# =============================================================================
+# Sweeps (2a on body-open, 2b + Tier-3 TTL on a coarse background timer —
+# never per-frame; the caller decides WHEN to invoke these, this file only
+# implements WHAT each sweep does)
+# =============================================================================
+
+
+## (2a) Time-since-last-visit — drop every non-floored entry whose
+## last_read_at exceeds STORAGE_TTL_SEC. Cheap: an index scan only, no bulk
+## payload-file stat()s beyond the deletes this triggers. Independent of
+## (2b)/Tier-3 — this NEVER touches a floored (Global) entry, and geometry
+## entries are storage-motivated only (never "wrong"), never Tier-3
+## staleness-motivated.
+func run_visit_sweep(body_id: String) -> void:
+	var idx := _load_index(body_id)
+	var now: int = Time.get_unix_time_from_system()
+	var to_drop: Array = []
+	for key: String in idx.keys():
+		var entry: Dictionary = idx[key]
+		if bool(entry.get("retention_floor", false)):
+			continue
+		if now - int(entry.get("last_read_at", now)) > STORAGE_TTL_SEC:
+			to_drop.append(key)
+	for key in to_drop:
+		_drop_entry(body_id, key)
+
+
+## (2b) LRU-capacity — if total sub-global (non-floored) bytes for this body
+## exceed SUB_GLOBAL_BYTE_BUDGET, evict oldest-touched entries first until
+## under budget. Independent of (2a)/Tier-3 — a byte-budget question only,
+## never a staleness one; never touches a floored entry.
+func run_capacity_sweep(body_id: String) -> void:
+	var idx := _load_index(body_id)
+	var sub_global_keys: Array = []
+	var total_bytes: int = 0
+	for key: String in idx.keys():
+		var entry: Dictionary = idx[key]
+		if bool(entry.get("retention_floor", false)):
+			continue
+		sub_global_keys.append(key)
+		total_bytes += int(entry.get("size_bytes", 0))
+	if total_bytes <= SUB_GLOBAL_BYTE_BUDGET:
+		return
+	sub_global_keys.sort_custom(
+		func(a: String, b: String) -> bool:
+			return int(idx[a].get("last_read_at", 0)) < int(idx[b].get("last_read_at", 0))
+	)
+	for key in sub_global_keys:
+		if total_bytes <= SUB_GLOBAL_BYTE_BUDGET:
+			break
+		total_bytes -= int(idx[key].get("size_bytes", 0))
+		_drop_entry(body_id, key)
+
+
+## Tier 3 — drop every SimState entry past its own sim_ttl, regardless of
+## capacity pressure or last_read_at. Structurally separate condition from
+## both sweeps above (`now > written_at + sim_ttl`) — a sim-state entry does
+## NOT get to live longer just because disk space is available, and is
+## never evicted early just because it was recently read.
+func run_staleness_sweep(body_id: String) -> void:
+	var idx := _load_index(body_id)
+	var now: int = Time.get_unix_time_from_system()
+	var to_drop: Array = []
+	for key: String in idx.keys():
+		var entry: Dictionary = idx[key]
+		if str(entry.get("tier", TIER_GEOMETRY)) != TIER_SIM_STATE:
+			continue
+		var ttl_raw: Variant = entry.get("sim_ttl")
+		if ttl_raw == null:
+			continue
+		if now > int(entry.get("written_at", now)) + int(ttl_raw):
+			to_drop.append(key)
+	for key in to_drop:
+		_drop_entry(body_id, key)
+
+
+## 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)
+
+
+# =============================================================================
+# Introspection (tests + diagnostics)
+# =============================================================================
+
+
+func entry_count(body_id: String) -> int:
+	return _load_index(body_id).size()
+
+
+func has_entry_for_key(body_id: String, key: String) -> bool:
+	return _load_index(body_id).has(key)
+
+
+## Removes the entire on-disk cache root (all bodies). D-227: this only
+## causes re-fetches on next touch — never a correctness change. Used by
+## tests for cleanup; also the structural answer to "manual clear" from the
+## class doc's D-227 guarantee.
+func clear_all() -> void:
+	_remove_dir_recursive(_root)
+	_indexes.clear()
+
+
+static func _remove_dir_recursive(path: String) -> void:
+	var dir := DirAccess.open(path)
+	if dir == null:
+		return
+	dir.list_dir_begin()
+	var entry := dir.get_next()
+	while entry != "":
+		if entry != "." and entry != "..":
+			var full := path.path_join(entry)
+			if dir.current_is_dir():
+				_remove_dir_recursive(full)
+			else:
+				DirAccess.remove_absolute(full)
+		entry = dir.get_next()
+	dir.list_dir_end()
+	DirAccess.remove_absolute(path)

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

@@ -1,11 +1,21 @@
 extends Node
 
 ## Step-canvas request orchestration for StepCanvasViewer (T-1182, D-255(c)/
-## (d)). Owns the in-memory LRU cache, fires StepCanvasRequest frames, and
-## retries on a Pending response — the same D-225 poll/cache/enqueue serving
-## model atlas_window_request.gd already established for the legacy
+## (d)). Owns the in-memory LRU cache (Tier 1) AND the disk-backed cache
+## (Tier 2/3, T-1183), fires StepCanvasRequest frames, and retries on a
+## Pending response — the same D-225 poll/cache/enqueue serving model
+## atlas_window_request.gd already established for the legacy
 ## district_window carrier, now pointed at the new tagged envelope.
 ##
+## **Three-tier read path (D-255(d), cheapest-first): Tier 1 (in-memory LRU)
+## -> Tier 2 (disk) -> wire.** request_now() checks Tier 1 first (unchanged,
+## synchronous); a Tier-1 miss now checks Tier 2 (also synchronous — a
+## FileAccess read, not a network round-trip) before firing a
+## SimBridge.request_step_canvas() wire request. A disk hit is promoted into
+## Tier 1 on read (matches step_canvas_cache.gd's own "cache-accelerated"
+## framing — the next request for the same key is a Tier-1 hit). A fresh
+## Ready wire response is written through to BOTH tiers in on_response().
+##
 ## No `class_name` on purpose, matching every other viewer-owned helper in
 ## this cluster (atlas_overlay_bar.gd/atlas_window_request.gd, established
 ## precedent): the owner (StepCanvasViewer) passes itself to `_init()`.
@@ -22,18 +32,23 @@ extends Node
 ## step_canvas_protocol.gd's doc) — callers reading `get_held_extent()` for
 ## a Global-held canvas must special-case it themselves (the viewer does,
 ## via StepCanvasTransport.RUNG_GLOBAL checks), matching this ticket's own
-## "Global rung ignores wire extent" instruction.
+## "Global rung ignores wire extent" instruction. **The disk-tier key uses
+## the SAME requested/echoed extent as Tier 1 and the server's own key**
+## (confirmed correct in the #203 review) — no separate extent convention
+## for the disk tier.
 
 signal canvas_ready(response: Dictionary)  # emitted on a cache hit OR a fresh Ready response
 
 const StepCanvasCache := preload("res://ui/implant/apps/atlas/step_canvas/step_canvas_cache.gd")
+const StepCanvasDiskCache := preload("res://ui/implant/apps/atlas/step_canvas/step_canvas_disk_cache.gd")
 
 const INITIAL_RETRY_DELAY: float = 0.5
 const MAX_RETRY_DELAY: float = 4.0
 const MAX_RETRIES: int = 30
 
 var _owner = null  # StepCanvasViewer (untyped to avoid cyclic ref)
-var _cache: Variant = null  # StepCanvasCache
+var _cache: Variant = null  # StepCanvasCache (Tier 1, in-memory)
+var _disk_cache: Variant = null  # StepCanvasDiskCache (Tier 2/3, T-1183)
 
 var _body_id: String = ""
 var _rung: String = ""
@@ -51,9 +66,19 @@ var _pending: bool = false
 var _retries: int = 0
 
 
-func _init(owner_ref = null) -> void:
+## `disk_cache_root` is test-injection-only (default "" -> StepCanvasDiskCache's
+## own production default, user://atlas_cache/): every production call site
+## (step_canvas_viewer.gd's `StepCanvasRequest.new(self)`) omits it, keeping
+## the real cache path unchanged; tests pass a disposable per-run subdirectory
+## so they never touch or leak into the real cache directory.
+func _init(owner_ref = null, disk_cache_root: String = "") -> void:
 	_owner = owner_ref
 	_cache = StepCanvasCache.new()
+	_disk_cache = (
+		StepCanvasDiskCache.new(disk_cache_root)
+		if not disk_cache_root.is_empty()
+		else StepCanvasDiskCache.new()
+	)
 
 
 ## Reset in-flight bookkeeping for a fresh body/rung entry — does NOT clear
@@ -63,9 +88,12 @@ func reset() -> void:
 	_retries = 0
 
 
-## Fire (or serve from cache) a step-canvas request. Cache hit -> immediate
-## synchronous canvas_ready emit, no network traffic. Cache miss -> send the
-## request now; the response (or a Pending retry chain) arrives later via
+## Fire (or serve from cache) a step-canvas request. Tier-1 hit -> immediate
+## synchronous canvas_ready emit, no disk or network I/O. Tier-1 miss checks
+## Tier 2 (disk, D-255(d)) next — also synchronous, a FileAccess read, no
+## wire traffic — and PROMOTES a disk hit into Tier 1 (the next request for
+## this exact key is a Tier-1 hit). Only a miss on BOTH tiers sends a wire
+## request; the response (or a Pending retry chain) arrives later via
 ## on_response().
 func request_now(
 	body_id: String, rung: String, center: Vector2i, extent: Vector2i, min_wl_m: int = 0
@@ -84,6 +112,15 @@ func request_now(
 		canvas_ready.emit(cached)
 		return
 
+	var disk_cached: Variant = _disk_cache.get_canvas(body_id, rung, center, extent, min_wl_m)
+	if disk_cached != null:
+		_pending = false
+		_retries = 0
+		_cache.put(body_id, rung, center, extent, disk_cached, min_wl_m)
+		_held_extent = _echoed_extent(disk_cached, rung, extent)
+		canvas_ready.emit(disk_cached)
+		return
+
 	_pending = true
 	_retries = 0
 	SimBridge.request_step_canvas(body_id, rung, center, extent, min_wl_m)
@@ -121,6 +158,10 @@ 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)
 
 
@@ -192,3 +233,7 @@ func get_held_extent() -> Vector2i:
 
 func get_cache() -> Variant:
 	return _cache
+
+
+func get_disk_cache() -> Variant:
+	return _disk_cache

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

@@ -46,6 +46,13 @@ const StepCanvasRequest := preload("res://ui/implant/apps/atlas/step_canvas/step
 const PANEL_MARGIN: float = 16.0
 const OVERLAY_BAR_HEADER_RESERVE: float = 360.0
 
+## Coarse background sweep interval (T-1183, D-255(d)): the disk cache's 2b
+## (LRU-capacity) + Tier-3 (sim-state TTL) sweeps run on this timer, NEVER
+## per-frame/per-step-cross (stig-round2.md "Sweep triggers": "proposed on
+## the order of minutes, tunable"). 5 minutes as the initial tunable — a
+## deferred, low-priority pass, never blocking a frame or a step-cross.
+const DISK_SWEEP_INTERVAL_SEC: float = 300.0
+
 const PAN_SPEED_PX_S: float = 220.0
 const EDGE_SCROLL_MARGIN_PX: float = 24.0
 
@@ -102,6 +109,7 @@ var _screen_header: ImplantHeader = null
 var _overlay_bar = null
 var _legend_panel = null
 var _request = null  # StepCanvasRequest
+var _disk_sweep_timer: Timer = null  # T-1183 coarse background sweep trigger
 
 
 func _ready() -> void:
@@ -138,6 +146,13 @@ func _ready() -> void:
 	_build_overlay_bar()
 	_build_legend_panel()
 
+	_disk_sweep_timer = Timer.new()
+	_disk_sweep_timer.name = "DiskSweepTimer"
+	_disk_sweep_timer.wait_time = DISK_SWEEP_INTERVAL_SEC
+	_disk_sweep_timer.autostart = true
+	_disk_sweep_timer.timeout.connect(_on_disk_sweep_timeout)
+	add_child(_disk_sweep_timer)
+
 	SimBridge.step_canvas_received.connect(_on_step_canvas_received)
 
 
@@ -168,6 +183,25 @@ func enter(body: Dictionary, system: Dictionary) -> void:
 	grab_focus()
 	queue_redraw()
 
+	# T-1183, D-255(d) sweep trigger (2a): "on body-open... this is the
+	# natural moment ('returning to a body') where stale-by-absence entries
+	# are most likely to exist" (stig-round2.md). Cheap — an index scan, no
+	# bulk payload I/O beyond the deletes it triggers.
+	var body_id: String = get_body_id()
+	if not body_id.is_empty():
+		_request.get_disk_cache().run_visit_sweep(body_id)
+
+
+func _on_disk_sweep_timeout() -> void:
+	# T-1183, D-255(d) sweep triggers (2b + Tier 3): a coarse background
+	# timer, never per-frame/per-step-cross. Only sweeps the CURRENTLY open
+	# body — a body the player isn't looking at doesn't need its disk cache
+	# swept on this viewer's own clock (it gets swept on its own next
+	# body-open, per (2a) above).
+	var body_id: String = get_body_id()
+	if not body_id.is_empty():
+		_request.get_disk_cache().run_background_sweep(body_id)
+
 
 func leave() -> void:
 	pass
@@ -181,6 +215,14 @@ func get_held_rung() -> String:
 	return _held_rung
 
 
+## T-1183 test seam: exposes the owned StepCanvasRequest (and, through it,
+## get_disk_cache()) so sweep-trigger wiring is directly testable, matching
+## the get_cache()/get_disk_cache() accessor pattern StepCanvasRequest
+## already exposes for the same reason.
+func get_request() -> Variant:
+	return _request
+
+
 func is_overlay_visible(overlay_id: String) -> bool:
 	return bool(_overlay_visibility.get(overlay_id, false))