cant wait to do all types of shi on ts hook

also turn all into Verifiable macros pls dont fail github

.env.example

@@ -0,0 +1 @@
+SQLX_OFFLINE=true

.github/plan-observationStoreAndSyncV1.prompt.md

@@ -0,0 +1,184 @@
+# Plan: Observation Store And Sync V1
+
+## Summary
+
+Build device observation support in layers:
+
+1. Local router observations are captured by `wakey observe ...` from OpenWrt hotplug.
+2. `wakey-linux` stores current observed DHCP/neigh state locally.
+3. `wakey-agent` forwards compact observations to the control plane.
+4. The control plane stores observations per agent and later joins them to known devices through `device_identifiers`.
+
+This is separate from durable identity. Observations are facts an agent saw at a time; known devices are manual/user-approved identity records.
+
+## Network/API Boundary
+
+The control plane has two route classes:
+
+- Public agent routes: intended to be reachable by agents.
+- Protected control routes: intended to sit behind Cloudflare Access/admin auth.
+
+Observation upload belongs on the public agent route surface because routers/agents must call it:
+
+```text
+POST /api/v1/agents/observations
+```
+
+The endpoint must require normal agent authentication, using the same persistent `agent_id`/`agent_token` trust model as WebSocket auth. It must not be a Cloudflare Access-only admin endpoint.
+
+Admin/UI read APIs belong under protected control routes:
+
+```text
+GET /api/v1/control/observations
+GET /api/v1/control/devices
+```
+
+## Step 1: Local Observation Store
+
+Replace the narrow `/tmp/wakey_mac_names.json` cache with a local observation store owned by `wakey-linux`.
+
+Minimum local tables/state:
+
+```text
+observed_dhcp_clients:
+  mac
+  ip
+  hostname
+  first_seen_unix
+  last_seen_unix
+  last_action
+
+observed_neighbors:
+  key
+  mac
+  ip
+  first_seen_unix
+  last_seen_unix
+  last_action
+```
+
+Behavior:
+
+- Hotplug scripts stay minimal and call `wakey-agent observe ...`.
+- `wakey-agent observe ...` delegates to `wakey observe ...`.
+- `wakey observe dhcp ...` writes DHCP observations.
+- `wakey observe neigh ...` writes neighbor observations.
+- Existing `wakey leases` and `wakey inventory` still read live sources first, then enrich from the local observation store.
+- Do not make local queries depend exclusively on hotplug events yet.
+
+## Step 2: Control-Plane Observation Tables
+
+Add observation tables early so future API/UI work does not need another major state refactor.
+
+Control-plane schema:
+
+```text
+agent_device_observations:
+  observation_key TEXT PRIMARY KEY
+  agent_id TEXT NOT NULL
+  kind TEXT NOT NULL
+  mac TEXT
+  ip TEXT
+  hostname TEXT
+  first_seen_unix INTEGER NOT NULL
+  last_seen_unix INTEGER NOT NULL
+  last_action TEXT NOT NULL
+
+agent_device_observation_events:
+  event_id TEXT PRIMARY KEY
+  agent_id TEXT NOT NULL
+  kind TEXT NOT NULL
+  action TEXT NOT NULL
+  mac TEXT
+  ip TEXT
+  hostname TEXT
+  ts_unix INTEGER NOT NULL
+```
+
+Keep `agent_device_observation_events` optional in behavior if needed, but create it in schema early. Current-state rows are the primary path; event history is for debugging/audit.
+
+Indexes:
+
+```text
+agent_device_observations(agent_id)
+agent_device_observations(mac)
+agent_device_observations(ip)
+agent_device_observations(hostname)
+agent_device_observations(last_seen_unix)
+agent_device_observation_events(agent_id, ts_unix)
+agent_device_observation_events(mac)
+```
+
+## Step 3: Agent Upload
+
+Add agent upload payload:
+
+```json
+{
+  "observations": [
+    {
+      "kind": "dhcp",
+      "action": "update",
+      "mac": "04:7c:16:79:6d:ee",
+      "ip": "192.168.100.94",
+      "hostname": "lda",
+      "first_seen_unix": 1770000000,
+      "last_seen_unix": 1770000123
+    }
+  ]
+}
+```
+
+Rules:
+
+- Agent may send snapshots periodically and after local observe events.
+- Control plane upserts current-state rows by stable observation key.
+- Observation keys should be deterministic from `agent_id`, `kind`, and the best available identifier:
+  - DHCP with MAC: `agent:{agent_id}:dhcp:mac:{mac}`
+  - Neigh with MAC: `agent:{agent_id}:neigh:mac:{mac}`
+  - Neigh without MAC: `agent:{agent_id}:neigh:ip:{ip}`
+- Uploads must not create known devices automatically.
+
+## Step 4: Join Observations To Known Devices
+
+Known devices already have durable IDs and manual identifiers:
+
+```text
+known_devices
+device_identifiers
+```
+
+Join rule:
+
+```text
+agent_device_observations.mac
+  -> device_identifiers(kind = 'mac', value = mac)
+  -> known_devices.device_id
+```
+
+Unknown observations are rows that do not match any manual `device_identifiers` row.
+
+This enables:
+
+- same known device observed by multiple agents;
+- same known device with multiple MACs;
+- UI flow to attach an unknown observed MAC to an existing known device;
+- wake flows that choose agent-local observed IP/MAC context for a known device.
+
+## Step 5: UI/API Later
+
+After storage and upload exist:
+
+- show known devices with all matching observations grouped by agent;
+- show unknown observations;
+- add action: attach observation identifier to known device;
+- add action: create known device from observation;
+- add wake action from known device using a selected agent/observation.
+
+## Defaults
+
+- `wakey` remains the local router/debugging CLI.
+- `wakey-agent` is the sync bridge to the control plane.
+- `wakey-control-plane` is durable identity and multi-agent view.
+- Local observation store is not authoritative; live sources still matter.
+- Control-plane observations are not durable identity; only manual known-device identifiers are.

.github/plan-sqlxQueryMacros.prompt.md

@@ -0,0 +1,69 @@
+# Plan: SQLx Query Macro Adoption
+
+## Summary
+
+Move control-plane SQL from unchecked `sqlx::query(...)` calls toward `sqlx::query!` / `query_as!` where the SQL shape is static.
+
+This gives compile-time checking for table/column names and Rust value types. Dynamic filter builders can stay on `QueryBuilder`.
+
+## Offline Setup
+
+Normal builds should not need a live database:
+
+```text
+SQLX_OFFLINE=true
+```
+
+`.env` is ignored, so copy:
+
+```sh
+cp .env.example .env
+```
+
+SQLx query metadata should be generated into `.sqlx/` and committed.
+
+Prepare metadata with:
+
+```sh
+cargo install sqlx-cli --no-default-features --features sqlite
+./scripts/prepare_sqlx_sqlite.sh
+```
+
+The script:
+
+- creates a temporary SQLite database;
+- runs `wakey-control-plane/migrations`;
+- runs `cargo sqlx prepare --workspace -- -p wakey-control-plane --all-targets --all-features`;
+- writes `.sqlx/`.
+
+## Conversion Rules
+
+Use `query!` / `query_as!` for static SQL:
+
+- inserts;
+- deletes;
+- simple selects by primary key;
+- fixed update statements;
+- count queries.
+
+Keep non-macro SQL for genuinely dynamic SQL:
+
+- audit event filters built with `QueryBuilder`;
+- observation listing with optional filters, unless split into fixed branches;
+- any SQL where table/column names are intentionally generated.
+
+## Test/CI
+
+After query macros are introduced, CI should add:
+
+```sh
+cargo install sqlx-cli --no-default-features --features sqlite
+./scripts/prepare_sqlx_sqlite.sh
+cargo sqlx prepare --check --workspace -- --all-targets --all-features
+```
+
+Do not add the CI check until the first `.sqlx/` metadata is committed, otherwise it adds dependency install cost without catching anything.
+
+## Notes
+
+The SQLx docs say offline mode needs `cargo sqlx prepare` output checked into version control, and that `DATABASE_URL` takes precedence unless `SQLX_OFFLINE=true` is set. Keep `.env` local; commit `.env.example` and `.sqlx/`.

.sqlx/query-020b33576bfc2d171a051f903d40ccc78c972fa9d5d983a5678f03abbc8b8488.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agents (agent_id, agent_token) VALUES (?1, ?2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "020b33576bfc2d171a051f903d40ccc78c972fa9d5d983a5678f03abbc8b8488"
+}

.sqlx/query-07d7f283daba2742e5522abf99074171ed484c03f70138be611e0c4b17e311b8.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT OR REPLACE INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "07d7f283daba2742e5522abf99074171ed484c03f70138be611e0c4b17e311b8"
+}

.sqlx/query-0c47aade9d5c71d6c7e2023c6ff66c6ab4b3a0dd2b78af3a600dedc89da508f2.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO audit_events\n             (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,\n              event_type, outcome, latency_ms, message, metadata_json)\n             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 12
+    },
+    "nullable": []
+  },
+  "hash": "0c47aade9d5c71d6c7e2023c6ff66c6ab4b3a0dd2b78af3a600dedc89da508f2"
+}

