Summary

Closes the last governance-waveplan observability gap on RockyComponent. After engine-v1.16.0 shipped the Wave B (rocky compliance) and Wave C-2 (rocky retention-status) surfaces and dagster-v1.12.0 shipped the matching Pydantic types, every RockyComponent adopter who wanted compliance/retention events on the asset graph had to hand-roll a wrapper asset that shelled out, parsed JSON, and yielded AssetCheckResult / AssetObservation — duplicating work the component already does for drift + anomalies.

This PR adds two opt-in YAML attributes (surface_compliance, surface_retention_status, both default False — zero behaviour change for deployments that don't flip them). When on, the component auto-wires both surfaces through the existing bridge pattern.

Adopters flip both on with two lines in defs.yaml:

type: dagster_rocky.RockyComponent
attributes:
  binary_path: rocky
  config_path: rocky/rocky.toml
  surface_compliance: true
  surface_retention_status: true

What got added

RockyResource methods (mirror state_health() shape):

• compliance(env=None) -> ComplianceOutput
• retention_status(env=None) -> RetentionStatusOutput

observability.py helpers + constants (mirror drift_observations() / anomaly_check_results()):

• compliance_check_results(output, *, key_resolver) -> Iterator[AssetCheckResult]
• retention_observations(output, *, key_resolver) -> Iterator[AssetObservation]
• COMPLIANCE_CHECK_NAME = "compliance_exception"
• COMPLIANCE_FALLBACK_ASSET_KEY = AssetKey(["_compliance"])
• RETENTION_OBSERVATION_NAME = "retention_drift"

RockyComponent wiring:

• Two Pydantic fields: surface_compliance: bool = False, surface_retention_status: bool = False.
• Pre-declared compliance_exception check spec per asset when opt-in is on (matches the anomaly pattern).
• _emit_governance_events helper invoked in both execution modes (streaming + pipes).
• Graceful error swallowing on binary failure (warning log, materialization continues).
• Invariant guard that drops sentinel-keyed compliance results with a warning rather than crashing Dagster when exceptions target a model outside the component's selection.

Public API re-exports via dagster_rocky/__init__.py for compliance_check_results, retention_observations, the three new constants, ComplianceOutput, RetentionStatusOutput, ComplianceException, ComplianceSummary, ModelRetentionStatus.

Tests (24 new, 417 total after rebase on #248, all green):

• Parse-guard tests for both output types.
• Helper-level tests: aggregation per asset, sentinel fallback, empty case, resolver-miss path, forward-compat for warehouse_days.
• Resource tests: argv shape with/without --env.
• Component integration tests: pre-declared specs, opt-in gating, end-to-end events alongside drift+anomaly, failure tolerance, unknown-model filtering.

Test plan

• cd integrations/dagster && uv run pytest — 417 passed
• uv run ruff check src/ tests/ — clean
• uv run ruff format --check src/ tests/ — clean
• Verify in a downstream consumer that flipping surface_compliance: true on their RockyComponent YAML surfaces compliance exceptions on the asset graph without any wrapper code

Deviations from the original design sketch

Several field names and call shapes in the original design sketch diverged from what the shipped schemas actually expose. Deliberate deviations:

• ComplianceException has no severity field — real fields are column / env / model / reason. The sketch's "passed=True when severity=info/acknowledged, passed=False when severity=warn/error" is unimplementable. Every exception yields passed=False, severity=WARN.
• ModelRetentionStatus fields differ from the sketch — real fields are configured_days / in_sync / model / warehouse_days. Metadata keys are named accordingly: rocky/retention_configured_days, rocky/retention_warehouse_days, rocky/retention_in_sync. warehouse_days is always None in v1 (schema doc) so the metadata field is omitted conditionally.
• Helpers take key_resolver: Callable[[str], AssetKey | None], not translator: RockyDagsterTranslator — matches the existing drift_observations() / anomaly_check_results() contract. Simpler signature, and translator.get_asset_key(source, table) doesn't fit a compliance exception (which has model + column, no source/TableInfo).
• Compliance aggregates per asset, not per exception — Dagster rejects duplicate (asset_key, check_name) pairs per materialization. Multiple exceptions for the same model (typically one per env) fold into one aggregated WARN result; metadata lists every (model, column, env) triple. This is the only deviation from a "one event per input row" helper shape; retention still yields one observation per row.
• No live-binary fixtures. The integration's fixture convention has moved past JSON files — conftest.py comment spells out that the legacy tests/fixtures/*.json was removed; scenarios now live as Python dicts in scenarios.py and flow through conftest.py via json.dumps. The playground POC also doesn't have governance config, so regen-fixtures couldn't capture a live capture without also changing the POC. Added COMPLIANCE + RETENTION_STATUS dicts to scenarios.py and compliance_json / retention_status_json fixtures to conftest.py — the parse-guard still fires via the new tests.