Summary

Adds a focused programmatic accessor for Rocky state-backend health to the Dagster integration. Dagster users observing Rocky state-backend health can now read RockyResource.state_health() to get a live snapshot of the configured backend + the most recent run's status; with probe_write=True it additionally exercises the engine's state_rw put/get/delete probe and surfaces the outcome as typed fields. Closes the reprioritization-doc wave 2 follow-up (FR-003) now that wave 1 (#224) has merged.

Design

• Data sources — both already-shipped in v1.13.0. The accessor aggregates two existing CLI surfaces:
  1. rocky history --output json for last_run_status / last_run_at (normalised from RunStatus's CamelCase wire values into snake_case literals so callers don't need to know the serialisation).
  2. rocky doctor for the optional state_rw probe (translated into a tri-state ok / timeout / error outcome).
     Plus a cheap tomllib.load of config_path for the backend field.
• No new engine CLI commands. The FR's proposed rocky state-health CLI verb stays out of scope — all the substrate (probe_state_backend helper, record_run wiring) already ships, so a new verb would just cascade through codegen without adding signal the existing surfaces don't already carry.
• No schema changes, no codegen cascade, no fixture regen. Pure Python.
• Sensor-tick safety. history() raising dg.Failure (missing binary, unreadable state store) degrades to last_run_status=None rather than crashing the sensor tick. A failing probe populates probe_outcome="error" + probe_error rather than raising. Mirrors the existing rocky_healthcheck tolerance pattern.
• Thin facade. The method on RockyResource delegates to health.state_health() — same split as the existing rocky_healthcheck function so the resource stays easy to diff against wave 1's additions.

Shape

Matches the FR's explicit Python sketch:

class StateHealthResult(BaseModel):
    backend: Literal["local", "tiered", "valkey", "s3", "gcs"]
    last_run_status: Literal["success", "partial_failure", "failure"] | None
    last_run_at: datetime | None
    probe_outcome: Literal["ok", "timeout", "error"] | None
    probe_duration_ms: int | None
    probe_error: str | None

The FR's stretch-goal "recent outcomes" rollup (persisted ring buffer of state-upload / state-download outcomes) stays deferred — those signals are tracing::info! events only today and would require new engine-side persistence. Flagging explicitly so the decision is visible in history rather than buried.

Known follow-up optimisation

RockyResource.doctor() today doesn't accept a --check filter — the probe path therefore invokes the full doctor (~1-2s cold). The engine CLI itself supports --check state_rw; wiring a check: str | None = None kwarg on RockyResource.doctor() would cut the probe cost to the engine's sub-second probe_state_backend helper and is a natural follow-up. Not landed here to keep the scope tight.

Test plan

• uv run pytest — 366 tests pass (24 new in test_state_health.py, covering the full matrix: all 5 backends, fresh state store, each RunStatus variant, unknown status → failure fallback, binary failure tolerance, probe-skipped-by-default, healthy probe → ok, critical + timeout message → timeout, critical generic → error, probe check missing, probe binary failure, probe warning severity, resource-method facade).
• uv run ruff check src/ tests/ — clean.
• uv run ruff format --check src/ tests/ — clean.
• No engine changes, no codegen cascade, no fixture regen.
• Wave 1 surface (setup_for_execution, run_pipes, RockyPipesMessageReader, strict_doctor*) untouched — the new method sits next to doctor() and does not intersect.

Wave 2 follow-up

This closes the wave 2 item in the 2026-04-22 reprio — the Python-only state-health accessor. Wave 3 (workspace-binding reconciliation, engine-side) and wave 4 (adapter-namespaced source metadata, codegen cascade) remain.