api! and some more wicked ahh sql shi 100% made by codex

Co-authored-by: Copilot <[email protected]>

.cargo/config.toml

@@ -16,6 +16,10 @@ cw = "c --target=x86_64-pc-windows-msvc"
 bw = "b --target=x86_64-pc-windows-msvc"
 tw = "t --target=x86_64-pc-windows-msvc"
 
+check-all = "check --workspace --all-features --all-targets"
+clippy-all = "clippy --workspace --all-features --all-targets"
+fmt-all = "fmt --all"
+
 # idk below
 ldabr = "b -r --target=armv7-unknown-linux-musleabihf"
 t = "test -- --nocapture --test-threads=1"

.sqlx/query-0a7efadab5487c6ebbf159b4fdd842faac795fe24901756ae67303acc8343e19.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": "0a7efadab5487c6ebbf159b4fdd842faac795fe24901756ae67303acc8343e19"
+}

.sqlx/query-190ea5014566f84255a8655599e03695fb2f5ce9b720700f93d625c1ae2780e6.json

@@ -0,0 +1,86 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT observations.observation_key as \"observation_key!\",\n                        observations.agent_id as \"agent_id!\",\n                        observations.kind as \"kind!\",\n                        observations.mac,\n                        observations.ip,\n                        observations.hostname,\n                        observations.first_seen_unix,\n                        observations.last_seen_unix,\n                        observations.last_action as \"last_action!\",\n                        known_devices.device_id,\n                        known_devices.display_name,\n                        known_devices.pinned\n                 FROM agent_device_observations observations\n                 LEFT JOIN device_identifiers identifiers\n                   ON identifiers.identifier_key =\n                      CASE\n                        WHEN observations.mac IS NOT NULL THEN 'mac:' || observations.mac\n                        WHEN observations.ip IS NOT NULL THEN 'ip:' || observations.ip\n                      END\n                 LEFT JOIN known_devices ON known_devices.device_id = identifiers.device_id\n                 WHERE observations.agent_id = ?1\n                 ORDER BY observations.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"
+      },
+      {
+        "name": "device_id",
+        "ordinal": 9,
+        "type_info": "Text"
+      },
+      {
+        "name": "display_name",
+        "ordinal": 10,
+        "type_info": "Text"
+      },
+      {
+        "name": "pinned",
+        "ordinal": 11,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 2
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      false,
+      true,
+      false,
+      false
+    ]
+  },
+  "hash": "190ea5014566f84255a8655599e03695fb2f5ce9b720700f93d625c1ae2780e6"
+}

.sqlx/query-19b5fc20ffc99589bbaf11b94661f7c0dfcf80748c60f8f3289bb3296114635a.json

@@ -46,5 +46,5 @@
       true
     ]
   },
-  "hash": "15da60f9e43fa7098edf7c9fb5becfb518e9b30c5139fdb23283ed302e3da00f"
+  "hash": "19b5fc20ffc99589bbaf11b94661f7c0dfcf80748c60f8f3289bb3296114635a"
 }

.sqlx/query-3b06698566cea17eed9d10a389ee12a95cee2db6c3d450dca7a3b2d08e8fed7b.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT mac, ip\n         FROM agent_device_observations\n         WHERE observation_key = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "mac",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "ip",
+        "ordinal": 1,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      true
+    ]
+  },
+  "hash": "3b06698566cea17eed9d10a389ee12a95cee2db6c3d450dca7a3b2d08e8fed7b"
+}

.sqlx/query-409e1e7aa2d9bc46030001e5c522d7c2812b4583dceadfe5c535a866ac76489c.json

@@ -0,0 +1,86 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT observations.observation_key as \"observation_key!\",\n                        observations.agent_id as \"agent_id!\",\n                        observations.kind as \"kind!\",\n                        observations.mac,\n                        observations.ip,\n                        observations.hostname,\n                        observations.first_seen_unix,\n                        observations.last_seen_unix,\n                        observations.last_action as \"last_action!\",\n                        known_devices.device_id,\n                        known_devices.display_name,\n                        known_devices.pinned\n                 FROM agent_device_observations observations\n                 LEFT JOIN device_identifiers identifiers\n                   ON identifiers.identifier_key =\n                      CASE\n                        WHEN observations.mac IS NOT NULL THEN 'mac:' || observations.mac\n                        WHEN observations.ip IS NOT NULL THEN 'ip:' || observations.ip\n                      END\n                 LEFT JOIN known_devices ON known_devices.device_id = identifiers.device_id\n                 ORDER BY observations.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"
+      },
+      {
+        "name": "device_id",
+        "ordinal": 9,
+        "type_info": "Text"
+      },
+      {
+        "name": "display_name",
+        "ordinal": 10,
+        "type_info": "Text"
+      },
+      {
+        "name": "pinned",
+        "ordinal": 11,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      false,
+      false,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "409e1e7aa2d9bc46030001e5c522d7c2812b4583dceadfe5c535a866ac76489c"
+}

.sqlx/query-908d763909c76fb6170b818784e360372c02295523ef9970c91199db7d4f535a.json

@@ -0,0 +1,86 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT events.event_id as \"event_id!\",\n                    ('agent:' || events.agent_id || ':' || events.kind || ':' ||\n                        CASE\n                            WHEN events.mac IS NOT NULL THEN 'mac:' || events.mac\n                            WHEN events.ip IS NOT NULL THEN 'ip:' || events.ip\n                            ELSE ''\n                        END) as \"observation_key!\",\n                    events.agent_id as \"agent_id!\",\n                    events.kind as \"kind!\",\n                    events.action as \"action!\",\n                    events.mac,\n                    events.ip,\n                    events.hostname,\n                    events.ts_unix,\n                    known_devices.device_id,\n                    known_devices.display_name,\n                    known_devices.pinned\n             FROM agent_device_observation_events events\n             LEFT JOIN device_identifiers identifiers\n               ON identifiers.identifier_key =\n                  CASE\n                    WHEN events.mac IS NOT NULL THEN 'mac:' || events.mac\n                    WHEN events.ip IS NOT NULL THEN 'ip:' || events.ip\n                  END\n             LEFT JOIN known_devices ON known_devices.device_id = identifiers.device_id\n             WHERE (?1 IS NULL OR events.agent_id = ?1)\n               AND (?2 IS NULL OR events.kind = ?2)\n               AND (?3 IS NULL OR events.mac = ?3)\n               AND (?4 IS NULL OR events.ip = ?4)\n               AND (?5 IS NULL OR ('agent:' || events.agent_id || ':' || events.kind || ':' ||\n                    CASE\n                        WHEN events.mac IS NOT NULL THEN 'mac:' || events.mac\n                        WHEN events.ip IS NOT NULL THEN 'ip:' || events.ip\n                        ELSE ''\n                    END) = ?5)\n             ORDER BY events.ts_unix DESC\n             LIMIT ?6",
+  "describe": {
+    "columns": [
+      {
+        "name": "event_id!",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "observation_key!",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "agent_id!",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "kind!",
+        "ordinal": 3,
+        "type_info": "Text"
+      },
+      {
+        "name": "action!",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "mac",
+        "ordinal": 5,
+        "type_info": "Text"
+      },
+      {
+        "name": "ip",
+        "ordinal": 6,
+        "type_info": "Text"
+      },
+      {
+        "name": "hostname",
+        "ordinal": 7,
+        "type_info": "Text"
+      },
+      {
+        "name": "ts_unix",
+        "ordinal": 8,
+        "type_info": "Integer"
+      },
+      {
+        "name": "device_id",
+        "ordinal": 9,
+        "type_info": "Text"
+      },
+      {
+        "name": "display_name",
+        "ordinal": 10,
+        "type_info": "Text"
+      },
+      {
+        "name": "pinned",
+        "ordinal": 11,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 6
+    },
+    "nullable": [
+      true,
+      true,
+      false,
+      false,
+      false,
+      true,
+      true,
+      true,
+      false,
+      true,
+      true,
+      true
+    ]
+  },
+  "hash": "908d763909c76fb6170b818784e360372c02295523ef9970c91199db7d4f535a"
+}

.sqlx/query-92775bf2e3b9f772d108bb8d0cd1aebbe01357ed70dc2af7b50ee348e668b319.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT mac, ip, hostname, first_seen_unix, last_seen_unix,\n                  last_action as \"last_action!\"\n           FROM agent_device_observations\n           WHERE observation_key = ?1",
+  "describe": {
+    "columns": [
+      {
+        "name": "mac",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "ip",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "hostname",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "first_seen_unix",
+        "ordinal": 3,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_seen_unix",
+        "ordinal": 4,
+        "type_info": "Integer"
+      },
+      {
+        "name": "last_action!",
+        "ordinal": 5,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      true,
+      true,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "92775bf2e3b9f772d108bb8d0cd1aebbe01357ed70dc2af7b50ee348e668b319"
+}

.sqlx/query-9fe9f41d6716305ffebc764a72a3aea11e4ba348fae664aa6ba4bffbb3a2bf34.json

@@ -46,5 +46,5 @@
       true
     ]
   },
-  "hash": "db255767f7f0aae53f685a88a6127a80b89601cdf1163881c5ec07160fbaeeab"
+  "hash": "9fe9f41d6716305ffebc764a72a3aea11e4ba348fae664aa6ba4bffbb3a2bf34"
 }

.sqlx/query-dd1d6ba80d6380cb07227eac0b90951c93f63dc4c89349cf1920405943e2093a.json

@@ -40,5 +40,5 @@
       false
     ]
   },
-  "hash": "2a7abe4d0ba22f991f490b45e2edcc980586116e92bac2c5394387b7052abe65"
+  "hash": "dd1d6ba80d6380cb07227eac0b90951c93f63dc4c89349cf1920405943e2093a"
 }

.sqlx/query-f94cb699c0b4202bc5611799c931ed5d566b3215fcaebdf564e614a417317871.json

@@ -1,12 +0,0 @@
-{
-  "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"
-}

Cargo.lock

@@ -2497,6 +2497,15 @@ dependencies = [
  "winapi-util",
 ]
 
+[[package]]
+name = "scc"
+version = "2.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "46e6f046b7fef48e2660c57ed794263155d713de679057f2d0c169bfc6e756cc"
+dependencies = [
+ "sdd",
+]
+
 [[package]]
 name = "schannel"
 version = "0.1.29"
@@ -2536,6 +2545,12 @@ version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
+[[package]]
+name = "sdd"
+version = "3.0.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "490dcfcbfef26be6800d11870ff2df8774fa6e86d047e3e8c8a76b25655e41ca"
+
 [[package]]
 name = "security-framework"
 version = "3.7.0"
@@ -2671,6 +2686,32 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serial_test"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "911bd979bf1070a3f3aa7b691a3b3e9968f339ceeec89e08c280a8a22207a32f"
+dependencies = [
+ "futures-executor",
+ "futures-util",
+ "log",
+ "once_cell",
+ "parking_lot 0.12.5",
+ "scc",
+ "serial_test_derive",
+]
+
+[[package]]
+name = "serial_test_derive"
+version = "3.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0a7d91949b85b0d2fb687445e448b40d322b6b3e4af6b44a29b21d9a5f33e6d9"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.6"
@@ -3752,6 +3793,7 @@ dependencies = [
  "macaddr",
  "serde",
  "serde_json",
+ "serial_test",
  "tokio",
  "wakey-core",
 ]

src/lib.rs

@@ -3,6 +3,7 @@
 
 pub mod service;
 pub mod utils;
