Frac load gen: local frac → MSK direct-SCRAM producer + sshuttle/SSM tunnel

> **Part of a 4-issue set for one remote agent** (Epic below → A → B → C → D). Implement the whole set in **one PR**, in order. Read the design spec: [https://github.com/aws-samples/sample-edge-to-cloud-digital-ops-workshop/blob/feat/frac-load-dashboard-timeseries/docs/superpowers/specs/2026-09-09-frac-load-and-dashboard-timeseries-design.md](https://github.com/aws-samples/sample-edge-to-cloud-digital-ops-workshop/blob/feat/frac-load-dashboard-timeseries/docs/superpowers/specs/2026-09-09-frac-load-and-dashboard-timeseries-design.md).

## A — Local frac → MSK load generator + sshuttle/SSM tunnel

Give an operator a **local** script that generates realistic well-fracturing telemetry at configurable volume and pushes it **directly into the cloud MSK** cluster.

### Context
- MSK is in the private `workshop-cloud` VPC (broker port 9096, SASL_SSL/SCRAM-SHA-512). A laptop cannot reach it without a tunnel.
- The SCRAM/bootstrap recipe already exists in `scripts/create-msk-topics.sh` (secret `AmazonMSK_workshop-`, brokers from `aws kafka get-bootstrap-brokers … BootstrapBrokerStringSaslScram`, `kafka-python` fallback) — **reuse it verbatim**.
- The frac sensor model already exists in `simulator/sensor-sim.py` (`SENSORS`, `SimulationEngine`, `build_payload`) and `simulator/frac-op-burst.py` (by-path import + hard-stop watchdog pattern) — **reuse, do not reinvent**.
- Target topic **`sensors.raw.sim`** with the frac schema `{sensor, value, unit, ts_ms, site_id}`. **`site_id` MUST start with the `ws-slotNN` prefix** (e.g. `ws-slot00-siteA-pump-03`) — downstream `deployment_id` is derived from that prefix.

### Deliverables
1. **`simulator/frac-msk-load.py`** (new): imports the sensor model from `sensor-sim.py`; produces JSON directly to MSK via `kafka-python` `KafkaProducer(security_protocol="SASL_SSL", sasl_mechanism="SCRAM-SHA-512", …)`. CLI: `--slot` (required), `--sites`, `--units-per-site`, `--hz`, `--duration`, `--topic` (default `sensors.raw.sim`), `--region`, `--dry-run`. Hard-stop watchdog + live achieved-rate summary like `frac-op-burst.py`.
2. **`scripts/msk-tunnel.sh`** (new): `sshuttle` over SSH-over-SSM to an EKS node in the cloud VPC, routing the VPC CIDR (`10.1.0.0/16`) **and DNS** (`sshuttle --dns … --ssh-cmd 'ssh -o ProxyCommand="aws ssm start-session --target %h --document-name AWS-StartSSHSession --parameters portNumber=%p"'`). A plain port-forward does NOT work (multi-broker advertised-DNS collision) — document that in a header comment.

### Acceptance (agent, offline)
- `python simulator/frac-msk-load.py --slot ws-slot00 --dry-run` prints payloads whose keys are exactly `sensor,value,unit,ts_ms,site_id`, every `site_id` starts with `ws-slot`, and `ts_ms` is epoch-ms.
- The producer config uses `SASL_SSL` + `SCRAM-SHA-512`; no MQTT.
- `shellcheck scripts/msk-tunnel.sh` is clean.

**Live-slot validation (real tunnel + real produce) stays with the local credentialed session — not required from the agent.**

Start with the linked design spec, then read scripts/create-msk-topics.sh, simulator/sensor-sim.py, and simulator/frac-op-burst.py for the existing MSK recipe and sensor/watchdog patterns. Implement and inspect simulator/frac-msk-load.py and scripts/msk-tunnel.sh against the listed offline acceptance checks, including exact dry-run payload keys, slot-prefixed site IDs, epoch-millisecond timestamps, SCRAM-SHA-512 configuration, and clean shellcheck output.

Written by the indexing model from the issue text.