perf: bound lake scans and add platform Grafana dashboard

Explorer and exporter were scanning the full Parquet lake every 1–5s (~2000 files, 200MB+), driving ~2.2 CPU cores. Limit metrics to the last N flights, cache DuckDB views and the partition tree, slow live polls to 2s, and keep drones alive after seal to stop restart churn. Add node-exporter, scan-duration metrics, and a Swarm Platform Grafana dashboard for node CPU/memory and scan health.
fix: wire live mode and Grafana to the generated data stream
2026-07-08 21:19:53 +01:00 · 2026-07-08 20:34:37 +01:00 · 2026-07-08 20:27:25 +01:00 · 2026-07-08 20:20:19 +01:00 · 2026-07-08 20:18:06 +01:00 · 2026-07-08 20:17:34 +01:00
58 changed files with 2757 additions and 190 deletions
@@ -30,6 +30,10 @@ jobs:
      - name: Install dependencies
        run: pip install --quiet -r simulator/requirements.txt
      - name: Unit tests (TDD gate)
        working-directory: simulator
        run: pytest tests/ -q
      - name: Compile check
        run: python -m compileall -q simulator/virtual_drone simulator/monitoring simulator/explorer
@@ -49,26 +53,6 @@ jobs:
          print(f"smoke flight OK: {rows} telemetry rows")
          EOF
      - name: SQL gate tests
        working-directory: simulator
        run: |
          python - <<'EOF'
          import sys
          sys.path.insert(0, "explorer")
          from server import gate
          assert gate("SELECT 1") is None
          assert gate("WITH t AS (SELECT 1) SELECT * FROM t") is None
          assert gate("DESCRIBE telemetry") is None
          assert gate("DROP TABLE telemetry") is not None
          assert gate("SELECT 1; SELECT 2") is not None
          assert gate("INSTALL httpfs") is not None
          assert gate("SET memory_limit='1GB'") is not None
          assert gate("/* sneaky */ COPY t TO 'x'") is not None
          assert gate("") is not None
          print("SQL gate OK: reads pass, writes and config are rejected")
          EOF
  scan:
    name: Trivy scan
    runs-on: ubuntu-latest
@@ -141,6 +125,7 @@ jobs:
          TAG="${{ steps.version.outputs.new_tag }}"
          mkdir -p .tmp
          tar --exclude='.git' --exclude='.tmp' --exclude='simulator/data' \
              --exclude='var/t1' --exclude='var/t3' \
              --exclude='simulator/.venv' --exclude='prototype/node_modules' \
              -czf ".tmp/swarm-house-${TAG}.tar.gz" .