.sqlx/query-0ca58d0cbc25c59d2c956222d41c02eb0c76a135f29a56a7570b750f7378738d.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) as \"count!: i64\" FROM enroll_tokens",
+  "describe": {
+    "columns": [
+      {
+        "name": "count!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "0ca58d0cbc25c59d2c956222d41c02eb0c76a135f29a56a7570b750f7378738d"
+}

.sqlx/query-15da60f9e43fa7098edf7c9fb5becfb518e9b30c5139fdb23283ed302e3da00f.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT device_id as \"device_id!\", display_name as \"display_name!\",\n                    pinned, created_at_unix, updated_at_unix, notes\n             FROM known_devices\n             ORDER BY pinned DESC, display_name, device_id",
+  "describe": {
+    "columns": [
+      {
+        "name": "device_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "display_name!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "pinned",
+        "ordinal": 2,
+        "type_info": "Integer"
+      },
+      {
+        "name": "created_at_unix",
+        "ordinal": 3,
+        "type_info": "Integer"
+      },
+      {
+        "name": "updated_at_unix",
+        "ordinal": 4,
+        "type_info": "Integer"
+      },
+      {
+        "name": "notes",
+        "ordinal": 5,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "15da60f9e43fa7098edf7c9fb5becfb518e9b30c5139fdb23283ed302e3da00f"
+}

.sqlx/query-181992e554afdd8f21bd5df71bfb7880a5a9de28af8cd0c766d8d4aa649b3c90.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_meta (agent_id, nickname) VALUES (?1, ?2)\n                 ON CONFLICT(agent_id) DO UPDATE SET nickname = excluded.nickname",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "181992e554afdd8f21bd5df71bfb7880a5a9de28af8cd0c766d8d4aa649b3c90"
+}

.sqlx/query-21e5986ad3e8efd3393f2bc4929ee79e664cbc2b55948c276d52ca2a6fdc6a48.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) as \"count!: i64\" FROM agents",
+  "describe": {
+    "columns": [
+      {
+        "name": "count!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "21e5986ad3e8efd3393f2bc4929ee79e664cbc2b55948c276d52ca2a6fdc6a48"
+}

.sqlx/query-2a7abe4d0ba22f991f490b45e2edcc980586116e92bac2c5394387b7052abe65.json

@@ -0,0 +1,44 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT identifier_key as \"identifier_key!\", device_id as \"device_id!\",\n                    kind as \"kind!\", value as \"value!\", created_at_unix\n             FROM device_identifiers\n             WHERE device_id = ?1\n             ORDER BY kind, value",
+  "describe": {
+    "columns": [
+      {
+        "name": "identifier_key!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "device_id!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "value!",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "created_at_unix",
+        "ordinal": 4,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "2a7abe4d0ba22f991f490b45e2edcc980586116e92bac2c5394387b7052abe65"
+}

.sqlx/query-2bff8b8b070f5b5dd946e754c5e15ea405f6c29baa4c2c31347f455227437ae2.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO known_devices\n             (device_id, display_name, pinned, created_at_unix, updated_at_unix, notes)\n             VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 6
+    },
+    "nullable": []
+  },
+  "hash": "2bff8b8b070f5b5dd946e754c5e15ea405f6c29baa4c2c31347f455227437ae2"
+}

.sqlx/query-2c0dd5d9be90cc716da219c75c2d3b75dae8c0902c64ef5baa96c902f23abc5d.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT 1 as \"ok!\"",
+  "describe": {
+    "columns": [
+      {
+        "name": "ok!",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "2c0dd5d9be90cc716da219c75c2d3b75dae8c0902c64ef5baa96c902f23abc5d"
+}

.sqlx/query-305f50ab2691e12c74e8a222ef12a16f14a65edd7b7c5400a196fd57a9788dcf.json

@@ -0,0 +1,68 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT observation_key as \"observation_key!\", agent_id as \"agent_id!\",\n                        kind as \"kind!\", mac, ip, hostname,\n                        first_seen_unix, last_seen_unix, last_action as \"last_action!\"\n                 FROM agent_device_observations\n                 WHERE agent_id = ?1\n                 ORDER BY last_seen_unix DESC\n                 LIMIT ?2",
+  "describe": {
+    "columns": [
+      {
+        "name": "observation_key!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "agent_id!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "mac",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "ip",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "hostname",
+        "ordinal": 5,
+        "type_info": "Text"
+      },
+      {
+        "name": "first_seen_unix",
+        "ordinal": 6,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_seen_unix",
+        "ordinal": 7,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_action!",
+        "ordinal": 8,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "305f50ab2691e12c74e8a222ef12a16f14a65edd7b7c5400a196fd57a9788dcf"
+}

.sqlx/query-5083fd5878c57f4d54e8b24ad0cef2771f7cac36afb7c370c65137fc0784f093.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) as \"count!: i64\" FROM enroll_tokens WHERE expires_at_unix <= ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "count!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "5083fd5878c57f4d54e8b24ad0cef2771f7cac36afb7c370c65137fc0784f093"
+}

.sqlx/query-51cb6ec47368803af60114a39dd9e70ba9d31946bc39980c9e85dbc4e842a2c8.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM agents WHERE agent_id = ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "51cb6ec47368803af60114a39dd9e70ba9d31946bc39980c9e85dbc4e842a2c8"
+}

.sqlx/query-52d22161e936b39b57d119aab592426e88f60387ccba58888cc7d440012777ed.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_device_observations\n                 (observation_key, agent_id, kind, mac, ip, hostname,\n                  first_seen_unix, last_seen_unix, last_action)\n                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)\n                 ON CONFLICT(observation_key) DO UPDATE SET\n                   mac = excluded.mac,\n                   ip = excluded.ip,\n                   hostname = excluded.hostname,\n                   first_seen_unix = MIN(agent_device_observations.first_seen_unix, excluded.first_seen_unix),\n                   last_seen_unix = MAX(agent_device_observations.last_seen_unix, excluded.last_seen_unix),\n                   last_action = excluded.last_action",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 9
+    },
+    "nullable": []
+  },
+  "hash": "52d22161e936b39b57d119aab592426e88f60387ccba58888cc7d440012777ed"
+}

.sqlx/query-540f57604aa987b97930bb0bcfe440094d97b769fd758fd5cffc32f6d5716bb4.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT expires_at_unix FROM enroll_tokens WHERE token = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "expires_at_unix",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "540f57604aa987b97930bb0bcfe440094d97b769fd758fd5cffc32f6d5716bb4"
+}

.sqlx/query-5a0f37b5da194e432c795972d5b3bf67c2d2a00f379143d1ae5871ea790d1bf2.json

@@ -0,0 +1,68 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT observation_key as \"observation_key!\", agent_id as \"agent_id!\",\n                        kind as \"kind!\", mac, ip, hostname,\n                        first_seen_unix, last_seen_unix, last_action as \"last_action!\"\n                 FROM agent_device_observations\n                 ORDER BY last_seen_unix DESC\n                 LIMIT ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "observation_key!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "agent_id!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "mac",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "ip",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "hostname",
+        "ordinal": 5,
+        "type_info": "Text"
+      },
+      {
+        "name": "first_seen_unix",
+        "ordinal": 6,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_seen_unix",
+        "ordinal": 7,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_action!",
+        "ordinal": 8,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "5a0f37b5da194e432c795972d5b3bf67c2d2a00f379143d1ae5871ea790d1bf2"
+}

.sqlx/query-60c6631ace995d7ca109d1bee1d878c823edd40f15f8571b42e00680a120ad74.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) as \"count!: i64\" FROM agents WHERE agent_id = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "count!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "60c6631ace995d7ca109d1bee1d878c823edd40f15f8571b42e00680a120ad74"
+}

.sqlx/query-72afd48584f822116318c623f7dd42ddf4c82b337d8a3d52ed5973d52e08c163.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM active_alerts",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": []
+  },
+  "hash": "72afd48584f822116318c623f7dd42ddf4c82b337d8a3d52ed5973d52e08c163"
+}

.sqlx/query-7dc7edbbe5539f89783cbccb0475e676cb6d369b676b0c5aaedf99854ffbd760.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO device_identifiers\n         (identifier_key, device_id, kind, value, created_at_unix)\n         VALUES (?1, ?2, ?3, ?4, ?5)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 5
+    },
+    "nullable": []
+  },
+  "hash": "7dc7edbbe5539f89783cbccb0475e676cb6d369b676b0c5aaedf99854ffbd760"
+}

.sqlx/query-7f6c5c108953cd4eda8f4e1d4f4db6b3b88270d3d7087b2948362d0ea94ad246.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT token as \"token!\", expires_at_unix FROM enroll_tokens ORDER BY expires_at_unix, token",
+  "describe": {
+    "columns": [
+      {
+        "name": "token!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "expires_at_unix",
+        "ordinal": 1,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true,
+      false
+    ]
+  },
+  "hash": "7f6c5c108953cd4eda8f4e1d4f4db6b3b88270d3d7087b2948362d0ea94ad246"
+}

.sqlx/query-8bf8470846bd228d2ea9b243f70f946f002915b9b1983bce818abc0ae3dd97f0.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM agent_meta WHERE agent_id = ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "8bf8470846bd228d2ea9b243f70f946f002915b9b1983bce818abc0ae3dd97f0"
+}

.sqlx/query-8f3499a2f461f738711b9942a9c9d6254c1f7ed0db54db8b3d65e10fa8dadfc9.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM enroll_tokens WHERE token = ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "8f3499a2f461f738711b9942a9c9d6254c1f7ed0db54db8b3d65e10fa8dadfc9"
+}

.sqlx/query-977adb8bdfb9e031a738055fb81c02c6180ca9c37c90e39b3e4efb9168bd1944.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT agents.agent_id as \"agent_id!\", agent_meta.nickname\n             FROM agents\n             LEFT JOIN agent_meta ON agent_meta.agent_id = agents.agent_id\n             ORDER BY agents.agent_id",
+  "describe": {
+    "columns": [
+      {
+        "name": "agent_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "nickname",
+        "ordinal": 1,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "977adb8bdfb9e031a738055fb81c02c6180ca9c37c90e39b3e4efb9168bd1944"
+}

.sqlx/query-9f5d340530083d040794bed2a45b01522f57dc4132184dab0eb32d1b6811992d.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) as \"count!: i64\" FROM meta WHERE key = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "count!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "9f5d340530083d040794bed2a45b01522f57dc4132184dab0eb32d1b6811992d"
+}

.sqlx/query-a1bca2de4e1328fa066d5a8703ef5b8bda560b234b9f1e75093827efeac10eec.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT device_id FROM device_identifiers WHERE identifier_key = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "device_id",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "a1bca2de4e1328fa066d5a8703ef5b8bda560b234b9f1e75093827efeac10eec"
+}

.sqlx/query-a34accb5d8ef1fc90e1029b808956c85fa7cb00c0e879f27bbb99b5a76b5e151.json

@@ -0,0 +1,68 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT transition_id as \"transition_id!\", ts_unix, alert_id as \"alert_id!\",\n                        kind as \"kind!\", agent_id, from_status, to_status as \"to_status!\",\n                        message as \"message!\", metadata_json as \"metadata_json!\"\n                 FROM alert_transitions\n                 ORDER BY transition_key DESC\n                 LIMIT ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "transition_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "ts_unix",
+        "ordinal": 1,
+        "type_info": "Integer"
+      },
+      {
+        "name": "alert_id!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "agent_id",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "from_status",
+        "ordinal": 5,
+        "type_info": "Text"
+      },
+      {
+        "name": "to_status!",
+        "ordinal": 6,
+        "type_info": "Text"
+      },
+      {
+        "name": "message!",
+        "ordinal": 7,
+        "type_info": "Text"
+      },
+      {
+        "name": "metadata_json!",
+        "ordinal": 8,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "a34accb5d8ef1fc90e1029b808956c85fa7cb00c0e879f27bbb99b5a76b5e151"
+}

