feat(server): dedup machines on machine_uid (SPEC-004 Task 2)

Persist the agent-reported machine_uid and dedup connect_machines on it so a
single physical machine can't register duplicate rows when its config-file
agent_id regenerates (the ghost-session root cause).

- migration 008: nullable connect_machines.machine_uid + partial unique index
  (WHERE machine_uid IS NOT NULL); idempotent, startup-applied.
- upsert_machine: two-path dedup (ON CONFLICT machine_uid when present, else
  the legacy ON CONFLICT agent_id path, unchanged).
- session reattach: a machine_uid index consulted before agent_id, with all
  removal paths purging it.
- security: keyed (cak_) agents stay authoritative — their claimed machine_uid
  is dropped (effective_machine_uid=None); uid is dedup-only for un-keyed /
  support-code agents. Startup restore skips uid-indexing keyed machines and
  fails closed if the keyed-set query errors.

74 server tests pass; clippy clean. Implements specs/v2-stable-identity/plan.md Task 2.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/src/db/machines.rs

@@ -50,6 +50,14 @@ pub struct Machine {
     /// impl) so it can never error at decode time.
     #[serde(default)]
     pub tags: Vec<String>,
+    /// Deterministic, recomputable hardware identity reported by the agent
+    /// (`AgentStatus.machine_uid` / connect query param). Column added in migration
+    /// 008. NULLABLE: legacy rows and agents that do not report a uid carry `None`.
+    /// For un-keyed agents this is the dedup key (`upsert_machine` keys
+    /// `ON CONFLICT (machine_uid)` when present); for `cak_`-keyed agents the key's
+    /// machine binding stays authoritative and the claimed uid is NOT used to dedup
+    /// (see `upsert_machine`).
+    pub machine_uid: Option<String>,
 }
 
 impl<'r> FromRow<'r, PgRow> for Machine {
@@ -65,6 +73,8 @@ impl<'r> FromRow<'r, PgRow> for Machine {
             tenant_id: row.try_get("tenant_id")?,
             organization: row.try_get("organization")?,
             site: row.try_get("site")?,
+            // Schema-nullable (migration 008); decode directly as Option.
+            machine_uid: row.try_get("machine_uid")?,
             // Nullable-with-default columns mapped to non-`Option` Rust types: read as
             // `Option<T>` and fall back to the type default so a NULL cell never errors.
             is_elevated: row
@@ -97,29 +107,88 @@ impl<'r> FromRow<'r, PgRow> for Machine {
     }
 }
 
-/// Get or create a machine by agent_id (upsert)
+/// Get or create a machine record (upsert), deduplicating on the most stable
+/// identity available (SPEC-004 / v2-stable-identity Task 2).
+///
+/// Two dedup paths, selected by whether the caller passes a `machine_uid`:
+///
+/// - **`machine_uid = Some(uid)` (the un-keyed dedup path):** key on the stable
+///   hardware identity — `ON CONFLICT (machine_uid)`. The SAME machine reconnecting
+///   with a DIFFERENT `agent_id` (e.g. after a config loss minted a fresh random id)
+///   updates its EXISTING row instead of inserting a duplicate. `agent_id` and
+///   `hostname` are refreshed to the latest reported values. This is what collapses
+///   the duplicate-registration fleet down to one row per real machine.
+///
+/// - **`machine_uid = None` (legacy / authoritative path):** preserve the original
+///   behavior exactly — `ON CONFLICT (agent_id)`. Used for agents that do not report
+///   a uid AND, critically, for `cak_`-keyed agents: the caller (`relay`) passes
+///   `None` for keyed agents so their AUTHORITATIVE key-bound `agent_id` is the dedup
+///   key and a client-claimed `machine_uid` can never repoint a keyed machine's row.
+///
+/// SECURITY: a client-asserted `machine_uid` is spoofable, so it is a *correctness*
+/// aid, not a trust boundary. Only the un-keyed path supplies it; the keyed path's
+/// authority lives in the key→machine binding upstream (see
+/// `relay::agent_ws_handler`), never here.
 pub async fn upsert_machine(
     pool: &PgPool,
     agent_id: &str,
     hostname: &str,
     is_persistent: bool,
+    machine_uid: Option<&str>,
 ) -> Result<Machine, sqlx::Error> {
-    sqlx::query_as::<_, Machine>(
-        r#"
-        INSERT INTO connect_machines (agent_id, hostname, is_persistent, status, last_seen)
-        VALUES ($1, $2, $3, 'online', NOW())
-        ON CONFLICT (agent_id) DO UPDATE SET
-            hostname = EXCLUDED.hostname,
-            status = 'online',
-            last_seen = NOW()
-        RETURNING *
-        "#,
-    )
-    .bind(agent_id)
-    .bind(hostname)
-    .bind(is_persistent)
-    .fetch_one(pool)
-    .await
+    match machine_uid {
+        // Un-keyed dedup path: stable hardware identity is the conflict arbiter.
+        // A new agent_id for the same physical machine updates the existing row.
+        //
+        // Edge case: if this INSERT's new random `agent_id` happens to collide with a
+        // DIFFERENT legacy row's value under the `agent_id UNIQUE` constraint, Postgres
+        // raises the unique violation on `agent_id` BEFORE the `ON CONFLICT
+        // (machine_uid)` arbiter is consulted, so the upsert errors instead of merging
+        // on uid. This is non-fatal: the caller logs the error and the live in-memory
+        // session is unaffected, and the agent simply retries with a freshly minted
+        // UUID. The window closes on its own as legacy rows age out (SPEC-004 Task 3).
+        Some(uid) => {
+            sqlx::query_as::<_, Machine>(
+                r#"
+                INSERT INTO connect_machines (agent_id, hostname, is_persistent, status, last_seen, machine_uid)
+                VALUES ($1, $2, $3, 'online', NOW(), $4)
+                ON CONFLICT (machine_uid) DO UPDATE SET
+                    agent_id = EXCLUDED.agent_id,
+                    hostname = EXCLUDED.hostname,
+                    status = 'online',
+                    last_seen = NOW()
+                RETURNING *
+                "#,
+            )
+            .bind(agent_id)
+            .bind(hostname)
+            .bind(is_persistent)
+            .bind(uid)
+            .fetch_one(pool)
+            .await
+        }
+        // Legacy / authoritative path: dedup on agent_id exactly as before. Leaves
+        // machine_uid NULL (the partial unique index excludes NULLs, so any number
+        // of these may coexist).
+        None => {
+            sqlx::query_as::<_, Machine>(
+                r#"
+                INSERT INTO connect_machines (agent_id, hostname, is_persistent, status, last_seen)
+                VALUES ($1, $2, $3, 'online', NOW())
+                ON CONFLICT (agent_id) DO UPDATE SET
+                    hostname = EXCLUDED.hostname,
+                    status = 'online',
+                    last_seen = NOW()
+                RETURNING *
+                "#,
+            )
+            .bind(agent_id)
+            .bind(hostname)
+            .bind(is_persistent)
+            .fetch_one(pool)
+            .await
+        }
+    }
 }
 
 /// Update machine status and info
@@ -239,3 +308,134 @@ pub async fn update_machine_metadata(
     .await?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use sqlx::postgres::PgPoolOptions;
+
+    /// Connect to a throwaway test Postgres and apply migrations, or return `None`
+    /// when `TEST_DATABASE_URL` is unset so the suite is a no-op on workstations
+    /// without a database. CI sets `TEST_DATABASE_URL` against an ephemeral Postgres,
+    /// where these run for real. (The server crate is Linux-targeted and validated
+    /// in Gitea CI; these DB tests run there.)
+    async fn test_pool() -> Option<PgPool> {
+        let url = std::env::var("TEST_DATABASE_URL").ok()?;
+        let pool = PgPoolOptions::new()
+            .max_connections(2)
+            .connect(&url)
+            .await
+            .expect("connect to TEST_DATABASE_URL");
+        sqlx::migrate!("./migrations")
+            .run(&pool)
+            .await
+            .expect("apply migrations to the test database");
+        Some(pool)
+    }
+
+    /// Remove any rows this test created so reruns are clean and tests don't collide.
+    async fn cleanup(pool: &PgPool, agent_ids: &[&str], machine_uids: &[&str]) {
+        for id in agent_ids {
+            let _ = sqlx::query("DELETE FROM connect_machines WHERE agent_id = $1")
+                .bind(id)
+                .execute(pool)
+                .await;
+        }
+        for uid in machine_uids {
+            let _ = sqlx::query("DELETE FROM connect_machines WHERE machine_uid = $1")
+                .bind(uid)
+                .execute(pool)
+                .await;
+        }
+    }
+
+    /// (a) Same `machine_uid` with two DIFFERENT `agent_id`s collapses to ONE row.
+    /// The second upsert updates the existing row (and repoints its agent_id) rather
+    /// than inserting a duplicate — the core dedup guarantee for the un-keyed fleet.
+    #[tokio::test]
+    async fn same_machine_uid_two_agent_ids_one_row() {
+        let Some(pool) = test_pool().await else {
+            return; // no TEST_DATABASE_URL: skip (runs in CI)
+        };
+        let uid = "test-muid-dedup-001";
+        cleanup(&pool, &["agent-A", "agent-B"], &[uid]).await;
+
+        let m1 = upsert_machine(&pool, "agent-A", "HOST-A", true, Some(uid))
+            .await
+            .expect("first upsert");
+        let m2 = upsert_machine(&pool, "agent-B", "HOST-A2", true, Some(uid))
+            .await
+            .expect("second upsert with same uid, different agent_id");
+
+        // Same physical row, agent_id and hostname refreshed to the latest.
+        assert_eq!(m1.id, m2.id, "same machine_uid must update the same row");
+        assert_eq!(m2.agent_id, "agent-B");
+        assert_eq!(m2.hostname, "HOST-A2");
+        assert_eq!(m2.machine_uid.as_deref(), Some(uid));
+
+        // Exactly one row carries this uid.
+        let count: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM connect_machines WHERE machine_uid = $1")
+                .bind(uid)
+                .fetch_one(&pool)
+                .await
+                .expect("count rows for uid");
+        assert_eq!(count, 1, "must be exactly one row for the machine_uid");
+
+        cleanup(&pool, &["agent-A", "agent-B"], &[uid]).await;
+    }
+
+    /// (b) Legacy NULL-`machine_uid` path is unchanged: dedup keys on `agent_id`,
+    /// the row's `machine_uid` stays NULL, and re-upserting the same agent_id with
+    /// no uid updates the same row (no crash, no duplicate).
+    #[tokio::test]
+    async fn legacy_null_machine_uid_dedups_on_agent_id() {
+        let Some(pool) = test_pool().await else {
+            return; // no TEST_DATABASE_URL: skip (runs in CI)
+        };
+        let agent = "test-legacy-agent-001";
+        cleanup(&pool, &[agent], &[]).await;
+
+        let m1 = upsert_machine(&pool, agent, "LEGACY-HOST", true, None)
+            .await
+            .expect("legacy upsert (no uid)");
+        assert_eq!(m1.machine_uid, None, "legacy row must have NULL machine_uid");
+
+        let m2 = upsert_machine(&pool, agent, "LEGACY-HOST-RENAMED", true, None)
+            .await
+            .expect("legacy re-upsert (no uid)");
+        assert_eq!(m1.id, m2.id, "legacy agent_id must dedup to the same row");
+        assert_eq!(m2.hostname, "LEGACY-HOST-RENAMED");
+        assert_eq!(m2.machine_uid, None);
+
+        let count: i64 =
+            sqlx::query_scalar("SELECT COUNT(*) FROM connect_machines WHERE agent_id = $1")
+                .bind(agent)
+                .fetch_one(&pool)
+                .await
+                .expect("count legacy rows");
+        assert_eq!(count, 1, "legacy path must not duplicate the row");
+
+        cleanup(&pool, &[agent], &[]).await;
+    }
+
+    /// Multiple legacy rows with NULL machine_uid coexist — the partial unique index
+    /// excludes NULLs, so distinct un-keyed agents are independent rows.
+    #[tokio::test]
+    async fn multiple_null_machine_uid_rows_coexist() {
+        let Some(pool) = test_pool().await else {
+            return; // no TEST_DATABASE_URL: skip (runs in CI)
+        };
+        cleanup(&pool, &["null-1", "null-2"], &[]).await;
+
+        let a = upsert_machine(&pool, "null-1", "H1", true, None)
+            .await
+            .expect("first null-uid row");
+        let b = upsert_machine(&pool, "null-2", "H2", true, None)
+            .await
+            .expect("second null-uid row");
+        assert_ne!(a.id, b.id, "distinct legacy agents must be distinct rows");
+
+        cleanup(&pool, &["null-1", "null-2"], &[]).await;
+    }
+}