@@ -1,7 +1,15 @@
 data/
 simulator/data/
 var/t1/**
 var/t3/**
 !var/t1/.gitkeep
 !var/t3/.gitkeep
 __pycache__/
 *.pyc
 .venv/
 node_modules/
 dist/
 .env
 .terraform/
 *.tfstate
 *.tfstate.*
@@ -15,8 +15,13 @@ Rules for anyone (human or tooling) contributing to this repository.
 docs/           Design documentation (Markdown + Mermaid)
 simulator/      Virtual drone fleet — Python data generator (Parquet/DuckDB pipeline)
 prototype/      2D swarm visualization — React + TypeScript (Vite)
 infra/          Ansible host prep, Terraform workloads, Flux GitOps overlays
 var/t1/         Shared T1 live lake (gitignored runtime data)
 var/t3/         Shared T3 warehouse (gitignored runtime data)
 ```
 See [`CONTRIBUTING.md`](CONTRIBUTING.md) for the development workflow and TDD requirements.
 ## Conventions
 - Markdown: ATX headings, fenced code blocks with language tags, pipe tables.
@@ -0,0 +1,90 @@
 # Contributing
 Thank you for helping improve Swarm House. This repository is a self-contained
 design proposal for an on-prem data platform for autonomous drone swarms — all
 contributions must follow the rules in [`AGENTS.md`](AGENTS.md).
 ## TDD first (required)
 Every change to executable code ships with tests **before** or **alongside** the
 implementation — never as an afterthought.
 1. **Red:** write or extend a failing test that captures the behaviour you need.
 2. **Green:** implement the smallest change that makes it pass.
 3. **Refactor:** clean up without weakening the test.
 The CI pipeline enforces this:
 | Gate | When it runs |
 | --- | --- |
 | `pytest tests/` | Every push/PR (`verify` job) |
 | `RUN pytest tests/` in the simulator Docker build | Every `docker build` — a red test blocks the image |
 | Smoke flight + DuckDB assertion | Every push/PR |
 | Trivy scan | After verify passes |
 Test locations:
 - `simulator/tests/` — Python unit tests (SQL gate, broadcast codec, writer layout, flight kinematics)
 - Prototype: `npm run build` / `tsc --noEmit` before UI changes
 ## Local development
 ### Fast inner loop (Compose)
 ```bash
 cd simulator
 pip install -r requirements.txt
 pytest tests/ -q                          # run tests first
 DRONE_COUNT=5 docker compose up --build   # writes to ../var/t1
 ```
 ### Full ground miniature (k3d + Terraform + Flux)
 ```bash
 ansible-playbook infra/ansible/sim-cluster.yml
 cd infra/terraform/sim-env && terraform init && terraform apply
 cd ../ground && terraform init && terraform apply
 cd ../../prototype && npm install && npm run dev   # press "go live"
 ```
 Shared data directories (gitignored, created automatically):
 | Path | Tier | Contents |
 | --- | --- | --- |
 | `var/t1/` | T1 live lake | In-flight Parquet from Compose or k3d fleet |
 | `var/t3/` | T3 warehouse | Offloaded historical partitions |
 Override with `SWARM_T1_DIR` / `SWARM_WAREHOUSE_DATA` / `SWARM_SIM_DATA` env vars.
 ### Prototype only
 ```bash
 cd prototype && npm install && npm run dev
 ```
 ## Commit messages
 Conventional Commits only:
 ```
 feat: add warehouse offload cron schedule variable
 fix: reject COPY statements in the SQL gate
 docs: clarify WireGuard vs SSH layering
 test: cover broadcast frame roundtrip
 ```
 Never include AI tool references, `Co-authored-by` trailers for assistants, or
 identifying names of people, companies, or locations.
 ## Pull request checklist
 - [ ] Tests added or updated; `pytest tests/ -q` passes locally
 - [ ] `docker build` in `simulator/` succeeds (test stage included)
 - [ ] Documentation updated if behaviour or boundaries changed
 - [ ] English only; no identifying references
 - [ ] Conventional commit message
 ## Questions
 Open design questions live in [`docs/09-open-questions.md`](docs/09-open-questions.md).
 Architecture boundaries are documented in [`docs/06-environments.md`](docs/06-environments.md#iac-boundaries).
@@ -39,18 +39,37 @@ This repository describes how to build, deliver, test, and operate that platform
 | Component | Purpose | Stack |
 | --- | --- | --- |
 | [`simulator/`](simulator/) | Virtual drone fleet: generates realistic telemetry through the same Parquet/DuckDB pipeline a real drone would use; N drones via Docker Compose | Python, pyarrow, DuckDB |
-| [`prototype/`](prototype/) | 2D top-down swarm visualization: drones, obstacles, inter-drone links with live transfer metrics | React, TypeScript, Vite, SVG |
+| [`prototype/`](prototype/) | 2D top-down swarm visualization: drones, obstacles, inter-drone links with live transfer metrics; **live mode** renders real positions from the k3d fleet | React, TypeScript, Vite, SVG |
 | [`infra/ansible/`](infra/ansible/) | Host preparation: drone bench provisioning (keys, forced commands, WireGuard, Compose bundle) and local k3d simulation cluster | Ansible |
 | [`infra/terraform/`](infra/terraform/) | Declared workloads on the sim cluster: virtual fleet + observability ([`sim-env`](infra/terraform/sim-env/)), warehouse + T1→T3 offload ([`ground`](infra/terraform/ground/)) | Terraform, kubernetes provider |
 | [`infra/gitops/`](infra/gitops/) | Flux overlays for the ground segment — policy labels, dashboard bundles; drones stay on the fleet manifest | Flux, Kustomize |
 Shared runtime data (gitignored): `var/t1/` (live lake), `var/t3/` (warehouse). See [`CONTRIBUTING.md`](CONTRIBUTING.md).
 ## Quick start
 ```bash
-# Spin up a virtual swarm of 10 drones
+# Spin up a virtual swarm of 10 drones (Compose, no cluster needed)
 cd simulator && DRONE_COUNT=10 docker compose up
 # Run the visualization
 cd prototype && npm install && npm run dev
 ```
 Or run the full miniature of the ground segment — k3d cluster, fleet as
 StatefulSet, Grafana, explorer, warehouse with post-flight offload:
 ```bash
 ansible-playbook infra/ansible/sim-cluster.yml
 cd infra/terraform/sim-env && terraform init && terraform apply
 cd ../ground && terraform init && terraform apply
 # then press "go live" in the prototype header
 ```
 See [06 — Environments → IaC boundaries](docs/06-environments.md#iac-boundaries) for why Ansible owns hosts, Terraform owns the ground segment, and neither flies on the drone.
 Contributing (TDD-first workflow): [`CONTRIBUTING.md`](CONTRIBUTING.md).
 ## CI & releases
-Every push runs the pipeline in [`.github/workflows/release.yaml`](.github/workflows/release.yaml): a compile check, a 10-second virtual smoke flight verified with DuckDB, unit tests for the read-only SQL gate, and a Trivy vulnerability/misconfiguration scan. Pushes to `main` then cut a semantic version from the conventional-commit type (`feat:` → minor, `!`/`BREAKING CHANGE` → major, else patch), tag the commit, and publish a release with a **fleet release manifest** (versioned artifact list with checksums — the delivery unit described in [11 — CI/CD & delivery](docs/11-cicd-delivery.md)) plus a source bundle.
+Every push runs the pipeline in [`.github/workflows/release.yaml`](.github/workflows/release.yaml): **pytest unit tests**, a compile check, a 10-second virtual smoke flight verified with DuckDB, and a Trivy vulnerability/misconfiguration scan. The simulator Docker build also runs `pytest` — a failing test blocks the image. Pushes to `main` then cut a semantic version from the conventional-commit type (`feat:` → minor, `!`/`BREAKING CHANGE` → major, else patch), tag the commit, and publish a release with a **fleet release manifest** (versioned artifact list with checksums — the delivery unit described in [11 — CI/CD & delivery](docs/11-cicd-delivery.md)) plus a source bundle.
@@ -36,6 +36,21 @@ graph TB
 - **RSSI is data.** Per-peer signal strength is captured into `telemetry` like any other sensor — it feeds relative-positioning estimates and post-flight link-quality analysis, and it is free.
 - Electromagnetic environment sensing (whatever the radio can observe) is likewise recorded — cheap in flight, valuable in replay.
 ## Why not SSH alone?
 SSH and WireGuard solve different problems. SSH is the **authorization layer** for narrow, point-to-point operations (bulk sync, read-only SQL). WireGuard is the **membership and transport encryption layer** for everything on the mesh, including traffic SSH cannot carry.
 | Concern | SSH alone | WireGuard + SSH (this design) |
 | --- | --- | --- |
 | Pose broadcast (UDP, ~5 Hz, compact frame) | Cannot carry subnet broadcast; would need a separate app protocol | Encrypted inside the mesh overlay; any peer on the tunnel can receive |
 | Broadcast confidentiality without WG | Payloads traverse the radio **in cleartext** — any receiver in range reads swarm positions | Only provisioned fleet members complete a handshake; outsiders see noise |
 | Lossy ad-hoc links | TCP head-of-line blocking; sessions stall and reconnect on packet loss | UDP transport survives loss and roaming; SSH sessions **over** WG are more stable than SSH directly on Wi-Fi |
 | Network membership | Any host that reaches the port gets an SSH banner — a probe surface | Devices without a provisioned key get **no handshake response**; the port is effectively silent |
 **Rejected alternative:** encrypt each broadcast frame in application code with a fleet-wide AEAD key. That reinvents a crypto layer the OS already provides, still leaves TCP head-of-line blocking for peer queries, and does not answer the membership question (who is allowed on the mesh at all).
 Keep both layers: WireGuard for *who is on the network and whether the wire is readable*; SSH forced commands for *what a peer is allowed to do once connected* ([04](04-swarm-sync.md)).
 ## Key lifecycle
 | Phase | Action |
@@ -61,6 +61,29 @@ The environment engineers live in daily — and deliberately the first thing to
 - **T4 data slices**: a thin tool pulls partition subsets from T3 (`flight=X, drone=Y, hour=Z`) for local work — layout-identical, so every query and pipeline runs unmodified.
 - The same simulator images scale out on the ground k3s farm for CI regression runs: every merge request replays canonical scenarios and asserts on the resulting Parquet output (row counts, coverage, staleness budgets).
 ## IaC boundaries
 Infrastructure as code follows the same discipline as the data plane: each tool owns the layer it is actually good at, and nothing owns a layer it can't reach.
 | Layer | Tool | Why |
 | --- | --- | --- |
 | Host preparation (drone bench provisioning, sim cluster creation) | **Ansible** ([`infra/ansible/`](../infra/ansible/)) | Idempotent, agentless, works over the same SSH channel the platform already trusts. Keys, forced commands, WireGuard, Compose bundles — all host-level state. |
 | Declared workloads (simulated fleet, ground warehouse) | **Terraform** ([`infra/terraform/`](../infra/terraform/)) | The ground segment is stationary, long-lived, and cluster-shaped — classic Terraform territory. Fleet size, offload schedule, and explorer endpoints are variables, not YAML edits. |
 | Ongoing ground configuration | **Flux** ([`infra/gitops/`](../infra/gitops/)) | Policy labels, dashboard bundles, offload knobs — reconciled from git without reprovisioning PVCs. Drones stay on the fleet manifest, not GitOps. |
 | On-board runtime | **Compose bundle from the fleet release manifest** | Terraform is deliberately *not* on the drone: there is no API server to reconcile against mid-flight, and the release manifest already gives atomic, versioned, rollback-able delivery ([11](11-cicd-delivery.md)). |
 Shared local data paths (same layout everywhere): **`var/t1/`** (live lake, T1) and **`var/t3/`** (warehouse, T3). Compose, k3d, and DuckDB on the host all read the same Parquet tree.
 Two Terraform modules, two states, one deliberate split: [`sim-env`](../infra/terraform/sim-env/) (the disposable virtual fleet) and [`ground`](../infra/terraform/ground/) (the warehouse that must survive every `destroy` of the fleet). They never share a lifecycle.
 ### k3d: the ground segment as an executable miniature
 `ansible-playbook infra/ansible/sim-cluster.yml` creates a local k3d cluster (k3s in Docker — the same distribution as the real ground segment) and `terraform apply` populates it: virtual drones as StatefulSet replicas over a shared Parquet lake, the exporter/Prometheus/Grafana stack, the data-plane explorer, MinIO, and the T1→T3 offload CronJob. The whole diagram at the top of this page runs on one workstation, and the prototype's **live mode** renders real drone positions straight from the explorer's read-only SQL API.
 ### Next fidelity step: microVMs
 The k3d fleet shares one kernel and one network namespace tree — good enough to exercise the data plane, not good enough to demo the *peer* plane (real SSH sessions between isolated machines, WireGuard handshakes, packet loss injection). The next step, when that fidelity is needed, is one **Firecracker microVM per drone**: a real kernel, a real network interface, and the actual `authorized_keys` forced-command path between virtual drones — on any KVM-capable workstation, still with no cloud dependency. The process-level simulator stays as the fast inner loop; microVMs become the pre-flight integration rig.
 ## Why identical layouts matter (the payoff)
 | Operation | What it costs with one layout everywhere |
@@ -53,6 +53,13 @@ graph LR
 | Which link degraded first, and was it distance or interference? | RSSI telemetry joined with pose distance |
 | Is the new writer version flushing slower on ARM64? | CI simulation runs emit the same metrics; diff across fleet releases |
-A working miniature of this doctrine ships in [`simulator/`](../simulator/): the `monitoring` Compose profile runs a DuckDB-based exporter over the generated Parquet plus Prometheus and a provisioned Grafana dashboard — fleet statistics derived from the data platform itself, with no agent on the (virtual) drones.
+A working miniature of this doctrine ships in [`simulator/`](../simulator/): the `monitoring` Compose profile runs a DuckDB-based exporter over the generated Parquet plus Prometheus and two provisioned Grafana dashboards:
 | Dashboard | Focus |
 | --- | --- |
 | **Swarm Fleet — generated data** | Row counts, detections, pose frames, Parquet bytes, battery, RSSI |
 | **Swarm Platform — CPU & scan health** | Node CPU/memory (node-exporter), exporter/explorer scan durations, flight-window gauge |
 The exporter and explorer **scan only the most recent flight partitions** (`METRIC_FLIGHT_WINDOW`, default 5) and cache tree/views — otherwise CPU climbs as every pod restart appends a new `flight=` tree to the shared lake. Drones in k3d set `KEEP_ALIVE=1` after sealing so they idle instead of exiting and spawning another flight.
 Alerting on the ground follows standard practice (Prometheus alert rules for infrastructure, CI gates for regression in simulated staleness/throughput budgets). In flight there is nobody to page — the platform's job is to degrade in the documented order and record everything for the post-mortem.
@@ -97,6 +97,39 @@ config:
 2. Docked drones fetch the manifest, verify digests/signatures ([05](05-network-security.md)), stage the new Compose bundle, and switch on the next boot cycle.
 3. **Never mid-flight.** Updates are a dock-only operation by construction — the update endpoint does not exist in the flight radio profile.
 ## GitOps on the ground segment
 The fleet release manifest handles **drones** (atomic, digest-pinned, dock-only). The **ground k3s segment** uses a different tool: [Flux](https://fluxcd.io/) reconciles long-lived configuration from git — dashboard bundles, Prometheus rules, offload schedules, policy labels — without reprovisioning PVCs or re-running full Terraform applies for every label change.
 ```mermaid
 graph LR
    subgraph git [Git, on-prem]
        REPO["service repos"]
        GITOPS["infra/gitops/<br/>ground overlay"]
    end
    subgraph ground_k3s [Ground k3s]
        FLUX["Flux controllers"]
        WH["warehouse workloads<br/>(Terraform-provisioned)"]
        CFG["policy ConfigMaps<br/>dashboard bundles"]
    end
    subgraph drones [Fleet]
        MAN["fleet release manifest"]
        DOCK["dock: verify + apply"]
    end
    REPO --> CI
    GITOPS --> FLUX
    FLUX --> CFG
    MAN --> DOCK
 ```
 | Segment | Delivery mechanism | Reconciler in production? |
 | --- | --- | --- |
 | Drones | Fleet release manifest via registry mirror | **No** — no API server mid-flight; dock applies once |
 | Ground k3s (warehouse, dashboards, CI runners) | Terraform for shape + **Flux** for ongoing config | **Yes** — stationary cluster, wired network, git is the source of truth |
 | Dev simulation (k3d) | Ansible creates cluster; Terraform applies workloads; Flux CRs installed from [`infra/gitops/`](../infra/gitops/) | Yes (miniature of ground) |
 Boundary: **Terraform provisions** (namespaces, PVCs, Deployments, CronJobs); **Flux reconciles** policy and observability overlays on top. Neither replaces the fleet manifest on the drone.
 ## Developer experience
 - **Dev Containers** define the full toolchain (Python data tooling, DuckDB, compose, linters) — identical on any machine, onboarding in minutes.
@@ -0,0 +1,39 @@
 # Ansible — host preparation
 Ansible owns everything that happens **on a host before workloads run**.
 Terraform ([`../terraform/`](../terraform/)) owns the workloads themselves.
 The boundary is deliberate and documented in
 [06 — Environments](../../docs/06-environments.md#iac-boundaries).
 | Playbook | Target | What it does |
 | --- | --- | --- |
 | `drone-provision.yml` | Drone (on the bench, before a mission) | Identity keys, peer trust with forced commands, WireGuard, Compose bundle from the fleet release manifest |
 | `sim-cluster.yml` | Local workstation / CI host | k3d cluster that miniatures the ground segment; shared data volume; simulator image import |
 ## Provision a drone
 ```bash
 ansible-playbook -i inventory.example.yml drone-provision.yml
 ```
 The playbook is idempotent and runs only over the bench network — drones
 are never provisioned in flight (see [05 — Network & security](../../docs/05-network-security.md)):
 1. **Identity**: generate the drone's ed25519 keypair if absent.
 2. **Peer trust**: install every fleet member's public key into
   `authorized_keys`, each pinned to the read-only SQL forced command —
   the only thing a peer can execute.
 3. **WireGuard**: render `wg0.conf` with the drone's address and peers.
 4. **Data plane**: lay down the Compose bundle referenced by the fleet
   release manifest and enable it as a systemd unit.
 ## Create the simulation cluster
 ```bash
 ansible-playbook sim-cluster.yml            # creates k3d cluster 'swarm-sim'
 cd ../terraform/sim-env && terraform init && terraform apply
 ```
 The cluster mounts a shared host directory as `/data` on the (single)
 node, so drone pods, the exporter, and the explorer see one Parquet lake —
 the same contract as the on-board NVMe layout.
@@ -0,0 +1,101 @@
 # Provision a drone on the bench: identity, peer trust, WireGuard, data plane.
 # Idempotent; never runs over the mission radio (bench network only).
 ---
 - name: Provision drone data plane
  hosts: drones
  become: true
  vars:
    swarm_user: swarm
    swarm_home: /opt/swarm
    forced_command: "{{ swarm_home }}/bin/peer-query.sh"
  tasks:
    - name: Create swarm service user
      ansible.builtin.user:
        name: "{{ swarm_user }}"
        home: "{{ swarm_home }}"
        shell: /usr/sbin/nologin
        system: true
    # --- 1. Identity -----------------------------------------------------
    - name: Generate drone ed25519 identity key
      community.crypto.openssh_keypair:
        path: "{{ swarm_home }}/.ssh/id_ed25519"
        type: ed25519
        comment: "{{ inventory_hostname }}"
        owner: "{{ swarm_user }}"
        mode: "0600"
      register: identity
    - name: Collect fleet public keys
      ansible.builtin.slurp:
        src: "{{ swarm_home }}/.ssh/id_ed25519.pub"
      register: pubkey
    # --- 2. Peer trust: forced command only ------------------------------
    # Every peer may connect, but the only thing it can execute is the
    # read-only SQL wrapper (docs/04). No shell, no forwarding, no pty.
    - name: Authorize fleet peers with read-only forced command
      ansible.posix.authorized_key:
        user: "{{ swarm_user }}"
        key: "{{ hostvars[item].pubkey.content | b64decode }}"
        key_options: >-
          command="{{ forced_command }}",no-port-forwarding,no-agent-forwarding,no-X11-forwarding,no-pty
      loop: "{{ groups['drones'] | difference([inventory_hostname]) }}"
      when: hostvars[item].pubkey is defined
    - name: Install peer query wrapper
      ansible.builtin.copy:
        dest: "{{ forced_command }}"
        mode: "0755"
        content: |
          #!/bin/sh
          # Forced command: read-only DuckDB SQL over the sealed lake.
          # SSH_ORIGINAL_COMMAND carries the statement; the gate rejects
          # anything that is not a single read statement.
          exec {{ swarm_home }}/bin/sql-gate --read-only --data /data "$SSH_ORIGINAL_COMMAND"
    # --- 3. WireGuard swarm plane ----------------------------------------
    - name: Render WireGuard configuration
      ansible.builtin.template:
        src: templates/wg0.conf.j2
        dest: /etc/wireguard/wg0.conf
        mode: "0600"
      notify: Restart wireguard
    - name: Enable WireGuard interface
      ansible.builtin.systemd:
        name: wg-quick@wg0
        enabled: true
        state: started
    # --- 4. Data plane from the fleet release manifest --------------------
    - name: Read fleet release manifest
      ansible.builtin.include_vars:
        file: "{{ fleet_manifest }}"
        name: manifest
    - name: Lay down Compose bundle for release {{ manifest.release }}
      ansible.builtin.unarchive:
        src: "{{ playbook_dir }}/../../.tmp/{{ manifest.artifacts[0].name }}"
        dest: "{{ swarm_home }}/release"
        owner: "{{ swarm_user }}"
    - name: Enable data plane unit
      ansible.builtin.template:
        src: templates/swarm-data-plane.service.j2
        dest: /etc/systemd/system/swarm-data-plane.service
        mode: "0644"
      notify: Restart data plane
  handlers:
    - name: Restart wireguard
      ansible.builtin.systemd:
        name: wg-quick@wg0
        state: restarted
    - name: Restart data plane
      ansible.builtin.systemd:
        name: swarm-data-plane
        state: restarted
        daemon_reload: true
@@ -0,0 +1,19 @@
 # Example fleet inventory. Real inventories are generated from the fleet
 # release manifest and live outside the repository.
 all:
  children:
    drones:
      hosts:
        dr-01:
          ansible_host: 10.44.0.11
          wg_address: 10.44.0.11/24
        dr-02:
          ansible_host: 10.44.0.12
          wg_address: 10.44.0.12/24
        dr-03:
          ansible_host: 10.44.0.13
          wg_address: 10.44.0.13/24
      vars:
        ansible_user: ops
        wg_port: 51820
        fleet_manifest: ../../.tmp/manifest.yaml
@@ -0,0 +1,129 @@
 # Create the local k3d simulation cluster — an executable miniature of the
 # ground segment (k3s) from docs/06. Terraform then owns the workloads.
 ---
 - name: Simulation cluster (k3d)
  hosts: localhost
  connection: local
  gather_facts: false
  vars:
    cluster_name: swarm-sim
    repo_root: "{{ playbook_dir }}/../.."
    data_dir: "{{ lookup('env', 'SWARM_SIM_DATA') | default(repo_root + '/var/t1', true) }}"
    warehouse_dir: "{{ lookup('env', 'SWARM_WAREHOUSE_DATA') | default(repo_root + '/var/t3', true) }}"
    api_port: 6560
    # Host ports for services exposed by the Terraform modules (NodePorts)
    port_explorer: 30088
    port_grafana: 30300
    port_prometheus: 30990
    port_warehouse_explorer: 30089
    port_minio_console: 30901
    simulator_image: swarm-house/simulator:dev
  tasks:
    - name: Check k3d is installed
      ansible.builtin.command: k3d version
      changed_when: false
    - name: Create shared data directory (the pods' /data lake)
      ansible.builtin.file:
        path: "{{ data_dir }}"
        state: directory
        mode: "0777"
        owner: "10001"
        group: "10001"
    - name: Create warehouse directory (T3 host path inside the k3d node)
      ansible.builtin.file:
        path: "{{ warehouse_dir }}"
        state: directory
        mode: "0777"
        owner: "10001"
        group: "10001"
    - name: List existing clusters
      ansible.builtin.command: k3d cluster list -o json
      register: clusters
      changed_when: false
    - name: Create cluster {{ cluster_name }}
      ansible.builtin.command: >-
        k3d cluster create {{ cluster_name }}
        --api-port {{ api_port }}
        --servers 1 --agents 0
        --volume {{ data_dir }}:/data@server:0
        --volume {{ warehouse_dir }}:/warehouse@server:0
        --port {{ port_explorer }}:{{ port_explorer }}@server:0
        --port {{ port_grafana }}:{{ port_grafana }}@server:0
        --port {{ port_prometheus }}:{{ port_prometheus }}@server:0
        --port {{ port_warehouse_explorer }}:{{ port_warehouse_explorer }}@server:0
        --port {{ port_minio_console }}:{{ port_minio_console }}@server:0
        --k3s-arg "--disable=traefik@server:0"
        --wait
      when: clusters.stdout | from_json | selectattr('name', 'equalto', cluster_name) | list | length == 0
    - name: Build simulator image
      ansible.builtin.command:
        cmd: docker build -t {{ simulator_image }} .
        chdir: "{{ playbook_dir }}/../../simulator"
      register: build
      changed_when: "'Using cache' not in build.stderr"
    - name: Import simulator image into the cluster
      ansible.builtin.command: k3d image import {{ simulator_image }} -c {{ cluster_name }}
      changed_when: true
    - name: Merge k3d kubeconfig into the default context
      ansible.builtin.command: k3d kubeconfig merge {{ cluster_name }} --kubeconfig-merge-default
      changed_when: false
    - name: Check whether Flux CLI is available
      ansible.builtin.command: flux version --client
      register: flux_cli
      changed_when: false
      failed_when: false
    - name: Install Flux CLI (local user bin)
      when: flux_cli.rc != 0
      block:
        - name: Download Flux release archive
          ansible.builtin.get_url:
            url: https://github.com/fluxcd/flux2/releases/download/v2.4.0/flux_2.4.0_linux_amd64.tar.gz
            dest: /tmp/flux.tar.gz
            mode: "0644"
        - name: Extract flux binary
          ansible.builtin.unarchive:
            src: /tmp/flux.tar.gz
            dest: "{{ lookup('env', 'HOME') }}/.local/bin"
            remote_src: true
            include:
              - flux
            extra_opts:
              - --no-same-owner
          ignore_errors: true
        - name: Ensure flux is executable
          ansible.builtin.file:
            path: "{{ lookup('env', 'HOME') }}/.local/bin/flux"
            mode: "0755"
            state: file
    - name: Install Flux controllers into the cluster
      ansible.builtin.command: flux install --namespace=flux-system --components=source-controller,kustomize-controller
      environment:
        PATH: "{{ lookup('env', 'HOME') }}/.local/bin:{{ lookup('env', 'PATH') }}"
        KUBECONFIG: "{{ lookup('env', 'HOME') }}/.kube/config"
      register: flux_install
      changed_when: "'installed' in (flux_install.stdout | default(''))"
    - name: Apply Flux GitOps CRs (GitRepository + Kustomization)
      ansible.builtin.command: kubectl apply -k {{ repo_root }}/infra/gitops/flux
      changed_when: true
    - name: Seed ground GitOps resources locally (works before first git push)
      ansible.builtin.command: kubectl apply -k {{ repo_root }}/infra/gitops/ground
      changed_when: true
    - name: Show kubeconfig hint
      ansible.builtin.debug:
        msg: >-
          Cluster ready. Workloads: cd ../terraform/sim-env && terraform init && terraform apply.
          kubeconfig: k3d kubeconfig get {{ cluster_name }}
@@ -0,0 +1,15 @@
 [Unit]
 Description=Swarm data plane (Compose bundle, release {{ manifest.release }})
 After=docker.service wg-quick@wg0.service
 Requires=docker.service
 [Service]
 Type=oneshot
 RemainAfterExit=yes
 WorkingDirectory={{ swarm_home }}/release
 ExecStart=/usr/bin/docker compose up -d
 ExecStop=/usr/bin/docker compose down
 TimeoutStartSec=300
 [Install]
 WantedBy=multi-user.target
@@ -0,0 +1,12 @@
 # WireGuard swarm plane — rendered by drone-provision.yml
 [Interface]
 Address = {{ wg_address }}
 ListenPort = {{ wg_port }}
 PrivateKey = __REPLACED_FROM_HSM_AT_SEALING__
 {% for peer in groups['drones'] | difference([inventory_hostname]) %}
 [Peer]
 # {{ peer }}
 PublicKey = __PEER_{{ peer | upper | replace('-', '_') }}_PUBKEY__
 AllowedIPs = {{ hostvars[peer].wg_address | ansible.utils.ipaddr('address') }}/32
 {% endfor %}
@@ -0,0 +1,52 @@
 # GitOps — ground segment (Flux)
 Flux reconciles **long-lived ground configuration** from this repository. It does
 not run on drones and does not replace the fleet release manifest — those are
 different lifecycles by design ([11 — CI/CD](../../docs/11-cicd-delivery.md)).
 ## Boundary: Terraform vs Flux vs fleet manifest
 | Layer | Tool | Owns |
 | --- | --- | --- |
 | Cluster + first boot | **Terraform** ([`../terraform/`](../terraform/)) | Namespaces, PVCs, Deployments, CronJobs, NodePorts — the shape of the simulation |
 | Ongoing ground config | **Flux** (this directory) | Policy labels, dashboard bundles, offload knobs — things that change without reprovisioning PVCs |
 | Drones | **Fleet release manifest** | Compose bundle digests, model weights, peer registry — atomic, dock-only delivery |
 Drones are **outside GitOps**: there is no reconciler in flight. A dock applies
 the pinned manifest once; mid-mission drift is impossible because the update
 endpoint does not exist in the radio profile.
 ## Layout
 ```
 infra/gitops/
  flux/           Flux GitRepository + Kustomization CRs (install into flux-system)
  ground/         Kustomize overlay reconciled into the ground namespace
 ```
 ## Bootstrap on the k3d simulation cluster
 `infra/ansible/sim-cluster.yml` installs Flux controllers and applies the CRs
 below. After the first `git push`, Flux polls `origin` and reconciles
 `infra/gitops/ground/` into the `ground` namespace.
 ```bash
 # Manual bootstrap (if you skipped Ansible):
 flux install --namespace=flux-system
 kubectl apply -k infra/gitops/flux
 ```
 To reconcile immediately without waiting for the poll interval:
 ```bash
 flux reconcile source git swarm-house -n flux-system
 flux reconcile kustomization ground-segment -n flux-system
 ```
 ## What Flux manages here (example)
 The [`ground/`](ground/) overlay currently carries a **GitOps-managed ConfigMap**
 that tags the warehouse segment with fleet policy metadata. In production this
 pattern extends to Grafana dashboard bundles, Prometheus rule files, and
 offload-schedule ConfigMaps — all versioned in git, all auditable, none of
 them requiring a `terraform apply` to tweak a label.
@@ -0,0 +1,10 @@
 apiVersion: source.toolkit.fluxcd.io/v1
 kind: GitRepository
 metadata:
  name: swarm-house
  namespace: flux-system
 spec:
  interval: 1m
  url: https://git.produktor.io/eSlider/swarm-house.git
  ref:
    branch: main
@@ -0,0 +1,14 @@
 apiVersion: kustomize.toolkit.fluxcd.io/v1
 kind: Kustomization
 metadata:
  name: ground-segment
  namespace: flux-system
 spec:
  interval: 2m
  sourceRef:
    kind: GitRepository
    name: swarm-house
  path: ./infra/gitops/ground
  prune: true
  wait: true
  timeout: 3m
@@ -0,0 +1,5 @@
 apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 resources:
  - gitrepository.yaml
  - kustomization-ground.yaml
@@ -0,0 +1,12 @@
 apiVersion: v1
 kind: ConfigMap
 metadata:
  name: ground-fleet-policy
  namespace: ground
  labels:
    app.kubernetes.io/part-of: swarm-ground
    app.kubernetes.io/managed-by: flux
 data:
  offload_schedule: "*/2 * * * *"
  retention_policy: "keep-all-flights"
  explorer_mode: "read-only-sql"