+pub use wakey_linux;
 
 pub use service::{
     broadcast_wake_targets, get_interface_summaries, get_interface_summary, get_ips, get_leases,

ui/src/App.tsx

@@ -20,6 +20,7 @@ import { AuditPage } from "@/pages/AuditPage";
 import { CommandsPage } from "@/pages/CommandsPage";
 import { DashboardPage } from "@/pages/DashboardPage";
 import { DevicesPage } from "@/pages/DevicesPage";
+import { ObservationsPage } from "@/pages/ObservationsPage";
 import { TokensPage } from "@/pages/TokensPage";
 
 type LoadState = "idle" | "loading" | "ready" | "error";
@@ -151,6 +152,7 @@ export function App() {
               />
             }
           />
+          <Route path="observations" element={<ObservationsPage />} />
           <Route
             path="agents"
             element={

ui/src/api.ts

@@ -72,6 +72,56 @@ export type SetAgentNicknameResponse = {
   updated: boolean;
 };
 
+export type DeviceIdentifier = {
+  identifier_key: string;
+  device_id: string;
+  kind: string;
+  value: string;
+  created_at_unix: number;
+};
+
+export type KnownDevice = {
+  device_id: string;
+  display_name: string;
+  pinned: boolean;
+  created_at_unix: number;
+  updated_at_unix: number;
+  notes: string | null;
+  identifiers: DeviceIdentifier[];
+};
+
+export type KnownDeviceSummary = {
+  device_id: string;
+  display_name: string;
+  pinned: boolean;
+};
+
+export type AgentDeviceObservation = {
+  observation_key: string;
+  agent_id: string;
+  kind: string;
+  mac: string | null;
+  ip: string | null;
+  hostname: string | null;
+  first_seen_unix: number;
+  last_seen_unix: number;
+  last_action: string;
+  known_device: KnownDeviceSummary | null;
+};
+
+export type AgentDeviceObservationEvent = {
+  event_id: string;
+  observation_key: string;
+  agent_id: string;
+  kind: string;
+  action: string;
+  mac: string | null;
+  ip: string | null;
+  hostname: string | null;
+  ts_unix: number;
+  known_device: KnownDeviceSummary | null;
+};
+
 async function request<T>(path: string, init?: RequestInit): Promise<T> {
   const res = await fetch(path, {
     ...init,
@@ -154,6 +204,72 @@ export function setAgentNickname(
   );
 }
 
+export function fetchKnownDevices(): Promise<KnownDevice[]> {
+  return request<KnownDevice[]>("/api/v1/control/devices");
+}
+
+export function fetchObservations(opts?: {
+  agentId?: string;
+  limit?: number;
+}): Promise<AgentDeviceObservation[]> {
+  const params = new URLSearchParams();
+  if (opts?.agentId) params.set("agent_id", opts.agentId);
+  params.set("limit", String(opts?.limit ?? 500));
+  return request<AgentDeviceObservation[]>(
+    `/api/v1/control/observations?${params.toString()}`,
+  );
+}
+
+export function fetchObservationHistory(opts?: {
+  agentId?: string;
+  kind?: string;
+  mac?: string;
+  ip?: string;
+  observationKey?: string;
+  limit?: number;
+}): Promise<AgentDeviceObservationEvent[]> {
+  const params = new URLSearchParams();
+  if (opts?.agentId) params.set("agent_id", opts.agentId);
+  if (opts?.kind) params.set("kind", opts.kind);
+  if (opts?.mac) params.set("mac", opts.mac);
+  if (opts?.ip) params.set("ip", opts.ip);
+  if (opts?.observationKey) params.set("observation_key", opts.observationKey);
+  params.set("limit", String(opts?.limit ?? 500));
+  return request<AgentDeviceObservationEvent[]>(
+    `/api/v1/control/observations/history?${params.toString()}`,
+  );
+}
+
+export function createKnownDevice(input: {
+  display_name: string;
+  pinned?: boolean;
+  notes?: string | null;
+  identifiers?: { kind: string; value: string }[];
+}): Promise<KnownDevice> {
+  return request<KnownDevice>("/api/v1/control/devices", {
+    method: "POST",
+    body: JSON.stringify({
+      display_name: input.display_name,
+      pinned: input.pinned ?? true,
+      notes: input.notes ?? null,
+      identifiers: input.identifiers ?? [],
+    }),
+  });
+}
+
+export function attachObservationIdentifier(
+  deviceId: string,
+  observationKey: string,
+): Promise<KnownDevice> {
+  return request<KnownDevice>(
+    `/api/v1/control/devices/${encodeURIComponent(deviceId)}/identifiers/from-observation`,
+    {
+      method: "POST",
+      body: JSON.stringify({ observation_key: observationKey }),
+    },
+  );
+}
+
 export function runCommand(
   agentId: string,
   kind: CommandKind,

ui/src/layout/AppLayout.tsx

@@ -6,6 +6,7 @@ import { Button } from "@/components/ui/button";
 
 const navItems = [
   ["/", "Devices"],
+  ["/observations", "Observed"],
   ["/commands", "Wake Tools"],
   ["/dashboard", "Fleet Health"],
   ["/agents", "Agents"],

ui/src/pages/ObservationsPage.tsx

@@ -0,0 +1,367 @@
+import { useEffect, useMemo, useState } from "react";
+
+import {
+  attachObservationIdentifier,
+  createKnownDevice,
+  fetchKnownDevices,
+  fetchObservations,
+  type AgentDeviceObservation,
+  type KnownDevice,
+} from "@/api";
+import { Badge } from "@/components/ui/badge";
+import { Button } from "@/components/ui/button";
+import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
+import { Input } from "@/components/ui/input";
+import {
+  Select,
+  SelectContent,
+  SelectItem,
+  SelectTrigger,
+} from "@/components/ui/select";
+
+type Filter = "all" | "unknown" | "known";
+
+function observationLabel(observation: AgentDeviceObservation): string {
+  return (
+    observation.hostname?.trim() ||
+    observation.mac?.trim() ||
+    observation.ip?.trim() ||
+    observation.observation_key
+  );
+}
+
+function formatSeen(tsUnix: number): string {
+  if (!Number.isFinite(tsUnix) || tsUnix <= 0) return "-";
+  return new Date(tsUnix * 1000).toLocaleString();
+}
+
+function identifierSummary(device: KnownDevice): string {
+  if (!device.identifiers.length) return "no identifiers";
+  return device.identifiers
+    .map((identifier) => `${identifier.kind}:${identifier.value}`)
+    .join(", ");
+}
+
+export function ObservationsPage() {
+  const [observations, setObservations] = useState<AgentDeviceObservation[]>(
+    [],
+  );
+  const [devices, setDevices] = useState<KnownDevice[]>([]);
+  const [filter, setFilter] = useState<Filter>("unknown");
+  const [query, setQuery] = useState("");
+  const [selectedDevices, setSelectedDevices] = useState<
+    Record<string, string>
+  >({});
+  const [newNames, setNewNames] = useState<Record<string, string>>({});
+  const [busyKey, setBusyKey] = useState("");
+  const [loading, setLoading] = useState(false);
+  const [status, setStatus] = useState("");
+  const [error, setError] = useState("");
+
+  async function load() {
+    setLoading(true);
+    setError("");
+    try {
+      const [nextObservations, nextDevices] = await Promise.all([
+        fetchObservations({ limit: 500 }),
+        fetchKnownDevices(),
+      ]);
+      setObservations(nextObservations);
+      setDevices(nextDevices);
+    } catch (err) {
+      setError(String(err));
+    } finally {
+      setLoading(false);
+    }
+  }
+
+  useEffect(() => {
+    void load();
+  }, []);
+
+  const filtered = useMemo(() => {
+    const q = query.trim().toLowerCase();
+    return observations.filter((observation) => {
+      if (filter === "known" && !observation.known_device) return false;
+      if (filter === "unknown" && observation.known_device) return false;
+      if (!q) return true;
+
+      const haystack = [
+        observation.hostname,
+        observation.mac,
+        observation.ip,
+        observation.agent_id,
+        observation.kind,
+        observation.last_action,
+        observation.known_device?.display_name,
+      ]
+        .filter((value): value is string => Boolean(value))
+        .join(" ")
+        .toLowerCase();
+      return haystack.includes(q);
+    });
+  }, [filter, observations, query]);
+
+  async function attachExisting(observation: AgentDeviceObservation) {
+    const deviceId = selectedDevices[observation.observation_key];
+    if (!deviceId) return;
+    setBusyKey(observation.observation_key);
+    setStatus("");
+    setError("");
+    try {
+      const device = await attachObservationIdentifier(
+        deviceId,
+        observation.observation_key,
+      );
+      setStatus(
+        `Attached ${observationLabel(observation)} to ${device.display_name}`,
+      );
+      await load();
+    } catch (err) {
+      setError(String(err));
+    } finally {
+      setBusyKey("");
+    }
+  }
+
+  async function createAndAttach(observation: AgentDeviceObservation) {
+    const displayName =
+      newNames[observation.observation_key]?.trim() ||
+      observationLabel(observation);
+    if (!displayName.trim()) return;
+    setBusyKey(observation.observation_key);
+    setStatus("");
+    setError("");
+    try {
+      const device = await createKnownDevice({
+        display_name: displayName,
+        pinned: true,
+        identifiers: [],
+      });
+      const updated = await attachObservationIdentifier(
+        device.device_id,
+        observation.observation_key,
+      );
+      setStatus(
+        `Created ${updated.display_name} and attached ${observationLabel(observation)}`,
+      );
+      await load();
+    } catch (err) {
+      setError(String(err));
+    } finally {
+      setBusyKey("");
+    }
+  }
+
+  return (
+    <Card>
+      <CardHeader className="flex flex-row items-center justify-between gap-2 space-y-0">
+        <CardTitle>Observed Devices</CardTitle>
+        <Button
+          variant="outline"
+          size="sm"
+          onClick={() => void load()}
+          disabled={loading}
+        >
+          {loading ? "Refreshing..." : "Refresh"}
+        </Button>
+      </CardHeader>
+      <CardContent className="space-y-3">
+        <div className="grid gap-2 md:grid-cols-[12rem_minmax(0,1fr)]">
+          <label className="grid gap-1 text-sm text-muted-foreground">
+            <span>Status</span>
+            <Select
+              value={filter}
+              onValueChange={(value: Filter | null) => {
+                if (value) setFilter(value);
+              }}
+            >
+              <SelectTrigger className="w-full">
+                <span>{filter}</span>
+              </SelectTrigger>
+              <SelectContent alignItemWithTrigger={false}>
+                <SelectItem value="unknown">unknown</SelectItem>
+                <SelectItem value="known">known</SelectItem>
+                <SelectItem value="all">all</SelectItem>
+              </SelectContent>
+            </Select>
+          </label>
+          <label className="grid gap-1 text-sm text-muted-foreground">
+            <span>Search</span>
+            <Input
+              value={query}
+              onChange={(event) => setQuery(event.target.value)}
+              placeholder="hostname, mac, ip, agent, known device"
+            />
+          </label>
+        </div>
+
+        <p className="text-sm text-muted-foreground">
+          Showing {filtered.length} of {observations.length}
+        </p>
+        {status && (
+          <pre className="max-h-80 overflow-auto rounded-md border bg-muted/40 p-3 text-xs">
+            {status}
+          </pre>
+        )}
+        {error && (
+          <pre className="max-h-80 overflow-auto rounded-md border border-destructive/60 bg-destructive/10 p-3 text-xs text-destructive">
+            {error}
+          </pre>
+        )}
+
+        <div className="grid gap-2 overflow-x-auto">
+          <div className="grid min-w-[68rem] grid-cols-[minmax(10rem,1.2fr)_minmax(8rem,0.8fr)_minmax(8rem,0.8fr)_minmax(9rem,0.8fr)_minmax(16rem,1.3fr)] gap-2 rounded-md border bg-muted/60 px-3 py-2 text-sm">
+            <span>Device</span>
+            <span>Network</span>
+            <span>Agent</span>
+            <span>Seen</span>
+            <span>Known Device</span>
+          </div>
+
+          {filtered.map((observation) => {
+            const busy = busyKey === observation.observation_key;
+            const attachable = Boolean(observation.mac || observation.ip);
+            return (
+              <div
+                key={observation.observation_key}
+                className="grid min-w-[68rem] grid-cols-[minmax(10rem,1.2fr)_minmax(8rem,0.8fr)_minmax(8rem,0.8fr)_minmax(9rem,0.8fr)_minmax(16rem,1.3fr)] gap-2 rounded-md border bg-card px-3 py-2 text-sm"
+              >
+                <div className="min-w-0">
+                  <div className="truncate font-medium">
+                    {observationLabel(observation)}
+                  </div>
+                  <div className="mt-1 flex flex-wrap gap-1 text-xs text-muted-foreground">
+                    <Badge variant="outline">{observation.kind}</Badge>
+                    <Badge variant="outline">{observation.last_action}</Badge>
+                  </div>
+                </div>
+                <div className="min-w-0 text-muted-foreground">
+                  <div className="truncate" title={observation.ip ?? "-"}>
+                    {observation.ip ?? "-"}
+                  </div>
+                  <div className="truncate" title={observation.mac ?? "-"}>
+                    {observation.mac ?? "-"}
+                  </div>
+                </div>
+                <div className="min-w-0 truncate text-muted-foreground">
+                  {observation.agent_id}
+                </div>
+                <div className="min-w-0 text-muted-foreground">
+                  <div className="truncate">
+                    {formatSeen(observation.last_seen_unix)}
+                  </div>
+                </div>
+                <div className="min-w-0">
+                  {observation.known_device ? (
+                    <div className="grid gap-1">
+                      <div className="flex min-w-0 items-center gap-2">
+                        <Badge variant="secondary">known</Badge>
+                        <span className="truncate font-medium">
+                          {observation.known_device.display_name}
+                        </span>
+                      </div>
+                      <span className="text-xs text-muted-foreground">
+                        {observation.known_device.pinned
+                          ? "pinned"
+                          : "not pinned"}
+                      </span>
+                    </div>
+                  ) : (
+                    <div className="grid min-w-0 gap-2">
+                      <div className="flex min-w-0 gap-2">
+                        <Select
+                          value={
+                            selectedDevices[observation.observation_key] ?? ""
+                          }
+                          onValueChange={(value: string | null) => {
+                            setSelectedDevices((prev) => ({
+                              ...prev,
+                              [observation.observation_key]: value ?? "",
+                            }));
+                          }}
+                        >
+                          <SelectTrigger className="min-w-0 flex-1">
+                            <span className="truncate">
+                              {selectedDevices[observation.observation_key]
+                                ? devices.find(
+                                    (device) =>
+                                      device.device_id ===
+                                      selectedDevices[
+                                        observation.observation_key
+                                      ],
+                                  )?.display_name
+                                : "Attach to existing"}
+                            </span>
+                          </SelectTrigger>
+                          <SelectContent
+                            className="max-w-[min(92vw,28rem)]"
+                            alignItemWithTrigger={false}
+                          >
+                            {devices.map((device) => (
+                              <SelectItem
+                                key={device.device_id}
+                                value={device.device_id}
+                              >
+                                <span className="grid min-w-0 gap-0.5">
+                                  <span className="truncate">
+                                    {device.display_name}
+                                  </span>
+                                  <span className="truncate text-xs text-muted-foreground">
+                                    {identifierSummary(device)}
+                                  </span>
+                                </span>
+                              </SelectItem>
+                            ))}
+                          </SelectContent>
+                        </Select>
+                        <Button
+                          variant="outline"
+                          size="sm"
+                          disabled={
+                            busy ||
+                            !attachable ||
+                            !selectedDevices[observation.observation_key]
+                          }
+                          onClick={() => void attachExisting(observation)}
+                        >
+                          Attach
+                        </Button>
+                      </div>
+                      <div className="flex min-w-0 gap-2">
+                        <Input
+                          value={newNames[observation.observation_key] ?? ""}
+                          onChange={(event) =>
+                            setNewNames((prev) => ({
+                              ...prev,
+                              [observation.observation_key]: event.target.value,
+                            }))
+                          }
+                          placeholder={`Create as ${observationLabel(observation)}`}
+                          disabled={busy || !attachable}
+                        />
+                        <Button
+                          size="sm"
+                          disabled={busy || !attachable}
+                          onClick={() => void createAndAttach(observation)}
+                        >
+                          Create
+                        </Button>
+                      </div>
+                    </div>
+                  )}
+                </div>
+              </div>
+            );
+          })}
+
+          {!filtered.length && (
+            <div className="px-1 py-2 text-sm text-muted-foreground">
+              No observations found
+            </div>
+          )}
+        </div>
+      </CardContent>
+    </Card>
+  );
+}

wakey-agent/src/cli.rs

@@ -14,6 +14,10 @@ pub struct Cli {
     #[arg(short = 'v', long = "verbose", action = ArgAction::Count, global = true)]
     pub verbose: u8,
 
+    /// Path to the agent config file for commands that use agent config.
+    #[arg(long, global = true)]
+    pub config: Option<PathBuf>,
+
     #[command(subcommand)]
     pub command: Command,
 }
@@ -28,6 +32,9 @@ pub enum Command {
     InitConfig(InitConfigArgs),
     /// Reload a running agent daemon by sending SIGHUP.
     Reload(ReloadArgs),
+    /// Upload local observations to the control plane once and exit.
+    #[command(visible_alias = "sync")]
+    SyncObservations(SyncObservationsArgs),
     /// Pass local hotplug observations through to the wakey CLI.
     Observe(ObserveArgs),
 }
@@ -76,6 +83,10 @@ pub struct InitConfigArgs {
     #[arg(long)]
     pub config: Option<PathBuf>,
 
+    /// Existing agent config to use as a base before applying explicit overrides.
+    #[arg(long)]
+    pub from_config: Option<PathBuf>,
+
     /// Print config to stdout.
     #[arg(long, conflicts_with = "config")]
     pub stdout: bool,
@@ -104,8 +115,19 @@ pub struct ReloadArgs {
     pub pid_file: PathBuf,
 }
 
+#[derive(Args)]
+pub struct SyncObservationsArgs {
+    /// Path to the agent config file.
+    #[arg(long, default_value = config::DEFAULT_CONFIG_PATH)]
+    pub config: PathBuf,
+}
+
 #[derive(Args)]
 pub struct ObserveArgs {
+    /// Path to the agent config file. If present, local path settings are passed through.
+    #[arg(long, global = true, default_value = config::DEFAULT_CONFIG_PATH)]
+    pub config: PathBuf,
+
     #[command(subcommand)]
     pub command: ObserveCommand,
 }

wakey-agent/src/config.rs

@@ -4,6 +4,12 @@ use std::fmt;
 use std::path::{Path, PathBuf};
 
 pub const DEFAULT_CONFIG_PATH: &str = "/etc/wakey-agent/config.toml";
+const WAKEY_DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
+const WAKEY_MAC_NAME_CACHE_ENV: &str = "WAKEY_MAC_NAME_CACHE";
+const WAKEY_OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE";
+const DEFAULT_DHCP_LEASES_PATH: &str = "/tmp/dhcp.leases";
+const DEFAULT_MAC_NAME_CACHE_PATH: &str = "/tmp/wakey_mac_names.json";
+const DEFAULT_OBSERVATION_STORE_PATH: &str = "/tmp/wakey_observations.json";
 
 #[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct AgentConfig {
@@ -14,6 +20,14 @@ pub struct AgentConfig {
     pub reconnect_base_ms: u64,
     #[serde(default = "default_reconnect_max_ms")]
     pub reconnect_max_ms: u64,
+    #[serde(default = "default_observation_sync_interval_seconds")]
+    pub observation_sync_interval_seconds: u64,
+    #[serde(default = "default_dhcp_leases_path")]
+    pub dhcp_leases_path: PathBuf,
+    #[serde(default = "default_mac_name_cache_path")]
+    pub mac_name_cache_path: PathBuf,
+    #[serde(default = "default_observation_store_path")]
+    pub observation_store_path: PathBuf,
 }
 
 impl fmt::Debug for AgentConfig {
@@ -24,6 +38,13 @@ impl fmt::Debug for AgentConfig {
             .field("agent_token", &"<redacted>")
             .field("reconnect_base_ms", &self.reconnect_base_ms)
             .field("reconnect_max_ms", &self.reconnect_max_ms)
+            .field(
+                "observation_sync_interval_seconds",
+                &self.observation_sync_interval_seconds,
+            )
+            .field("dhcp_leases_path", &self.dhcp_leases_path)
+            .field("mac_name_cache_path", &self.mac_name_cache_path)
+            .field("observation_store_path", &self.observation_store_path)
             .finish()
     }
 }
@@ -36,6 +57,41 @@ const fn default_reconnect_max_ms() -> u64 {
     30_000
 }
 
+const fn default_observation_sync_interval_seconds() -> u64 {
+    60
+}
+
+fn default_dhcp_leases_path() -> PathBuf {
+    DEFAULT_DHCP_LEASES_PATH.into()
+}
+
+fn default_mac_name_cache_path() -> PathBuf {
+    DEFAULT_MAC_NAME_CACHE_PATH.into()
+}
+
+fn default_observation_store_path() -> PathBuf {
+    DEFAULT_OBSERVATION_STORE_PATH.into()
+}
+
+impl AgentConfig {
+    pub fn local_path_envs(&self) -> Vec<(&'static str, &Path)> {
+        vec![
+            (WAKEY_DHCP_LEASES_ENV, self.dhcp_leases_path.as_path()),
+            (WAKEY_MAC_NAME_CACHE_ENV, self.mac_name_cache_path.as_path()),
+            (
+                WAKEY_OBSERVATION_STORE_ENV,
+                self.observation_store_path.as_path(),
+            ),
+        ]
+    }
+}
+
+pub fn apply_local_path_env_to_command(cmd: &mut std::process::Command, config: &AgentConfig) {
+    for (key, path) in config.local_path_envs() {
+        cmd.env(key, path);
+    }
+}
+
 pub fn load_config(path: &Path) -> Result<AgentConfig> {
     let content = std::fs::read_to_string(path)
         .with_context(|| format!("failed to read agent config {}", path.display()))?;
@@ -118,12 +174,22 @@ mod tests {
             agent_token: "secret".into(),
             reconnect_base_ms: 123,
             reconnect_max_ms: 456,
+            observation_sync_interval_seconds: 7,
+            dhcp_leases_path: "/tmp/test-dhcp.leases".into(),
+            mac_name_cache_path: "/tmp/test-names.json".into(),
+            observation_store_path: "/tmp/test-observations.json".into(),
         };
 
         save_config(&path, &config).expect("save");
         let loaded = load_config(&path).expect("load");
         assert_eq!(loaded, config);
         assert!(format!("{:?}", loaded).contains("<redacted>"));
+        assert!(
+            loaded
+                .local_path_envs()
+                .iter()
+                .any(|(key, _)| *key == WAKEY_OBSERVATION_STORE_ENV)
+        );
         let _ = std::fs::remove_file(&path);
         let _ = std::fs::remove_dir_all(&dir);
     }

wakey-agent/src/dispatch.rs

@@ -1,27 +1,38 @@
 use anyhow::Result;
 use tracing::{debug, info, instrument};
 
+use crate::config::AgentConfig;
 use crate::protocol::{
     AgentCommand, CommandResult, DevsRequest, InventoryRequest, LeasesRequest, WakeRequest,
 };
 
 #[instrument(skip_all)]
-pub async fn dispatch_command(command: AgentCommand) -> Result<CommandResult> {
+pub async fn dispatch_command(
+    command: AgentCommand,
+    config: &AgentConfig,
+) -> Result<CommandResult> {
     let kind = command_kind(&command);
     info!(command = %kind, "dispatching command into local wakey services");
     match command {
-        AgentCommand::Leases(req) => dispatch_leases(req).await,
+        AgentCommand::Leases(req) => dispatch_leases(req, config).await,
         AgentCommand::Devs(req) => dispatch_devs(req).await,
-        AgentCommand::Inventory(req) => dispatch_inventory(req).await,
+        AgentCommand::Inventory(req) => dispatch_inventory(req, config).await,
         AgentCommand::Wake(req) => dispatch_wake(req).await,
     }
 }
 
-async fn dispatch_leases(req: LeasesRequest) -> Result<CommandResult> {
+async fn dispatch_leases(req: LeasesRequest, config: &AgentConfig) -> Result<CommandResult> {
-    let leases = wakey::get_leases(wakey_core::LeaseQuery {
+    let leases = wakey::wakey_linux::dhcp::read_dhcp_leases_with_names_from_paths(
-        include_state: req.include_state,
+        &config.dhcp_leases_path,
-    })
+        &config.observation_store_path,
+        &config.mac_name_cache_path,
+    )
     .await?;
+    let leases = if req.include_state {
+        wakey::wakey_linux::dhcp::enrich_leases_with_nud_state(leases).await
+    } else {
+        wakey::leases_without_state(leases)
+    };
     debug!(
         rows = leases.len(),
         include_state = req.include_state,
@@ -50,8 +61,18 @@ async fn dispatch_devs(req: DevsRequest) -> Result<CommandResult> {
     Ok(CommandResult::Devs { rows: devs })
 }
 
-async fn dispatch_inventory(req: InventoryRequest) -> Result<CommandResult> {
+async fn dispatch_inventory(req: InventoryRequest, config: &AgentConfig) -> Result<CommandResult> {
-    let inventory = wakey::inventory(req.into_inventory_query()).await?;
+    let query = req.into_inventory_query();
+    let neighbors = wakey::wakey_linux::devices::query_neighbors(&query).await?;
+    let leases = wakey::wakey_linux::dhcp::read_dhcp_leases_with_names_from_paths(
+        &config.dhcp_leases_path,
+        &config.observation_store_path,
+        &config.mac_name_cache_path,
+    )
+    .await?;
+    let inventory = wakey_core::DeviceInventory {
+        devices: wakey::merge_devices(neighbors, wakey::leases_without_state(leases), &query),
+    };
     debug!(
         rows = inventory.devices.len(),
         "dispatched inventory command"

wakey-agent/src/enroll.rs

@@ -60,6 +60,10 @@ pub async fn enroll(
         agent_token: payload.agent_token,
         reconnect_base_ms: 1_000,
         reconnect_max_ms: 30_000,
+        observation_sync_interval_seconds: 60,
+        dhcp_leases_path: "/tmp/dhcp.leases".into(),
+        mac_name_cache_path: "/tmp/wakey_mac_names.json".into(),
+        observation_store_path: "/tmp/wakey_observations.json".into(),
     };
     let backup_path = save_config_with_backup(config_path, &config)?;
     info!(agent_id = %config.agent_id, config_path = %config_path.display(), "agent enrollment succeeded and config was written");

wakey-agent/src/main.rs

@@ -15,13 +15,20 @@ use cli::{Cli, Command, InitConfigArgs, ObserveCommand};
 async fn main() -> Result<()> {
     let cli = Cli::parse();
     tracing::init(cli.verbose);
+    let global_config = cli.config.as_deref();
 
     match cli.command {
-        Command::Serve(args) => {
+        Command::Serve(mut args) => {
+            if let Some(config) = global_config {
+                args.config = config.to_path_buf();
+            }
             ::tracing::info!("wakey-agent command: serve");
             serve::serve(args).await?
         }
-        Command::Enroll(args) => {
+        Command::Enroll(mut args) => {
+            if let Some(config) = global_config {
+                args.config = config.to_path_buf();
+            }
             let resolved_server_url = if let Some(server_url) = args.server_url.as_deref() {
                 server_url.to_string()
             } else {
@@ -81,12 +88,34 @@ async fn main() -> Result<()> {
                 );
             }
         }
-        Command::InitConfig(args) => init_config(args)?,
+        Command::InitConfig(mut args) => {
+            if let Some(config) = global_config
+                && args.config.is_none()
+                && !args.stdout
+            {
+                args.config = Some(config.to_path_buf());
+            }
+            init_config(args)?
+        }
         Command::Reload(args) => {
             ::tracing::info!(pid_file = %args.pid_file.display(), "wakey-agent command: reload");
             serve::reload_daemon(&args.pid_file)?
         }
-        Command::Observe(args) => observe(args)?,
+        Command::SyncObservations(mut args) => {
+            if let Some(config) = global_config {
+                args.config = config.to_path_buf();
+            }
+            ::tracing::info!(config = %args.config.display(), "wakey-agent command: sync-observations");
+            let cfg = config::load_config(&args.config)?;
+            let accepted = session::sync_observations_once(&cfg).await?;
+            println!("observations_synced={accepted}");
+        }
+        Command::Observe(mut args) => {
+            if let Some(config) = global_config {
+                args.config = config.to_path_buf();
+            }
+            observe(args)?
+        }
     }
 
     Ok(())
@@ -95,6 +124,9 @@ async fn main() -> Result<()> {
 fn observe(args: cli::ObserveArgs) -> Result<()> {
     let mut cmd = std::process::Command::new(resolve_wakey_binary());
     cmd.arg("observe");
+    if let Ok(config) = config::load_config(&args.config) {
+        config::apply_local_path_env_to_command(&mut cmd, &config);
+    }
 
     match args.command {
         ObserveCommand::Dhcp(args) => {
@@ -161,20 +193,32 @@ fn init_config(args: InitConfigArgs) -> Result<()> {
         );
     }
 
-    let cfg = config::AgentConfig {
+    let mut cfg = if let Some(from_config) = &args.from_config {
-        server_url: args
+        config::load_config(from_config)?
-            .server_url
+    } else {
-            .unwrap_or_else(|| "https://wakey.ldlda.com".to_string()),
+        config::AgentConfig {
-        agent_id: args
+            server_url: "https://wakey.ldlda.com".to_string(),
-            .agent_id
+            agent_id: "REPLACE_ME_AGENT_ID".to_string(),
-            .unwrap_or_else(|| "REPLACE_ME_AGENT_ID".to_string()),
+            agent_token: "REPLACE_ME_AGENT_TOKEN".to_string(),
-        agent_token: args
-            .agent_token
-            .unwrap_or_else(|| "REPLACE_ME_AGENT_TOKEN".to_string()),
             reconnect_base_ms: 1_000,
             reconnect_max_ms: 30_000,
+            observation_sync_interval_seconds: 60,
+            dhcp_leases_path: "/tmp/dhcp.leases".into(),
+            mac_name_cache_path: "/tmp/wakey_mac_names.json".into(),
+            observation_store_path: "/tmp/wakey_observations.json".into(),
+        }
     };
 
+    if let Some(server_url) = args.server_url {
+        cfg.server_url = server_url;
+    }
+    if let Some(agent_id) = args.agent_id {
+        cfg.agent_id = agent_id;
+    }
+    if let Some(agent_token) = args.agent_token {
+        cfg.agent_token = agent_token;
+    }
+
     if let Some(path) = &args.config {
         config::save_config(path, &cfg)?;
         println!("config={}", path.display());

wakey-agent/src/session.rs

@@ -88,7 +88,9 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
     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));
+    let mut observation_sync = interval(Duration::from_secs(
+        config.observation_sync_interval_seconds.max(1),
+    ));
     observation_sync.set_missed_tick_behavior(MissedTickBehavior::Skip);
 
     loop {
@@ -117,7 +119,7 @@ async fn run_once(config: &AgentConfig) -> Result<()> {
                     Message::Text(text) => {
                         match serde_json::from_str::<ServerMessage>(&text) {
                             Ok(message) => {
-                                handle_server_message(&mut sink, message).await?;
+                                handle_server_message(config, &mut sink, message).await?;
                             }
                             Err(err) => {
                                 // Allow the server to introduce extra frame types without
@@ -162,12 +164,18 @@ struct AgentObservationRequest {
     last_seen_unix: u64,
 }
 
-async fn send_agent_observations(client: &reqwest::Client, config: &AgentConfig) -> Result<()> {
+pub async fn sync_observations_once(config: &AgentConfig) -> Result<usize> {
-    let observations = wakey::list_local_observations()
+    let client = reqwest::Client::new();
+    send_agent_observations(&client, config).await
+}
+
+async fn send_agent_observations(client: &reqwest::Client, config: &AgentConfig) -> Result<usize> {
+    let observations =
+        wakey::wakey_linux::dhcp::list_local_observations_from_path(&config.observation_store_path)
             .await
             .context("failed to read local observations")?;
     if observations.is_empty() {
-        return Ok(());
+        return Ok(0);
     }
 
     let url = observations_url(&config.server_url)?;
@@ -207,7 +215,7 @@ async fn send_agent_observations(client: &reqwest::Client, config: &AgentConfig)
         observations = payload.observations.len(),
         "synced local observations"
     );
-    Ok(())
+    Ok(payload.observations.len())
 }
 
 pub fn next_backoff_ms(current_ms: u64, max_ms: u64) -> u64 {
@@ -215,7 +223,11 @@ pub fn next_backoff_ms(current_ms: u64, max_ms: u64) -> u64 {
     current_ms.saturating_mul(2).min(cap)
 }
 
-async fn handle_server_message<S>(sink: &mut S, message: ServerMessage) -> Result<()>
+async fn handle_server_message<S>(
+    config: &AgentConfig,
+    sink: &mut S,
+    message: ServerMessage,
+) -> Result<()>
 where
     S: SinkExt<Message> + Unpin,
     <S as futures_util::Sink<Message>>::Error: std::error::Error + Send + Sync + 'static,
@@ -227,7 +239,7 @@ where
         } => {
             let kind = command_kind(&command);
             info!(request_id = %request_id, command = %kind, "received command from control-plane");
-            match dispatch_command(command).await {
+            match dispatch_command(command, config).await {
                 Ok(result) => {
                     info!(request_id = %request_id, command = %kind, "command execution completed");
                     send_command_result(sink, request_id, kind, result).await?;

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

@@ -8,8 +8,8 @@ use tracing::{info, warn};
 use crate::api::json_error;
 use crate::runtime::{AppState, SessionEvent};
 use crate::state::{
-    AgentDeviceObservation, AgentDeviceObservationInput, AuditEventInput, DeviceIdentifierInput,
+    AgentDeviceObservationEvent, AgentDeviceObservationInput, AgentDeviceObservationView,
-    KnownDeviceInput,
+    AuditEventInput, DeviceIdentifierInput, KnownDeviceInput,
 };
 
 #[derive(Debug, Deserialize)]
@@ -91,6 +91,11 @@ pub struct DeviceIdentifierRequest {
     pub value: String,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct AttachObservationIdentifierRequest {
+    pub observation_key: String,
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct KnownDeviceResponse {
     pub device_id: String,
@@ -147,6 +152,16 @@ pub struct ListObservationsQuery {
     pub limit: Option<usize>,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct ListObservationHistoryQuery {
+    pub agent_id: Option<String>,
+    pub kind: Option<String>,
+    pub mac: Option<String>,
+    pub ip: Option<String>,
+    pub observation_key: Option<String>,
+    pub limit: Option<usize>,
+}
+
 pub async fn healthz() -> &'static str {
     "ok"
 }
@@ -592,6 +607,33 @@ pub async fn attach_device_identifier(
     }
 }
 
+pub async fn attach_observation_identifier(
+    State(state): State<AppState>,
+    AxumPath(device_id): AxumPath<String>,
+    Json(req): Json<AttachObservationIdentifierRequest>,
+) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
+    match state
+        .store
+        .attach_observation_identifier(&device_id, &req.observation_key)
+        .await
+    {
+        Ok(Some(device)) => Ok((StatusCode::OK, Json(known_device_response(device)))),
+        Ok(None) => Err(json_error(
+            StatusCode::NOT_FOUND,
+            "known_device_not_found",
+            "known device not found",
+        )),
+        Err(err) => {
+            warn!(error = %err, "failed to attach observation identifier");
+            Err(json_error(
+                StatusCode::BAD_REQUEST,
+                "attach_observation_identifier_failed",
+                &err.to_string(),
+            ))
+        }
+    }
+}
+
 pub async fn upload_agent_observations(
     State(state): State<AppState>,
     Json(req): Json<UploadAgentObservationsRequest>,
@@ -648,7 +690,7 @@ pub async fn list_agent_observations(
 ) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
     match state
         .store
-        .list_agent_observations(query.agent_id.as_deref(), query.limit.unwrap_or(500))
+        .list_agent_observation_views(query.agent_id.as_deref(), query.limit.unwrap_or(500))
         .await
     {
         Ok(observations) => Ok((
@@ -671,6 +713,46 @@ pub async fn list_agent_observations(
     }
 }
 
+pub async fn list_agent_observation_history(
+    State(state): State<AppState>,
+    Query(query): Query<ListObservationHistoryQuery>,
+) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
+    let kind = normalized_query_value(query.kind).map(|value| value.to_ascii_lowercase());
+    let mac = normalized_query_value(query.mac).map(|value| value.to_ascii_lowercase());
+    let ip = normalized_query_value(query.ip);
+    let observation_key = normalized_query_value(query.observation_key);
+    match state
+        .store
+        .list_agent_observation_events(
+            query.agent_id.as_deref(),
+            kind.as_deref(),
+            mac.as_deref(),
+            ip.as_deref(),
+            observation_key.as_deref(),
+            query.limit.unwrap_or(500),
+        )
+        .await
+    {
+        Ok(events) => Ok((
+            StatusCode::OK,
+            Json(
+                events
+                    .into_iter()
+                    .map(agent_observation_event_response) // no-op premium
+                    .collect::<Vec<_>>(),
+            ),
+        )),
+        Err(err) => {
+            warn!(error = %err, "failed to list agent observation history");
+            Err(json_error(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "list_observation_history_failed",
+                &err.to_string(),
+            ))
+        }
+    }
+}
+
 pub async fn state_stats(
     State(state): State<AppState>,
 ) -> Result<impl IntoResponse, (StatusCode, Json<serde_json::Value>)> {
@@ -696,10 +778,24 @@ pub async fn state_stats(
     }
 }
 
-fn agent_observation_response(observation: AgentDeviceObservation) -> AgentDeviceObservation {
+fn agent_observation_response(
+    observation: AgentDeviceObservationView,
+) -> AgentDeviceObservationView {
     observation
 }
 
+fn agent_observation_event_response(
+    event: AgentDeviceObservationEvent,
+) -> AgentDeviceObservationEvent {
+    event
+}
+
+fn normalized_query_value(value: Option<String>) -> Option<String> {
+    value
+        .map(|value| value.trim().to_string())
+        .filter(|value| !value.is_empty())
+}
+
 fn known_device_response(device: crate::state::KnownDevice) -> KnownDeviceResponse {
     KnownDeviceResponse {
         device_id: device.device_id,

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

@@ -11,9 +11,11 @@ pub use audit::list_audit_events;
 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,
+    StateStatsResponse, attach_device_identifier, attach_observation_identifier,
-    healthz, issue_enroll_token, list_agent_observations, list_enroll_tokens, list_known_devices,
+    create_known_device, enroll, forget_known_device, healthz, issue_enroll_token,
-    revoke_agent, revoke_enroll_token, set_agent_nickname, state_stats, upload_agent_observations,
+    list_agent_observation_history, list_agent_observations, list_enroll_tokens,
+    list_known_devices, revoke_agent, revoke_enroll_token, set_agent_nickname, state_stats,
+    upload_agent_observations,
 };
 
 pub fn json_error(

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

@@ -97,6 +97,10 @@ fn control_api_routes() -> Router<AppState> {
             "/api/v1/control/observations",
             get(api::list_agent_observations),
         )
+        .route(
+            "/api/v1/control/observations/history",
+            get(api::list_agent_observation_history),
+        )
         .route(
             "/api/v1/control/devices",
             get(api::list_known_devices).post(api::create_known_device),
@@ -109,6 +113,10 @@ fn control_api_routes() -> Router<AppState> {
             "/api/v1/control/devices/{device_id}/identifiers",
             post(api::attach_device_identifier),
         )
+        .route(
+            "/api/v1/control/devices/{device_id}/identifiers/from-observation",
+            post(api::attach_observation_identifier),
+        )
         .route("/api/v1/control/audit/events", get(api::list_audit_events))
         .route("/api/v1/control/alerts", get(api::active_alerts))
         .route("/api/v1/control/alerts/history", get(api::alert_history))

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

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

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

@@ -0,0 +1,132 @@
+use super::*;
+
+impl Store {
+    pub async fn sync_alert_transitions(
+        &self,
+        current: &[AlertState],
+    ) -> Result<Vec<AlertTransition>> {
+        let mut previous = std::collections::HashMap::<String, AlertState>::new();
+        let rows = sqlx::query_as!(
+            AlertStateRow,
+            r#"SELECT alert_id as "alert_id!", kind as "kind!", severity as "severity!",
+                    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
+        .context("failed iterating active_alerts table")?;
+        for row in rows {
+            let alert = alert_state_from_row(row).context("failed decoding active alert")?;
+            previous.insert(alert.alert_id.clone(), alert);
+        }
+
+        let mut current_map = std::collections::HashMap::<String, AlertState>::new();
+        for alert in current {
+            current_map.insert(alert.alert_id.clone(), alert.clone());
+        }
+
+        let now = now_unix();
+        let mut transitions = Vec::new();
+
+        for (alert_id, current_alert) in &current_map {
+            if previous.contains_key(alert_id) {
+                continue;
+            }
+            transitions.push(AlertTransition {
+                transition_id: format!("atr-{}", Uuid::new_v4()),
+                ts_unix: now,
+                alert_id: current_alert.alert_id.clone(),
+                kind: current_alert.kind.clone(),
+                agent_id: current_alert.agent_id.clone(),
+                from_status: None,
+                to_status: "active".into(),
+                message: current_alert.message.clone(),
+                metadata: current_alert.metadata.clone(),
+            });
+        }
+
+        for (alert_id, previous_alert) in &previous {
+            if current_map.contains_key(alert_id) {
+                continue;
+            }
+            transitions.push(AlertTransition {
+                transition_id: format!("atr-{}", Uuid::new_v4()),
+                ts_unix: now,
+                alert_id: previous_alert.alert_id.clone(),
+                kind: previous_alert.kind.clone(),
+                agent_id: previous_alert.agent_id.clone(),
+                from_status: Some("active".into()),
+                to_status: "resolved".into(),
+                message: format!("resolved alert {}", previous_alert.alert_id),
+                metadata: previous_alert.metadata.clone(),
+            });
+        }
+
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting alert transaction")?;
+        sqlx::query!("DELETE FROM active_alerts")
+            .execute(&mut *tx)
+            .await
+            .context("failed clearing active alert snapshot")?;
+
+        for alert in current {
+            insert_active_alert(&mut tx, alert).await?;
+        }
+
+        for transition in &transitions {
+            let key = format!("{:020}:{}", transition.ts_unix, transition.transition_id);
+            insert_alert_transition(&mut tx, &key, transition).await?;
+        }
+
+        tx.commit()
+            .await
+            .context("failed committing state db after alert sync")?;
+        Ok(transitions)
+    }
+
+    pub async fn list_alert_transitions(
+        &self,
+        since_unix: Option<u64>,
+        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 {
+            let since = i64::try_from(since).context("since_unix overflow")?;
+            sqlx::query_as!(
+                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"#,
+                since,
+                limit
+            )
+            .fetch_all(&self.pool)
+            .await
+        } else {
+            sqlx::query_as!(
+                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
+                 ORDER BY transition_key DESC
+                 LIMIT ?1"#,
+                limit
+            )
+            .fetch_all(&self.pool)
+            .await
+        }
+        .context("failed listing alert transitions")?;
+
+        rows.into_iter().map(alert_transition_from_row).collect()
+    }
+}

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

@@ -0,0 +1,96 @@
+use super::*;
+
+impl Store {
+    pub async fn append_audit_event(&self, input: AuditEventInput) -> Result<AuditEvent> {
+        let event = AuditEvent {
+            event_id: format!("evt-{}", Uuid::new_v4()),
+            ts_unix: now_unix(),
+            actor_type: input.actor_type,
+            actor_id: input.actor_id,
+            agent_id: input.agent_id,
+            request_id: input.request_id,
+            event_type: input.event_type,
+            outcome: input.outcome,
+            latency_ms: input.latency_ms,
+            message: input.message,
+            metadata: input.metadata,
+        };
+
+        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")?;
+        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
+        )
+        .execute(&self.pool)
+        .await
+        .context("failed persisting audit event")?;
+        Ok(event)
+    }
+
+    pub async fn list_audit_events(&self, filter: AuditEventFilter) -> Result<Vec<AuditEvent>> {
+        let limit = filter.limit.clamp(1, 500);
+        let mut builder = sqlx::QueryBuilder::new(
+            "SELECT event_id, ts_unix, actor_type, actor_id, agent_id, request_id,
+                    event_type, outcome, latency_ms, message, metadata_json
+             FROM audit_events
+             WHERE 1 = 1",
+        );
+
+        if let Some(agent_id) = filter.agent_id.as_deref() {
+            builder.push(" AND agent_id = ");
+            builder.push_bind(agent_id);
+        }
+        if let Some(request_id) = filter.request_id.as_deref() {
+            builder.push(" AND request_id = ");
+            builder.push_bind(request_id);
+        }
+        if let Some(event_type) = filter.event_type.as_deref() {
+            builder.push(" AND event_type = ");
+            builder.push_bind(event_type);
+        }
+        if let Some(outcome) = filter.outcome.as_deref() {
+            builder.push(" AND outcome = ");
+            builder.push_bind(outcome);
+        }
+        if let Some(since_unix) = filter.since_unix {
+            builder.push(" AND ts_unix >= ");
+            builder.push_bind(i64::try_from(since_unix).context("since_unix overflow")?);
+        }
+        if let Some(until_unix) = filter.until_unix {
+            builder.push(" AND ts_unix <= ");
+            builder.push_bind(i64::try_from(until_unix).context("until_unix overflow")?);
+        }
+        builder.push(" ORDER BY event_key DESC LIMIT ");
+        builder.push_bind(i64::try_from(limit).context("audit limit overflow")?);
+
+        let rows = builder
+            .build()
+            .fetch_all(&self.pool)
+            .await
+            .context("failed listing audit events")?;
+
+        rows.into_iter().map(audit_event_from_row).collect()
+    }
+}

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

@@ -0,0 +1,205 @@
+use super::*;
+
+impl Store {
+    pub async fn load_or_init(
+        path: &Path,
+        enroll_tokens: Vec<String>,
+        seed_ttl: Duration,
+    ) -> Result<Self> {
+        if path.is_dir() {
+            anyhow::bail!(
+                "state_file {} is a directory, which looks like a legacy sled store; run `wakey-control-plane import-sled-state --from-sled-state {} --to-state-file <sqlite-file>` and update state_file",
+                path.display(),
+                path.display()
+            );
+        }
+
+        let db_path = path.to_path_buf();
+        if let Some(parent) = db_path.parent() {
+            std::fs::create_dir_all(parent)
+                .with_context(|| format!("failed to create state dir {}", parent.display()))?;
+        }
+
+        let pool = open_sqlite_pool(&db_path).await?;
+        sqlx::migrate!("./migrations")
+            .run(&pool)
+            .await
+            .with_context(|| format!("failed to migrate state db {}", db_path.display()))?;
+
+        let store = Self { db_path, pool };
+
+        store.ensure_schema_version().await?;
+        store
+            .seed_bootstrap_enroll_tokens(&enroll_tokens, seed_ttl)
+            .await?;
+        store.gc_expired_enroll_tokens_inner().await?;
+
+        let enroll_tokens = sql_count(&store.pool, "enroll_tokens").await?;
+        let agents = sql_count(&store.pool, "agents").await?;
+        let audit_events = sql_count(&store.pool, "audit_events").await?;
+        info!(
+            path = %store.db_path.display(),
+            enroll_tokens,
+            agents,
+            audit_events,
+            "control-plane store ready"
+        );
+        Ok(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!(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!(
+            r#"SELECT COUNT(*) as "count!: i64" FROM enroll_tokens WHERE expires_at_unix <= ?1"#,
+            now
+        )
+        .fetch_one(&self.pool)
+        .await
+        .context("failed counting expired enroll tokens")?;
+        let agent_count = sqlx::query_scalar!(r#"SELECT COUNT(*) as "count!: i64" FROM agents"#)
+            .fetch_one(&self.pool)
+            .await
+            .context("failed counting agents")?;
+
+        Ok(StateStats {
+            db_path: self.db_path.clone(),
+            schema_version: self.schema_version().await?,
+            agent_count: usize::try_from(agent_count).context("agent count overflow")?,
+            enroll_token_count: usize::try_from(enroll_token_count)
+                .context("enroll token count overflow")?,
+            expired_enroll_token_count: usize::try_from(expired_enroll_token_count)
+                .context("expired enroll token count overflow")?,
+        })
+    }
+
+    #[cfg_attr(not(unix), allow(dead_code))]
+    pub async fn gc_expired_enroll_tokens(&self) -> Result<u64> {
+        self.gc_expired_enroll_tokens_inner().await
+    }
+
+    #[cfg_attr(not(unix), allow(dead_code))]
+    pub async fn reload_from_disk(&self) -> Result<()> {
+        sqlx::query!(r#"SELECT 1 as "ok!""#)
+            .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");
+        Ok(())
+    }
+
+    async fn seed_bootstrap_enroll_tokens(
+        &self,
+        enroll_tokens: &[String],
+        seed_ttl: Duration,
+    ) -> Result<()> {
+        for token in enroll_tokens {
+            let token = token.trim();
+            if token.is_empty() {
+                continue;
+            }
+
+            let marker_key = seeded_enroll_token_key(token);
+            let marker_exists = sqlx::query_scalar!(
+                r#"SELECT COUNT(*) as "count!: i64" FROM meta WHERE key = ?1"#,
+                marker_key
+            )
+            .fetch_one(&self.pool)
+            .await
+            .context("failed reading bootstrap marker")?;
+            if marker_exists > 0 {
+                continue;
+            }
+
+            let expires_at = now_unix().saturating_add(seed_ttl.as_secs().max(1));
+            let mut tx = self
+                .pool
+                .begin()
+                .await
+                .context("failed starting bootstrap token transaction")?;
+            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
+            )
+            .execute(&mut *tx)
+            .await
+            .context("failed seeding enroll token")?;
+            let marker_value = expires_at.to_le_bytes().to_vec();
+            sqlx::query!(
+                "INSERT INTO meta (key, value) VALUES (?1, ?2)",
+                marker_key,
+                marker_value
+            )
+            .execute(&mut *tx)
+            .await
+            .context("failed persisting bootstrap marker")?;
+            tx.commit()
+                .await
+                .context("failed committing bootstrap token transaction")?;
+        }
+        Ok(())
+    }
+
+    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", now)
+            .execute(&self.pool)
+            .await
+            .context("failed removing expired enroll tokens")?;
+        let removed = result.rows_affected();
+        if removed > 0 {
+            info!(removed, "garbage-collected expired enroll tokens");
+        }
+        Ok(removed)
+    }
+
+    async fn ensure_schema_version(&self) -> Result<()> {
+        match sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?1", SCHEMA_VERSION_KEY)
+            .fetch_optional(&self.pool)
+            .await
+            .context("failed reading schema version")?
+        {
+            Some(raw) => {
+                let schema =
+                    decode_schema(raw.as_ref()).context("failed decoding schema version")?;
+                if schema != SCHEMA_VERSION {
+                    anyhow::bail!(
+                        "unsupported db schema version {}; expected {}",
+                        schema,
+                        SCHEMA_VERSION
+                    );
+                }
+            }
+            None => {
+                let schema_version = SCHEMA_VERSION.to_le_bytes().to_vec();
+                sqlx::query!(
+                    "INSERT INTO meta (key, value) VALUES (?1, ?2)",
+                    SCHEMA_VERSION_KEY,
+                    schema_version
+                )
+                .execute(&self.pool)
+                .await
+                .context("failed writing schema version")?;
+                info!(
+                    schema_version = SCHEMA_VERSION,
+                    "initialized state schema version"
+                );
+            }
+        }
+        Ok(())
+    }
+
+    async fn schema_version(&self) -> Result<u32> {
+        let raw = sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?1", SCHEMA_VERSION_KEY)
+            .fetch_one(&self.pool)
+            .await
+            .context("failed reading schema version")?;
+        decode_schema(raw.as_ref()).context("failed decoding schema version")
+    }
+}

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

@@ -0,0 +1,163 @@
+use super::*;
+impl Store {
+    pub async fn create_known_device(&self, input: KnownDeviceInput) -> Result<KnownDevice> {
+        let display_name = normalize_required_text(&input.display_name, "display_name")?;
+        let notes = normalize_optional_text(input.notes.as_deref());
+        let identifiers = input
+            .identifiers
+            .into_iter()
+            .map(normalize_device_identifier)
+            .collect::<Result<Vec<_>>>()?;
+        let device_id = format!("dev-{}", Uuid::new_v4());
+        let now = now_unix();
+
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting known device transaction")?;
+        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
+        )
+        .execute(&mut *tx)
+        .await
+        .context("failed persisting known device")?;
+
+        for identifier in &identifiers {
+            insert_device_identifier_tx(&mut tx, &device_id, identifier, now).await?;
+        }
+
+        tx.commit()
+            .await
+            .context("failed committing known device transaction")?;
+        self.get_known_device(&device_id)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("created known device disappeared"))
+    }
+
+    pub async fn list_known_devices(&self) -> Result<Vec<KnownDevice>> {
+        let rows = list_known_device_rows(&self.pool).await?;
+
+        let mut out = Vec::with_capacity(rows.len());
+        for row in rows {
+            let device_id = known_device_row_device_id(&row).to_string();
+            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 = get_known_device_row(&self.pool, device_id).await?;
+
+        match row {
+            Some(row) => self.known_device_from_row(row, device_id).await.map(Some),
+            None => Ok(None),
+        }
+    }
+
+    pub async fn forget_known_device(&self, device_id: &str) -> Result<bool> {
+        let result = sqlx::query!("DELETE FROM known_devices WHERE device_id = ?1", device_id)
+            .execute(&self.pool)
+            .await
+            .context("failed deleting known device")?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    pub async fn attach_device_identifier(
+        &self,
+        device_id: &str,
+        input: DeviceIdentifierInput,
+    ) -> Result<Option<KnownDevice>> {
+        let identifier = normalize_device_identifier(input)?;
+        let now = now_unix();
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting device identifier transaction")?;
+        let exists = sqlx::query_scalar!(
+            r#"SELECT COUNT(*) as "count!: i64" FROM known_devices WHERE device_id = ?1"#,
+            device_id
+        )
+        .fetch_one(&mut *tx)
+        .await
+        .context("failed checking known device existence")?;
+        if exists == 0 {
+            return Ok(None);
+        }
+
+        insert_device_identifier_tx(&mut tx, device_id, &identifier, now).await?;
+        let now_i64 = i64::try_from(now).context("known device timestamp overflow")?;
+        sqlx::query!(
+            "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")?;
+        tx.commit()
+            .await
+            .context("failed committing device identifier transaction")?;
+        self.get_known_device(device_id).await
+    }
+
+    pub async fn attach_observation_identifier(
+        &self,
+        device_id: &str,
+        observation_key: &str,
+    ) -> Result<Option<KnownDevice>> {
+        let observation = get_observation_identifier_row(&self.pool, observation_key).await?;
+
+        let Some(observation) = observation else {
+            anyhow::bail!("observation not found");
+        };
+        let input = observation_identifier_to_input(observation)?;
+
+        self.attach_device_identifier(device_id, input).await
+    }
+
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub async fn lookup_known_device_by_identifier(
+        &self,
+        input: DeviceIdentifierInput,
+    ) -> Result<Option<KnownDevice>> {
+        let identifier = normalize_device_identifier(input)?;
+        let identifier_key = normalized_identifier_key_owned(identifier);
+        let device_id = sqlx::query_scalar!(
+            "SELECT device_id FROM device_identifiers WHERE identifier_key = ?1",
+            identifier_key
+        )
+        .fetch_optional(&self.pool)
+        .await
+        .context("failed looking up known device identifier")?;
+        match device_id {
+            Some(device_id) => self.get_known_device(&device_id).await,
+            None => Ok(None),
+        }
+    }
+
+    async fn known_device_from_row(
+        &self,
+        row: KnownDeviceRow,
+        device_id: &str,
+    ) -> Result<KnownDevice> {
+        let identifiers = self.list_device_identifiers(device_id).await?;
+        known_device_from_row_and_identifiers(row, identifiers)
+    }
+
+    async fn list_device_identifiers(&self, device_id: &str) -> Result<Vec<DeviceIdentifier>> {
+        let rows = list_device_identifier_rows(&self.pool, device_id).await?;
+        rows.into_iter().map(device_identifier_from_row).collect()
+    }
+}

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

@@ -0,0 +1,216 @@
+use super::*;
+
+impl Store {
+    pub async fn enroll(&self, enroll_token: &str) -> Result<IssuedAgent> {
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting enroll transaction")?;
+        let expires_at_unix = sqlx::query_scalar!(
+            "SELECT expires_at_unix FROM enroll_tokens WHERE token = ?1",
+            enroll_token
+        )
+        .fetch_optional(&mut *tx)
+        .await
+        .context("failed reading enroll token")?;
+
+        let Some(expires_at_unix) = expires_at_unix else {
+            warn!("rejecting enroll attempt with invalid, expired, or consumed token");
+            anyhow::bail!("invalid or already-used enroll token");
+        };
+
+        let now = now_unix();
+        if expires_at_unix as u64 <= now {
+            sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", enroll_token)
+                .execute(&mut *tx)
+                .await
+                .context("failed removing expired enroll token")?;
+            tx.commit()
+                .await
+                .context("failed committing expired enroll token removal")?;
+            warn!(
+                expires_at_unix,
+                now_unix = now,
+                "rejecting expired enroll token"
+            );
+            anyhow::bail!("enroll token has expired");
+        }
+
+        sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", enroll_token)
+            .execute(&mut *tx)
+            .await
+            .context("failed consuming enroll token")?;
+
+        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)",
+            agent_id,
+            agent_token
+        )
+        .execute(&mut *tx)
+        .await
+        .context("failed persisting agent credentials")?;
+
+        tx.commit()
+            .await
+            .context("failed committing state db after enroll")?;
+        info!(agent_id = %agent_id, "issued persistent agent credentials");
+
+        Ok(IssuedAgent {
+            agent_id,
+            agent_token,
+        })
+    }
+
+    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));
+        let expires_at_unix_i64 =
+            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")?;
+        info!(expires_at_unix, "persisted new enroll token");
+        Ok(IssuedEnrollToken {
+            enroll_token: token,
+            expires_at_unix,
+        })
+    }
+
+    pub async fn list_enroll_tokens(&self) -> Result<Vec<EnrollTokenInfo>> {
+        let now = now_unix();
+        let rows = list_enroll_token_rows(&self.pool).await?;
+        rows.into_iter()
+            .map(|row| enroll_token_info_from_row(row, now))
+            .collect()
+    }
+
+    pub async fn revoke_enroll_token(&self, token: &str) -> Result<bool> {
+        let result = sqlx::query!("DELETE FROM enroll_tokens WHERE token = ?1", token)
+            .execute(&self.pool)
+            .await
+            .context("failed removing enroll token")?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    pub async fn revoke_agent(&self, agent_id: &str) -> Result<bool> {
+        let mut tx = self
+            .pool
+            .begin()
+            .await
+            .context("failed starting revoke transaction")?;
+        let result = sqlx::query!("DELETE FROM agents WHERE agent_id = ?1", 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", agent_id)
+                .execute(&mut *tx)
+                .await
+                .context("failed removing agent metadata")?;
+        }
+        tx.commit()
+            .await
+            .context("failed committing db after agent revoke")?;
+        Ok(result.rows_affected() > 0)
+    }
+
+    pub async fn set_agent_nickname(&self, agent_id: &str, nickname: Option<&str>) -> Result<bool> {
+        let exists = sqlx::query_scalar!(
+            r#"SELECT COUNT(*) as "count!: i64" FROM agents WHERE agent_id = ?1"#,
+            agent_id
+        )
+        .fetch_one(&self.pool)
+        .await
+        .context("failed checking agent existence")?;
+        if exists == 0 {
+            return Ok(false);
+        }
+
+        let normalized = nickname.map(str::trim).filter(|v| !v.is_empty());
+        if let Some(value) = normalized {
+            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
+            )
+            .execute(&self.pool)
+            .await
+            .context("failed persisting agent nickname")?;
+        } else {
+            sqlx::query!("DELETE FROM agent_meta WHERE agent_id = ?1", agent_id)
+                .execute(&self.pool)
+                .await
+                .context("failed clearing agent nickname")?;
+        }
+
+        Ok(true)
+    }
+
+    pub async fn verify_agent_token(&self, agent_id: &str, token: &str) -> bool {
+        match sqlx::query_scalar!(
+            "SELECT agent_token FROM agents WHERE agent_id = ?1",
+            agent_id
+        )
+        .fetch_optional(&self.pool)
+        .await
+        {
+            Ok(Some(value)) => value == token,
+            Ok(None) => false,
+            Err(err) => {
+                warn!(error = %err, agent_id = %agent_id, "failed to read agent token from state db");
+                false
+            }
+        }
+    }
+
+    pub async fn list_agents(&self) -> Vec<String> {
+        match sqlx::query_scalar!(r#"SELECT agent_id as "agent_id!" FROM agents ORDER BY agent_id"#)
+            .fetch_all(&self.pool)
+            .await
+        {
+            Ok(out) => out,
+            Err(err) => {
+                warn!(error = %err, "failed to list agents from state db");
+                Vec::new()
+            }
+        }
+    }
+
+    pub async fn list_agents_with_nicknames(&self) -> Vec<(String, Option<String>)> {
+        let rows = sqlx::query!(
+            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"#,
+        )
+        .fetch_all(&self.pool)
+        .await;
+
+        match rows {
+            Ok(rows) => rows
+                .into_iter()
+                .map(|row| {
+                    let nickname = row
+                        .nickname
+                        .map(|v| v.trim().to_string())
+                        .filter(|v| !v.is_empty());
+                    (row.agent_id, nickname)
+                })
+                .collect(),
+            Err(err) => {
+                warn!(error = %err, "failed to list agents from state db");
+                Vec::new()
+            }
+        }
+    }
+}

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

@@ -0,0 +1,6 @@
+use super::*;
+
+pub(super) mod alerts_audit;
+pub(super) mod core;
+pub(super) mod legacy;
+pub(super) mod rows;

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

@@ -0,0 +1,211 @@
+use super::*;
+
+pub(in crate::state::store) async fn insert_audit_event(
+    pool: &SqlitePool,
+    key: &str,
+    event: &AuditEvent,
+) -> Result<()> {
+    let metadata_json =
+        serde_json::to_string(&event.metadata).context("failed to encode audit metadata")?;
+    sqlx::query(
+        "INSERT OR REPLACE 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)",
+    )
+    .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(pool)
+    .await
+    .context("failed persisting audit event")?;
+    Ok(())
+}
+
+pub(in crate::state::store) async fn insert_active_alert(
+    tx: &mut Transaction<'_, Sqlite>,
+    alert: &AlertState,
+) -> Result<()> {
+    let metadata_json =
+        serde_json::to_string(&alert.metadata).context("failed to encode active alert metadata")?;
+    sqlx::query(
+        "INSERT INTO active_alerts
+         (alert_id, kind, severity, status, agent_id, message, value, threshold,
+          last_seen_unix, metadata_json)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
+    )
+    .bind(&alert.alert_id)
+    .bind(&alert.kind)
+    .bind(&alert.severity)
+    .bind(&alert.status)
+    .bind(&alert.agent_id)
+    .bind(&alert.message)
+    .bind(i64::try_from(alert.value).context("active alert value overflow")?)
+    .bind(i64::try_from(alert.threshold).context("active alert threshold overflow")?)
+    .bind(i64::try_from(alert.last_seen_unix).context("active alert timestamp overflow")?)
+    .bind(metadata_json)
+    .execute(&mut **tx)
+    .await
+    .context("failed writing active alert snapshot")?;
+    Ok(())
+}
+
+pub(in crate::state::store) async fn insert_active_alert_pool(
+    pool: &SqlitePool,
+    alert: &AlertState,
+) -> Result<()> {
+    let metadata_json =
+        serde_json::to_string(&alert.metadata).context("failed to encode active alert metadata")?;
+    sqlx::query(
+        "INSERT OR REPLACE INTO active_alerts
+         (alert_id, kind, severity, status, agent_id, message, value, threshold,
+          last_seen_unix, metadata_json)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
+    )
+    .bind(&alert.alert_id)
+    .bind(&alert.kind)
+    .bind(&alert.severity)
+    .bind(&alert.status)
+    .bind(&alert.agent_id)
+    .bind(&alert.message)
+    .bind(i64::try_from(alert.value).context("active alert value overflow")?)
+    .bind(i64::try_from(alert.threshold).context("active alert threshold overflow")?)
+    .bind(i64::try_from(alert.last_seen_unix).context("active alert timestamp overflow")?)
+    .bind(metadata_json)
+    .execute(pool)
+    .await
+    .context("failed writing active alert snapshot")?;
+    Ok(())
+}
+
+pub(in crate::state::store) async fn insert_alert_transition(
+    tx: &mut Transaction<'_, Sqlite>,
+    key: &str,
+    transition: &AlertTransition,
+) -> Result<()> {
+    let metadata_json = serde_json::to_string(&transition.metadata)
+        .context("failed to encode alert transition metadata")?;
+    sqlx::query(
+        "INSERT INTO alert_transitions
+         (transition_key, transition_id, ts_unix, alert_id, kind, agent_id,
+          from_status, to_status, message, metadata_json)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
+    )
+    .bind(key)
+    .bind(&transition.transition_id)
+    .bind(i64::try_from(transition.ts_unix).context("alert timestamp overflow")?)
+    .bind(&transition.alert_id)
+    .bind(&transition.kind)
+    .bind(&transition.agent_id)
+    .bind(&transition.from_status)
+    .bind(&transition.to_status)
+    .bind(&transition.message)
+    .bind(metadata_json)
+    .execute(&mut **tx)
+    .await
+    .context("failed persisting alert transition")?;
+    Ok(())
+}
+
+pub(in crate::state::store) async fn insert_alert_transition_pool(
+    pool: &SqlitePool,
+    key: &str,
+    transition: &AlertTransition,
+) -> Result<()> {
+    let metadata_json = serde_json::to_string(&transition.metadata)
+        .context("failed to encode alert transition metadata")?;
+    sqlx::query(
+        "INSERT OR REPLACE INTO alert_transitions
+         (transition_key, transition_id, ts_unix, alert_id, kind, agent_id,
+          from_status, to_status, message, metadata_json)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
+    )
+    .bind(key)
+    .bind(&transition.transition_id)
+    .bind(i64::try_from(transition.ts_unix).context("alert timestamp overflow")?)
+    .bind(&transition.alert_id)
+    .bind(&transition.kind)
+    .bind(&transition.agent_id)
+    .bind(&transition.from_status)
+    .bind(&transition.to_status)
+    .bind(&transition.message)
+    .bind(metadata_json)
+    .execute(pool)
+    .await
+    .context("failed persisting alert transition")?;
+    Ok(())
+}
+
+pub(in crate::state::store) fn audit_event_from_row(
+    row: sqlx::sqlite::SqliteRow,
+) -> Result<AuditEvent> {
+    let ts_unix: i64 = row.try_get("ts_unix")?;
+    let latency_ms: Option<i64> = row.try_get("latency_ms")?;
+    let metadata_json: String = row.try_get("metadata_json")?;
+    Ok(AuditEvent {
+        event_id: row.try_get("event_id")?,
+        ts_unix: u64::try_from(ts_unix).context("negative audit timestamp in state db")?,
+        actor_type: row.try_get("actor_type")?,
+        actor_id: row.try_get("actor_id")?,
+        agent_id: row.try_get("agent_id")?,
+        request_id: row.try_get("request_id")?,
+        event_type: row.try_get("event_type")?,
+        outcome: row.try_get("outcome")?,
+        latency_ms: latency_ms
+            .map(u64::try_from)
+            .transpose()
+            .context("negative audit latency in state db")?,
+        message: row.try_get("message")?,
+        metadata: serde_json::from_str(&metadata_json).context("failed decoding audit metadata")?,
+    })
+}
+
+pub(in crate::state::store) fn alert_state_from_row(row: AlertStateRow) -> Result<AlertState> {
+    Ok(AlertState {
+        alert_id: row.alert_id,
+        kind: row.kind,
+        severity: row.severity,
+        status: row.status,
+        agent_id: row.agent_id,
+        message: row.message,
+        value: u64::try_from(row.value).context("negative active alert value in state db")?,
+        threshold: u64::try_from(row.threshold)
+            .context("negative active alert threshold in state db")?,
+        last_seen_unix: u64::try_from(row.last_seen_unix)
+            .context("negative active alert timestamp in state db")?,
+        metadata: serde_json::from_str(&row.metadata_json)
+            .context("failed decoding active alert metadata")?,
+    })
+}
+
+pub(in crate::state::store) fn alert_transition_from_row(
+    row: AlertTransitionRow,
+) -> Result<AlertTransition> {
+    Ok(AlertTransition {
+        transition_id: row.transition_id,
+        ts_unix: u64::try_from(row.ts_unix).context("negative alert timestamp in state db")?,
+        alert_id: row.alert_id,
+        kind: row.kind,
+        agent_id: row.agent_id,
+        from_status: row.from_status,
+        to_status: row.to_status,
+        message: row.message,
+        metadata: serde_json::from_str(&row.metadata_json)
+            .context("failed decoding alert transition metadata")?,
+    })
+}

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

@@ -0,0 +1,393 @@
+use super::*;
+
+pub(in crate::state::store) async fn open_sqlite_pool(path: &Path) -> Result<SqlitePool> {
+    let options = SqliteConnectOptions::new()
+        .filename(path)
+        .create_if_missing(true)
+        .foreign_keys(true)
+        .journal_mode(SqliteJournalMode::Wal)
+        .synchronous(SqliteSynchronous::Normal);
+    SqlitePoolOptions::new()
+        .max_connections(5)
+        .connect_with(options)
+        .await
+        .with_context(|| format!("failed to open SQLite state db {}", path.display()))
+}
+
+pub(in crate::state::store) async fn sql_count(pool: &SqlitePool, table: &str) -> Result<i64> {
+    let sql = format!("SELECT COUNT(*) FROM {table}");
+    sqlx::query_scalar::<_, i64>(&sql)
+        .fetch_one(pool)
+        .await
+        .with_context(|| format!("failed counting {table}"))
+}
+
+pub(in crate::state::store) fn now_unix() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+pub(in crate::state::store) fn decode_expiry(raw: &[u8]) -> Result<u64> {
+    if raw.len() != 8 {
+        anyhow::bail!("invalid token expiry length {}", raw.len());
+    }
+    let mut arr = [0u8; 8];
+    arr.copy_from_slice(raw);
+    Ok(u64::from_le_bytes(arr))
+}
+
+pub(in crate::state::store) fn decode_schema(raw: &[u8]) -> Result<u32> {
+    if raw.len() != 4 {
+        anyhow::bail!("invalid schema version length {}", raw.len());
+    }
+    let mut arr = [0u8; 4];
+    arr.copy_from_slice(raw);
+    Ok(u32::from_le_bytes(arr))
+}
+
+pub(in crate::state::store) fn seeded_enroll_token_key(token: &str) -> String {
+    format!("{SEEDED_ENROLL_TOKEN_PREFIX}{token}")
+}
+
+pub(in crate::state::store) fn normalize_required_text(value: &str, field: &str) -> Result<String> {
+    let normalized = value.trim();
+    if normalized.is_empty() {
+        anyhow::bail!("{field} must not be empty");
+    }
+    Ok(normalized.to_string())
+}
+
+pub(in crate::state::store) fn normalize_optional_text(value: Option<&str>) -> Option<String> {
+    value
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(ToOwned::to_owned)
+}
+
+pub(in crate::state::store) fn normalize_device_identifier(
+    input: DeviceIdentifierInput,
+) -> Result<NormalizedDeviceIdentifier> {
+    let kind = normalize_required_text(&input.kind, "identifier kind")?.to_ascii_lowercase();
+    let value = normalize_required_text(&input.value, "identifier value")?.to_ascii_lowercase();
+    let identifier_key = format!("{kind}:{value}");
+    Ok(NormalizedDeviceIdentifier {
+        identifier_key,
+        kind,
+        value,
+    })
+}
+
+#[cfg_attr(not(test), allow(dead_code))]
+pub(in crate::state::store) fn normalized_identifier_key_owned(
+    identifier: NormalizedDeviceIdentifier,
+) -> String {
+    identifier.identifier_key
+}
+
+pub(in crate::state::store) async fn list_enroll_token_rows(
+    pool: &SqlitePool,
+) -> Result<Vec<EnrollTokenRow>> {
+    sqlx::query_as!(
+        EnrollTokenRow,
+        r#"SELECT token as "token!", expires_at_unix FROM enroll_tokens ORDER BY expires_at_unix, token"#,
+    )
+    .fetch_all(pool)
+    .await
+    .context("failed listing enroll tokens")
+}
+
+pub(in crate::state::store) fn enroll_token_info_from_row(
+    row: EnrollTokenRow,
+    now: u64,
+) -> Result<EnrollTokenInfo> {
+    let expires_at_unix =
+        u64::try_from(row.expires_at_unix).context("negative token expiry in state db")?;
+    Ok(EnrollTokenInfo {
+        enroll_token: row.token,
+        expires_at_unix,
+        expired: expires_at_unix <= now,
+    })
+}
+
+pub(in crate::state::store) async fn list_known_device_rows(
+    pool: &SqlitePool,
+) -> Result<Vec<KnownDeviceRow>> {
+    sqlx::query_as!(
+        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"#,
+    )
+    .fetch_all(pool)
+    .await
+    .context("failed listing known devices")
+}
+
+pub(in crate::state::store) async fn get_known_device_row(
+    pool: &SqlitePool,
+    device_id: &str,
+) -> Result<Option<KnownDeviceRow>> {
+    sqlx::query_as!(
+        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"#,
+        device_id
+    )
+    .fetch_optional(pool)
+    .await
+    .context("failed reading known device")
+}
+
+pub(in crate::state::store) fn known_device_row_device_id(row: &KnownDeviceRow) -> &str {
+    &row.device_id
+}
+
+pub(in crate::state::store) fn known_device_from_row_and_identifiers(
+    row: KnownDeviceRow,
+    identifiers: Vec<DeviceIdentifier>,
+) -> Result<KnownDevice> {
+    Ok(KnownDevice {
+        device_id: row.device_id,
+        display_name: row.display_name,
+        pinned: row.pinned != 0,
+        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(row.updated_at_unix)
+            .context("negative known device updated timestamp in state db")?,
+        notes: row.notes,
+        identifiers,
+    })
+}
+
+pub(in crate::state::store) async fn list_device_identifier_rows(
+    pool: &SqlitePool,
+    device_id: &str,
+) -> Result<Vec<DeviceIdentifierRow>> {
+    sqlx::query_as!(
+        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"#,
+        device_id
+    )
+    .fetch_all(pool)
+    .await
+    .context("failed listing device identifiers")
+}
+
+pub(in crate::state::store) async fn get_observation_identifier_row(
+    pool: &SqlitePool,
+    observation_key: &str,
+) -> Result<Option<ObservationIdentifierRow>> {
+    sqlx::query_as!(
+        ObservationIdentifierRow,
+        r#"SELECT mac, ip
+         FROM agent_device_observations
+         WHERE observation_key = ?1"#,
+        observation_key
+    )
+    .fetch_optional(pool)
+    .await
+    .context("failed reading observation identifier")
+}
+
+pub(in crate::state::store) async fn get_observation_current_row(
+    tx: &mut Transaction<'_, Sqlite>,
+    observation_key: &str,
+) -> Result<Option<ObservationCurrentRow>> {
+    sqlx::query_as!(
+        ObservationCurrentRow,
+        r#"SELECT mac, ip, hostname, first_seen_unix, last_seen_unix,
+                  last_action as "last_action!"
+           FROM agent_device_observations
+           WHERE observation_key = ?1"#,
+        observation_key
+    )
+    .fetch_optional(&mut **tx)
+    .await
+    .context("failed reading current observation")
+}
+
+pub(in crate::state::store) fn observation_current_changed(
+    current: Option<&ObservationCurrentRow>,
+    next: &AgentDeviceObservation,
+    first_seen_unix: i64,
+    last_seen_unix: i64,
+) -> bool {
+    let Some(current) = current else {
+        return true;
+    };
+    current.mac != next.mac
+        || current.ip != next.ip
+        || current.hostname != next.hostname
+        || current.first_seen_unix != first_seen_unix
+        || current.last_seen_unix != last_seen_unix
+        || current.last_action != next.last_action
+}
+
+pub(in crate::state::store) fn observation_identifier_to_input(
+    observation: ObservationIdentifierRow,
+) -> Result<DeviceIdentifierInput> {
+    observation
+        .mac
+        .map(|value| DeviceIdentifierInput {
+            kind: "mac".into(),
+            value,
+        })
+        .or_else(|| {
+            observation.ip.map(|value| DeviceIdentifierInput {
+                kind: "ip".into(),
+                value,
+            })
+        })
+        .ok_or_else(|| anyhow::anyhow!("observation has no attachable mac or ip"))
+}
+
+pub(in crate::state::store) 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,
+    })
+}
+
+pub(in crate::state::store) async fn insert_device_identifier_tx(
+    tx: &mut Transaction<'_, Sqlite>,
+    device_id: &str,
+    identifier: &NormalizedDeviceIdentifier,
+    created_at_unix: u64,
+) -> Result<()> {
+    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
+    )
+    .execute(&mut **tx)
+    .await
+    .with_context(|| {
+        format!(
+            "failed attaching device identifier {} to {}",
+            identifier.identifier_key, device_id
+        )
+    })?;
+    Ok(())
+}
+
+pub(in crate::state::store) fn device_identifier_from_row(
+    row: DeviceIdentifierRow,
+) -> Result<DeviceIdentifier> {
+    Ok(DeviceIdentifier {
+        identifier_key: row.identifier_key,
+        device_id: row.device_id,
+        kind: row.kind,
+        value: row.value,
+        created_at_unix: u64::try_from(row.created_at_unix)
+            .context("negative device identifier timestamp in state db")?,
+    })
+}
+
+#[cfg_attr(not(test), allow(dead_code))]
+pub(in crate::state::store) 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,
+    })
+}
+
+pub(in crate::state::store) fn agent_observation_view_from_row(
+    row: AgentObservationViewRow,
+) -> Result<AgentDeviceObservationView> {
+    let known_device = match (row.device_id, row.display_name, row.pinned) {
+        (Some(device_id), Some(display_name), Some(pinned)) => Some(KnownDeviceSummary {
+            device_id,
+            display_name,
+            pinned: pinned != 0,
+        }),
+        _ => None,
+    };
+    Ok(AgentDeviceObservationView {
+        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,
+        known_device,
+    })
+}
+
+pub(in crate::state::store) fn agent_observation_event_from_row(
+    row: AgentObservationEventRow,
+) -> Result<AgentDeviceObservationEvent> {
+    let known_device = match (row.device_id, row.display_name, row.pinned) {
+        (Some(device_id), Some(display_name), Some(pinned)) => Some(KnownDeviceSummary {
+            device_id,
+            display_name,
+            pinned: pinned != 0,
+        }),
+        _ => None,
+    };
+    Ok(AgentDeviceObservationEvent {
+        event_id: row.event_id,
+        observation_key: row.observation_key,
+        agent_id: row.agent_id,
+        kind: row.kind,
+        action: row.action,
+        mac: row.mac,
+        ip: row.ip,
+        hostname: row.hostname,
+        ts_unix: u64::try_from(row.ts_unix)
+            .context("negative observation event timestamp in state db")?,
+        known_device,
+    })
+}

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

@@ -0,0 +1,156 @@
+use super::*;
+
+pub(in crate::state::store) async fn import_tree_raw(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+    sled_tree: &str,
+    sql_table: &str,
+    key_col: &str,
+    value_col: &str,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree(sled_tree)
+        .with_context(|| format!("failed to open legacy {sled_tree} tree"))?;
+    let sql =
+        format!("INSERT OR REPLACE INTO {sql_table} ({key_col}, {value_col}) VALUES (?1, ?2)");
+    for item in tree.iter() {
+        let (key, value) =
+            item.with_context(|| format!("failed reading legacy {sled_tree} tree"))?;
+        let key = String::from_utf8(key.to_vec())
+            .with_context(|| format!("legacy {sled_tree} key is not utf-8"))?;
+        sqlx::query(&sql)
+            .bind(key)
+            .bind(value.to_vec())
+            .execute(pool)
+            .await
+            .with_context(|| format!("failed importing legacy {sled_tree} row"))?;
+    }
+    Ok(())
+}
+
+pub(in crate::state::store) async fn import_enroll_tokens(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree("enroll_tokens")
+        .context("failed to open legacy enroll_tokens tree")?;
+    for item in tree.iter() {
+        let (token, expiry) = item.context("failed reading legacy enroll token")?;
+        let token =
+            String::from_utf8(token.to_vec()).context("legacy enroll token key is not utf-8")?;
+        let expires_at_unix = decode_expiry(expiry.as_ref())?;
+        sqlx::query(
+            "INSERT OR REPLACE INTO enroll_tokens (token, expires_at_unix) VALUES (?1, ?2)",
+        )
+        .bind(token)
+        .bind(i64::try_from(expires_at_unix).context("legacy token expiry overflow")?)
+        .execute(pool)
+        .await
+        .context("failed importing legacy enroll token")?;
+    }
+    Ok(())
+}
+
+pub(in crate::state::store) async fn import_agents(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree("agents")
+        .context("failed to open legacy agents tree")?;
+    for item in tree.iter() {
+        let (agent_id, agent_token) = item.context("failed reading legacy agent")?;
+        let agent_id =
+            String::from_utf8(agent_id.to_vec()).context("legacy agent id is not utf-8")?;
+        let agent_token =
+            String::from_utf8(agent_token.to_vec()).context("legacy agent token is not utf-8")?;
+        sqlx::query("INSERT OR REPLACE INTO agents (agent_id, agent_token) VALUES (?1, ?2)")
+            .bind(agent_id)
+            .bind(agent_token)
+            .execute(pool)
+            .await
+            .context("failed importing legacy agent")?;
+    }
+    Ok(())
+}
+
+pub(in crate::state::store) async fn import_agent_meta(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree("agent_meta")
+        .context("failed to open legacy agent_meta tree")?;
+    for item in tree.iter() {
+        let (agent_id, nickname) = item.context("failed reading legacy agent metadata")?;
+        let agent_id =
+            String::from_utf8(agent_id.to_vec()).context("legacy agent id is not utf-8")?;
+        let nickname =
+            String::from_utf8(nickname.to_vec()).context("legacy nickname is not utf-8")?;
+        sqlx::query("INSERT OR REPLACE INTO agent_meta (agent_id, nickname) VALUES (?1, ?2)")
+            .bind(agent_id)
+            .bind(nickname)
+            .execute(pool)
+            .await
+            .context("failed importing legacy agent metadata")?;
+    }
+    Ok(())
+}
+
+pub(in crate::state::store) async fn import_audit_events(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree("audit_events")
+        .context("failed to open legacy audit_events tree")?;
+    for item in tree.iter() {
+        let (key, value) = item.context("failed reading legacy audit event")?;
+        let key = String::from_utf8(key.to_vec()).context("legacy audit key is not utf-8")?;
+        let event: AuditEvent =
+            serde_json::from_slice(value.as_ref()).context("failed decoding legacy audit event")?;
+        insert_audit_event(pool, &key, &event)
+            .await
+            .context("failed importing legacy audit event")?;
+    }
+    Ok(())
+}
+
+pub(in crate::state::store) async fn import_active_alerts(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree("active_alerts")
+        .context("failed to open legacy active_alerts tree")?;
+    for item in tree.iter() {
+        let (_, value) = item.context("failed reading legacy active alert")?;
+        let alert: AlertState = serde_json::from_slice(value.as_ref())
+            .context("failed decoding legacy active alert")?;
+        insert_active_alert_pool(pool, &alert)
+            .await
+            .context("failed importing legacy active alert")?;
+    }
+    Ok(())
+}
+
+pub(in crate::state::store) async fn import_alert_transitions(
+    pool: &SqlitePool,
+    legacy: &sled::Db,
+) -> Result<()> {
+    let tree = legacy
+        .open_tree("alert_transitions")
+        .context("failed to open legacy alert_transitions tree")?;
+    for item in tree.iter() {
+        let (key, value) = item.context("failed reading legacy alert transition")?;
+        let key =
+            String::from_utf8(key.to_vec()).context("legacy alert transition key is not utf-8")?;
+        let transition: AlertTransition = serde_json::from_slice(value.as_ref())
+            .context("failed decoding legacy alert transition")?;
+        insert_alert_transition_pool(pool, &key, &transition)
+            .await
+            .context("failed importing legacy alert transition")?;
+    }
+    Ok(())
+}

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

@@ -0,0 +1,108 @@
+pub(in crate::state::store) struct NormalizedDeviceIdentifier {
+    pub(in crate::state::store) identifier_key: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) value: String,
+}
+
+pub(in crate::state::store) struct EnrollTokenRow {
+    pub(in crate::state::store) token: String,
+    pub(in crate::state::store) expires_at_unix: i64,
+}
+
+pub(in crate::state::store) struct KnownDeviceRow {
+    pub(in crate::state::store) device_id: String,
+    pub(in crate::state::store) display_name: String,
+    pub(in crate::state::store) pinned: i64,
+    pub(in crate::state::store) created_at_unix: i64,
+    pub(in crate::state::store) updated_at_unix: i64,
+    pub(in crate::state::store) notes: Option<String>,
+}
+
+pub(in crate::state::store) struct DeviceIdentifierRow {
+    pub(in crate::state::store) identifier_key: String,
+    pub(in crate::state::store) device_id: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) value: String,
+    pub(in crate::state::store) created_at_unix: i64,
+}
+
+pub(in crate::state::store) struct AgentObservationRow {
+    pub(in crate::state::store) observation_key: String,
+    pub(in crate::state::store) agent_id: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) mac: Option<String>,
+    pub(in crate::state::store) ip: Option<String>,
+    pub(in crate::state::store) hostname: Option<String>,
+    pub(in crate::state::store) first_seen_unix: i64,
+    pub(in crate::state::store) last_seen_unix: i64,
+    pub(in crate::state::store) last_action: String,
+}
+
+pub(in crate::state::store) struct AgentObservationViewRow {
+    pub(in crate::state::store) observation_key: String,
+    pub(in crate::state::store) agent_id: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) mac: Option<String>,
+    pub(in crate::state::store) ip: Option<String>,
+    pub(in crate::state::store) hostname: Option<String>,
+    pub(in crate::state::store) first_seen_unix: i64,
+    pub(in crate::state::store) last_seen_unix: i64,
+    pub(in crate::state::store) last_action: String,
+    pub(in crate::state::store) device_id: Option<String>,
+    pub(in crate::state::store) display_name: Option<String>,
+    pub(in crate::state::store) pinned: Option<i64>,
+}
+
+pub(in crate::state::store) struct AgentObservationEventRow {
+    pub(in crate::state::store) event_id: String,
+    pub(in crate::state::store) observation_key: String,
+    pub(in crate::state::store) agent_id: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) action: String,
+    pub(in crate::state::store) mac: Option<String>,
+    pub(in crate::state::store) ip: Option<String>,
+    pub(in crate::state::store) hostname: Option<String>,
+    pub(in crate::state::store) ts_unix: i64,
+    pub(in crate::state::store) device_id: Option<String>,
+    pub(in crate::state::store) display_name: Option<String>,
+    pub(in crate::state::store) pinned: Option<i64>,
+}
+
+pub(in crate::state::store) struct ObservationIdentifierRow {
+    pub(in crate::state::store) mac: Option<String>,
+    pub(in crate::state::store) ip: Option<String>,
+}
+
+pub(in crate::state::store) struct ObservationCurrentRow {
+    pub(in crate::state::store) mac: Option<String>,
+    pub(in crate::state::store) ip: Option<String>,
+    pub(in crate::state::store) hostname: Option<String>,
+    pub(in crate::state::store) first_seen_unix: i64,
+    pub(in crate::state::store) last_seen_unix: i64,
+    pub(in crate::state::store) last_action: String,
+}
+
+pub(in crate::state::store) struct AlertStateRow {
+    pub(in crate::state::store) alert_id: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) severity: String,
+    pub(in crate::state::store) status: String,
+    pub(in crate::state::store) agent_id: Option<String>,
+    pub(in crate::state::store) message: String,
+    pub(in crate::state::store) value: i64,
+    pub(in crate::state::store) threshold: i64,
+    pub(in crate::state::store) last_seen_unix: i64,
+    pub(in crate::state::store) metadata_json: String,
+}
+
+pub(in crate::state::store) struct AlertTransitionRow {
+    pub(in crate::state::store) transition_id: String,
+    pub(in crate::state::store) ts_unix: i64,
+    pub(in crate::state::store) alert_id: String,
+    pub(in crate::state::store) kind: String,
+    pub(in crate::state::store) agent_id: Option<String>,
+    pub(in crate::state::store) from_status: Option<String>,
+    pub(in crate::state::store) to_status: String,
+    pub(in crate::state::store) message: String,
+    pub(in crate::state::store) metadata_json: String,
+}

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

@@ -0,0 +1,61 @@
+use super::*;
+
+impl Store {
+    pub async fn import_sled_state(
+        from_sled_state: &Path,
+        to_state_file: &Path,
+        force: bool,
+    ) -> Result<()> {
+        if !from_sled_state.is_dir() {
+            anyhow::bail!(
+                "legacy sled state {} does not exist or is not a directory",
+                from_sled_state.display()
+            );
+        }
+        if to_state_file.is_dir() {
+            anyhow::bail!(
+                "target state file {} is a directory",
+                to_state_file.display()
+            );
+        }
+        if to_state_file.exists()
+            && to_state_file
+                .metadata()
+                .with_context(|| format!("failed to stat {}", to_state_file.display()))?
+                .len()
+                > 0
+        {
+            if !force {
+                anyhow::bail!(
+                    "target SQLite state file {} already exists and is non-empty; re-run with --force to overwrite",
+                    to_state_file.display()
+                );
+            }
+            std::fs::remove_file(to_state_file)
+                .with_context(|| format!("failed to remove {}", to_state_file.display()))?;
+        }
+
+        let legacy = sled::open(from_sled_state).with_context(|| {
+            format!(
+                "failed to open legacy sled state {}",
+                from_sled_state.display()
+            )
+        })?;
+        let store = Store::load_or_init(to_state_file, Vec::new(), Duration::from_secs(1)).await?;
+
+        import_tree_raw(&store.pool, &legacy, "meta", "meta", "key", "value").await?;
+        import_enroll_tokens(&store.pool, &legacy).await?;
+        import_agents(&store.pool, &legacy).await?;
+        import_agent_meta(&store.pool, &legacy).await?;
+        import_audit_events(&store.pool, &legacy).await?;
+        import_active_alerts(&store.pool, &legacy).await?;
+        import_alert_transitions(&store.pool, &legacy).await?;
+
+        info!(
+            from = %from_sled_state.display(),
+            to = %to_state_file.display(),
+            "imported legacy sled state into SQLite"
+        );
+        Ok(())
+    }
+}

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

@@ -0,0 +1,262 @@
+use super::*;
+
+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")?;
+            let current = get_observation_current_row(&mut tx, &observation.observation_key)
+                .await
+                .context("failed checking existing observation")?;
+            let append_event = observation_current_changed(
+                current.as_ref(),
+                &observation,
+                first_seen_unix,
+                last_seen_unix,
+            );
+            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")?;
+
+            if append_event {
+                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)
+    }
+
+    #[cfg_attr(not(test), allow(dead_code))]
+    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 list_agent_observation_views(
+        &self,
+        agent_id: Option<&str>,
+        limit: usize,
+    ) -> Result<Vec<AgentDeviceObservationView>> {
+        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!(
+                AgentObservationViewRow,
+                r#"SELECT observations.observation_key as "observation_key!",
+                        observations.agent_id as "agent_id!",
+                        observations.kind as "kind!",
+                        observations.mac,
+                        observations.ip,
+                        observations.hostname,
+                        observations.first_seen_unix,
+                        observations.last_seen_unix,
+                        observations.last_action as "last_action!",
+                        known_devices.device_id,
+                        known_devices.display_name,
+                        known_devices.pinned
+                 FROM agent_device_observations observations
+                 LEFT JOIN device_identifiers identifiers
+                   ON identifiers.identifier_key =
+                      CASE
+                        WHEN observations.mac IS NOT NULL THEN 'mac:' || observations.mac
+                        WHEN observations.ip IS NOT NULL THEN 'ip:' || observations.ip
+                      END
+                 LEFT JOIN known_devices ON known_devices.device_id = identifiers.device_id
+                 WHERE observations.agent_id = ?1
+                 ORDER BY observations.last_seen_unix DESC
+                 LIMIT ?2"#,
+                agent_id,
+                limit
+            )
+            .fetch_all(&self.pool)
+            .await
+        } else {
+            sqlx::query_as!(
+                AgentObservationViewRow,
+                r#"SELECT observations.observation_key as "observation_key!",
+                        observations.agent_id as "agent_id!",
+                        observations.kind as "kind!",
+                        observations.mac,
+                        observations.ip,
+                        observations.hostname,
+                        observations.first_seen_unix,
+                        observations.last_seen_unix,
+                        observations.last_action as "last_action!",
+                        known_devices.device_id,
+                        known_devices.display_name,
+                        known_devices.pinned
+                 FROM agent_device_observations observations
+                 LEFT JOIN device_identifiers identifiers
+                   ON identifiers.identifier_key =
+                      CASE
+                        WHEN observations.mac IS NOT NULL THEN 'mac:' || observations.mac
+                        WHEN observations.ip IS NOT NULL THEN 'ip:' || observations.ip
+                      END
+                 LEFT JOIN known_devices ON known_devices.device_id = identifiers.device_id
+                 ORDER BY observations.last_seen_unix DESC
+                 LIMIT ?1"#,
+                limit
+            )
+            .fetch_all(&self.pool)
+            .await
+        }
+        .context("failed listing agent observation views")?;
+        rows.into_iter()
+            .map(agent_observation_view_from_row)
+            .collect()
+    }
+
+    pub async fn list_agent_observation_events(
+        &self,
+        agent_id: Option<&str>,
+        kind: Option<&str>,
+        mac: Option<&str>,
+        ip: Option<&str>,
+        observation_key: Option<&str>,
+        limit: usize,
+    ) -> Result<Vec<AgentDeviceObservationEvent>> {
+        let limit = limit.clamp(1, 1000);
+        let limit = i64::try_from(limit).context("observation event limit overflow")?;
+        let rows = sqlx::query_as!(
+            AgentObservationEventRow,
+            r#"SELECT events.event_id as "event_id!",
+                    ('agent:' || events.agent_id || ':' || events.kind || ':' ||
+                        CASE
+                            WHEN events.mac IS NOT NULL THEN 'mac:' || events.mac
+                            WHEN events.ip IS NOT NULL THEN 'ip:' || events.ip
+                            ELSE ''
+                        END) as "observation_key!",
+                    events.agent_id as "agent_id!",
+                    events.kind as "kind!",
+                    events.action as "action!",
+                    events.mac,
+                    events.ip,
+                    events.hostname,
+                    events.ts_unix,
+                    known_devices.device_id,
+                    known_devices.display_name,
+                    known_devices.pinned
+             FROM agent_device_observation_events events
+             LEFT JOIN device_identifiers identifiers
+               ON identifiers.identifier_key =
+                  CASE
+                    WHEN events.mac IS NOT NULL THEN 'mac:' || events.mac
+                    WHEN events.ip IS NOT NULL THEN 'ip:' || events.ip
+                  END
+             LEFT JOIN known_devices ON known_devices.device_id = identifiers.device_id
+             WHERE (?1 IS NULL OR events.agent_id = ?1)
+               AND (?2 IS NULL OR events.kind = ?2)
+               AND (?3 IS NULL OR events.mac = ?3)
+               AND (?4 IS NULL OR events.ip = ?4)
+               AND (?5 IS NULL OR ('agent:' || events.agent_id || ':' || events.kind || ':' ||
+                    CASE
+                        WHEN events.mac IS NOT NULL THEN 'mac:' || events.mac
+                        WHEN events.ip IS NOT NULL THEN 'ip:' || events.ip
+                        ELSE ''
+                    END) = ?5)
+             ORDER BY events.ts_unix DESC
+             LIMIT ?6"#,
+            agent_id,
+            kind,
+            mac,
+            ip,
+            observation_key,
+            limit
+        )
+        .fetch_all(&self.pool)
+        .await
+        .context("failed listing agent observation events")?;
+        rows.into_iter()
+            .map(agent_observation_event_from_row)
+            .collect()
+    }
+}

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

@@ -77,6 +77,41 @@ pub struct AgentDeviceObservation {
     pub last_action: String,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AgentDeviceObservationView {
+    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,
+    pub known_device: Option<KnownDeviceSummary>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AgentDeviceObservationEvent {
+    pub event_id: String,
+    pub observation_key: String,
+    pub agent_id: String,
+    pub kind: String,
+    pub action: String,
+    pub mac: Option<String>,
+    pub ip: Option<String>,
+    pub hostname: Option<String>,
+    pub ts_unix: u64,
+    pub known_device: Option<KnownDeviceSummary>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct KnownDeviceSummary {
+    pub device_id: String,
+    pub display_name: String,
+    pub pinned: bool,
+}
+
 #[derive(Debug, Clone)]
 pub struct AgentDeviceObservationInput {
     pub kind: String,

wakey-linux/Cargo.toml

@@ -13,6 +13,9 @@ serde_json = "1"
 tokio = { version = "1", features = ["fs", "net", "rt", "sync"] }
 wakey-core = { path = "../wakey-core", registry = "gitea", version = "0"}
 
+[dev-dependencies]
+serial_test = "3"
+
 [dependencies.lda-ipjs]
 path = "../ipjs"
 registry = "gitea"

wakey-linux/src/dhcp.rs

@@ -1,13 +1,18 @@
 use std::io::{self, ErrorKind};
 use std::net::IpAddr;
+use std::path::PathBuf;
 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 DEFAULT_DHCP_LEASES: &str = "/tmp/dhcp.leases";
-const OBSERVATION_STORE: &str = "/tmp/wakey_observations.json";
+const DEFAULT_MAC_NAME_CACHE: &str = "/tmp/wakey_mac_names.json";
+const DEFAULT_OBSERVATION_STORE: &str = "/tmp/wakey_observations.json";
+const DHCP_LEASES_ENV: &str = "WAKEY_DHCP_LEASES";
+const MAC_NAME_CACHE_ENV: &str = "WAKEY_MAC_NAME_CACHE";
+const OBSERVATION_STORE_ENV: &str = "WAKEY_OBSERVATION_STORE";
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct LocalObservationStore {
@@ -50,7 +55,13 @@ pub struct LocalDeviceObservation {
 
 /// 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>> {
-    match tokio::fs::read_to_string(MAC_NAME_CACHE).await {
+    load_mac_name_cache_from_path(mac_name_cache_path()).await
+}
+
+pub async fn load_mac_name_cache_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<std::collections::BTreeMap<String, String>> {
+    match tokio::fs::read_to_string(path).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),
@@ -59,13 +70,30 @@ pub async fn load_mac_name_cache() -> io::Result<std::collections::BTreeMap<Stri
 
 /// Persist the MAC-to-name cache back to disk.
 async fn save_mac_name_cache(map: &std::collections::BTreeMap<String, String>) -> io::Result<()> {
+    save_mac_name_cache_to_path(mac_name_cache_path(), map).await
+}
+
+async fn save_mac_name_cache_to_path(
+    path: impl AsRef<std::path::Path>,
+    map: &std::collections::BTreeMap<String, String>,
+) -> io::Result<()> {
     let s = serde_json::to_string(map).map_err(io::Error::other)?;
-    let _ = tokio::fs::write(MAC_NAME_CACHE, s).await;
+    let _ = tokio::fs::write(path, s).await;
     Ok(())
 }
 
 pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
-    match tokio::fs::read_to_string(OBSERVATION_STORE).await {
+    match tokio::fs::read_to_string(observation_store_path()).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),
+    }
+}
+
+pub async fn load_observation_store_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<LocalObservationStore> {
+    match tokio::fs::read_to_string(path).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),
@@ -74,11 +102,24 @@ pub async fn load_observation_store() -> io::Result<LocalObservationStore> {
 
 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
+    tokio::fs::write(observation_store_path(), s).await
 }
 
 pub async fn list_local_observations() -> io::Result<Vec<LocalDeviceObservation>> {
     let store = load_observation_store().await?;
+    list_local_observations_from_store(store)
+}
+
+pub async fn list_local_observations_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<Vec<LocalDeviceObservation>> {
+    let store = load_observation_store_from_path(path).await?;
+    list_local_observations_from_store(store)
+}
+
+fn list_local_observations_from_store(
+    store: LocalObservationStore,
+) -> io::Result<Vec<LocalDeviceObservation>> {
     let mut out = Vec::with_capacity(store.dhcp_clients.len() + store.neighbors.len());
     out.extend(
         store
@@ -126,6 +167,7 @@ pub async fn observe_dhcp_event(
     hostname: Option<&str>,
 ) -> io::Result<bool> {
     if !matches!(action, "add" | "update" | "old" | "remove") {
+        // old not emitted by hotplug
         return Ok(false);
     }
 
@@ -134,7 +176,7 @@ pub async fn observe_dhcp_event(
         .filter(|v| !v.is_empty() && *v != "*")
         .map(ToOwned::to_owned);
     let now = now_unix();
-    let mac_s = mac.to_string();
+    let mac_s = mac.to_string().to_ascii_lowercase();
 
     let mut store = load_observation_store().await.unwrap_or_default();
     let mut changed = false;
@@ -187,17 +229,18 @@ pub async fn observe_neighbor_event(
     ip: Option<IpAddr>,
 ) -> io::Result<bool> {
     if !matches!(action, "add" | "update" | "old" | "remove") {
+        // update and remove not emitted by hotplug
         return Ok(false);
     }
     let Some(key) = mac
-        .map(|value| format!("mac:{}", value))
+        .map(|value| format!("mac:{}", value.to_string().to_ascii_lowercase()))
         .or_else(|| ip.map(|value| format!("ip:{}", value)))
     else {
         return Ok(false);
     };
 
     let now = now_unix();
-    let mac = mac.map(|value| value.to_string());
+    let mac = mac.map(|value| value.to_string().to_ascii_lowercase());
     let mut store = load_observation_store().await.unwrap_or_default();
     let mut changed = false;
     store
@@ -248,9 +291,15 @@ pub fn parse_dhcp_lease_line(line: &str) -> Option<DhcpLease> {
     })
 }
 
-/// Read raw DHCP leases from `/tmp/dhcp.leases`.
+/// Read raw DHCP leases from the configured dnsmasq lease file.
 pub async fn read_dhcp_leases() -> io::Result<Vec<DhcpLease>> {
-    match tokio::fs::read_to_string("/tmp/dhcp.leases").await {
+    read_dhcp_leases_from_path(dhcp_leases_path()).await
+}
+
+pub async fn read_dhcp_leases_from_path(
+    path: impl AsRef<std::path::Path>,
+) -> io::Result<Vec<DhcpLease>> {
+    match tokio::fs::read_to_string(path).await {
         Ok(file) => Ok(file.lines().filter_map(parse_dhcp_lease_line).collect()),
         Err(e) if e.kind() == ErrorKind::NotFound => Ok(Vec::new()),
         Err(e) => Err(e),
@@ -259,9 +308,27 @@ 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?;
+    read_dhcp_leases_with_names_from_paths(
-    let observations = load_observation_store().await.unwrap_or_default();
+        dhcp_leases_path(),
-    let mut cache = load_mac_name_cache().await.unwrap_or_default();
+        observation_store_path(),
+        mac_name_cache_path(),
+    )
+    .await
+}
+
+pub async fn read_dhcp_leases_with_names_from_paths(
+    leases_path: impl AsRef<std::path::Path>,
+    observation_store_path: impl AsRef<std::path::Path>,
+    mac_name_cache_path: impl AsRef<std::path::Path>,
+) -> io::Result<Vec<DhcpLease>> {
+    let leases = read_dhcp_leases_from_path(leases_path).await?;
+    let observations = load_observation_store_from_path(observation_store_path)
+        .await
+        .unwrap_or_default();
+    let mac_name_cache_path = mac_name_cache_path.as_ref();
+    let mut cache = load_mac_name_cache_from_path(mac_name_cache_path)
+        .await
+        .unwrap_or_default();
     let mut changed = false;
     let mut leases_with_names = Vec::with_capacity(leases.len());
     for mut l in leases {
@@ -283,7 +350,7 @@ pub async fn read_dhcp_leases_with_names() -> io::Result<Vec<DhcpLease>> {
         leases_with_names.push(l);
     }
     if changed {
-        let _ = save_mac_name_cache(&cache).await;
+        let _ = save_mac_name_cache_to_path(mac_name_cache_path, &cache).await;
     }
     Ok(leases_with_names)
 }
@@ -295,6 +362,25 @@ fn now_unix() -> u64 {
         .unwrap_or(0)
 }
 
+fn dhcp_leases_path() -> PathBuf {
+    configured_path(DHCP_LEASES_ENV, DEFAULT_DHCP_LEASES)
+}
+
+fn mac_name_cache_path() -> PathBuf {
+    configured_path(MAC_NAME_CACHE_ENV, DEFAULT_MAC_NAME_CACHE)
+}
+
+fn observation_store_path() -> PathBuf {
+    configured_path(OBSERVATION_STORE_ENV, DEFAULT_OBSERVATION_STORE)
+}
+
+fn configured_path(env_key: &str, default: &str) -> PathBuf {
+    std::env::var_os(env_key)
+        .filter(|value| !value.is_empty())
+        .map(PathBuf::from)
+        .unwrap_or_else(|| PathBuf::from(default))
+}
+
 /// 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();
@@ -323,3 +409,92 @@ pub async fn enrich_leases_with_nud_state(leases: Vec<DhcpLease>) -> Vec<DhcpLea
         })
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serial_test::serial;
+
+    struct EnvGuard {
+        keys: Vec<&'static str>,
+    }
+
+    impl EnvGuard {
+        fn set(key: &'static str, value: &std::path::Path) -> Self {
+            let guard = Self { keys: vec![key] };
+            // SAFETY: these tests are serialized and do not spawn work that reads these
+            // environment variables outside the test body.
+            unsafe {
+                std::env::set_var(key, value);
+            }
+            guard
+        }
+    }
+
+    impl Drop for EnvGuard {
+        fn drop(&mut self) {
+            for key in &self.keys {
+                // SAFETY: these tests are serialized and do not spawn work that reads these
+                // environment variables outside the test body.
+                unsafe {
+                    std::env::remove_var(key);
+                }
+            }
+        }
+    }
+
+    fn temp_file(name: &str) -> PathBuf {
+        let nonce = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|duration| duration.as_nanos())
+            .unwrap_or(0);
+        std::env::temp_dir().join(format!("wakey-linux-{name}-{}-{nonce}", std::process::id()))
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn dhcp_lease_file_path_can_be_overridden() {
+        let path = temp_file("leases");
+        let _guard = EnvGuard::set(DHCP_LEASES_ENV, &path);
+        tokio::fs::write(&path, "1893456000 aa:bb:cc:dd:ee:ff 192.168.1.2 lda *\n")
+            .await
+            .expect("lease fixture should write");
+
+        let leases = read_dhcp_leases().await.expect("leases should read");
+        assert_eq!(leases.len(), 1);
+        assert_eq!(leases[0].name.as_deref(), Some("lda"));
+        assert_eq!(leases[0].ip.to_string(), "192.168.1.2");
+
+        let _ = tokio::fs::remove_file(path).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn observation_and_name_cache_paths_can_be_overridden() {
+        let observation_path = temp_file("observations");
+        let cache_path = temp_file("names");
+        let _observation_guard = EnvGuard::set(OBSERVATION_STORE_ENV, &observation_path);
+        let _cache_guard = EnvGuard::set(MAC_NAME_CACHE_ENV, &cache_path);
+
+        let changed = observe_dhcp_event(
+            "add",
+            "aa:bb:cc:dd:ee:ff".parse().expect("mac should parse"),
+            Some("192.168.1.2".parse().expect("ip should parse")),
+            Some("lda"),
+        )
+        .await
+        .expect("observation should write");
+        assert!(changed);
+
+        let store = load_observation_store()
+            .await
+            .expect("observation store should read");
+        assert!(store.dhcp_clients.contains_key("aa:bb:cc:dd:ee:ff"));
+
+        let cache = load_mac_name_cache().await.expect("name cache should read");
+        assert_eq!(cache.get("aa:bb:cc:dd:ee:ff"), Some(&"lda".to_string()));
+
+        let _ = tokio::fs::remove_file(observation_path).await;
+        let _ = tokio::fs::remove_file(cache_path).await;
+    }
+}