.sqlx/query-a5baac2d2f6dc38946ff15f6cba46ac085780e6521ba23a605a78b4b92a62073.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT agent_id as \"agent_id!\" FROM agents ORDER BY agent_id",
+  "describe": {
+    "columns": [
+      {
+        "name": "agent_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "a5baac2d2f6dc38946ff15f6cba46ac085780e6521ba23a605a78b4b92a62073"
+}

.sqlx/query-b031d0fa0f9d0640555fc78b400b7bb951a3f013715fb5be51030c2be63e864d.json

@@ -0,0 +1,74 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT alert_id as \"alert_id!\", kind as \"kind!\", severity as \"severity!\",\n                    status as \"status!\", agent_id, message as \"message!\",\n                    value, threshold, last_seen_unix, metadata_json as \"metadata_json!\"\n             FROM active_alerts",
+  "describe": {
+    "columns": [
+      {
+        "name": "alert_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "severity!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "status!",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "agent_id",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "message!",
+        "ordinal": 5,
+        "type_info": "Text"
+      },
+      {
+        "name": "value",
+        "ordinal": 6,
+        "type_info": "Integer"
+      },
+      {
+        "name": "threshold",
+        "ordinal": 7,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_seen_unix",
+        "ordinal": 8,
+        "type_info": "Integer"
+      },
+      {
+        "name": "metadata_json!",
+        "ordinal": 9,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      false,
+      true,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "b031d0fa0f9d0640555fc78b400b7bb951a3f013715fb5be51030c2be63e864d"
+}

.sqlx/query-b1c6150ef7bd67ce78f8934fc5d451b36f5bed834d39bdde01356e331c5e5ddb.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM known_devices WHERE device_id = ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "b1c6150ef7bd67ce78f8934fc5d451b36f5bed834d39bdde01356e331c5e5ddb"
+}

.sqlx/query-b805f8dada38af1c79a7b3bbc063fd949bf81757705ace7173658de9f185868b.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT value FROM meta WHERE key = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "value",
+        "ordinal": 0,
+        "type_info": "Blob"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "b805f8dada38af1c79a7b3bbc063fd949bf81757705ace7173658de9f185868b"
+}

.sqlx/query-bee62978a2ce59095f9bc679258b5747e074a55d52c43b982ea019c79aff1479.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "bee62978a2ce59095f9bc679258b5747e074a55d52c43b982ea019c79aff1479"
+}

.sqlx/query-ca117e134d45464a8930d0483ff8670d2798f38718c67f49adce06f74656bf2c.json

@@ -0,0 +1,68 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT transition_id as \"transition_id!\", ts_unix, alert_id as \"alert_id!\",\n                        kind as \"kind!\", agent_id, from_status, to_status as \"to_status!\",\n                        message as \"message!\", metadata_json as \"metadata_json!\"\n                 FROM alert_transitions\n                 WHERE ts_unix >= ?1\n                 ORDER BY transition_key DESC\n                 LIMIT ?2",
+  "describe": {
+    "columns": [
+      {
+        "name": "transition_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "ts_unix",
+        "ordinal": 1,
+        "type_info": "Integer"
+      },
+      {
+        "name": "alert_id!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "agent_id",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "from_status",
+        "ordinal": 5,
+        "type_info": "Text"
+      },
+      {
+        "name": "to_status!",
+        "ordinal": 6,
+        "type_info": "Text"
+      },
+      {
+        "name": "message!",
+        "ordinal": 7,
+        "type_info": "Text"
+      },
+      {
+        "name": "metadata_json!",
+        "ordinal": 8,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "ca117e134d45464a8930d0483ff8670d2798f38718c67f49adce06f74656bf2c"
+}

.sqlx/query-d2c2ce1761d22477164b3e5cfde9718877b3b472886ae1dc7e64dc36ad75b2ea.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE known_devices SET updated_at_unix = ?1 WHERE device_id = ?2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "d2c2ce1761d22477164b3e5cfde9718877b3b472886ae1dc7e64dc36ad75b2ea"
+}

.sqlx/query-db255767f7f0aae53f685a88a6127a80b89601cdf1163881c5ec07160fbaeeab.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT device_id as \"device_id!\", display_name as \"display_name!\",\n                    pinned, created_at_unix, updated_at_unix, notes\n             FROM known_devices\n             WHERE device_id = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "device_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "display_name!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "pinned",
+        "ordinal": 2,
+        "type_info": "Integer"
+      },
+      {
+        "name": "created_at_unix",
+        "ordinal": 3,
+        "type_info": "Integer"
+      },
+      {
+        "name": "updated_at_unix",
+        "ordinal": 4,
+        "type_info": "Integer"
+      },
+      {
+        "name": "notes",
+        "ordinal": 5,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      false,
+      false,
+      true
+    ]
+  },
+  "hash": "db255767f7f0aae53f685a88a6127a80b89601cdf1163881c5ec07160fbaeeab"
+}

.sqlx/query-e0def02317412d21994af8d7770a9e717f4be97e022c0f2022983d59391c775b.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT agent_token FROM agents WHERE agent_id = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "agent_token",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "e0def02317412d21994af8d7770a9e717f4be97e022c0f2022983d59391c775b"
+}

.sqlx/query-ee6d28dc501f6bb8cf1249ad03b82e3554f1167d5056f02a0cc823601f480e53.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "DELETE FROM enroll_tokens WHERE expires_at_unix <= ?1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "ee6d28dc501f6bb8cf1249ad03b82e3554f1167d5056f02a0cc823601f480e53"
+}

.sqlx/query-f18ab386ab1961e938ae7fec3e752bdd958118aa4ca81cab3421ac5827b04029.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT COUNT(*) as \"count!: i64\" FROM known_devices WHERE device_id = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "count!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "f18ab386ab1961e938ae7fec3e752bdd958118aa4ca81cab3421ac5827b04029"
+}

.sqlx/query-f94cb699c0b4202bc5611799c931ed5d566b3215fcaebdf564e614a417317871.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO agent_device_observation_events\n                 (event_id, agent_id, kind, action, mac, ip, hostname, ts_unix)\n                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 8
+    },
+    "nullable": []
+  },
+  "hash": "f94cb699c0b4202bc5611799c931ed5d566b3215fcaebdf564e614a417317871"
+}

.sqlx/query-ffed939cfeecca1d46bdf77b67aca2919b34d496e729ad08f87ac9782fb8e382.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO meta (key, value) VALUES (?1, ?2)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": []
+  },
+  "hash": "ffed939cfeecca1d46bdf77b67aca2919b34d496e729ad08f87ac9782fb8e382"
+}

Cargo.lock