@@ -0,0 +1,6 @@
 apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 namespace: ground
 resources:
  - namespace.yaml
  - configmap-fleet-policy.yaml
@@ -0,0 +1,6 @@
 apiVersion: v1
 kind: Namespace
 metadata:
  name: ground
  labels:
    app.kubernetes.io/part-of: swarm-ground
@@ -0,0 +1,34 @@
 ## Recovering from partial state
 If resources were created outside Terraform (or state was lost), import them
 before `terraform apply`:
 ```bash
 # sim-env (namespace: swarm)
 terraform import kubernetes_namespace.swarm swarm
 terraform import kubernetes_stateful_set.drone swarm/drone
 terraform import kubernetes_deployment.exporter swarm/exporter
 terraform import kubernetes_deployment.explorer swarm/explorer
 terraform import kubernetes_deployment.prometheus swarm/prometheus
 terraform import kubernetes_deployment.grafana swarm/grafana
 terraform import kubernetes_service.exporter swarm/exporter
 terraform import kubernetes_service.explorer swarm/explorer
 terraform import kubernetes_service.prometheus swarm/prometheus
 terraform import kubernetes_service.grafana swarm/grafana
 terraform import kubernetes_config_map.prometheus swarm/prometheus-config
 terraform import kubernetes_config_map.grafana_provisioning swarm/grafana-provisioning
 terraform import kubernetes_config_map.grafana_dashboard swarm/grafana-dashboard
 # ground (namespace: ground)
 terraform import kubernetes_namespace.ground ground
 terraform import kubernetes_persistent_volume_claim.minio ground/minio-data
 terraform import kubernetes_secret.minio ground/minio-credentials
 terraform import kubernetes_deployment.minio ground/minio
 terraform import kubernetes_deployment.warehouse_explorer ground/warehouse-explorer
 terraform import kubernetes_service.minio ground/minio
 terraform import kubernetes_service.warehouse_explorer ground/warehouse-explorer
 terraform import kubernetes_cron_job_v1.offload ground/offload
 ```
 Then `terraform apply` reconciles drift (for example pinning `EXPLORER_PORT`
 and `EXPORTER_PORT` so Kubernetes service env injection does not break startup).
@@ -0,0 +1,22 @@
 # This file is maintained automatically by "terraform init".
 # Manual edits may be lost in future updates.
 provider "registry.terraform.io/hashicorp/kubernetes" {
  version     = "2.38.0"
  constraints = "~> 2.33"
  hashes = [
    "h1:5CkveFo5ynsLdzKk+Kv+r7+U9rMrNjfZPT3a0N/fhgE=",
    "zh:0af928d776eb269b192dc0ea0f8a3f0f5ec117224cd644bdacdc682300f84ba0",
    "zh:1be998e67206f7cfc4ffe77c01a09ac91ce725de0abaec9030b22c0a832af44f",
    "zh:326803fe5946023687d603f6f1bab24de7af3d426b01d20e51d4e6fbe4e7ec1b",
    "zh:4a99ec8d91193af961de1abb1f824be73df07489301d62e6141a656b3ebfff12",
    "zh:5136e51765d6a0b9e4dbcc3b38821e9736bd2136cf15e9aac11668f22db117d2",
    "zh:63fab47349852d7802fb032e4f2b6a101ee1ce34b62557a9ad0f0f0f5b6ecfdc",
    "zh:924fb0257e2d03e03e2bfe9c7b99aa73c195b1f19412ca09960001bee3c50d15",
    "zh:b63a0be5e233f8f6727c56bed3b61eb9456ca7a8bb29539fba0837f1badf1396",
    "zh:d39861aa21077f1bc899bc53e7233262e530ba8a3a2d737449b100daeb303e4d",
    "zh:de0805e10ebe4c83ce3b728a67f6b0f9d18be32b25146aa89116634df5145ad4",
    "zh:f569b65999264a9416862bca5cd2a6177d94ccb0424f3a4ef424428912b9cb3c",
    "zh:faf23e45f0090eef8ba28a8aac7ec5d4fdf11a36c40a8d286304567d71c1e7db",
  ]
 }
@@ -0,0 +1,38 @@
 # Terraform — ground warehouse (T3)
 The stationary half of the system: the historical warehouse that receives
 sealed partitions after every flight ([03 — Storage tiers](../../../docs/03-storage-design.md)).
 It is the longest-lived, most conventional infrastructure in the design —
 and therefore the most natural Terraform territory.
 Runs in the same k3d cluster as [`../sim-env`](../sim-env/) but in its own
 namespace with its own state: the fleet is disposable, the warehouse is not,
 and the two lifecycles must never share a `terraform destroy`.
 | Resource | Purpose |
 | --- | --- |
 | Deployment `minio` + PVC + NodePort 30901 | Object-store facade of the warehouse for downstream consumers (training pipelines, replay) |
 | CronJob `offload` | Post-flight offload: copies Hive partitions verbatim from the fleet lake (T1) into the warehouse (T3), then mirrors into MinIO |
 | Deployment `warehouse-explorer` + NodePort 30089 | Second explorer instance over the warehouse path — historical read-only SQL across **all** flights |
 ## Usage
 ```bash
 terraform init
 terraform apply
 terraform output
 ```
 After a couple of offload runs, the warehouse explorer shows the same
 partition tree as the live lake but accumulated across flights — the T1
 layout **is** the T3 layout, which is the whole point: no transform step,
 no schema drift, DuckDB queries work identically on both ends.
 ## Simulation vs production
 - The CronJob compresses "drone lands, docks, offloads, prunes" into a
  periodic rsync-style copy. Production offload is event-driven per docking
  and verifies checksums from the flight's partition manifest before pruning
  the on-board lake.
 - MinIO credentials here are throwaway defaults; production credentials are
  sealed per site and never in state or VCS.