@@ -3750,6 +3750,7 @@ dependencies = [
  "futures",
  "lda-ipjs",
  "macaddr",
+ "serde",
  "serde_json",
  "tokio",
  "wakey-core",

scripts/prepare_sqlx_sqlite.sh

@@ -0,0 +1,21 @@
+#!/usr/bin/env sh
+set -eu
+
+ROOT="$(CDPATH='' cd -- "$(dirname -- "$0")/.." && pwd)"
+DB="${SQLX_PREPARE_DB:-/tmp/wakey-sqlx-prepare.sqlite3}"
+DATABASE_URL="sqlite://$DB"
+
+if ! cargo sqlx --version >/dev/null 2>&1; then
+  echo "cargo-sqlx is not installed." >&2
+  echo "Install it with: cargo install sqlx-cli --no-default-features --features sqlite" >&2
+  exit 1
+fi
+
+rm -f "$DB" "$DB-shm" "$DB-wal"
+
+cd "$ROOT"
+DATABASE_URL="$DATABASE_URL" cargo sqlx database create
+DATABASE_URL="$DATABASE_URL" cargo sqlx migrate run --source wakey-control-plane/migrations
+DATABASE_URL="$DATABASE_URL" cargo sqlx prepare --workspace -- -p wakey-control-plane --all-targets --all-features
+
+echo "SQLx metadata prepared in $ROOT/.sqlx"

src/lib.rs

@@ -9,7 +9,7 @@ pub use service::{
     inventory, leases_without_state, merge_devices, resolve_devices, resolve_query,
     resolve_selector, resolve_wake_targets, wake_explicit, wake_from_query, wake_targets,
 };
-pub use wakey_linux::dhcp::{observe_dhcp_event, observe_neighbor_event};
+pub use wakey_linux::dhcp::{list_local_observations, observe_dhcp_event, observe_neighbor_event};
 
 #[cfg(test)]
 mod tests {

wakey-agent/src/session.rs

@@ -1,5 +1,6 @@
 use anyhow::{Context, Result};
 use futures_util::{SinkExt, StreamExt};
+use serde::Serialize;
 use std::net::IpAddr;
 use std::time::Instant;
 use tokio::time::{Duration, MissedTickBehavior, interval, sleep};
@@ -84,8 +85,11 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
     .await?;
     info!(agent_id = %config.agent_id, "agent websocket session authenticated");
 
+    let http_client = reqwest::Client::new();
     let mut heartbeat = interval(Duration::from_secs(30));
     heartbeat.set_missed_tick_behavior(MissedTickBehavior::Skip);
+    let mut observation_sync = interval(Duration::from_secs(60));
+    observation_sync.set_missed_tick_behavior(MissedTickBehavior::Skip);
 
     loop {
         tokio::select! {
@@ -95,6 +99,11 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
                 }).await?;
                 debug!(agent_id = %config.agent_id, "heartbeat sent");
             }
+            _ = observation_sync.tick() => {
+                if let Err(err) = send_agent_observations(&http_client, config).await {
+                    warn!(agent_id = %config.agent_id, error = %err, "failed to sync local observations");
+                }
+            }
             maybe_msg = source.next() => {
                 let msg = match maybe_msg {
                     Some(msg) => msg.context("websocket frame failed")?,
@@ -135,6 +144,72 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
     }
 }
 
+#[derive(Debug, Serialize)]
+struct UploadAgentObservationsRequest {
+    agent_id: String,
+    agent_token: String,
+    observations: Vec<AgentObservationRequest>,
+}
+
+#[derive(Debug, Serialize)]
+struct AgentObservationRequest {
+    kind: String,
+    action: String,
+    mac: Option<String>,
+    ip: Option<IpAddr>,
+    hostname: Option<String>,
+    first_seen_unix: u64,
+    last_seen_unix: u64,
+}
+
+async fn send_agent_observations(client: &reqwest::Client, config: &AgentConfig) -> Result<()> {
+    let observations = wakey::list_local_observations()
+        .await
+        .context("failed to read local observations")?;
+    if observations.is_empty() {
+        return Ok(());
+    }
+
+    let url = observations_url(&config.server_url)?;
+    let payload = UploadAgentObservationsRequest {
+        agent_id: config.agent_id.clone(),
+        agent_token: config.agent_token.clone(),
+        observations: observations
+            .into_iter()
+            .map(|observation| AgentObservationRequest {
+                kind: observation.kind,
+                action: observation.action,
+                mac: observation.mac,
+                ip: observation.ip,
+                hostname: observation.hostname,
+                first_seen_unix: observation.first_seen_unix,
+                last_seen_unix: observation.last_seen_unix,
+            })
+            .collect(),
+    };
+
+    let response = client
+        .post(url.clone())
+        .json(&payload)
+        .send()
+        .await
+        .with_context(|| format!("failed to call observation endpoint {url}"))?;
+    if !response.status().is_success() {
+        let status = response.status();
+        let body = response
+            .text()
+            .await
+            .unwrap_or_else(|_| "<unreadable error body>".to_string());
+        anyhow::bail!("observation upload failed with {status}: {body}");
+    }
+    debug!(
+        agent_id = %config.agent_id,
+        observations = payload.observations.len(),
+        "synced local observations"
+    );
+    Ok(())
+}
+
 pub fn next_backoff_ms(current_ms: u64, max_ms: u64) -> u64 {
     let cap = max_ms.max(current_ms);
     current_ms.saturating_mul(2).min(cap)
@@ -273,6 +348,14 @@ pub fn websocket_url(server_url: &str) -> Result<url::Url> {
     Ok(url)
 }
 
+pub fn observations_url(server_url: &str) -> Result<url::Url> {
+    let mut url = url::Url::parse(server_url).context("invalid server_url")?;
+    url.set_path("/api/v1/agents/observations");
+    url.set_query(None);
+    url.set_fragment(None);
+    Ok(url)
+}
+
 async fn dns_resolution_diagnostics(ws_url: &url::Url) -> Option<(u64, usize)> {
     let host = ws_url.host_str()?;
     if host.parse::<IpAddr>().is_ok() {

wakey-control-plane/migrations/20260426000000_initial_sqlite_state.sql

@@ -38,6 +38,38 @@ CREATE TABLE device_identifiers (
 
 CREATE INDEX device_identifiers_device_id_idx ON device_identifiers(device_id);
 
+CREATE TABLE agent_device_observations (
+    observation_key TEXT PRIMARY KEY,
+    agent_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    mac TEXT,
+    ip TEXT,
+    hostname TEXT,
+    first_seen_unix INTEGER NOT NULL,
+    last_seen_unix INTEGER NOT NULL,
+    last_action TEXT NOT NULL
+);
+
+CREATE INDEX agent_device_observations_agent_id_idx ON agent_device_observations(agent_id);
+CREATE INDEX agent_device_observations_mac_idx ON agent_device_observations(mac);
+CREATE INDEX agent_device_observations_ip_idx ON agent_device_observations(ip);
+CREATE INDEX agent_device_observations_hostname_idx ON agent_device_observations(hostname);
+CREATE INDEX agent_device_observations_last_seen_unix_idx ON agent_device_observations(last_seen_unix);
+
+CREATE TABLE agent_device_observation_events (
+    event_id TEXT PRIMARY KEY,
+    agent_id TEXT NOT NULL,
+    kind TEXT NOT NULL,
+    action TEXT NOT NULL,
+    mac TEXT,
+    ip TEXT,
+    hostname TEXT,
+    ts_unix INTEGER NOT NULL
+);
+
+CREATE INDEX agent_device_observation_events_agent_ts_idx ON agent_device_observation_events(agent_id, ts_unix);
+CREATE INDEX agent_device_observation_events_mac_idx ON agent_device_observation_events(mac);
+
 CREATE TABLE audit_events (
     event_key TEXT PRIMARY KEY,
     event_id TEXT NOT NULL UNIQUE,

wakey-control-plane/src/api/control.rs

@@ -7,7 +7,10 @@ use tracing::{info, warn};
 
 use crate::api::json_error;
 use crate::runtime::{AppState, SessionEvent};
-use crate::state::{AuditEventInput, DeviceIdentifierInput, KnownDeviceInput};
+use crate::state::{
+    AgentDeviceObservation, AgentDeviceObservationInput, AuditEventInput, DeviceIdentifierInput,
+    KnownDeviceInput,
+};
 
 #[derive(Debug, Deserialize)]
 pub struct EnrollRequest {
@@ -114,6 +117,36 @@ pub struct ForgetKnownDeviceResponse {
     pub forgotten: bool,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct UploadAgentObservationsRequest {
+    pub agent_id: String,
+    pub agent_token: String,
+    #[serde(default)]
+    pub observations: Vec<AgentObservationRequest>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct AgentObservationRequest {
+    pub kind: String,
+    pub action: String,
+    pub mac: Option<String>,
+    pub ip: Option<String>,
+    pub hostname: Option<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct UploadAgentObservationsResponse {
+    pub accepted: usize,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ListObservationsQuery {
+    pub agent_id: Option<String>,
+    pub limit: Option<usize>,
+}
+
 pub async fn healthz() -> &'static str {
     "ok"
 }
@@ -559,6 +592,85 @@ pub async fn attach_device_identifier(
     }
 }
 
+pub async fn upload_agent_observations(
+    State(state): State<AppState>,
+    Json(req): Json<UploadAgentObservationsRequest>,
+) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
+    if !state
+        .store
+        .verify_agent_token(&req.agent_id, &req.agent_token)
+        .await
+    {
+        return Err(json_error(
+            StatusCode::UNAUTHORIZED,
+            "agent_auth_rejected",
+            "agent credentials rejected",
+        ));
+    }
+
+    let observations = req
+        .observations
+        .into_iter()
+        .map(|observation| AgentDeviceObservationInput {
+            kind: observation.kind,
+            action: observation.action,
+            mac: observation.mac,
+            ip: observation.ip,
+            hostname: observation.hostname,
+            first_seen_unix: observation.first_seen_unix,
+            last_seen_unix: observation.last_seen_unix,
+        })
+        .collect();
+
+    match state
+        .store
+        .upsert_agent_observations(&req.agent_id, observations)
+        .await
+    {
+        Ok(accepted) => Ok((
+            StatusCode::OK,
+            Json(UploadAgentObservationsResponse { accepted }),
+        )),
+        Err(err) => {
+            warn!(error = %err, agent_id = %req.agent_id, "failed to upload agent observations");
+            Err(json_error(
+                StatusCode::BAD_REQUEST,
+                "upload_observations_failed",
+                &err.to_string(),
+            ))
+        }
+    }
+}
+
+pub async fn list_agent_observations(
+    State(state): State<AppState>,
+    Query(query): Query<ListObservationsQuery>,
+) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
+    match state
+        .store
+        .list_agent_observations(query.agent_id.as_deref(), query.limit.unwrap_or(500))
+        .await
+    {
+        Ok(observations) => Ok((
+            StatusCode::OK,
+            Json(
+                observations
+                    .into_iter()
+                    .map(agent_observation_response) // no-op premium
+                    .collect::<Vec<_>>(),
+            ),
+        )),
+        Err(err) => {
+            warn!(error = %err, "failed to list agent observations");
+            Err(json_error(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "list_observations_failed",
+                &err.to_string(),
+            ))
+        }
+    }
+}
+
 pub async fn state_stats(
     State(state): State<AppState>,
 ) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
@@ -584,6 +696,10 @@ pub async fn state_stats(
     }
 }
 
+fn agent_observation_response(observation: AgentDeviceObservation) -> AgentDeviceObservation {
+    observation
+}
+
 fn known_device_response(device: crate::state::KnownDevice) -> KnownDeviceResponse {
     KnownDeviceResponse {
         device_id: device.device_id,

wakey-control-plane/src/api/mod.rs

@@ -12,8 +12,8 @@ pub use commands::{list_agents, run_command};
 pub use control::{
     EnrollTokenStatus, IssueEnrollTokenResponse, RevokeAgentResponse, RevokeEnrollTokenResponse,
     StateStatsResponse, attach_device_identifier, create_known_device, enroll, forget_known_device,
-    healthz, issue_enroll_token, list_enroll_tokens, list_known_devices, revoke_agent,
+    healthz, issue_enroll_token, list_agent_observations, list_enroll_tokens, list_known_devices,
-    revoke_enroll_token, set_agent_nickname, state_stats,
+    revoke_agent, revoke_enroll_token, set_agent_nickname, state_stats, upload_agent_observations,
 };
 
 pub fn json_error(

wakey-control-plane/src/runtime/mod.rs

@@ -71,6 +71,10 @@ fn public_api_routes(ui_dist_dir: std::path::PathBuf) -> Router<AppState> {
         )
         .route("/healthz", get(api::healthz))
         .route("/api/v1/agents/enroll", post(api::enroll))
+        .route(
+            "/api/v1/agents/observations",
+            post(api::upload_agent_observations),
+        )
         .route("/api/v1/agent/ws", get(ws::agent_ws))
 }
 
@@ -89,6 +93,10 @@ fn control_api_routes() -> Router<AppState> {
             axum::routing::delete(api::revoke_enroll_token),
         )
         .route("/api/v1/control/state-stats", get(api::state_stats))
+        .route(
+            "/api/v1/control/observations",
+            get(api::list_agent_observations),
+        )
         .route(
             "/api/v1/control/devices",
             get(api::list_known_devices).post(api::create_known_device),

wakey-control-plane/src/state/mod.rs

@@ -3,6 +3,6 @@ mod types;
 
 pub use store::Store;
 pub use types::{
-    AlertState, AuditEvent, AuditEventFilter, AuditEventInput, DeviceIdentifierInput, KnownDevice,
+    AgentDeviceObservation, AgentDeviceObservationInput, AlertState, AuditEvent, AuditEventFilter,
-    KnownDeviceInput,
+    AuditEventInput, DeviceIdentifierInput, KnownDevice, KnownDeviceInput,
 };

wakey-control-plane/src/state/store.rs

@@ -8,9 +8,9 @@ use tracing::{info, warn};
 use uuid::Uuid;
 
 use crate::state::types::{
-    AlertState, AlertTransition, AuditEvent, AuditEventFilter, AuditEventInput, DeviceIdentifier,
+    AgentDeviceObservation, AgentDeviceObservationInput, AlertState, AlertTransition, AuditEvent,
-    DeviceIdentifierInput, EnrollTokenInfo, IssuedAgent, IssuedEnrollToken, KnownDevice,
+    AuditEventFilter, AuditEventInput, DeviceIdentifier, DeviceIdentifierInput, EnrollTokenInfo,
-    KnownDeviceInput, StateStats,
+    IssuedAgent, IssuedEnrollToken, KnownDevice, KnownDeviceInput, StateStats,
 };
 
 pub struct Store {
@@ -133,10 +133,10 @@ impl Store {
             .begin()
             .await
             .context("failed starting enroll transaction")?;
-        let expires_at_unix = sqlx::query_scalar::<_, i64>(
+        let expires_at_unix = sqlx::query_scalar!(
             "SELECT expires_at_unix FROM enroll_tokens WHERE token = ?1",
+            enroll_token
         )
-        .bind(enroll_token)
         .fetch_optional(&mut *tx)
         .await
         .context("failed reading enroll token")?;
@@ -148,8 +148,7 @@ impl Store {
 
         let now = now_unix();
         if expires_at_unix as u64 <= now {
-            sqlx::query("DELETE FROM enroll_tokens WHERE token = ?1")
+            sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", enroll_token)
-                .bind(enroll_token)
                 .execute(&mut *tx)
                 .await
                 .context("failed removing expired enroll token")?;
@@ -164,8 +163,7 @@ impl Store {
             anyhow::bail!("enroll token has expired");
         }
 
-        sqlx::query("DELETE FROM enroll_tokens WHERE token = ?1")
+        sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", enroll_token)
-            .bind(enroll_token)
             .execute(&mut *tx)
             .await
             .context("failed consuming enroll token")?;
@@ -173,9 +171,11 @@ impl Store {
         let agent_id = format!("agent-{}", Uuid::new_v4());
         let agent_token = format!("tok-{}", Uuid::new_v4());
 
-        sqlx::query("INSERT INTO agents (agent_id, agent_token) VALUES (?1, ?2)")
+        sqlx::query!(
-            .bind(&agent_id)
+            "INSERT INTO agents (agent_id, agent_token) VALUES (?1, ?2)",
-            .bind(&agent_token)
+            agent_id,
+            agent_token
+        )
         .execute(&mut *tx)
         .await
         .context("failed persisting agent credentials")?;
@@ -194,9 +194,13 @@ impl Store {
     pub async fn issue_enroll_token(&self, ttl: Duration) -> Result<IssuedEnrollToken> {
         let token = format!("enr-{}", Uuid::new_v4());
         let expires_at_unix = now_unix().saturating_add(ttl.as_secs().max(1));
-        sqlx::query("INSERT INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)")
+        let expires_at_unix_i64 =
-            .bind(&token)
+            i64::try_from(expires_at_unix).context("expiry does not fit SQLite integer")?;
-            .bind(i64::try_from(expires_at_unix).context("expiry does not fit SQLite integer")?)
+        sqlx::query!(
+            "INSERT INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
+            token,
+            expires_at_unix_i64
+        )
         .execute(&self.pool)
         .await
         .context("failed persisting enroll token")?;
@@ -209,8 +213,9 @@ impl Store {
 
     pub async fn list_enroll_tokens(&self) -> Result<Vec<EnrollTokenInfo>> {
         let now = now_unix();
-        let rows = sqlx::query(
+        let rows = sqlx::query_as!(
-            "SELECT token, expires_at_unix FROM enroll_tokens ORDER BY expires_at_unix, token",
+            EnrollTokenRow,
+            r#"SELECT token as "token!", expires_at_unix FROM enroll_tokens ORDER BY expires_at_unix, token"#,
         )
         .fetch_all(&self.pool)
         .await
@@ -218,12 +223,10 @@ impl Store {
 
         rows.into_iter()
             .map(|row| {
-                let enroll_token: String = row.try_get("token")?;
+                let expires_at_unix = u64::try_from(row.expires_at_unix)
-                let expires_at_unix: i64 = row.try_get("expires_at_unix")?;
+                    .context("negative token expiry in state db")?;
-                let expires_at_unix =
-                    u64::try_from(expires_at_unix).context("negative token expiry in state db")?;
                 Ok(EnrollTokenInfo {
-                    enroll_token,
+                    enroll_token: row.token,
                     expires_at_unix,
                     expired: expires_at_unix <= now,
                 })
@@ -232,8 +235,7 @@ impl Store {
     }
 
     pub async fn revoke_enroll_token(&self, token: &str) -> Result<bool> {
-        let result = sqlx::query("DELETE FROM enroll_tokens WHERE token = ?1")
+        let result = sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", token)
-            .bind(token)
             .execute(&self.pool)
             .await
             .context("failed removing enroll token")?;
@@ -246,14 +248,12 @@ impl Store {
             .begin()
             .await
             .context("failed starting revoke transaction")?;
-        let result = sqlx::query("DELETE FROM agents WHERE agent_id = ?1")
+        let result = sqlx::query!("DELETE FROM agents WHERE agent_id = ?1", agent_id)
-            .bind(agent_id)
             .execute(&mut *tx)
             .await
             .context("failed removing agent credentials")?;
         if result.rows_affected() > 0 {
-            sqlx::query("DELETE FROM agent_meta WHERE agent_id = ?1")
+            sqlx::query!("DELETE FROM agent_meta WHERE agent_id = ?1", agent_id)
-                .bind(agent_id)
                 .execute(&mut *tx)
                 .await
                 .context("failed removing agent metadata")?;
@@ -265,9 +265,10 @@ impl Store {
     }
 
     pub async fn set_agent_nickname(&self, agent_id: &str, nickname: Option<&str>) -> Result<bool> {
-        let exists =
+        let exists = sqlx::query_scalar!(
-            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents WHERE agent_id = ?1")
+            r#"SELECT COUNT(*) as "count!: i64" FROM agents WHERE agent_id = ?1"#,
-                .bind(agent_id)
+            agent_id
+        )
         .fetch_one(&self.pool)
         .await
         .context("failed checking agent existence")?;
@@ -277,18 +278,17 @@ impl Store {
 
         let normalized = nickname.map(str::trim).filter(|v| !v.is_empty());
         if let Some(value) = normalized {
-            sqlx::query(
+            sqlx::query!(
                 "INSERT INTO agent_meta (agent_id, nickname) VALUES (?1, ?2)
                  ON CONFLICT(agent_id) DO UPDATE SET nickname = excluded.nickname",
+                agent_id,
+                value
             )
-            .bind(agent_id)
-            .bind(value)
             .execute(&self.pool)
             .await
             .context("failed persisting agent nickname")?;
         } else {
-            sqlx::query("DELETE FROM agent_meta WHERE agent_id = ?1")
+            sqlx::query!("DELETE FROM agent_meta WHERE agent_id = ?1", agent_id)
-                .bind(agent_id)
                 .execute(&self.pool)
                 .await
                 .context("failed clearing agent nickname")?;
@@ -299,18 +299,19 @@ impl Store {
 
     pub async fn stats(&self) -> Result<StateStats> {
         let now = i64::try_from(now_unix()).context("current time does not fit SQLite integer")?;
-        let enroll_token_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM enroll_tokens")
+        let enroll_token_count =
+            sqlx::query_scalar!(r#"SELECT COUNT(*) as "count!: i64" FROM enroll_tokens"#)
                 .fetch_one(&self.pool)
                 .await
                 .context("failed counting enroll tokens")?;
-        let expired_enroll_token_count = sqlx::query_scalar::<_, i64>(
+        let expired_enroll_token_count = sqlx::query_scalar!(
-            "SELECT COUNT(*) FROM enroll_tokens WHERE expires_at_unix <= ?1",
+            r#"SELECT COUNT(*) as "count!: i64" FROM enroll_tokens WHERE expires_at_unix <= ?1"#,
+            now
         )
-        .bind(now)
         .fetch_one(&self.pool)
         .await
         .context("failed counting expired enroll tokens")?;
-        let agent_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents")
+        let agent_count = sqlx::query_scalar!(r#"SELECT COUNT(*) as "count!: i64" FROM agents"#)
             .fetch_one(&self.pool)
             .await
             .context("failed counting agents")?;
@@ -333,8 +334,8 @@ impl Store {
 
     #[cfg_attr(not(unix), allow(dead_code))]
     pub async fn reload_from_disk(&self) -> Result<()> {
-        sqlx::query("SELECT 1")
+        sqlx::query!(r#"SELECT 1 as "ok!""#)
-            .execute(&self.pool)
+            .fetch_one(&self.pool)
             .await
             .context("failed validating SQLite state connection")?;
         info!(path = %self.db_path.display(), "reload requested; SQLite backend does not require in-memory reload");
@@ -342,8 +343,10 @@ impl Store {
     }
 
     pub async fn verify_agent_token(&self, agent_id: &str, token: &str) -> bool {
-        match sqlx::query_scalar::<_, String>("SELECT agent_token FROM agents WHERE agent_id = ?1")
+        match sqlx::query_scalar!(
-            .bind(agent_id)
+            "SELECT agent_token FROM agents WHERE agent_id = ?1",
+            agent_id
+        )
         .fetch_optional(&self.pool)
         .await
         {
@@ -357,7 +360,7 @@ impl Store {
     }
 
     pub async fn list_agents(&self) -> Vec<String> {
-        match sqlx::query_scalar::<_, String>("SELECT agent_id FROM agents ORDER BY agent_id")
+        match sqlx::query_scalar!(r#"SELECT agent_id as "agent_id!" FROM agents ORDER BY agent_id"#)
             .fetch_all(&self.pool)
             .await
         {
@@ -370,11 +373,11 @@ impl Store {
     }
 
     pub async fn list_agents_with_nicknames(&self) -> Vec<(String, Option<String>)> {
-        let rows = sqlx::query(
+        let rows = sqlx::query!(
-            "SELECT agents.agent_id, agent_meta.nickname
+            r#"SELECT agents.agent_id as "agent_id!", agent_meta.nickname
              FROM agents
              LEFT JOIN agent_meta ON agent_meta.agent_id = agents.agent_id
-             ORDER BY agents.agent_id",
+             ORDER BY agents.agent_id"#,
         )
         .fetch_all(&self.pool)
         .await;
@@ -382,15 +385,12 @@ impl Store {
         match rows {
             Ok(rows) => rows
                 .into_iter()
-                .filter_map(|row| {
+                .map(|row| {
-                    let agent_id: String = row.try_get("agent_id").ok()?;
+                    let nickname = row
-                    let nickname: Option<String> = row
+                        .nickname
-                        .try_get::<Option<String>, _>("nickname")
-                        .ok()
-                        .flatten()
                         .map(|v| v.trim().to_string())
                         .filter(|v| !v.is_empty());
-                    Some((agent_id, nickname))
+                    (row.agent_id, nickname)
                 })
                 .collect(),
             Err(err) => {
@@ -416,17 +416,19 @@ impl Store {
             .begin()
             .await
             .context("failed starting known device transaction")?;
-        sqlx::query(
+        let pinned = if input.pinned { 1_i64 } else { 0_i64 };
+        let now_i64 = i64::try_from(now).context("known device timestamp overflow")?;
+        sqlx::query!(
             "INSERT INTO known_devices
              (device_id, display_name, pinned, created_at_unix, updated_at_unix, notes)
              VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+            device_id,
+            display_name,
+            pinned,
+            now_i64,
+            now_i64,
+            notes
         )
-        .bind(&device_id)
-        .bind(&display_name)
-        .bind(if input.pinned { 1_i64 } else { 0_i64 })
-        .bind(i64::try_from(now).context("known device timestamp overflow")?)
-        .bind(i64::try_from(now).context("known device timestamp overflow")?)
-        .bind(&notes)
         .execute(&mut *tx)
         .await
         .context("failed persisting known device")?;
@@ -444,10 +446,12 @@ impl Store {
     }
 
     pub async fn list_known_devices(&self) -> Result<Vec<KnownDevice>> {
-        let rows = sqlx::query(
+        let rows = sqlx::query_as!(
-            "SELECT device_id, display_name, pinned, created_at_unix, updated_at_unix, notes
+            KnownDeviceRow,
+            r#"SELECT device_id as "device_id!", display_name as "display_name!",
+                    pinned, created_at_unix, updated_at_unix, notes
              FROM known_devices
-             ORDER BY pinned DESC, display_name, device_id",
+             ORDER BY pinned DESC, display_name, device_id"#,
         )
         .fetch_all(&self.pool)
         .await
@@ -455,19 +459,21 @@ impl Store {
 
         let mut out = Vec::with_capacity(rows.len());
         for row in rows {
-            let device_id: String = row.try_get("device_id")?;
+            let device_id = row.device_id.clone();
             out.push(self.known_device_from_row(row, &device_id).await?);
         }
         Ok(out)
     }
 
     pub async fn get_known_device(&self, device_id: &str) -> Result<Option<KnownDevice>> {
-        let row = sqlx::query(
+        let row = sqlx::query_as!(
-            "SELECT device_id, display_name, pinned, created_at_unix, updated_at_unix, notes
+            KnownDeviceRow,
+            r#"SELECT device_id as "device_id!", display_name as "display_name!",
+                    pinned, created_at_unix, updated_at_unix, notes
              FROM known_devices
-             WHERE device_id = ?1",
+             WHERE device_id = ?1"#,
+            device_id
         )
-        .bind(device_id)
         .fetch_optional(&self.pool)
         .await
         .context("failed reading known device")?;
@@ -479,8 +485,7 @@ impl Store {
     }
 
     pub async fn forget_known_device(&self, device_id: &str) -> Result<bool> {
-        let result = sqlx::query("DELETE FROM known_devices WHERE device_id = ?1")
+        let result = sqlx::query!("DELETE FROM known_devices WHERE device_id = ?1", device_id)
-            .bind(device_id)
             .execute(&self.pool)
             .await
             .context("failed deleting known device")?;
@@ -499,9 +504,10 @@ impl Store {
             .begin()
             .await
             .context("failed starting device identifier transaction")?;
-        let exists =
+        let exists = sqlx::query_scalar!(
-            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM known_devices WHERE device_id = ?1")
+            r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#,
-                .bind(device_id)
+            device_id
+        )
         .fetch_one(&mut *tx)
         .await
         .context("failed checking known device existence")?;
@@ -510,9 +516,12 @@ impl Store {
         }
 
         insert_device_identifier_tx(&mut tx, device_id, &identifier, now).await?;
-        sqlx::query("UPDATE known_devices SET updated_at_unix = ?1 WHERE device_id = ?2")
+        let now_i64 = i64::try_from(now).context("known device timestamp overflow")?;
-            .bind(i64::try_from(now).context("known device timestamp overflow")?)
+        sqlx::query!(
-            .bind(device_id)
+            "UPDATE known_devices SET updated_at_unix = ?1 WHERE device_id = ?2",
+            now_i64,
+            device_id
+        )
         .execute(&mut *tx)
         .await
         .context("failed updating known device timestamp")?;
@@ -522,15 +531,16 @@ impl Store {
         self.get_known_device(device_id).await
     }
 
+    #[allow(unused)] // we'll get to this
     pub async fn lookup_known_device_by_identifier(
         &self,
         input: DeviceIdentifierInput,
     ) -> Result<Option<KnownDevice>> {
         let identifier = normalize_device_identifier(input)?;
-        let device_id = sqlx::query_scalar::<_, String>(
+        let device_id = sqlx::query_scalar!(
             "SELECT device_id FROM device_identifiers WHERE identifier_key = ?1",
+            identifier.identifier_key
         )
-        .bind(identifier.identifier_key)
         .fetch_optional(&self.pool)
         .await
         .context("failed looking up known device identifier")?;
@@ -540,6 +550,114 @@ impl Store {
         }
     }
 
+    pub async fn upsert_agent_observations(
+        &self,
+        agent_id: &str,
+        observations: Vec<AgentDeviceObservationInput>,
+    ) -> Result<usize> {
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting observation transaction")?;
+        let mut written = 0usize;
+        for observation in observations {
+            let observation = normalize_agent_observation(agent_id, observation)?;
+            let first_seen_unix = i64::try_from(observation.first_seen_unix)
+                .context("observation first_seen overflow")?;
+            let last_seen_unix = i64::try_from(observation.last_seen_unix)
+                .context("observation last_seen overflow")?;
+            sqlx::query!(
+                "INSERT INTO agent_device_observations
+                 (observation_key, agent_id, kind, mac, ip, hostname,
+                  first_seen_unix, last_seen_unix, last_action)
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
+                 ON CONFLICT(observation_key) DO UPDATE SET
+                   mac = excluded.mac,
+                   ip = excluded.ip,
+                   hostname = excluded.hostname,
+                   first_seen_unix = MIN(agent_device_observations.first_seen_unix, excluded.first_seen_unix),
+                   last_seen_unix = MAX(agent_device_observations.last_seen_unix, excluded.last_seen_unix),
+                   last_action = excluded.last_action",
+                observation.observation_key,
+                observation.agent_id,
+                observation.kind,
+                observation.mac,
+                observation.ip,
+                observation.hostname,
+                first_seen_unix,
+                last_seen_unix,
+                observation.last_action
+            )
+            .execute(&mut *tx)
+            .await
+            .context("failed upserting agent device observation")?;
+
+            let event_id = format!("ode-{}", Uuid::new_v4());
+            sqlx::query!(
+                "INSERT INTO agent_device_observation_events
+                 (event_id, agent_id, kind, action, mac, ip, hostname, ts_unix)
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
+                event_id,
+                observation.agent_id,
+                observation.kind,
+                observation.last_action,
+                observation.mac,
+                observation.ip,
+                observation.hostname,
+                last_seen_unix
+            )
+            .execute(&mut *tx)
+            .await
+            .context("failed appending agent device observation event")?;
+            written = written.saturating_add(1);
+        }
+        tx.commit()
+            .await
+            .context("failed committing observation transaction")?;
+        Ok(written)
+    }
+
+    pub async fn list_agent_observations(
+        &self,
+        agent_id: Option<&str>,
+        limit: usize,
+    ) -> Result<Vec<AgentDeviceObservation>> {
+        let limit = limit.clamp(1, 1000);
+        let limit = i64::try_from(limit).context("observation limit overflow")?;
+        let rows = if let Some(agent_id) = agent_id {
+            sqlx::query_as!(
+                AgentObservationRow,
+                r#"SELECT observation_key as "observation_key!", agent_id as "agent_id!",
+                        kind as "kind!", mac, ip, hostname,
+                        first_seen_unix, last_seen_unix, last_action as "last_action!"
+                 FROM agent_device_observations
+                 WHERE agent_id = ?1
+                 ORDER BY last_seen_unix DESC
+                 LIMIT ?2"#,
+                agent_id,
+                limit
+            )
+            .fetch_all(&self.pool)
+            .await
+        } else {
+            sqlx::query_as!(
+                AgentObservationRow,
+                r#"SELECT observation_key as "observation_key!", agent_id as "agent_id!",
+                        kind as "kind!", mac, ip, hostname,
+                        first_seen_unix, last_seen_unix, last_action as "last_action!"
+                 FROM agent_device_observations
+                 ORDER BY last_seen_unix DESC
+                 LIMIT ?1"#,
+                limit
+            )
+            .fetch_all(&self.pool)
+            .await
+        }
+        .context("failed listing agent observations")?;
+        rows.into_iter().map(agent_observation_from_row).collect()
+    }
+
     pub async fn append_audit_event(&self, input: AuditEventInput) -> Result<AuditEvent> {
         let event = AuditEvent {
             event_id: format!("evt-{}", Uuid::new_v4()),
@@ -558,30 +676,30 @@ impl Store {
         let key = format!("{:020}:{}", event.ts_unix, event.event_id);
         let metadata_json =
             serde_json::to_string(&event.metadata).context("failed to encode audit metadata")?;
-        sqlx::query(
+        let ts_unix = i64::try_from(event.ts_unix).context("audit timestamp overflow")?;
+        let latency_ms = event
+            .latency_ms
+            .map(i64::try_from)
+            .transpose()
+            .context("audit latency overflow")?;
+        sqlx::query!(
             "INSERT INTO audit_events
              (event_key, event_id, ts_unix, actor_type, actor_id, agent_id, request_id,
               event_type, outcome, latency_ms, message, metadata_json)
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
+            key,
+            event.event_id,
+            ts_unix,
+            event.actor_type,
+            event.actor_id,
+            event.agent_id,
+            event.request_id,
+            event.event_type,
+            event.outcome,
+            latency_ms,
+            event.message,
+            metadata_json
         )
-        .bind(&key)
-        .bind(&event.event_id)
-        .bind(i64::try_from(event.ts_unix).context("audit timestamp overflow")?)
-        .bind(&event.actor_type)
-        .bind(&event.actor_id)
-        .bind(&event.agent_id)
-        .bind(&event.request_id)
-        .bind(&event.event_type)
-        .bind(&event.outcome)
-        .bind(
-            event
-                .latency_ms
-                .map(i64::try_from)
-                .transpose()
-                .context("audit latency overflow")?,
-        )
-        .bind(&event.message)
-        .bind(metadata_json)
         .execute(&self.pool)
         .await
         .context("failed persisting audit event")?;
@@ -638,10 +756,12 @@ impl Store {
         current: &[AlertState],
     ) -> Result<Vec<AlertTransition>> {
         let mut previous = std::collections::HashMap::<String, AlertState>::new();
-        let rows = sqlx::query(
+        let rows = sqlx::query_as!(
-            "SELECT alert_id, kind, severity, status, agent_id, message,
+            AlertStateRow,
-                    value, threshold, last_seen_unix, metadata_json
+            r#"SELECT alert_id as "alert_id!", kind as "kind!", severity as "severity!",
-             FROM active_alerts",
+                    status as "status!", agent_id, message as "message!",
+                    value, threshold, last_seen_unix, metadata_json as "metadata_json!"
+             FROM active_alerts"#,
         )
         .fetch_all(&self.pool)
         .await
@@ -698,7 +818,7 @@ impl Store {
             .begin()
             .await
             .context("failed starting alert transaction")?;
-        sqlx::query("DELETE FROM active_alerts")
+        sqlx::query!("DELETE FROM active_alerts")
             .execute(&mut *tx)
             .await
             .context("failed clearing active alert snapshot")?;
@@ -724,28 +844,34 @@ impl Store {
         limit: usize,
     ) -> Result<Vec<AlertTransition>> {
         let limit = limit.clamp(1, 500);
+        let limit = i64::try_from(limit).context("alert limit overflow")?;
         let rows = if let Some(since) = since_unix {
-            sqlx::query(
+            let since = i64::try_from(since).context("since_unix overflow")?;
-                "SELECT transition_id, ts_unix, alert_id, kind, agent_id,
+            sqlx::query_as!(
-                        from_status, to_status, message, metadata_json
+                AlertTransitionRow,
+                r#"SELECT transition_id as "transition_id!", ts_unix, alert_id as "alert_id!",
+                        kind as "kind!", agent_id, from_status, to_status as "to_status!",
+                        message as "message!", metadata_json as "metadata_json!"
                  FROM alert_transitions
                  WHERE ts_unix >= ?1
                  ORDER BY transition_key DESC
-                 LIMIT ?2",
+                 LIMIT ?2"#,
+                since,
+                limit
             )
-            .bind(i64::try_from(since).context("since_unix overflow")?)
-            .bind(i64::try_from(limit).context("alert limit overflow")?)
             .fetch_all(&self.pool)
             .await
         } else {
-            sqlx::query(
+            sqlx::query_as!(
-                "SELECT transition_id, ts_unix, alert_id, kind, agent_id,
+                AlertTransitionRow,
-                        from_status, to_status, message, metadata_json
+                r#"SELECT transition_id as "transition_id!", ts_unix, alert_id as "alert_id!",
+                        kind as "kind!", agent_id, from_status, to_status as "to_status!",
+                        message as "message!", metadata_json as "metadata_json!"
                  FROM alert_transitions
                  ORDER BY transition_key DESC
-                 LIMIT ?1",
+                 LIMIT ?1"#,
+                limit
             )
-            .bind(i64::try_from(limit).context("alert limit overflow")?)
             .fetch_all(&self.pool)
             .await
         }
@@ -766,9 +892,10 @@ impl Store {
             }
 
             let marker_key = seeded_enroll_token_key(token);
-            let marker_exists =
+            let marker_exists = sqlx::query_scalar!(
-                sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM meta WHERE key = ?1")
+                r#"SELECT COUNT(*) as "count!: i64" FROM meta WHERE key = ?1"#,
-                    .bind(&marker_key)
+                marker_key
+            )
             .fetch_one(&self.pool)
             .await
             .context("failed reading bootstrap marker")?;
@@ -782,17 +909,21 @@ impl Store {
                 .begin()
                 .await
                 .context("failed starting bootstrap token transaction")?;
-            sqlx::query(
+            let expires_at_i64 = i64::try_from(expires_at).context("token expiry overflow")?;
+            sqlx::query!(
                 "INSERT OR REPLACE INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
+                token,
+                expires_at_i64
             )
-            .bind(token)
-            .bind(i64::try_from(expires_at).context("token expiry overflow")?)
             .execute(&mut *tx)
             .await
             .context("failed seeding enroll token")?;
-            sqlx::query("INSERT INTO meta (key, value) VALUES (?1, ?2)")
+            let marker_value = expires_at.to_le_bytes().to_vec();
-                .bind(marker_key)
+            sqlx::query!(
-                .bind(expires_at.to_le_bytes().to_vec())
+                "INSERT INTO meta (key, value) VALUES (?1, ?2)",
+                marker_key,
+                marker_value
+            )
             .execute(&mut *tx)
             .await
             .context("failed persisting bootstrap marker")?;
@@ -805,8 +936,7 @@ impl Store {
 
     async fn gc_expired_enroll_tokens_inner(&self) -> Result<u64> {
         let now = i64::try_from(now_unix()).context("current time does not fit SQLite integer")?;
-        let result = sqlx::query("DELETE FROM enroll_tokens WHERE expires_at_unix <= ?1")
+        let result = sqlx::query!("DELETE FROM enroll_tokens WHERE expires_at_unix <= ?1", now)
-            .bind(now)
             .execute(&self.pool)
             .await
             .context("failed removing expired enroll tokens")?;
@@ -818,8 +948,7 @@ impl Store {
     }
 
     async fn ensure_schema_version(&self) -> Result<()> {
-        match sqlx::query_scalar::<_, Vec<u8>>("SELECT value FROM meta WHERE key = ?1")
+        match sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?1", SCHEMA_VERSION_KEY)
-            .bind(SCHEMA_VERSION_KEY)
             .fetch_optional(&self.pool)
             .await
             .context("failed reading schema version")?
@@ -836,9 +965,12 @@ impl Store {
                 }
             }
             None => {
-                sqlx::query("INSERT INTO meta (key, value) VALUES (?1, ?2)")
+                let schema_version = SCHEMA_VERSION.to_le_bytes().to_vec();
-                    .bind(SCHEMA_VERSION_KEY)
+                sqlx::query!(
-                    .bind(SCHEMA_VERSION.to_le_bytes().to_vec())
+                    "INSERT INTO meta (key, value) VALUES (?1, ?2)",
+                    SCHEMA_VERSION_KEY,
+                    schema_version
+                )
                 .execute(&self.pool)
                 .await
                 .context("failed writing schema version")?;
@@ -852,8 +984,7 @@ impl Store {
     }
 
     async fn schema_version(&self) -> Result<u32> {
-        let raw = sqlx::query_scalar::<_, Vec<u8>>("SELECT value FROM meta WHERE key = ?1")
+        let raw = sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?1", SCHEMA_VERSION_KEY)
-            .bind(SCHEMA_VERSION_KEY)
             .fetch_one(&self.pool)
             .await
             .context("failed reading schema version")?;
@@ -862,34 +993,33 @@ impl Store {
 
     async fn known_device_from_row(
         &self,
-        row: sqlx::sqlite::SqliteRow,
+        row: KnownDeviceRow,
         device_id: &str,
     ) -> Result<KnownDevice> {
-        let pinned: i64 = row.try_get("pinned")?;
-        let created_at_unix: i64 = row.try_get("created_at_unix")?;
-        let updated_at_unix: i64 = row.try_get("updated_at_unix")?;
         let identifiers = self.list_device_identifiers(device_id).await?;
         Ok(KnownDevice {
-            device_id: row.try_get("device_id")?,
+            device_id: row.device_id,
-            display_name: row.try_get("display_name")?,
+            display_name: row.display_name,
-            pinned: pinned != 0,
+            pinned: row.pinned != 0,
-            created_at_unix: u64::try_from(created_at_unix)
+            created_at_unix: u64::try_from(row.created_at_unix)
                 .context("negative known device created timestamp in state db")?,
-            updated_at_unix: u64::try_from(updated_at_unix)
+            updated_at_unix: u64::try_from(row.updated_at_unix)
                 .context("negative known device updated timestamp in state db")?,
-            notes: row.try_get("notes")?,
+            notes: row.notes,
             identifiers,
         })
     }
 
     async fn list_device_identifiers(&self, device_id: &str) -> Result<Vec<DeviceIdentifier>> {
-        let rows = sqlx::query(
+        let rows = sqlx::query_as!(
-            "SELECT identifier_key, device_id, kind, value, created_at_unix
+            DeviceIdentifierRow,
+            r#"SELECT identifier_key as "identifier_key!", device_id as "device_id!",
+                    kind as "kind!", value as "value!", created_at_unix
              FROM device_identifiers
              WHERE device_id = ?1
-             ORDER BY kind, value",
+             ORDER BY kind, value"#,
+            device_id
         )
-        .bind(device_id)
         .fetch_all(&self.pool)
         .await
         .context("failed listing device identifiers")?;
@@ -904,6 +1034,65 @@ struct NormalizedDeviceIdentifier {
     value: String,
 }
 
+struct EnrollTokenRow {
+    token: String,
+    expires_at_unix: i64,
+}
+
+struct KnownDeviceRow {
+    device_id: String,
+    display_name: String,
+    pinned: i64,
+    created_at_unix: i64,
+    updated_at_unix: i64,
+    notes: Option<String>,
+}
+
+struct DeviceIdentifierRow {
+    identifier_key: String,
+    device_id: String,
+    kind: String,
+    value: String,
+    created_at_unix: i64,
+}
+
+struct AgentObservationRow {
+    observation_key: String,
+    agent_id: String,
+    kind: String,
+    mac: Option<String>,
+    ip: Option<String>,
+    hostname: Option<String>,
+    first_seen_unix: i64,
+    last_seen_unix: i64,
+    last_action: String,
+}
+
+struct AlertStateRow {
+    alert_id: String,
+    kind: String,
+    severity: String,
+    status: String,
+    agent_id: Option<String>,
+    message: String,
+    value: i64,
+    threshold: i64,
+    last_seen_unix: i64,
+    metadata_json: String,
+}
+
+struct AlertTransitionRow {
+    transition_id: String,
+    ts_unix: i64,
+    alert_id: String,
+    kind: String,
+    agent_id: Option<String>,
+    from_status: Option<String>,
+    to_status: String,
+    message: String,
+    metadata_json: String,
+}
+
 async fn open_sqlite_pool(path: &Path) -> Result<SqlitePool> {
     let options = SqliteConnectOptions::new()
         .filename(path)
@@ -981,22 +1170,52 @@ fn normalize_device_identifier(input: DeviceIdentifierInput) -> Result<Normalize
     })
 }
 
+fn normalize_agent_observation(
+    agent_id: &str,
+    input: AgentDeviceObservationInput,
+) -> Result<AgentDeviceObservation> {
+    let kind = normalize_required_text(&input.kind, "observation kind")?.to_ascii_lowercase();
+    let action = normalize_required_text(&input.action, "observation action")?.to_ascii_lowercase();
+    let mac = normalize_optional_text(input.mac.as_deref()).map(|value| value.to_ascii_lowercase());
+    let ip = normalize_optional_text(input.ip.as_deref());
+    let hostname = normalize_optional_text(input.hostname.as_deref());
+    let identifier = mac
+        .as_ref()
+        .map(|value| format!("mac:{value}"))
+        .or_else(|| ip.as_ref().map(|value| format!("ip:{value}")))
+        .ok_or_else(|| anyhow::anyhow!("observation requires mac or ip"))?;
+    let observation_key = format!("agent:{agent_id}:{kind}:{identifier}");
+    Ok(AgentDeviceObservation {
+        observation_key,
+        agent_id: agent_id.to_string(),
+        kind,
+        mac,
+        ip,
+        hostname,
+        first_seen_unix: input.first_seen_unix,
+        last_seen_unix: input.last_seen_unix,
+        last_action: action,
+    })
+}
+
 async fn insert_device_identifier_tx(
     tx: &mut Transaction<'_, Sqlite>,
     device_id: &str,
     identifier: &NormalizedDeviceIdentifier,
     created_at_unix: u64,
 ) -> Result<()> {
-    sqlx::query(
+    let created_at_unix =
+        i64::try_from(created_at_unix).context("device identifier timestamp overflow")?;
+    sqlx::query!(
         "INSERT INTO device_identifiers
          (identifier_key, device_id, kind, value, created_at_unix)
          VALUES (?1, ?2, ?3, ?4, ?5)",
+        identifier.identifier_key,
+        device_id,
+        identifier.kind,
+        identifier.value,
+        created_at_unix
     )
-    .bind(&identifier.identifier_key)
-    .bind(device_id)
-    .bind(&identifier.kind)
-    .bind(&identifier.value)
-    .bind(i64::try_from(created_at_unix).context("device identifier timestamp overflow")?)
     .execute(&mut **tx)
     .await
     .with_context(|| {
@@ -1008,18 +1227,33 @@ async fn insert_device_identifier_tx(
     Ok(())
 }
 
-fn device_identifier_from_row(row: sqlx::sqlite::SqliteRow) -> Result<DeviceIdentifier> {
+fn device_identifier_from_row(row: DeviceIdentifierRow) -> Result<DeviceIdentifier> {
-    let created_at_unix: i64 = row.try_get("created_at_unix")?;
     Ok(DeviceIdentifier {
-        identifier_key: row.try_get("identifier_key")?,
+        identifier_key: row.identifier_key,
-        device_id: row.try_get("device_id")?,
+        device_id: row.device_id,
-        kind: row.try_get("kind")?,
+        kind: row.kind,
-        value: row.try_get("value")?,
+        value: row.value,
-        created_at_unix: u64::try_from(created_at_unix)
+        created_at_unix: u64::try_from(row.created_at_unix)
             .context("negative device identifier timestamp in state db")?,
     })
 }
 
+fn agent_observation_from_row(row: AgentObservationRow) -> Result<AgentDeviceObservation> {
+    Ok(AgentDeviceObservation {
+        observation_key: row.observation_key,
+        agent_id: row.agent_id,
+        kind: row.kind,
+        mac: row.mac,
+        ip: row.ip,
+        hostname: row.hostname,
+        first_seen_unix: u64::try_from(row.first_seen_unix)
+            .context("negative observation first_seen timestamp in state db")?,
+        last_seen_unix: u64::try_from(row.last_seen_unix)
+            .context("negative observation last_seen timestamp in state db")?,
+        last_action: row.last_action,
+    })
+}
+
 async fn import_tree_raw(
     pool: &SqlitePool,
     legacy: &sled::Db,
@@ -1320,41 +1554,35 @@ fn audit_event_from_row(row: sqlx::sqlite::SqliteRow) -> Result<AuditEvent> {
     })
 }
 
-fn alert_state_from_row(row: sqlx::sqlite::SqliteRow) -> Result<AlertState> {
+fn alert_state_from_row(row: AlertStateRow) -> Result<AlertState> {
-    let value: i64 = row.try_get("value")?;
-    let threshold: i64 = row.try_get("threshold")?;
-    let last_seen_unix: i64 = row.try_get("last_seen_unix")?;
-    let metadata_json: String = row.try_get("metadata_json")?;
     Ok(AlertState {
-        alert_id: row.try_get("alert_id")?,
+        alert_id: row.alert_id,
-        kind: row.try_get("kind")?,
+        kind: row.kind,
-        severity: row.try_get("severity")?,
+        severity: row.severity,
-        status: row.try_get("status")?,
+        status: row.status,
-        agent_id: row.try_get("agent_id")?,
+        agent_id: row.agent_id,
-        message: row.try_get("message")?,
+        message: row.message,
-        value: u64::try_from(value).context("negative active alert value in state db")?,
+        value: u64::try_from(row.value).context("negative active alert value in state db")?,
-        threshold: u64::try_from(threshold)
+        threshold: u64::try_from(row.threshold)
             .context("negative active alert threshold in state db")?,
-        last_seen_unix: u64::try_from(last_seen_unix)
+        last_seen_unix: u64::try_from(row.last_seen_unix)
             .context("negative active alert timestamp in state db")?,
-        metadata: serde_json::from_str(&metadata_json)
+        metadata: serde_json::from_str(&row.metadata_json)
             .context("failed decoding active alert metadata")?,
     })
 }
 
-fn alert_transition_from_row(row: sqlx::sqlite::SqliteRow) -> Result<AlertTransition> {
+fn alert_transition_from_row(row: AlertTransitionRow) -> Result<AlertTransition> {
-    let ts_unix: i64 = row.try_get("ts_unix")?;
-    let metadata_json: String = row.try_get("metadata_json")?;
     Ok(AlertTransition {
-        transition_id: row.try_get("transition_id")?,
+        transition_id: row.transition_id,
-        ts_unix: u64::try_from(ts_unix).context("negative alert timestamp in state db")?,
+        ts_unix: u64::try_from(row.ts_unix).context("negative alert timestamp in state db")?,
-        alert_id: row.try_get("alert_id")?,
+        alert_id: row.alert_id,
-        kind: row.try_get("kind")?,
+        kind: row.kind,
-        agent_id: row.try_get("agent_id")?,
+        agent_id: row.agent_id,
-        from_status: row.try_get("from_status")?,
+        from_status: row.from_status,
-        to_status: row.try_get("to_status")?,
+        to_status: row.to_status,
-        message: row.try_get("message")?,
+        message: row.message,
-        metadata: serde_json::from_str(&metadata_json)
+        metadata: serde_json::from_str(&row.metadata_json)
             .context("failed decoding alert transition metadata")?,
     })
 }
@@ -1659,6 +1887,38 @@ mod tests {
         cleanup_dir(&dir);
     }
 
+    #[tokio::test]
+    async fn agent_observations_upsert_current_state_and_events() {
+        let (store, dir) = make_store().await;
+
+        let accepted = store
+            .upsert_agent_observations(
+                "agent-a",
+                vec![crate::state::AgentDeviceObservationInput {
+                    kind: "dhcp".into(),
+                    action: "update".into(),
+                    mac: Some("AA:BB:CC:DD:EE:FF".into()),
+                    ip: Some("192.168.1.10".into()),
+                    hostname: Some("lda".into()),
+                    first_seen_unix: 10,
+                    last_seen_unix: 20,
+                }],
+            )
+            .await
+            .expect("observation upsert should succeed");
+        assert_eq!(accepted, 1);
+
+        let rows = store
+            .list_agent_observations(Some("agent-a"), 10)
+            .await
+            .expect("observations should list");
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].mac.as_deref(), Some("aa:bb:cc:dd:ee:ff"));
+        assert_eq!(rows[0].hostname.as_deref(), Some("lda"));
+
+        cleanup_dir(&dir);
+    }
+
     #[tokio::test]
     async fn audit_events_append_and_filter() {
         let (store, dir) = make_store().await;

wakey-control-plane/src/state/types.rs

@@ -64,6 +64,30 @@ pub struct DeviceIdentifierInput {
     pub value: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AgentDeviceObservation {
+    pub observation_key: String,
+    pub agent_id: String,
+    pub kind: String,
+    pub mac: Option<String>,
+    pub ip: Option<String>,
+    pub hostname: Option<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+    pub last_action: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct AgentDeviceObservationInput {
+    pub kind: String,
+    pub action: String,
+    pub mac: Option<String>,
+    pub ip: Option<String>,
+    pub hostname: Option<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AuditEvent {
     pub event_id: String,

wakey-linux/Cargo.toml

@@ -8,6 +8,7 @@ publish = ["gitea"]
 anyhow = "1"
 futures = "0"
 macaddr = { version = "1", features = ["serde", "serde_std"] }
+serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1", features = ["fs", "net", "rt", "sync"] }
 wakey-core = { path = "../wakey-core", registry = "gitea", version = "0"}

wakey-linux/src/dhcp.rs

@@ -1,10 +1,52 @@
 use std::io::{self, ErrorKind};
 use std::net::IpAddr;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use macaddr::MacAddr;
+use serde::{Deserialize, Serialize};
 use wakey_core::{DhcpLease, DhcpLeaseWithState};
 
 const MAC_NAME_CACHE: &str = "/tmp/wakey_mac_names.json";
+const OBSERVATION_STORE: &str = "/tmp/wakey_observations.json";
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct LocalObservationStore {
+    #[serde(default)]
+    pub dhcp_clients: std::collections::BTreeMap<String, ObservedDhcpClient>,
+    #[serde(default)]
+    pub neighbors: std::collections::BTreeMap<String, ObservedNeighbor>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ObservedDhcpClient {
+    pub mac: String,
+    pub ip: Option<IpAddr>,
+    pub hostname: Option<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+    pub last_action: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ObservedNeighbor {
+    pub key: String,
+    pub mac: Option<String>,
+    pub ip: Option<IpAddr>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+    pub last_action: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct LocalDeviceObservation {
+    pub kind: String,
+    pub action: String,
+    pub mac: Option<String>,
+    pub ip: Option<IpAddr>,
+    pub hostname: Option<String>,
+    pub first_seen_unix: u64,
+    pub last_seen_unix: u64,
+}
 
 /// Load the MAC-to-name cache used to preserve useful names across lease churn.
 pub async fn load_mac_name_cache() -> io::Result<std::collections::BTreeMap<String, String>> {
@@ -22,43 +64,173 @@ async fn save_mac_name_cache(map: &std::collections::BTreeMap<String, String>) -
     Ok(())
 }
 
+pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
+    match tokio::fs::read_to_string(OBSERVATION_STORE).await {
+        Ok(s) => serde_json::from_str(&s).map_err(io::Error::other),
+        Err(e) if e.kind() == ErrorKind::NotFound => Ok(Default::default()),
+        Err(e) => Err(e),
+    }
+}
+
+async fn save_observation_store(store: &LocalObservationStore) -> io::Result<()> {
+    let s = serde_json::to_string(store).map_err(io::Error::other)?;
+    tokio::fs::write(OBSERVATION_STORE, s).await
+}
+
+pub async fn list_local_observations() -> io::Result<Vec<LocalDeviceObservation>> {
+    let store = load_observation_store().await?;
+    let mut out = Vec::with_capacity(store.dhcp_clients.len() + store.neighbors.len());
+    out.extend(
+        store
+            .dhcp_clients
+            .into_values()
+            .map(|row| LocalDeviceObservation {
+                kind: "dhcp".into(),
+                action: row.last_action,
+                mac: Some(row.mac),
+                ip: row.ip,
+                hostname: row.hostname,
+                first_seen_unix: row.first_seen_unix,
+                last_seen_unix: row.last_seen_unix,
+            }),
+    );
+    out.extend(
+        store
+            .neighbors
+            .into_values()
+            .map(|row| LocalDeviceObservation {
+                kind: "neigh".into(),
+                action: row.last_action,
+                mac: row.mac,
+                ip: row.ip,
+                hostname: None,
+                first_seen_unix: row.first_seen_unix,
+                last_seen_unix: row.last_seen_unix,
+            }),
+    );
+    out.sort_by(|a, b| {
+        b.last_seen_unix
+            .cmp(&a.last_seen_unix)
+            .then(a.kind.cmp(&b.kind))
+            .then(a.mac.cmp(&b.mac))
+            .then(a.ip.cmp(&b.ip))
+    });
+    Ok(out)
+}
+
 /// Observe a DHCP hotplug event and update the local MAC-to-name cache.
 pub async fn observe_dhcp_event(
     action: &str,
     mac: MacAddr,
-    _ip: Option<IpAddr>,
+    ip: Option<IpAddr>,
     hostname: Option<&str>,
 ) -> io::Result<bool> {
-    if !matches!(action, "add" | "update" | "old") {
+    if !matches!(action, "add" | "update" | "old" | "remove") {
         return Ok(false);
     }
 
-    let Some(hostname) = hostname
+    let hostname = hostname
         .map(str::trim)
         .filter(|v| !v.is_empty() && *v != "*")
+        .map(ToOwned::to_owned);
+    let now = now_unix();
+    let mac_s = mac.to_string();
+
+    let mut store = load_observation_store().await.unwrap_or_default();
+    let mut changed = false;
+    store
+        .dhcp_clients
+        .entry(mac_s.clone())
+        .and_modify(|row| {
+            if row.ip != ip
+                || row.hostname != hostname
+                || row.last_action != action
+                || row.last_seen_unix != now
+            {
+                row.ip = ip;
+                row.hostname = hostname.clone();
+                row.last_action = action.to_string();
+                row.last_seen_unix = now;
+                changed = true;
+            }
+        })
+        .or_insert_with(|| {
+            changed = true;
+            ObservedDhcpClient {
+                mac: mac_s.clone(),
+                ip,
+                hostname: hostname.clone(),
+                first_seen_unix: now,
+                last_seen_unix: now,
+                last_action: action.to_string(),
+            }
+        });
+    if changed {
+        save_observation_store(&store).await?;
+    }
+
+    if let Some(hostname) = hostname {
+        let mut cache = load_mac_name_cache().await.unwrap_or_default();
+        if cache.get(&mac_s).map(|v| v != &hostname).unwrap_or(true) {
+            cache.insert(mac_s, hostname);
+            save_mac_name_cache(&cache).await?;
+            changed = true;
+        }
+    }
+
+    Ok(changed)
+}
+
+pub async fn observe_neighbor_event(
+    action: &str,
+    mac: Option<MacAddr>,
+    ip: Option<IpAddr>,
+) -> io::Result<bool> {
+    if !matches!(action, "add" | "update" | "old" | "remove") {
+        return Ok(false);
+    }
+    let Some(key) = mac
+        .map(|value| format!("mac:{}", value))
+        .or_else(|| ip.map(|value| format!("ip:{}", value)))
     else {
         return Ok(false);
     };
 
-    let mut cache = load_mac_name_cache().await.unwrap_or_default();
+    let now = now_unix();
-    let mac_s = mac.to_string();
+    let mac = mac.map(|value| value.to_string());
-    if cache.get(&mac_s).map(|v| v == hostname).unwrap_or(false) {
+    let mut store = load_observation_store().await.unwrap_or_default();
-        return Ok(false);
+    let mut changed = false;
+    store
+        .neighbors
+        .entry(key.clone())
+        .and_modify(|row| {
+            if row.mac != mac
+                || row.ip != ip
+                || row.last_action != action
+                || row.last_seen_unix != now
+            {
+                row.mac = mac.clone();
+                row.ip = ip;
+                row.last_action = action.to_string();
+                row.last_seen_unix = now;
+                changed = true;
             }
-
+        })
-    cache.insert(mac_s, hostname.to_string());
+        .or_insert_with(|| {
-    save_mac_name_cache(&cache).await?;
+            changed = true;
-    Ok(true)
+            ObservedNeighbor {
-}
+                key,
-
+                mac,
-/// Observe a neighbor hotplug event. This is currently a no-op placeholder for
+                ip,
-/// keeping DHCP and neighbor hook commands symmetrical.
+                first_seen_unix: now,
-pub async fn observe_neighbor_event(
+                last_seen_unix: now,
-    _action: &str,
+                last_action: action.to_string(),
-    _mac: Option<MacAddr>,
+            }
-    _ip: Option<IpAddr>,
+        });
-) -> io::Result<bool> {
+    if changed {
-    Ok(false)
+        save_observation_store(&store).await?;
+    }
+    Ok(changed)
 }
 
 /// Parse one `dnsmasq`-style DHCP lease line.
@@ -88,6 +260,7 @@ pub async fn read_dhcp_leases() -> io::Result<Vec<DhcpLease>> {
 /// Read DHCP leases and fill missing names from the MAC-name cache.
 pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
     let leases = read_dhcp_leases().await?;
+    let observations = load_observation_store().await.unwrap_or_default();
     let mut cache = load_mac_name_cache().await.unwrap_or_default();
     let mut changed = false;
     let mut leases_with_names = Vec::with_capacity(leases.len());
@@ -98,6 +271,12 @@ pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
                 cache.insert(mac_s, name.clone());
                 changed = true;
             }
+        } else if let Some(prev) = observations
+            .dhcp_clients
+            .get(&mac_s)
+            .and_then(|row| row.hostname.as_ref())
+        {
+            l.name = Some(prev.clone());
         } else if let Some(prev) = cache.get(&mac_s) {
             l.name = Some(prev.clone());
         }
@@ -109,6 +288,13 @@ pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
     Ok(leases_with_names)
 }
 
+fn now_unix() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_secs())
+        .unwrap_or(0)
+}
+
 /// Enrich DHCP leases with the best currently known neighbor state per IP.
 pub async fn enrich_leases_with_nud_state(leases: Vec<DhcpLease>) -> Vec<DhcpLeaseWithState> {
     let ips: Vec<IpAddr> = leases.iter().map(|l| l.ip).collect();