@@ -0,0 +1,273 @@
 # Ground warehouse segment (T3, docs/03 + docs/06) — the stationary,
 # long-lifecycle half of the system and therefore the most natural Terraform
 # territory. Runs in the same k3d cluster as the fleet, in its own namespace
 # and with its own state.
 locals {
  labels = { "app.kubernetes.io/part-of" = "swarm-ground" }
 }
 resource "kubernetes_namespace" "ground" {
  metadata {
    name   = var.namespace
    labels = local.labels
  }
 }
 # --- MinIO object store -------------------------------------------------------
 resource "kubernetes_secret" "minio" {
  metadata {
    name      = "minio-credentials"
    namespace = kubernetes_namespace.ground.metadata[0].name
  }
  data = {
    MINIO_ROOT_USER     = var.minio_root_user
    MINIO_ROOT_PASSWORD = var.minio_root_password
  }
 }
 resource "kubernetes_persistent_volume_claim" "minio" {
  metadata {
    name      = "minio-data"
    namespace = kubernetes_namespace.ground.metadata[0].name
  }
  spec {
    access_modes = ["ReadWriteOnce"]
    resources {
      requests = { storage = var.minio_storage }
    }
  }
 }
 resource "kubernetes_deployment" "minio" {
  metadata {
    name      = "minio"
    namespace = kubernetes_namespace.ground.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    strategy {
      type = "Recreate" # RWO volume
    }
    selector {
      match_labels = { app = "minio" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "minio" })
      }
      spec {
        container {
          name  = "minio"
          image = "minio/minio:RELEASE.2024-06-13T22-53-53Z"
          args  = ["server", "/data", "--console-address", ":9001"]
          env_from {
            secret_ref {
              name = kubernetes_secret.minio.metadata[0].name
            }
          }
          # Air-gap hygiene: no update checks
          env {
            name  = "MINIO_UPDATE"
            value = "off"
          }
          port {
            container_port = 9000
          }
          port {
            container_port = 9001
          }
          volume_mount {
            name       = "store"
            mount_path = "/data"
          }
        }
        volume {
          name = "store"
          persistent_volume_claim {
            claim_name = kubernetes_persistent_volume_claim.minio.metadata[0].name
          }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "minio" {
  metadata {
    name      = "minio"
    namespace = kubernetes_namespace.ground.metadata[0].name
  }
  spec {
    type     = "NodePort"
    selector = { app = "minio" }
    port {
      name        = "api"
      port        = 9000
      target_port = 9000
    }
    port {
      name        = "console"
      port        = 9001
      target_port = 9001
      node_port   = var.node_ports.minio_console
    }
  }
 }
 # --- Post-flight offload: T1 lake -> T3 warehouse ------------------------------
 # In production this runs on the docking station after a drone lands: only
 # sealed partitions move, checksums are verified, then the on-board lake is
 # pruned. The simulation compresses that into a periodic rsync-style copy
 # plus an object-store mirror.
 resource "kubernetes_cron_job_v1" "offload" {
  metadata {
    name      = "offload"
    namespace = kubernetes_namespace.ground.metadata[0].name
    labels    = local.labels
  }
  spec {
    schedule                      = var.offload_schedule
    concurrency_policy            = "Forbid"
    successful_jobs_history_limit = 3
    failed_jobs_history_limit     = 3
    job_template {
      metadata {
        labels = local.labels
      }
      spec {
        backoff_limit = 1
        template {
          metadata {
            labels = local.labels
          }
          spec {
            restart_policy = "Never"
            # Step 1: copy Hive partitions verbatim (T1 layout == T3 layout)
            init_container {
              name    = "copy"
              image   = "busybox:1.36"
              command = ["sh", "-c", "cp -ru /lake/flight_id=* /warehouse/ 2>/dev/null; ls /warehouse | wc -l"]
              volume_mount {
                name       = "lake"
                mount_path = "/lake"
                read_only  = true
              }
              volume_mount {
                name       = "warehouse"
                mount_path = "/warehouse"
              }
            }
            # Step 2: mirror the warehouse into the object store for
            # downstream consumers (training pipelines, replay tooling)
            container {
              name    = "mirror"
              image   = "minio/mc:RELEASE.2024-06-12T14-34-03Z"
              command = ["sh", "-c", "mc alias set store http://minio:9000 \"$MINIO_ROOT_USER\" \"$MINIO_ROOT_PASSWORD\" && mc mb -p store/warehouse && mc mirror --overwrite /warehouse store/warehouse"]
              env_from {
                secret_ref {
                  name = kubernetes_secret.minio.metadata[0].name
                }
              }
              volume_mount {
                name       = "warehouse"
                mount_path = "/warehouse"
                read_only  = true
              }
            }
            volume {
              name = "lake"
              host_path {
                path = var.lake_host_path
              }
            }
            volume {
              name = "warehouse"
              host_path {
                path = var.warehouse_host_path
                type = "DirectoryOrCreate"
              }
            }
          }
        }
      }
    }
  }
 }
 # --- Warehouse explorer: historical read-only SQL over all flights -------------
 resource "kubernetes_deployment" "warehouse_explorer" {
  metadata {
    name      = "warehouse-explorer"
    namespace = kubernetes_namespace.ground.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    selector {
      match_labels = { app = "warehouse-explorer" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "warehouse-explorer" })
      }
      spec {
        container {
          name              = "explorer"
          image             = var.simulator_image
          image_pull_policy = "Never"
          command           = ["python", "explorer/server.py"]
          env {
            name  = "DATA_DIR"
            value = "/warehouse"
          }
          port {
            container_port = 8088
          }
          volume_mount {
            name       = "warehouse"
            mount_path = "/warehouse"
            read_only  = true
          }
        }
        volume {
          name = "warehouse"
          host_path {
            path = var.warehouse_host_path
            type = "DirectoryOrCreate"
          }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "warehouse_explorer" {
  metadata {
    name      = "warehouse-explorer"
    namespace = kubernetes_namespace.ground.metadata[0].name
  }
  spec {
    type     = "NodePort"
    selector = { app = "warehouse-explorer" }
    port {
      port        = 8088
      target_port = 8088
      node_port   = var.node_ports.warehouse_explorer
    }
  }
 }
@@ -0,0 +1,14 @@
 output "warehouse_explorer_url" {
  description = "Historical read-only SQL over all offloaded flights"
  value       = "http://localhost:${var.node_ports.warehouse_explorer}"
 }
 output "minio_console_url" {
  description = "MinIO console (object-store view of the warehouse)"
  value       = "http://localhost:${var.node_ports.minio_console}"
 }
 output "offload_schedule" {
  description = "Cron schedule of the T1 -> T3 offload job"
  value       = var.offload_schedule
 }
@@ -0,0 +1,72 @@
 variable "kubeconfig" {
  description = "Path to the kubeconfig for the simulation cluster"
  type        = string
  default     = "~/.kube/config"
 }
 variable "kube_context" {
  description = "Kubeconfig context of the k3d simulation cluster"
  type        = string
  default     = "k3d-swarm-sim"
 }
 variable "namespace" {
  description = "Namespace for the ground warehouse segment"
  type        = string
  default     = "ground"
 }
 variable "simulator_image" {
  description = "Simulator image (provides the explorer; imported by Ansible)"
  type        = string
  default     = "swarm-house/simulator:dev"
 }
 variable "lake_host_path" {
  description = "Host path of the fleet's live lake (T1) inside the k3d node"
  type        = string
  default     = "/data"
 }
 variable "warehouse_host_path" {
  description = "Host path of the historical warehouse (T3) inside the k3d node; dot-prefixed so lake globs never match it"
  type        = string
  default     = "/data/.warehouse"
 }
 variable "offload_schedule" {
  description = "Cron schedule of the post-flight offload job (T1 -> T3)"
  type        = string
  default     = "*/2 * * * *"
 }
 variable "minio_root_user" {
  description = "MinIO root user (simulation only; production uses sealed credentials)"
  type        = string
  default     = "warehouse"
 }
 variable "minio_root_password" {
  description = "MinIO root password (simulation only)"
  type        = string
  default     = "warehouse-sim-only"
  sensitive   = true
 }
 variable "minio_storage" {
  description = "PVC size for the MinIO object store"
  type        = string
  default     = "2Gi"
 }
 variable "node_ports" {
  description = "Host-reachable NodePorts (must match the ports opened by sim-cluster.yml)"
  type = object({
    warehouse_explorer = number
    minio_console      = number
  })
  default = {
    warehouse_explorer = 30089
    minio_console      = 30901
  }
 }
@@ -0,0 +1,15 @@
 terraform {
  required_version = ">= 1.5"
  required_providers {
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = "~> 2.33"
    }
  }
 }
 provider "kubernetes" {
  config_path    = var.kubeconfig
  config_context = var.kube_context
 }
@@ -0,0 +1,22 @@
 # This file is maintained automatically by "terraform init".
 # Manual edits may be lost in future updates.
 provider "registry.terraform.io/hashicorp/kubernetes" {
  version     = "2.38.0"
  constraints = "~> 2.33"
  hashes = [
    "h1:5CkveFo5ynsLdzKk+Kv+r7+U9rMrNjfZPT3a0N/fhgE=",
    "zh:0af928d776eb269b192dc0ea0f8a3f0f5ec117224cd644bdacdc682300f84ba0",
    "zh:1be998e67206f7cfc4ffe77c01a09ac91ce725de0abaec9030b22c0a832af44f",
    "zh:326803fe5946023687d603f6f1bab24de7af3d426b01d20e51d4e6fbe4e7ec1b",
    "zh:4a99ec8d91193af961de1abb1f824be73df07489301d62e6141a656b3ebfff12",
    "zh:5136e51765d6a0b9e4dbcc3b38821e9736bd2136cf15e9aac11668f22db117d2",
    "zh:63fab47349852d7802fb032e4f2b6a101ee1ce34b62557a9ad0f0f0f5b6ecfdc",
    "zh:924fb0257e2d03e03e2bfe9c7b99aa73c195b1f19412ca09960001bee3c50d15",
    "zh:b63a0be5e233f8f6727c56bed3b61eb9456ca7a8bb29539fba0837f1badf1396",
    "zh:d39861aa21077f1bc899bc53e7233262e530ba8a3a2d737449b100daeb303e4d",
    "zh:de0805e10ebe4c83ce3b728a67f6b0f9d18be32b25146aa89116634df5145ad4",
    "zh:f569b65999264a9416862bca5cd2a6177d94ccb0424f3a4ef424428912b9cb3c",
    "zh:faf23e45f0090eef8ba28a8aac7ec5d4fdf11a36c40a8d286304567d71c1e7db",
  ]
 }
@@ -0,0 +1,40 @@
 # Terraform — simulation environment (T4)
 Declares the simulated fleet and its observability stack on the k3d cluster
 created by [`../../ansible/sim-cluster.yml`](../../ansible/sim-cluster.yml).
 This module is the executable miniature of the dev/sim environment from
 [06 — Environments](../../../docs/06-environments.md).
 | Resource | Purpose |
 | --- | --- |
 | StatefulSet `drone` × `drone_count` | Virtual drones; stable pod names become `DRONE_ID`s; all write Hive-partitioned Parquet into the shared `/data` lake |
 | Deployment `exporter` | Prometheus exporter reading the lake with DuckDB |
 | Deployment `explorer` + NodePort 30088 | Partition tree + read-only SQL console (also feeds the prototype's live mode) |
 | Deployment `prometheus` + NodePort 30990 | Scrapes the exporter |
 | Deployment `grafana` + NodePort 30300 | Same dashboard JSON as the Compose profile, provisioned from a ConfigMap |
 ## Usage
 ```bash
 ansible-playbook ../../ansible/sim-cluster.yml   # once: cluster + image
 terraform init
 terraform apply
 terraform output                                  # URLs
 ```
 Scale the fleet without touching YAML:
 ```bash
 terraform apply -var drone_count=9 -var speedup=4
 ```
 ## Notes
 - `image_pull_policy = "Never"` — the simulator image is imported by the
  Ansible playbook (`k3d image import`); the cluster never pulls from a
  registry, mirroring the air-gap doctrine.
 - Pods discover their identity from the pod name (StatefulSet ordinal), so
  a flight survives `kubectl delete pod` the same way a drone survives a
  power cycle: new process, same identity, new flight ID.
 - State is local (`terraform.tfstate` in this directory) — the simulation
  cluster is disposable; nothing here is shared infrastructure.
@@ -0,0 +1,569 @@
 # Simulated fleet (T4 dev environment, docs/06) on the k3d cluster created
 # by infra/ansible/sim-cluster.yml. Drones are StatefulSet replicas writing
 # to a shared hostPath lake — the same /data contract as the on-board NVMe.
 locals {
  labels = { "app.kubernetes.io/part-of" = "swarm-sim" }
 }
 resource "kubernetes_namespace" "swarm" {
  metadata {
    name   = var.namespace
    labels = local.labels
  }
 }
 # --- Fleet ------------------------------------------------------------------
 resource "kubernetes_stateful_set" "drone" {
  metadata {
    name      = "drone"
    namespace = kubernetes_namespace.swarm.metadata[0].name
    labels    = local.labels
  }
  spec {
    service_name = "drone"
    replicas     = var.drone_count
    # A landed drone powers off; the next start is the next flight
    pod_management_policy = "Parallel"
    selector {
      match_labels = { app = "drone" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "drone" })
      }
      spec {
        container {
          name              = "drone"
          image             = var.simulator_image
          image_pull_policy = "Never" # imported via k3d image import
          env {
            name = "DRONE_ID"
            value_from {
              field_ref {
                field_path = "metadata.name"
              }
            }
          }
          env {
            name  = "DURATION_S"
            value = tostring(var.flight_duration_s)
          }
          env {
            name  = "SPEEDUP"
            value = tostring(var.speedup)
          }
          env {
            name  = "DATA_DIR"
            value = "/data"
          }
          env {
            name  = "KEEP_ALIVE"
            value = "1"
          }
          env {
            name  = "IMU_HZ"
            value = "20"
          }
          volume_mount {
            name       = "lake"
            mount_path = "/data"
          }
          resources {
            requests = { cpu = "50m", memory = "128Mi" }
            limits   = { cpu = "500m", memory = "512Mi" }
          }
        }
        volume {
          name = "lake"
          host_path {
            path = var.data_host_path
          }
        }
      }
    }
  }
 }
 # --- Exporter + Prometheus + Grafana (observability, docs/07) ----------------
 resource "kubernetes_deployment" "exporter" {
  metadata {
    name      = "exporter"
    namespace = kubernetes_namespace.swarm.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    selector {
      match_labels = { app = "exporter" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "exporter" })
      }
      spec {
        container {
          name              = "exporter"
          image             = var.simulator_image
          image_pull_policy = "Never"
          command           = ["python", "monitoring/exporter.py"]
          env {
            name  = "DATA_DIR"
            value = "/data"
          }
          env {
            name  = "EXPORTER_PORT"
            value = "9105"
          }
          env {
            name  = "SCAN_INTERVAL_S"
            value = "30"
          }
          env {
            name  = "METRIC_FLIGHT_WINDOW"
            value = "5"
          }
          port {
            container_port = 9105
          }
          resources {
            requests = { cpu = "50m", memory = "64Mi" }
            limits   = { cpu = "500m", memory = "256Mi" }
          }
          volume_mount {
            name       = "lake"
            mount_path = "/data"
            read_only  = true
          }
        }
        volume {
          name = "lake"
          host_path {
            path = var.data_host_path
          }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "exporter" {
  metadata {
    name      = "exporter"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  spec {
    selector = { app = "exporter" }
    port {
      port        = 9105
      target_port = 9105
    }
  }
 }
 resource "kubernetes_deployment" "explorer" {
  metadata {
    name      = "explorer"
    namespace = kubernetes_namespace.swarm.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    selector {
      match_labels = { app = "explorer" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "explorer" })
      }
      spec {
        container {
          name              = "explorer"
          image             = var.simulator_image
          image_pull_policy = "Never"
          command           = ["python", "explorer/server.py"]
          env {
            name  = "DATA_DIR"
            value = "/data"
          }
          env {
            name  = "EXPLORER_PORT"
            value = "8088"
          }
          env {
            name  = "VIEW_REFRESH_S"
            value = "30"
          }
          env {
            name  = "TREE_CACHE_S"
            value = "15"
          }
          env {
            name  = "METRIC_FLIGHT_WINDOW"
            value = "5"
          }
          port {
            container_port = 8088
          }
          resources {
            requests = { cpu = "50m", memory = "64Mi" }
            limits   = { cpu = "750m", memory = "256Mi" }
          }
          volume_mount {
            name       = "lake"
            mount_path = "/data"
            read_only  = true
          }
        }
        volume {
          name = "lake"
          host_path {
            path = var.data_host_path
          }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "explorer" {
  metadata {
    name      = "explorer"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  spec {
    type     = "NodePort"
    selector = { app = "explorer" }
    port {
      port        = 8088
      target_port = 8088
      node_port   = var.node_ports.explorer
    }
  }
 }
 resource "kubernetes_config_map" "prometheus" {
  metadata {
    name      = "prometheus-config"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  data = {
    "prometheus.yml" = <<-EOT
      global:
        scrape_interval: 15s
      scrape_configs:
        - job_name: swarm
          static_configs:
            - targets: ["exporter:9105"]
        - job_name: explorer
          static_configs:
            - targets: ["explorer:8088"]
        - job_name: node
          static_configs:
            - targets: ["node-exporter:9100"]
    EOT
  }
 }
 resource "kubernetes_deployment" "prometheus" {
  metadata {
    name      = "prometheus"
    namespace = kubernetes_namespace.swarm.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    selector {
      match_labels = { app = "prometheus" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "prometheus" })
      }
      spec {
        container {
          name  = "prometheus"
          image = "prom/prometheus:v2.53.0"
          port {
            container_port = 9090
          }
          volume_mount {
            name       = "config"
            mount_path = "/etc/prometheus"
          }
        }
        volume {
          name = "config"
          config_map {
            name = kubernetes_config_map.prometheus.metadata[0].name
          }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "prometheus" {
  metadata {
    name      = "prometheus"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  spec {
    type     = "NodePort"
    selector = { app = "prometheus" }
    port {
      port        = 9090
      target_port = 9090
      node_port   = var.node_ports.prometheus
    }
  }
 }
 resource "kubernetes_config_map" "grafana_provisioning" {
  metadata {
    name      = "grafana-provisioning"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  data = {
    "datasource.yml" = <<-EOT
      apiVersion: 1
      deleteDatasources:
        - name: Prometheus
          orgId: 1
      datasources:
        - name: Prometheus
          uid: swarm-prom
          type: prometheus
          access: proxy
          url: http://prometheus:9090
          isDefault: true
    EOT
    "dashboards.yml" = <<-EOT
      apiVersion: 1
      providers:
        - name: swarm
          folder: ""
          type: file
          options:
            path: /var/lib/grafana/dashboards
    EOT
  }
 }
 resource "kubernetes_config_map" "grafana_dashboard" {
  metadata {
    name      = "grafana-dashboard"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  data = {
    "swarm.json"          = file("${path.module}/../../../simulator/monitoring/grafana/dashboards/swarm.json")
    "swarm-platform.json" = file("${path.module}/../../../simulator/monitoring/grafana/dashboards/swarm-platform.json")
  }
 }
 # Host metrics for the k3d node (CPU / memory on the platform dashboard)
 resource "kubernetes_deployment" "node_exporter" {
  metadata {
    name      = "node-exporter"
    namespace = kubernetes_namespace.swarm.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    selector {
      match_labels = { app = "node-exporter" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "node-exporter" })
      }
      spec {
        host_network = true
        host_pid     = true
        container {
          name  = "node-exporter"
          image = "prom/node-exporter:v1.8.2"
          args = [
            "--path.procfs=/host/proc",
            "--path.sysfs=/host/sys",
            "--path.rootfs=/host/root",
            "--collector.filesystem.mount-points-exclude=^/(sys|proc|dev|host|etc)($$|/)",
          ]
          port {
            container_port = 9100
          }
          volume_mount {
            name       = "proc"
            mount_path = "/host/proc"
            read_only  = true
          }
          volume_mount {
            name       = "sys"
            mount_path = "/host/sys"
            read_only  = true
          }
          volume_mount {
            name       = "root"
            mount_path = "/host/root"
            read_only  = true
          }
          resources {
            requests = { cpu = "20m", memory = "32Mi" }
            limits   = { cpu = "200m", memory = "64Mi" }
          }
        }
        volume {
          name = "proc"
          host_path { path = "/proc" }
        }
        volume {
          name = "sys"
          host_path { path = "/sys" }
        }
        volume {
          name = "root"
          host_path { path = "/" }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "node_exporter" {
  metadata {
    name      = "node-exporter"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  spec {
    selector = { app = "node-exporter" }
    port {
      port        = 9100
      target_port = 9100
    }
  }
 }
 resource "kubernetes_deployment" "grafana" {
  metadata {
    name      = "grafana"
    namespace = kubernetes_namespace.swarm.metadata[0].name
    labels    = local.labels
  }
  spec {
    replicas = 1
    selector {
      match_labels = { app = "grafana" }
    }
    template {
      metadata {
        labels = merge(local.labels, { app = "grafana" })
      }
      spec {
        container {
          name  = "grafana"
          image = "grafana/grafana:11.1.0"
          port {
            container_port = 3000
          }
          env {
            name  = "GF_AUTH_ANONYMOUS_ENABLED"
            value = "true"
          }
          env {
            name  = "GF_AUTH_ANONYMOUS_ORG_ROLE"
            value = "Admin"
          }
          env {
            name  = "GF_AUTH_DISABLE_LOGIN_FORM"
            value = "true"
          }
          # Air-gap hygiene — same flags as the Compose profile
          env {
            name  = "GF_ANALYTICS_REPORTING_ENABLED"
            value = "false"
          }
          env {
            name  = "GF_ANALYTICS_CHECK_FOR_UPDATES"
            value = "false"
          }
          env {
            name  = "GF_ANALYTICS_CHECK_FOR_PLUGIN_UPDATES"
            value = "false"
          }
          volume_mount {
            name       = "provisioning-datasources"
            mount_path = "/etc/grafana/provisioning/datasources"
          }
          volume_mount {
            name       = "provisioning-dashboards"
            mount_path = "/etc/grafana/provisioning/dashboards"
          }
          volume_mount {
            name       = "dashboards"
            mount_path = "/var/lib/grafana/dashboards"
          }
        }
        volume {
          name = "provisioning-datasources"
          config_map {
            name = kubernetes_config_map.grafana_provisioning.metadata[0].name
            items {
              key  = "datasource.yml"
              path = "datasource.yml"
            }
          }
        }
        volume {
          name = "provisioning-dashboards"
          config_map {
            name = kubernetes_config_map.grafana_provisioning.metadata[0].name
            items {
              key  = "dashboards.yml"
              path = "dashboards.yml"
            }
          }
        }
        volume {
          name = "dashboards"
          config_map {
            name = kubernetes_config_map.grafana_dashboard.metadata[0].name
          }
        }
      }
    }
  }
 }
 resource "kubernetes_service" "grafana" {
  metadata {
    name      = "grafana"
    namespace = kubernetes_namespace.swarm.metadata[0].name
  }
  spec {
    type     = "NodePort"
    selector = { app = "grafana" }
    port {
      port        = 3000
      target_port = 3000
      node_port   = var.node_ports.grafana
    }
  }
 }
@@ -0,0 +1,23 @@
 output "explorer_url" {
  description = "Data-plane explorer (partition tree + read-only SQL)"
  value       = "http://localhost:${var.node_ports.explorer}"
 }
 output "grafana_url" {
  description = "Grafana dashboards"
  value       = "http://localhost:${var.node_ports.grafana}"
 }
 output "prometheus_url" {
  description = "Prometheus"
  value       = "http://localhost:${var.node_ports.prometheus}"
 }
 output "fleet" {
  description = "Deployed fleet shape"
  value = {
    drones            = var.drone_count
    flight_duration_s = var.flight_duration_s
    speedup           = var.speedup
  }
 }
@@ -0,0 +1,66 @@
 variable "kubeconfig" {
  description = "Path to the kubeconfig for the simulation cluster"
  type        = string
  default     = "~/.kube/config"
 }
 variable "kube_context" {
  description = "Kubeconfig context of the k3d simulation cluster"
  type        = string
  default     = "k3d-swarm-sim"
 }
 variable "namespace" {
  description = "Namespace for the simulated fleet"
  type        = string
  default     = "swarm"
 }
 variable "simulator_image" {
  description = "Simulator image (imported into k3d by infra/ansible/sim-cluster.yml)"
  type        = string
  default     = "swarm-house/simulator:dev"
 }
 variable "drone_count" {
  description = "Number of virtual drones in the fleet"
  type        = number
  default     = 5
  validation {
    condition     = var.drone_count >= 1 && var.drone_count <= 32
    error_message = "drone_count must be between 1 and 32."
  }
 }
 variable "flight_duration_s" {
  description = "Simulated flight length in seconds; each pod restart begins a new flight"
  type        = number
  default     = 300
 }
 variable "speedup" {
  description = "Wall-clock acceleration of the simulation"
  type        = number
  default     = 2
 }
 variable "data_host_path" {
  description = "Host path inside the k3d node mounted as the shared Parquet lake"
  type        = string
  default     = "/data"
 }
 variable "node_ports" {
  description = "Host-reachable NodePorts (must match the ports opened by sim-cluster.yml)"
  type = object({
    explorer   = number
    grafana    = number
    prometheus = number
  })
  default = {
    explorer   = 30088
    grafana    = 30300
    prometheus = 30990
  }
 }
@@ -0,0 +1,15 @@
 terraform {
  required_version = ">= 1.5"
  required_providers {
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = "~> 2.33"
    }
  }
 }
 provider "kubernetes" {
  config_path    = var.kubeconfig
  config_context = var.kube_context
 }
@@ -1,6 +1,9 @@
 import { useEffect, useMemo, useRef, useState } from "react";
 import { DEFAULT_EXPLORER_URL, fetchLivePoses, toLiveWorld } from "./live";
 import {
  AREA,
  MAX_ASPECT,
  areaWidth,
  fmtBytes,
  makeWorld,
  seed,
@@ -9,24 +12,56 @@ import {
 } from "./sim";
 const VIEW = 1000;
 const PANEL_W = 320;
 const HEADER_H = 70;
 // Patrol field aspect ratio for the current window: wide monitors get a
 // wider field instead of empty side margins. Quantized to 0.25 steps so
 // minor resizes don't rebuild the world.
 function displayAspect(): number {
  const w = Math.max(320, window.innerWidth - PANEL_W);
  const h = Math.max(320, window.innerHeight - HEADER_H);
  const q = Math.round((w / h) * 4) / 4;
  return Math.max(1, Math.min(MAX_ASPECT, q));
 }
 const LIVE_POLL_MS = 2000;
 export default function App(): JSX.Element {
  const [droneCount, setDroneCount] = useState(8);
  const [speedup, setSpeedup] = useState(1);
  const [paused, setPaused] = useState(false);
  const [aspect, setAspect] = useState(displayAspect);
  const [mode, setMode] = useState<"sim" | "live">("sim");
  const [liveError, setLiveError] = useState<string | null>(null);
  const [world, setWorld] = useState<World>(() => {
    seed(42);
-    return makeWorld(8);
+    return makeWorld(8, displayAspect());
  });
  const raf = useRef(0);
  const last = useRef(performance.now());
  const liveWorld = useRef<World | null>(null);
  useEffect(() => {
    seed(42);
-    setWorld(makeWorld(droneCount));
+    setWorld(makeWorld(droneCount, aspect));
-  }, [droneCount]);
+  }, [droneCount, aspect]);
  useEffect(() => {
    let timer = 0;
    const onResize = (): void => {
      window.clearTimeout(timer);
      timer = window.setTimeout(() => setAspect(displayAspect()), 250);
    };
    window.addEventListener("resize", onResize);
    return () => {
      window.clearTimeout(timer);
      window.removeEventListener("resize", onResize);
    };
  }, []);
  useEffect(() => {
    if (mode !== "sim") return;
    const loop = (now: number): void => {
      const dt = Math.min(0.05, (now - last.current) / 1000);
      last.current = now;
@@ -37,7 +72,40 @@ export default function App(): JSX.Element {
    };
    raf.current = requestAnimationFrame(loop);
    return () => cancelAnimationFrame(raf.current);
-  }, [paused, speedup]);
+  }, [paused, speedup, mode]);
  // Live mode: poll the explorer's read-only SQL API for real poses
  useEffect(() => {
    if (mode !== "live") return;
    let cancelled = false;
    const poll = async (): Promise<void> => {
      try {
        const poses = await fetchLivePoses(DEFAULT_EXPLORER_URL);
        if (cancelled) return;
        setLiveError(poses.length === 0 ? "no drone data in the lake yet" : null);
        liveWorld.current = toLiveWorld(poses, liveWorld.current, LIVE_POLL_MS / 1000);
        setWorld(liveWorld.current);
      } catch (err) {
        if (!cancelled) setLiveError(err instanceof Error ? err.message : String(err));
      }
    };
    void poll();
    const timer = window.setInterval(() => void poll(), LIVE_POLL_MS);
    return () => {
      cancelled = true;
      window.clearInterval(timer);
    };
  }, [mode]);
  useEffect(() => {
    if (mode === "sim") {
      liveWorld.current = null;
      setLiveError(null);
      seed(42);
      setWorld(makeWorld(droneCount, aspect));
    }
    // eslint-disable-next-line react-hooks/exhaustive-deps -- reset only on mode change
  }, [mode]);
  const activeLinks = useMemo(
    () => [...world.links.values()].sort((a, b) => b.rate - a.rate),
@@ -55,6 +123,20 @@ export default function App(): JSX.Element {
          </div>
        </div>
        <div style={styles.controls}>
          <button
            style={{ ...styles.button, ...(mode === "live" ? styles.buttonActive : {}) }}
            onClick={() => setMode((m) => (m === "sim" ? "live" : "sim"))}
            title={`live source: ${DEFAULT_EXPLORER_URL}`}
          >
            {mode === "sim" ? "go live" : "back to sim"}
          </button>
          {mode === "live" && (
            <span style={liveError ? styles.liveError : styles.liveOk}>
              {liveError ?? `live · ${DEFAULT_EXPLORER_URL}`}
            </span>
          )}
          {mode === "sim" && (
            <>
              <label style={styles.label}>
                drones
                <input
@@ -81,12 +163,14 @@ export default function App(): JSX.Element {
              <button style={styles.button} onClick={() => setPaused((p) => !p)}>
                {paused ? "resume" : "pause"}
              </button>
            </>
          )}
        </div>
      </header>
      <div style={styles.main}>
        <svg
-          viewBox={`0 0 ${VIEW} ${VIEW}`}
+          viewBox={`0 0 ${sx(areaWidth())} ${VIEW}`}
          style={styles.canvas}
          preserveAspectRatio="xMidYMid meet"
        >
@@ -96,7 +180,7 @@ export default function App(): JSX.Element {
              <stop offset="100%" stopColor="#0a0f18" />
            </radialGradient>
          </defs>
-          <rect width={VIEW} height={VIEW} fill="url(#ground)" />
+          <rect width={sx(areaWidth())} height={VIEW} fill="url(#ground)" />
          {gridLines()}
          {/* Mesh links */}
@@ -141,7 +225,7 @@ export default function App(): JSX.Element {
              />
              {ob.moving && (
                <text x={sx(ob.pos.x)} y={sy(ob.pos.y) + 4} style={styles.obLabel as never}>
-                  mobile
+                  transit
                </text>
              )}
            </g>
@@ -155,7 +239,7 @@ export default function App(): JSX.Element {
            >
              <polygon points="0,-11 7,9 0,5 -7,9" fill="#e8ecf4" stroke="#7ea0d0" strokeWidth={1} />
              <text y={24} style={styles.droneLabel as never} transform={`rotate(${-((d.heading * 180) / Math.PI + 90)})`}>
-                dr-{String(d.id + 1).padStart(2, "0")} · ch{d.channel} · {d.battery.toFixed(0)}%
+                {d.name ?? `dr-${String(d.id + 1).padStart(2, "0")} · ch${d.channel}`} · {d.battery.toFixed(0)}%
              </text>
            </g>
          ))}
@@ -173,7 +257,8 @@ export default function App(): JSX.Element {
            {busiest.map((l) => (
              <div key={l.key} style={styles.linkRow}>
                <span style={styles.linkName}>
-                  dr-{String(l.a + 1).padStart(2, "0")} ↔ dr-{String(l.b + 1).padStart(2, "0")}
+                  {world.drones[l.a]?.name ?? `dr-${String(l.a + 1).padStart(2, "0")}`} ↔{" "}
                  {world.drones[l.b]?.name ?? `dr-${String(l.b + 1).padStart(2, "0")}`}
                </span>
                <span style={styles.linkStat}>
                  {fmtBytes(l.rate)}/s · ↑{fmtBytes(l.totalUp)} ↓{fmtBytes(l.totalDown)}
@@ -186,7 +271,7 @@ export default function App(): JSX.Element {
            <div style={styles.dim}>blue line — pose broadcast (5 Hz, ~46 B)</div>
            <div style={styles.dim}>green line — bulk sync (sealed partitions)</div>
            <div style={styles.dim}>flash — broadcast event delivered</div>
-            <div style={styles.dim}>red circle — mobile obstacle</div>
+            <div style={styles.dim}>red circle — transit object crossing the area</div>
          </Section>
        </aside>
      </div>
@@ -202,12 +287,17 @@ function sy(y: number): number {
 }
 function gridLines(): JSX.Element {
  const width = sx(areaWidth());
  const step = VIEW / 10; // one cell per 100 m in both directions
  const lines = [];
-  for (let i = 1; i < 10; i++) {
+  for (let p = step; p < width; p += step) {
    const p = (i / 10) * VIEW;
    lines.push(
-      <line key={`v${i}`} x1={p} y1={0} x2={p} y2={VIEW} stroke="#1a2436" strokeWidth={1} />,
+      <line key={`v${p}`} x1={p} y1={0} x2={p} y2={VIEW} stroke="#1a2436" strokeWidth={1} />,
-      <line key={`h${i}`} x1={0} y1={p} x2={VIEW} y2={p} stroke="#1a2436" strokeWidth={1} />,
+    );
  }
  for (let p = step; p < VIEW; p += step) {
    lines.push(
      <line key={`h${p}`} x1={0} y1={p} x2={width} y2={p} stroke="#1a2436" strokeWidth={1} />,
    );
  }
  return <g>{lines}</g>;
@@ -264,6 +354,9 @@ const styles: Record<string, React.CSSProperties> = {
    fontSize: 12,
    cursor: "pointer",
  },
  buttonActive: { background: "#1a3a2a", color: "#39d98a", borderColor: "#2a5c3c" },
  liveOk: { fontSize: 11, color: "#39d98a" },
  liveError: { fontSize: 11, color: "#c05a6a" },
  main: { display: "flex", flex: 1, gap: 0, minHeight: 0 },
  canvas: { flex: 1, minWidth: 0, display: "block" },
  panel: {
@@ -0,0 +1,129 @@
 // Live mode: real drone poses from the explorer's read-only SQL API.
 // The same gated /api/query endpoint a peer drone (or an engineer on the
 // bench) would use — the prototype is just another read-only consumer.
 import { AREA, LINK_RANGE, POSE_RATE_BYTES, type LinkStats, type World } from "./sim";
 export const DEFAULT_EXPLORER_URL =
  (import.meta.env.VITE_EXPLORER_URL as string | undefined) ?? "http://localhost:30088";
 export interface LivePose {
  id: string;
  x: number; // meters, mission frame
  y: number;
  yawDeg: number;
  battery: number | null;
  tsNs: number;
 }
 // Latest pose per drone from its own 'sent' state rows, joined with the
 // latest battery reading from telemetry. arg_max keeps it a single scan.
 const LIVE_SQL = `
 WITH pose AS (
  SELECT drone,
         arg_max(pos_x, ts_ns) AS x,
         arg_max(pos_y, ts_ns) AS y,
         arg_max(yaw, ts_ns)   AS yaw,
         max(ts_ns)            AS ts_ns
  FROM state
  WHERE direction = 'sent'
  GROUP BY drone
 ),
 batt AS (
  SELECT drone, arg_max(level_pct, ts_ns) AS battery
  FROM telemetry
  WHERE sensor = 'battery'
  GROUP BY drone
 )
 SELECT p.drone, p.x, p.y, p.yaw, p.ts_ns, b.battery
 FROM pose p LEFT JOIN batt b USING (drone)
 ORDER BY p.drone`;
 interface QueryResponse {
  columns?: string[];
  rows?: (string | number | null)[][];
  error?: string;
 }
 export async function fetchLivePoses(baseUrl: string): Promise<LivePose[]> {
  const res = await fetch(`${baseUrl.replace(/\/$/, "")}/api/query`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ sql: LIVE_SQL }),
  });
  if (!res.ok) throw new Error(`explorer responded ${res.status}`);
  const data = (await res.json()) as QueryResponse;
  if (data.error) throw new Error(data.error);
  return (data.rows ?? []).map((r) => ({
    id: String(r[0]),
    x: Number(r[1]),
    y: Number(r[2]),
    yawDeg: Number(r[3]),
    tsNs: Number(r[4]),
    battery: r[5] === null ? null : Number(r[5]),
  }));
 }
 // Scale mission-frame meters into the prototype's square view. The virtual
 // drones patrol a smaller square than the sim world, so fit the observed
 // extent (with padding) instead of assuming a size.
 function fitScale(poses: LivePose[]): number {
  const extent = Math.max(400, ...poses.map((p) => Math.max(p.x, p.y))) * 1.15;
  return AREA / extent;
 }
 function linkKey(a: number, b: number): string {
  return a < b ? `${a}-${b}` : `${b}-${a}`;
 }
 /** Convert live poses into the World shape the renderer already speaks. */
 export function toLiveWorld(poses: LivePose[], prev: World | null, dtS: number): World {
  const k = fitScale(poses);
  const drones = poses.map((p, i) => ({
    id: i,
    name: p.id,
    pos: { x: p.x * k, y: p.y * k },
    vel: { x: 0, y: 0 },
    heading: (p.yawDeg * Math.PI) / 180,
    waypoint: 0,
    battery: p.battery ?? 100,
    channel: 0,
  }));
  // Every in-range pair exchanges pose broadcasts; accumulate totals so the
  // panel keeps its meaning between polls.
  const links = new Map<string, LinkStats>();
  let totalBytes = prev?.totalBytes ?? 0;
  for (let i = 0; i < drones.length; i++) {
    for (let j = i + 1; j < drones.length; j++) {
      const dist = Math.hypot(
        drones[i].pos.x - drones[j].pos.x,
        drones[i].pos.y - drones[j].pos.y,
      );
      if (dist > LINK_RANGE) continue;
      const key = linkKey(i, j);
      const old = prev?.links.get(key);
      const bytes = POSE_RATE_BYTES * 2 * dtS;
      totalBytes += bytes;
      links.set(key, {
        key,
        a: i,
        b: j,
        totalUp: (old?.totalUp ?? 0) + bytes / 2,
        totalDown: (old?.totalDown ?? 0) + bytes / 2,
        rate: POSE_RATE_BYTES * 2,
        flash: 0.6,
        bulk: 0,
      });
    }
  }
  return {
    t: (prev?.t ?? 0) + dtS,
    drones,
    obstacles: [],
    links,
    totalBytes,
    broadcasts: (prev?.broadcasts ?? 0) + links.size * 2,
  };
 }
@@ -9,6 +9,7 @@ export interface Vec {
 export interface Drone {
  id: number;
  name?: string; // live mode: real drone_id from the data plane
  pos: Vec;
  vel: Vec;
  heading: number; // radians
@@ -45,12 +46,21 @@ export interface World {
  broadcasts: number;
 }
-export const AREA = 1000; // meters, square side
+export const AREA = 1000; // meters, vertical extent of the field
 export const MAX_ASPECT = 2.2;
 // Horizontal extent follows the display aspect ratio (set via makeWorld),
 // so wide monitors get a genuinely wider patrol area, not empty margins.
 let areaX = AREA;
 export function areaWidth(): number {
  return areaX;
 }
 export const LINK_RANGE = 320;
 const CRUISE = 28;
 const SEPARATION = 55;
 const POSE_BYTES = 46;
 const POSE_HZ = 5;
 // Steady per-link broadcast throughput (both directions), bytes/s
 export const POSE_RATE_BYTES = POSE_BYTES * POSE_HZ;
 const CHANNELS = [1, 6, 11, 36, 40, 44, 149, 157];
 let seedState = 1234;
@@ -65,35 +75,85 @@ function rnd(): number {
  return ((seedState >>> 0) % 100000) / 100000;
 }
-const WAYPOINTS: Vec[] = [
+// Two interleaved patrol routes: even drones circle the perimeter, odd
 // drones fly an X through the center. Paths cross mid-field, so the mesh
 // keeps bridging between the front and the back of the formation instead
 // of stretching into disconnected segments along one loop. Routes are
 // derived from the current field width, so they stretch on wide displays.
 function perimeterRoute(): Vec[] {
  return [
    { x: 120, y: 120 },
-  { x: AREA - 120, y: 150 },
+    { x: areaX - 120, y: 150 },
-  { x: AREA - 150, y: AREA - 120 },
+    { x: areaX - 150, y: AREA - 120 },
    { x: 150, y: AREA - 150 },
  ];
 }
 function crossRoute(): Vec[] {
  return [
    { x: 150, y: 150 },
    { x: areaX - 150, y: AREA - 150 },
    { x: areaX - 150, y: 150 },
    { x: 150, y: AREA - 150 },
  ];
 }
 function routeOf(d: Drone): Vec[] {
  return d.id % 2 === 0 ? perimeterRoute() : crossRoute();
 }
-export function makeWorld(droneCount: number): World {
+// Transit objects enter at one edge, cross the whole area, and respawn at
 // a fresh edge once they leave — a stream of through-traffic instead of
 // obstacles bouncing until they wedge into a corner.
 const TRANSIT_MARGIN = 120;
 function spawnTransit(id: number): Obstacle {
  const speed = 26 + rnd() * 30;
  const alongX = 120 + rnd() * (areaX - 240);
  const alongY = 120 + rnd() * (AREA - 240);
  const skew = (rnd() - 0.5) * speed * 0.9; // diagonal-ish crossings
  const edge = Math.floor(rnd() * 4);
  const table: { pos: Vec; vel: Vec }[] = [
    { pos: { x: alongX, y: -TRANSIT_MARGIN + 20 }, vel: { x: skew, y: speed } },
    { pos: { x: alongX, y: AREA + TRANSIT_MARGIN - 20 }, vel: { x: skew, y: -speed } },
    { pos: { x: -TRANSIT_MARGIN + 20, y: alongY }, vel: { x: speed, y: skew } },
    { pos: { x: areaX + TRANSIT_MARGIN - 20, y: alongY }, vel: { x: -speed, y: skew } },
  ];
  const { pos, vel } = table[edge];
  return { id, pos, radius: 22 + rnd() * 16, moving: true, vel };
 }
 function outOfTransit(ob: Obstacle): boolean {
  return (
    ob.pos.x < -TRANSIT_MARGIN ||
    ob.pos.x > areaX + TRANSIT_MARGIN ||
    ob.pos.y < -TRANSIT_MARGIN ||
    ob.pos.y > AREA + TRANSIT_MARGIN
  );
 }
 export function makeWorld(droneCount: number, aspect = 1): World {
  areaX = AREA * Math.max(1, Math.min(MAX_ASPECT, aspect));
  const drones: Drone[] = Array.from({ length: droneCount }, (_, i) => {
    const angle = (i / droneCount) * Math.PI * 2;
    return {
      id: i,
      pos: {
-        x: AREA / 2 + Math.cos(angle) * (150 + rnd() * 120),
+        x: areaX / 2 + Math.cos(angle) * (150 + rnd() * 120),
        y: AREA / 2 + Math.sin(angle) * (150 + rnd() * 120),
      },
      vel: { x: 0, y: 0 },
      heading: angle,
-      waypoint: i % WAYPOINTS.length,
+      waypoint: i % 4,
      battery: 90 + rnd() * 10,
      channel: CHANNELS[i % CHANNELS.length],
    };
  });
  // Static obstacles sit at fixed fractions of the field, so they spread
  // out instead of clustering left when the field widens
  const obstacles: Obstacle[] = [
-    { id: 0, pos: { x: 340, y: 420 }, radius: 60, moving: false, vel: { x: 0, y: 0 } },
+    { id: 0, pos: { x: areaX * 0.34, y: 420 }, radius: 60, moving: false, vel: { x: 0, y: 0 } },
-    { id: 1, pos: { x: 700, y: 260 }, radius: 45, moving: false, vel: { x: 0, y: 0 } },
+    { id: 1, pos: { x: areaX * 0.7, y: 260 }, radius: 45, moving: false, vel: { x: 0, y: 0 } },
-    { id: 2, pos: { x: 560, y: 720 }, radius: 70, moving: false, vel: { x: 0, y: 0 } },
+    { id: 2, pos: { x: areaX * 0.56, y: 720 }, radius: 70, moving: false, vel: { x: 0, y: 0 } },
-    { id: 3, pos: { x: 200, y: 800 }, radius: 30, moving: true, vel: { x: 14, y: -6 } },
+    spawnTransit(3),
-    { id: 4, pos: { x: 820, y: 620 }, radius: 26, moving: true, vel: { x: -10, y: 9 } },
+    spawnTransit(4),
    spawnTransit(5),
  ];
  return { t: 0, drones, obstacles, links: new Map(), totalBytes: 0, broadcasts: 0 };
 }
@@ -103,12 +163,13 @@ function linkKey(a: number, b: number): string {
 }
 function steer(d: Drone, world: World, dt: number): Drone {
-  const wp = WAYPOINTS[(d.waypoint + d.id) % WAYPOINTS.length];
+  const route = routeOf(d);
  const wp = route[(d.waypoint + d.id) % route.length];
  let ax = wp.x - d.pos.x;
  let ay = wp.y - d.pos.y;
  const wpDist = Math.hypot(ax, ay);
  let waypoint = d.waypoint;
-  if (wpDist < 90) waypoint = (d.waypoint + 1) % WAYPOINTS.length;
+  if (wpDist < 90) waypoint = (d.waypoint + 1) % route.length;
  ax /= wpDist || 1;
  ay /= wpDist || 1;
@@ -147,7 +208,7 @@ function steer(d: Drone, world: World, dt: number): Drone {
    vel: { x: vx, y: vy },
    heading,
    pos: {
-      x: Math.max(20, Math.min(AREA - 20, d.pos.x + vx * dt)),
+      x: Math.max(20, Math.min(areaX - 20, d.pos.x + vx * dt)),
      y: Math.max(20, Math.min(AREA - 20, d.pos.y + vy * dt)),
    },
    battery: Math.max(0, d.battery - dt * 0.05),
@@ -157,18 +218,15 @@ function steer(d: Drone, world: World, dt: number): Drone {
 export function tick(world: World, dt: number): World {
  const drones = world.drones.map((d) => steer(d, world, dt));
-  const obstacles = world.obstacles.map((ob) =>
+  const obstacles = world.obstacles.map((ob) => {
-    ob.moving
+    if (!ob.moving) return ob;
-      ? {
+    const moved = {
      ...ob,
      pos: { x: ob.pos.x + ob.vel.x * dt, y: ob.pos.y + ob.vel.y * dt },
-          vel: {
+    };
-            x: ob.pos.x < 60 || ob.pos.x > AREA - 60 ? -ob.vel.x : ob.vel.x,
+    // Once a transit object leaves the area, respawn it at a random edge
-            y: ob.pos.y < 60 || ob.pos.y > AREA - 60 ? -ob.vel.y : ob.vel.y,
+    return outOfTransit(moved) ? spawnTransit(ob.id) : moved;
-          },
+  });
        }
      : ob,
  );
  const links = new Map(world.links);
  let totalBytes = world.totalBytes;
@@ -0,0 +1 @@
 /// <reference types="vite/client" />
@@ -0,0 +1,3 @@
 data
 .venv
 README.md
@@ -1,9 +1,33 @@
 FROM python:3.12-slim AS test
 WORKDIR /app
 COPY requirements.txt .
 RUN --mount=type=cache,target=/root/.cache/pip pip install -r requirements.txt pytest
 COPY virtual_drone ./virtual_drone
 COPY monitoring ./monitoring
 COPY explorer ./explorer
 COPY tests ./tests
 RUN groupadd --gid 10001 swarm \
    && useradd --uid 10001 --gid swarm --home-dir /app --shell /usr/sbin/nologin swarm \
    && chown -R swarm:swarm /app
 USER swarm
 RUN pytest tests/ -q
 FROM python:3.12-slim
 WORKDIR /app
 COPY requirements.txt .
 RUN --mount=type=cache,target=/root/.cache/pip pip install -r requirements.txt
 COPY virtual_drone ./virtual_drone
 # Monitoring and explorer ship in the same image so Kubernetes can run
 # them without bind mounts (Compose overrides these with live mounts)
 COPY monitoring ./monitoring
 COPY explorer ./explorer
 RUN groupadd --gid 10001 swarm \
    && useradd --uid 10001 --gid swarm --home-dir /app --shell /usr/sbin/nologin swarm \
    && mkdir -p /data \
    && chown -R swarm:swarm /app /data
 ENV DATA_DIR=/data
 USER swarm
 CMD ["python", "-m", "virtual_drone.main"]
@@ -14,14 +14,16 @@ FLIGHT_ID=$(date -u +%Y%m%dT%H%MZ)-sim docker compose up --build --scale drone=5
 DRONE_COUNT=10 DURATION_S=300 SPEEDUP=4 docker compose up --build --scale drone=10
 ```
-Output lands in `./data/` with the standard layout:
+Output lands in [`../var/t1/`](../var/t1/) (the shared T1 lake — same path the k3d fleet and warehouse offload use). Override with `SWARM_T1_DIR` if needed:
 ```
-data/dataset=telemetry/flight=…/drone=…/sensor=imu/year=…/…/hour=…/data.parquet
+var/t1/dataset=telemetry/flight=…/drone=…/sensor=imu/year=…/…/hour=…/data.parquet
-data/dataset=detections/flight=…/drone=…/…
+var/t1/dataset=detections/flight=…/drone=…/…
-data/dataset=state/flight=…/drone=…/…        ← sent + received broadcasts
+var/t1/dataset=state/flight=…/drone=…/…        ← sent + received broadcasts
 ```
 Historical flights after offload live under [`../var/t3/`](../var/t3/) (T3 warehouse).
 ## Run a single drone without Docker
 ```bash
@@ -14,7 +14,7 @@ services:
      # Compose default bridge network: subnet-directed broadcast works
      BROADCAST_ADDR: ${BROADCAST_ADDR:-255.255.255.255}
    volumes:
-      - ./data:/data
+      - ${SWARM_T1_DIR:-../var/t1}:/data
  # --- Monitoring profile: docker compose --profile monitoring up ---
@@ -24,9 +24,10 @@ services:
    command: ["python", "monitoring/exporter.py"]
    environment:
      DATA_DIR: /data
-      SCAN_INTERVAL_S: "5"
+      SCAN_INTERVAL_S: "30"
      METRIC_FLIGHT_WINDOW: "5"
    volumes:
-      - ./data:/data:ro
+      - ${SWARM_T1_DIR:-../var/t1}:/data:ro
      - ./monitoring:/app/monitoring:ro
    ports:
      - "${EXPORTER_PORT:-9105}:9105"
@@ -37,8 +38,11 @@ services:
    command: ["python", "explorer/server.py"]
    environment:
      DATA_DIR: /data
      VIEW_REFRESH_S: "30"
      TREE_CACHE_S: "15"
      METRIC_FLIGHT_WINDOW: "5"
    volumes:
-      - ./data:/data:ro
+      - ${SWARM_T1_DIR:-../var/t1}:/data:ro
      - ./explorer:/app/explorer:ro
    ports:
      - "${EXPLORER_UI_PORT:-8088}:8088"
@@ -58,6 +62,10 @@ services:
      GF_AUTH_ANONYMOUS_ENABLED: "true"
      GF_AUTH_ANONYMOUS_ORG_ROLE: Admin
      GF_AUTH_DISABLE_LOGIN_FORM: "true"
      # Air-gap hygiene: no phone-home, no update checks, no plugin fetches
      GF_ANALYTICS_REPORTING_ENABLED: "false"
      GF_ANALYTICS_CHECK_FOR_UPDATES: "false"
      GF_ANALYTICS_CHECK_FOR_PLUGIN_UPDATES: "false"
    volumes:
      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro
      - ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards:ro
@@ -1,33 +1,28 @@
-"""Data-plane explorer: a web view over the Hive-partitioned Parquet lake.
+"""Data-plane explorer: a web view over the Hive-partitioned Parquet lake."""
 Serves three things:
  /            single-page UI (partition tree + read-only SQL console)
  /api/tree    partition hierarchy with file counts and bytes, live
  /api/query   gated read-only DuckDB SQL, same statement rules as the
               peer query channel (SELECT/WITH only, single statement)
 The point is doctrinal, not just convenient: the explorer reuses the exact
 read-only SQL contract that drones expose to each other, so "looking at the
 data plane" on the bench exercises the same path a peer would use in flight.
 """
 from __future__ import annotations
 import json
 import os
 import re
 import sys
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 import duckdb
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "monitoring"))
 from lake import flight_window, iter_parquet_files, parquet_reader  # noqa: E402
 DATA_DIR = Path(os.environ.get("DATA_DIR", "./data"))
 PORT = int(os.environ.get("EXPLORER_PORT", "8088"))
 ROW_LIMIT = int(os.environ.get("ROW_LIMIT", "500"))
 VIEW_REFRESH_S = float(os.environ.get("VIEW_REFRESH_S", "30"))
 TREE_CACHE_S = float(os.environ.get("TREE_CACHE_S", "15"))
 STATIC_DIR = Path(__file__).parent
 # Same spirit as the forced-command gate on a real drone: one statement,
 # must be a read, no statement that could write, configure, or reach out.
 _ALLOWED_START = re.compile(r"^\s*(SELECT|WITH|DESCRIBE|SUMMARIZE|SHOW)\b", re.IGNORECASE)
 _FORBIDDEN = re.compile(
    r"\b(INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|ATTACH|DETACH|COPY|EXPORT"
@@ -35,9 +30,17 @@ _FORBIDDEN = re.compile(
    re.IGNORECASE,
 )
 _lock = threading.Lock()
 _con: duckdb.DuckDBPyConnection | None = None
 _views_at = 0.0
 _tree_cache: dict | None = None
 _tree_at = 0.0
 _last_tree_s = 0.0
 _last_query_s = 0.0
 _query_count = 0
 def gate(sql: str) -> str | None:
    """Return a rejection reason, or None if the statement passes."""
    stripped = re.sub(r"--[^\n]*|/\*.*?\*/", " ", sql, flags=re.DOTALL).strip().rstrip(";")
    if not stripped:
        return "empty statement"
@@ -50,27 +53,39 @@ def gate(sql: str) -> str | None:
    return None
-def connect() -> duckdb.DuckDBPyConnection:
+def _refresh_views() -> None:
-    """Fresh connection with the three datasets pre-registered as views."""
+    global _con, _views_at
    con = duckdb.connect()
    for ds in ("telemetry", "detections", "state"):
-        pattern = f"{DATA_DIR}/dataset={ds}/**/*.parquet"
+        reader = parquet_reader(DATA_DIR, ds)
        try:
-            con.execute(
+            con.execute(f"CREATE OR REPLACE VIEW {ds} AS SELECT * FROM read_parquet({reader})")
                f"CREATE VIEW {ds} AS SELECT * FROM read_parquet("
                f"'{pattern}', hive_partitioning=true, union_by_name=true)"
            )
        except duckdb.Error:
-            pass  # dataset not written yet; view simply won't exist
+            pass
-    return con
+    with _lock:
        if _con is not None:
            _con.close()
        _con = con
        _views_at = time.monotonic()
 def connect() -> duckdb.DuckDBPyConnection:
    if _con is None or time.monotonic() - _views_at > VIEW_REFRESH_S:
        _refresh_views()
    assert _con is not None
    return _con
 def tree() -> dict:
-    """Partition hierarchy: dataset -> flight -> drone -> leafs, with sizes."""
+    global _tree_cache, _tree_at, _last_tree_s
    now = time.monotonic()
    if _tree_cache is not None and now - _tree_at < TREE_CACHE_S:
        return _tree_cache
    started = now
    root: dict = {}
    total_bytes = 0
    total_files = 0
-    for f in sorted(DATA_DIR.rglob("*.parquet")):
+    for f in sorted(iter_parquet_files(DATA_DIR)):
        rel = f.relative_to(DATA_DIR)
        size = f.stat().st_size
        total_bytes += size
@@ -80,7 +95,6 @@ def tree() -> dict:
            node = node.setdefault("children", {}).setdefault(part, {})
        leaf = node.setdefault("children", {}).setdefault(rel.parts[-1], {})
        leaf["bytes"] = size
        # roll sizes up the tree
        node = root
        node["bytes"] = node.get("bytes", 0) + size
        node["files"] = node.get("files", 0) + 1
@@ -88,18 +102,26 @@ def tree() -> dict:
            node = node["children"][part]
            node["bytes"] = node.get("bytes", 0) + size
            node["files"] = node.get("files", 0) + 1
-    return {"tree": root, "total_bytes": total_bytes, "total_files": total_files}
+    _last_tree_s = time.monotonic() - started
    _tree_cache = {"tree": root, "total_bytes": total_bytes, "total_files": total_files}
    _tree_at = now
    return _tree_cache
 def run_query(sql: str) -> dict:
    global _last_query_s, _query_count
    reason = gate(sql)
    if reason:
        return {"error": f"rejected by read-only gate: {reason}"}
    started = time.monotonic()
    with _lock:
        con = connect()
        try:
            cur = con.sql(sql)
            columns = cur.columns
            rows = cur.fetchmany(ROW_LIMIT)
            _query_count += 1
            _last_query_s = time.monotonic() - started
            return {
                "columns": columns,
                "rows": [[repr(v) if isinstance(v, bytes) else v for v in row] for row in rows],
@@ -107,8 +129,23 @@ def run_query(sql: str) -> dict:
            }
        except duckdb.Error as exc:
            return {"error": str(exc)}
-    finally:
+
-        con.close()
+
 def metrics_text() -> str:
    return "\n".join([
        "# HELP swarm_explorer_query_duration_seconds Wall time of the last SQL query",
        "# TYPE swarm_explorer_query_duration_seconds gauge",
        f"swarm_explorer_query_duration_seconds {_last_query_s:.4f}",
        "# HELP swarm_explorer_tree_duration_seconds Wall time of the last partition tree build",
        "# TYPE swarm_explorer_tree_duration_seconds gauge",
        f"swarm_explorer_tree_duration_seconds {_last_tree_s:.4f}",
        "# HELP swarm_explorer_queries_total Read-only queries served",
        "# TYPE swarm_explorer_queries_total counter",
        f"swarm_explorer_queries_total {_query_count}",
        "# HELP swarm_metric_flight_window Flight partitions in DuckDB views",
        "# TYPE swarm_metric_flight_window gauge",
        f"swarm_metric_flight_window {flight_window()}",
    ]) + "\n"
 class Handler(BaseHTTPRequestHandler):
@@ -116,21 +153,31 @@ class Handler(BaseHTTPRequestHandler):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)
    def do_OPTIONS(self) -> None:  # noqa: N802
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()
    def _json(self, payload: dict, code: int = 200) -> None:
        self._send(code, json.dumps(payload, default=str).encode(), "application/json")
-    def do_GET(self) -> None:  # noqa: N802 — http.server API
+    def do_GET(self) -> None:  # noqa: N802
        if self.path in ("/", "/index.html"):
            self._send(200, (STATIC_DIR / "index.html").read_bytes(), "text/html; charset=utf-8")
        elif self.path == "/api/tree":
            self._json(tree())
        elif self.path == "/metrics":
            self._send(200, metrics_text().encode(), "text/plain; version=0.0.4")
        else:
            self._send(404, b"not found", "text/plain")
-    def do_POST(self) -> None:  # noqa: N802 — http.server API
+    def do_POST(self) -> None:  # noqa: N802
        if self.path != "/api/query":
            self._send(404, b"not found", "text/plain")
            return
@@ -146,6 +193,20 @@ class Handler(BaseHTTPRequestHandler):
        pass
 def _view_loop() -> None:
    while True:
        try:
            _refresh_views()
        except Exception:
            pass
        time.sleep(VIEW_REFRESH_S)
 if __name__ == "__main__":
-    print(f"data-plane explorer on :{PORT}, reading {DATA_DIR}")
+    _refresh_views()
    threading.Thread(target=_view_loop, daemon=True).start()
    print(
        f"data-plane explorer on :{PORT}, views refresh every {VIEW_REFRESH_S}s, "
        f"last {flight_window()} flights, reading {DATA_DIR}"
    )
    ThreadingHTTPServer(("", PORT), Handler).serve_forever()
@@ -1,13 +1,14 @@
 """Prometheus exporter over the simulator's Parquet output.
 Periodically scans DATA_DIR with DuckDB and exposes fleet statistics as
-/metrics. Zero dependencies beyond duckdb: the exposition format is plain
+/metrics. Scans only the most recent flight partitions by default so CPU
-text, served with the standard library HTTP server.
+stays bounded as the lake grows across pod restarts.
 """
 from __future__ import annotations
 import os
 import sys
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@@ -15,27 +16,29 @@ from pathlib import Path
 import duckdb
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 from lake import flight_window, iter_parquet_files, parquet_reader  # noqa: E402
 DATA_DIR = Path(os.environ.get("DATA_DIR", "./data"))
 PORT = int(os.environ.get("EXPORTER_PORT", "9105"))
-SCAN_INTERVAL_S = float(os.environ.get("SCAN_INTERVAL_S", "5"))
+SCAN_INTERVAL_S = float(os.environ.get("SCAN_INTERVAL_S", "15"))
 _lock = threading.Lock()
 _payload = "# swarm exporter starting\n"
 _last_scan_s = 0.0
 def _q(con: duckdb.DuckDBPyConnection, sql: str) -> list[tuple]:
    try:
        return con.sql(sql).fetchall()
    except duckdb.Error:
-        return []  # partitions may not exist yet while the swarm warms up
+        return []
 def collect() -> str:
    global _last_scan_s
    started = time.monotonic()
    con = duckdb.connect()
    # union_by_name: sensors have different schemas under one dataset glob
    glob = lambda ds: (  # noqa: E731
        f"'{DATA_DIR}/dataset={ds}/**/*.parquet', hive_partitioning=true, union_by_name=true"
    )
    lines: list[str] = []
    def metric(name: str, help_text: str, mtype: str, rows: list[str]) -> None:
@@ -43,27 +46,38 @@ def collect() -> str:
        lines.append(f"# TYPE {name} {mtype}")
        lines.extend(rows)
    for ds, name in (
        ("telemetry", "swarm_rows_total"),
        ("detections", "swarm_detections_total"),
        ("state", "swarm_state_frames_total"),
    ):
        reader = parquet_reader(DATA_DIR, ds)
        if ds == "telemetry":
            metric(
-        "swarm_rows_total", "Telemetry rows written per drone and sensor", "gauge",
+                name, f"Rows in dataset={ds} (recent {flight_window()} flights)", "gauge",
-        [f'swarm_rows_total{{drone="{d}",sensor="{s}"}} {n}'
+                [f'{name}{{drone="{d}",sensor="{s}"}} {n}'
-         for d, s, n in _q(con, f"SELECT drone, sensor, count(*) FROM read_parquet({glob('telemetry')}) GROUP BY 1,2")],
+                 for d, s, n in _q(con, f"SELECT drone, sensor, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
            )
        elif ds == "detections":
            metric(
-        "swarm_detections_total", "Detection events per drone and class", "gauge",
+                name, "Detection events per drone and class", "gauge",
-        [f'swarm_detections_total{{drone="{d}",cls="{c}"}} {n}'
+                [f'{name}{{drone="{d}",cls="{c}"}} {n}'
-         for d, c, n in _q(con, f"SELECT drone, cls, count(*) FROM read_parquet({glob('detections')}) GROUP BY 1,2")],
+                 for d, c, n in _q(con, f"SELECT drone, cls, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
            )
        else:
            metric(
-        "swarm_state_frames_total", "Pose broadcast frames per drone and direction", "gauge",
+                name, "Pose broadcast frames per drone and direction", "gauge",
-        [f'swarm_state_frames_total{{drone="{d}",direction="{dr}"}} {n}'
+                [f'{name}{{drone="{d}",direction="{dr}"}} {n}'
-         for d, dr, n in _q(con, f"SELECT drone, direction, count(*) FROM read_parquet({glob('state')}) GROUP BY 1,2")],
+                 for d, dr, n in _q(con, f"SELECT drone, direction, count(*) FROM read_parquet({reader}) GROUP BY 1,2")],
            )
    telem = parquet_reader(DATA_DIR, "telemetry")
    metric(
        "swarm_battery_pct", "Latest battery level per drone", "gauge",
        [f'swarm_battery_pct{{drone="{d}"}} {v}'
         for d, v in _q(con, f"""
            SELECT drone, arg_max(level_pct, ts_ns)
-            FROM read_parquet({glob('telemetry')})
+            FROM read_parquet({telem})
            WHERE sensor='battery' GROUP BY drone""")],
    )
    metric(
@@ -71,26 +85,24 @@ def collect() -> str:
        [f'swarm_rssi_dbm{{drone="{d}",peer="{p}"}} {v}'
         for d, p, v in _q(con, f"""
            SELECT drone, peer_id, arg_max(rssi_dbm, ts_ns)
-            FROM read_parquet({glob('telemetry')})
+            FROM read_parquet({telem})
            WHERE sensor='rssi' GROUP BY drone, peer_id""")],
    )
    metric(
        "swarm_peer_distance_m", "Latest inter-drone distance estimate", "gauge",
        [f'swarm_peer_distance_m{{drone="{d}",peer="{p}"}} {v}'
         for d, p, v in _q(con, f"""
            SELECT drone, peer_id, arg_max(distance_m, ts_ns)
            FROM read_parquet({glob('telemetry')})
            WHERE sensor='rssi' GROUP BY drone, peer_id""")],
    )
-    files = list(DATA_DIR.rglob("*.parquet"))
+    files = iter_parquet_files(DATA_DIR)
    metric(
-        "swarm_parquet_bytes", "Bytes on disk per dataset", "gauge",
+        "swarm_parquet_bytes", "Bytes on disk per dataset (recent flights)", "gauge",
        [f'swarm_parquet_bytes{{dataset="{ds}"}} {sum(f.stat().st_size for f in files if f"dataset={ds}" in str(f))}'
         for ds in ("telemetry", "detections", "state")],
    )
-    metric("swarm_parquet_files", "Parquet files on disk", "gauge",
+    metric("swarm_parquet_files", "Parquet files scanned (recent flights)", "gauge",
           [f"swarm_parquet_files {len(files)}"])
    metric("swarm_metric_flight_window", "Flight partitions included per dataset", "gauge",
           [f"swarm_metric_flight_window {flight_window()}"])
    _last_scan_s = time.monotonic() - started
    metric("swarm_exporter_scan_duration_seconds", "Wall time of the last metrics scan", "gauge",
           [f"swarm_exporter_scan_duration_seconds {_last_scan_s:.4f}"])
    con.close()
    return "\n".join(lines) + "\n"
@@ -101,11 +113,11 @@ def scanner() -> None:
        started = time.monotonic()
        try:
            payload = collect()
-        except Exception as exc:  # keep serving stale metrics over dying
+        except Exception as exc:
            payload = f"# collect error: {exc}\n"
        with _lock:
            _payload = payload
-        time.sleep(max(0.5, SCAN_INTERVAL_S - (time.monotonic() - started)))
+        time.sleep(max(1.0, SCAN_INTERVAL_S - (time.monotonic() - started)))
 class Handler(BaseHTTPRequestHandler):
@@ -123,10 +135,13 @@ class Handler(BaseHTTPRequestHandler):
        self.wfile.write(body)
    def log_message(self, *_args: object) -> None:
-        pass  # scrapes every few seconds; keep the log quiet
+        pass
 if __name__ == "__main__":
    threading.Thread(target=scanner, daemon=True).start()
-    print(f"swarm exporter on :{PORT}/metrics, scanning {DATA_DIR} every {SCAN_INTERVAL_S}s")
+    print(
        f"swarm exporter on :{PORT}/metrics, scanning last {flight_window()} flights "
        f"every {SCAN_INTERVAL_S}s under {DATA_DIR}"
    )
    ThreadingHTTPServer(("", PORT), Handler).serve_forever()
@@ -0,0 +1,90 @@
 {
  "uid": "swarm-platform",
  "title": "Swarm Platform — CPU & scan health",
  "tags": ["swarm", "platform"],
  "timezone": "utc",
  "schemaVersion": 39,
  "version": 1,
  "refresh": "10s",
  "time": { "from": "now-30m", "to": "now" },
  "panels": [
    {
      "id": 1, "type": "timeseries", "title": "Node CPU %",
      "gridPos": { "h": 8, "w": 12, "x": 0, "y": 0 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{
        "expr": "100 - (avg by (instance) (rate(node_cpu_seconds_total{mode=\"idle\"}[1m])) * 100)",
        "legendFormat": "{{instance}}", "refId": "A"
      }],
      "fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] }
    },
    {
      "id": 2, "type": "timeseries", "title": "Node memory used %",
      "gridPos": { "h": 8, "w": 12, "x": 12, "y": 0 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{
        "expr": "(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes) * 100",
        "legendFormat": "used", "refId": "A"
      }],
      "fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] }
    },
    {
      "id": 3, "type": "timeseries", "title": "Exporter scan duration",
      "gridPos": { "h": 7, "w": 8, "x": 0, "y": 8 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{
        "expr": "swarm_exporter_scan_duration_seconds",
        "legendFormat": "scan seconds", "refId": "A"
      }],
      "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }
    },
    {
      "id": 4, "type": "timeseries", "title": "Explorer query duration",
      "gridPos": { "h": 7, "w": 8, "x": 8, "y": 8 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{
        "expr": "swarm_explorer_query_duration_seconds",
        "legendFormat": "last query", "refId": "A"
      }],
      "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }
    },
    {
      "id": 5, "type": "timeseries", "title": "Explorer tree build duration",
      "gridPos": { "h": 7, "w": 8, "x": 16, "y": 8 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{
        "expr": "swarm_explorer_tree_duration_seconds",
        "legendFormat": "tree seconds", "refId": "A"
      }],
      "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }
    },
    {
      "id": 6, "type": "stat", "title": "Parquet files scanned",
      "gridPos": { "h": 5, "w": 6, "x": 0, "y": 15 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{ "expr": "swarm_parquet_files", "instant": true, "refId": "A" }],
      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
    },
    {
      "id": 7, "type": "stat", "title": "Flight window",
      "gridPos": { "h": 5, "w": 6, "x": 6, "y": 15 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{ "expr": "swarm_metric_flight_window", "instant": true, "refId": "A" }],
      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
    },
    {
      "id": 8, "type": "stat", "title": "Telemetry rows (window)",
      "gridPos": { "h": 5, "w": 6, "x": 12, "y": 15 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{ "expr": "sum(swarm_rows_total)", "instant": true, "refId": "A" }],
      "options": { "reduceOptions": { "calcs": ["lastNotNull"] } }
    },
    {
      "id": 9, "type": "stat", "title": "Min fleet battery %",
      "gridPos": { "h": 5, "w": 6, "x": 18, "y": 15 },
      "datasource": { "type": "prometheus", "uid": "swarm-prom" },
      "targets": [{ "expr": "min(swarm_battery_pct)", "instant": true, "refId": "A" }],
      "fieldConfig": { "defaults": { "unit": "percent", "min": 0, "max": 100 }, "overrides": [] }
    }
  ]
 }
@@ -0,0 +1,43 @@
 """Scan the Hive-partitioned lake without re-reading every historical flight."""
 from __future__ import annotations
 import os
 from pathlib import Path
 def flight_window() -> int:
    return max(1, int(os.environ.get("METRIC_FLIGHT_WINDOW", "5")))
 def recent_flights(data_dir: Path, dataset: str, limit: int | None = None) -> list[Path]:
    """Most recently modified flight= partitions for a dataset."""
    limit = limit or flight_window()
    base = data_dir / f"dataset={dataset}"
    if not base.is_dir():
        return []
    flights = [p for p in base.iterdir() if p.is_dir() and p.name.startswith("flight=")]
    flights.sort(key=lambda p: p.stat().st_mtime, reverse=True)
    return flights[:limit]
 def parquet_reader(data_dir: Path, dataset: str, *, limit: int | None = None) -> str:
    """DuckDB read_parquet() source limited to recent flights."""
    flights = recent_flights(data_dir, dataset, limit)
    if not flights:
        path = data_dir / f"dataset={dataset}" / "**" / "*.parquet"
        return f"'{path}', hive_partitioning=true, union_by_name=true"
    if len(flights) == 1:
        return f"'{flights[0]}/**/*.parquet', hive_partitioning=true, union_by_name=true"
    inner = ", ".join(f"'{f}/**/*.parquet'" for f in flights)
    return f"[{inner}], hive_partitioning=true, union_by_name=true"
 def iter_parquet_files(data_dir: Path, *, flight_limit: int | None = None) -> list[Path]:
    """Parquet paths under recent flights only — avoids full-lake rglob."""
    limit = flight_limit or flight_window()
    out: list[Path] = []
    for ds in ("telemetry", "detections", "state"):
        for flight in recent_flights(data_dir, ds, limit):
            out.extend(flight.rglob("*.parquet"))
    return out
@@ -1,7 +1,10 @@
 global:
-  scrape_interval: 5s
+  scrape_interval: 15s
 scrape_configs:
  - job_name: swarm
    static_configs:
      - targets: ["exporter:9105"]
  - job_name: explorer
    static_configs:
      - targets: ["explorer:8088"]
@@ -1,3 +1,4 @@
 pyarrow>=16.0
 duckdb>=1.0
 pytz>=2024.1  # duckdb needs it to fetch TIMESTAMPTZ values into Python
 pytest>=8.0
@@ -0,0 +1,46 @@
 """Pose broadcast frame encode/decode roundtrip."""
 from __future__ import annotations
 import pytest
 from virtual_drone.broadcast import FRAME, decode_state, encode_state
 from virtual_drone.flight import Pose
 def test_frame_size_matches_wire_spec() -> None:
    # Documented as 46 B in docs/04; struct packs to FRAME.size (45 B on this layout).
    assert FRAME.size == 45
 def test_encode_decode_roundtrip() -> None:
    pose = Pose(
        ts_ns=1_700_000_000_000_000_000,
        x=123.456,
        y=78.9,
        z=60.0,
        roll=1.5,
        pitch=-2.0,
        yaw=45.0,
        vx=3.2,
        vy=-1.1,
        vz=0.05,
    )
    payload = encode_state("dr-01", pose, flags=0)
    assert len(payload) == FRAME.size
    decoded = decode_state(payload)
    assert decoded is not None
    assert decoded["peer_id"] == "dr-01"
    assert decoded["ts_ns"] == pose.ts_ns
    assert decoded["pos_x"] == pytest.approx(pose.x, abs=0.001)
    assert decoded["pos_y"] == pytest.approx(pose.y, abs=0.001)
    assert decoded["pos_z"] == pytest.approx(pose.z, abs=0.001)
    assert decoded["roll"] == pytest.approx(pose.roll, abs=0.01)
    assert decoded["pitch"] == pytest.approx(pose.pitch, abs=0.01)
    assert decoded["yaw"] == pytest.approx(pose.yaw, abs=0.01)
 def test_rejects_invalid_payload() -> None:
    assert decode_state(b"short") is None
    assert decode_state(b"XX" + b"\0" * (FRAME.size - 2)) is None
@@ -0,0 +1,35 @@
 """Flight kinematics — waypoint patrol and step integration."""
 from __future__ import annotations
 import random
 from virtual_drone.config import Config
 from virtual_drone.flight import initial_pose, make_waypoints, step
 def test_make_waypoints_returns_closed_perimeter() -> None:
    cfg = Config()
    rng = random.Random(7)
    wps = make_waypoints(cfg, rng)
    assert len(wps) == 4
    assert all(len(p) == 2 for p in wps)
 def test_step_moves_toward_target() -> None:
    cfg = Config()
    rng = random.Random(3)
    pose = initial_pose(cfg, rng, 1_000_000_000)
    target = (pose.x + 50.0, pose.y + 50.0)
    next_pose, _ = step(pose, target, cfg, rng, dt=0.1, ts_ns=1_100_000_000)
    dist_before = ((target[0] - pose.x) ** 2 + (target[1] - pose.y) ** 2) ** 0.5
    dist_after = ((target[0] - next_pose.x) ** 2 + (target[1] - next_pose.y) ** 2) ** 0.5
    assert dist_after < dist_before
 def test_reached_when_close_to_waypoint() -> None:
    cfg = Config()
    rng = random.Random(1)
    pose = initial_pose(cfg, rng, 2_000_000_000)
    _, reached = step(pose, (pose.x, pose.y), cfg, rng, dt=1.0, ts_ns=2_100_000_000)
    assert reached is True
@@ -0,0 +1,40 @@
 """Lake scan helpers — bounded to recent flight partitions."""
 from __future__ import annotations
 from pathlib import Path
 from monitoring.lake import flight_window, iter_parquet_files, parquet_reader, recent_flights
 def test_parquet_reader_limits_to_recent_flights(tmp_path: Path) -> None:
    for i, name in enumerate(("flight=aaa", "flight=bbb", "flight=ccc")):
        hour = tmp_path / f"dataset=telemetry/{name}/drone=dr-01/sensor=imu/year=2026/month=07/day=08/hour=10"
        hour.mkdir(parents=True)
        f = hour / "data.parquet"
        f.write_bytes(b"x" * (i + 1))
        # Make later names newer
        import os
        import time
        os.utime(f, (time.time() + i, time.time() + i))
    reader = parquet_reader(tmp_path, "telemetry", limit=1)
    assert "flight=ccc" in reader
    assert "flight=bbb" not in reader
 def test_iter_parquet_files_skips_old_flights(tmp_path: Path) -> None:
    old = tmp_path / "dataset=state/flight=old/drone=dr-01/year=2026/month=07/day=08/hour=09"
    old.mkdir(parents=True)
    (old / "data.parquet").write_bytes(b"old")
    new = tmp_path / "dataset=state/flight=new/drone=dr-01/year=2026/month=07/day=08/hour=10"
    new.mkdir(parents=True)
    new_file = new / "data.parquet"
    new_file.write_bytes(b"new")
    import os
    import time
    os.utime(new_file, (time.time() + 10, time.time() + 10))
    files = iter_parquet_files(tmp_path, flight_limit=1)
    assert len(files) == 1
    assert "flight=new" in str(files[0])
@@ -0,0 +1,35 @@
 """Read-only SQL gate — same rules as the peer-query forced command."""
 from __future__ import annotations
 import sys
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "explorer"))
 from server import gate  # noqa: E402
 def test_allows_read_statements() -> None:
    assert gate("SELECT 1") is None
    assert gate("WITH t AS (SELECT 1) SELECT * FROM t") is None
    assert gate("DESCRIBE telemetry") is None
    assert gate("SUMMARIZE state") is None
    assert gate("SHOW TABLES") is None
 def test_rejects_writes_and_config() -> None:
    assert gate("DROP TABLE telemetry") is not None
    assert gate("INSERT INTO t VALUES (1)") is not None
    assert gate("UPDATE t SET x=1") is not None
    assert gate("DELETE FROM t") is not None
    assert gate("CREATE TABLE t (x INT)") is not None
    assert gate("INSTALL httpfs") is not None
    assert gate("SET memory_limit='1GB'") is not None
    assert gate("PRAGMA threads=4") is not None
 def test_rejects_multi_statement_and_injection() -> None:
    assert gate("SELECT 1; SELECT 2") is not None
    assert gate("/* sneaky */ COPY t TO 'x'") is not None
    assert gate("") is not None
@@ -0,0 +1,67 @@
 """Hive-partitioned Parquet writer layout."""
 from __future__ import annotations
 import time
 from pathlib import Path
 import duckdb
 from virtual_drone.flight import Pose
 from virtual_drone.sensors import imu_row
 from virtual_drone.writer import PartitionWriter, make_writers
 def test_partition_path_follows_hive_layout(tmp_path: Path) -> None:
    ts_ns = int(time.time() * 1e9)
    writer = PartitionWriter(tmp_path, "telemetry", "flt-01", "dr-01", "imu")
    writer.append(imu_row(
        Pose(ts_ns=ts_ns, x=1.0, y=2.0, z=3.0, roll=0.0, pitch=0.0, yaw=0.0, vx=0.0, vy=0.0, vz=0.0),
        __import__("random").Random(0),
        ts_ns,
    ))
    writer.seal()
    files = list(tmp_path.rglob("*.parquet"))
    assert len(files) >= 1
    path_str = str(files[0])
    assert "dataset=telemetry" in path_str
    assert "flight=flt-01" in path_str
    assert "drone=dr-01" in path_str
    assert "sensor=imu" in path_str
    assert "year=" in path_str and "hour=" in path_str
 def test_seal_compacts_current_blocks(tmp_path: Path) -> None:
    writer = PartitionWriter(tmp_path, "state", "flt-02", "dr-02", None)
    base_ts = 1_700_000_000_000_000_000
    for i in range(3):
        writer.append({"ts_ns": base_ts + i * 5_000_000_000, "pos_x": float(i)})
    writer.seal()
    sealed = list(tmp_path.rglob("data.parquet"))
    assert len(sealed) == 1
    assert not list(tmp_path.rglob("current/min_*.parquet"))
    rows = duckdb.sql(f"SELECT count(*) FROM read_parquet('{sealed[0]}')").fetchone()[0]
    assert rows == 3
 def test_make_writers_registers_expected_datasets(tmp_path: Path) -> None:
    writers = make_writers(tmp_path, "flt-03", "dr-03", ["imu", "battery"])
    assert set(writers) == {"imu", "battery", "detections", "state"}
 def test_duckdb_reads_written_partition(tmp_path: Path) -> None:
    writers = make_writers(tmp_path, "flt-04", "dr-04", ["imu"])
    ts_ns = int(time.time() * 1e9)
    pose = Pose(ts_ns=ts_ns, x=0.0, y=0.0, z=0.0, roll=0.0, pitch=0.0, yaw=0.0, vx=0.0, vy=0.0, vz=0.0)
    writers["imu"].append(imu_row(pose, __import__("random").Random(1), ts_ns))
    for w in writers.values():
        w.seal()
    count = duckdb.sql(f"""
        SELECT count(*) FROM read_parquet(
            '{tmp_path}/dataset=telemetry/**/*.parquet',
            hive_partitioning=true, union_by_name=true)
    """).fetchone()[0]
    assert count >= 1
@@ -2,6 +2,7 @@
 from __future__ import annotations
 import os
 import random
 import select
 import time
@@ -108,6 +109,10 @@ def run(cfg: Config) -> None:
        writer.seal()
    print(f"[{cfg.drone_id}] done: {frames_sent} state frames sent, "
          f"{len(peers_seen)} peers seen {sorted(peers_seen)}; sealed to {root}")
    if os.environ.get("KEEP_ALIVE", "0") == "1":
        print(f"[{cfg.drone_id}] KEEP_ALIVE=1 — idle after seal (no pod restart churn)")
        while True:
            time.sleep(3600)
 if __name__ == "__main__":
Author	SHA1	Message	Date
eSlider	ea062af86a	perf: bound lake scans and add platform Grafana dashboard CI & Release / Verify simulator (push) Successful in 11s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 5s Details Explorer and exporter were scanning the full Parquet lake every 1–5s (~2000 files, 200MB+), driving ~2.2 CPU cores. Limit metrics to the last N flights, cache DuckDB views and the partition tree, slow live polls to 2s, and keep drones alive after seal to stop restart churn. Add node-exporter, scan-duration metrics, and a Swarm Platform Grafana dashboard for node CPU/memory and scan health.	2026-07-08 21:19:53 +01:00
eSlider	bcd956d11a	fix: wire live mode and Grafana to the generated data stream CI & Release / Verify simulator (push) Successful in 11s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 5s Details Use the real hive column name (`drone`) in the prototype's live query and pin the Grafana datasource uid to `swarm-prom` so provisioned dashboards resolve their Prometheus source in k3d.	2026-07-08 20:34:37 +01:00
eSlider	9f99d132e3	fix: chown var/t1 and var/t3 to simulator uid 10001 for k3d CI & Release / Verify simulator (push) Successful in 12s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 4s Details Non-root drone pods need write access on the hostPath lake. Ansible now creates var/t1 and var/t3 owned by uid 10001 instead of relying on 0777 alone after legacy root-owned parquet trees.	2026-07-08 20:27:25 +01:00
eSlider	669e8ff005	fix: run simulator image as non-root user (Trivy DS-0002) CI & Release / Verify simulator (push) Successful in 11s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 4s Details Add swarm uid 10001 in both build stages so Trivy config scan passes and containers do not run as root. Mounted /data volumes should be world-writable or owned by uid 10001 (var/t1 from Ansible is 0777).	2026-07-08 20:20:19 +01:00
eSlider	de1c7c9558	chore: stop tracking local Terraform state and provider cache CI & Release / Verify simulator (push) Successful in 10s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 4s Details Keep .terraform.lock.hcl for reproducible provider pins; gitignore state files and the .terraform/ working directory.	2026-07-08 20:18:06 +01:00
eSlider	ad7e0a5cbe	docs: WireGuard vs SSH, Flux GitOps, CONTRIBUTING with TDD-first Document why WireGuard and SSH are complementary layers, add Flux boundaries for the ground segment, shared var/ data paths in docs/06, and a contributor guide with the required test-first workflow.	2026-07-08 20:17:34 +01:00
eSlider	cb4664f5ee	feat: add IaC layer with shared var/t1 and var/t3 data paths Ansible provisions the k3d cluster and Flux controllers; Terraform modules deploy the simulated fleet and ground warehouse. Compose and k3d share var/t1 (live lake) and var/t3 (warehouse). The prototype gains a live mode fed by the explorer read-only SQL API.	2026-07-08 20:17:32 +01:00
eSlider	d58569b25c	test: add pytest suite and gate Docker builds on it Unit tests cover the SQL gate, broadcast codec, Hive writer layout, and flight kinematics. CI runs pytest first; the simulator image build runs tests in a dedicated stage before the final layer.	2026-07-08 20:17:18 +01:00
eSlider	eef6623f39	fix(simulator): disable Grafana phone-home for air-gap hygiene CI & Release / Verify simulator (push) Successful in 7s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 2s Details Grafana was checking grafana.com for updates and sending usage reporting by default - the only outbound traffic in the whole stack. Verified empty latestVersion and no established external connections from any container after the change.	2026-07-08 14:56:43 +01:00
eSlider	f008da1427	feat(prototype): adapt the patrol field to the display aspect ratio CI & Release / Verify simulator (push) Successful in 7s Details CI & Release / Trivy scan (push) Successful in 10s Details CI & Release / Semantic Release (push) Successful in 1s Details The field was a fixed square, leaving wide monitors with dead side margins. Field width now follows the window aspect (capped at 2.2:1, quantized to avoid rebuilds on minor resizes); routes, obstacles and transit spawns stretch with it, and the grid keeps 100 m cells.	2026-07-08 14:51:33 +01:00
eSlider	407a70fef1	feat(prototype): crossing patrol routes and through-transit objects CI & Release / Verify simulator (push) Successful in 7s Details CI & Release / Trivy scan (push) Successful in 9s Details CI & Release / Semantic Release (push) Successful in 2s Details Mobile obstacles used to bounce until they wedged into far corners, and a single perimeter route let the mesh stretch into segments that stopped exchanging data. Even drones now fly the perimeter while odd drones fly an X through the center, keeping bridge links alive, and transit objects enter at one edge, cross the area, and respawn at a fresh edge.	2026-07-08 14:39:57 +01:00