Reduce redundant intent writes while preserving liveness

packages/user-intent-kit/README.md

@@ -378,3 +378,24 @@ its actual supervisor process instead of a bare timer, e.g.
 ## License
 
 AGPL-3.0
+
+### Request coalescing and liveness
+
+DesktopAdapter now uses one telemetry timer rather than also starting a separate
+30-second client heartbeat. Sampling is capped at 60 seconds and device writes
+explicitly retain a 90-second TTL. Every sample refreshes the full device payload;
+this preserves changing vitals and avoids stale rows for old 120-second settings.
+Agent status sends on status/task changes and otherwise refreshes every 120
+seconds (300-second TTL). Agent vitals are collected on those actual sends.
+One request per adapter is in flight; pending updates coalesce to the latest
+state. Failed writes do not advance the successful checkpoint and retry on the
+next sample. Network outages and event-loop delays can still make a device stale.
+Room polling, message delivery and command handling are unaffected.
+
+For the old 120-second daemon configuration, a primary desktop publisher formerly
+made 2,880 standalone heartbeats plus 720 device and 720 agent writes per day.
+The new steady schedule is 1,440 device plus 720 agent writes: 4,320 → 2,160/day,
+a 50% reduction for that daemon, excluding startup/status changes and retries.
+These are schedule-derived counts, not measured Vercel billing or fleet totals.
+The test suite separately verifies 720 unchanged-agent writes over a simulated day.
+Standalone vitals scripts and secondary agents still contribute additional traffic.

packages/user-intent-kit/bin/uik-daemon.js

@@ -82,6 +82,11 @@ const modelProbe = new ServedModelProbe({ guard: id => availabilityProbe.couldLo
 const publishDevice = process.env.INTENT_DEVICE_PUBLISH !== '0';
 const pollIntervalMs = Number(process.env.POLL_INTERVAL_MS || 30000);
 
+if (!Number.isFinite(pollIntervalMs) || pollIntervalMs <= 0) {
+  console.error('uik-daemon: POLL_INTERVAL_MS must be positive and finite');
+  process.exit(1);
+}
+
 if (!apiKey || !userId) {
   console.error('uik-daemon: INTENT_API_KEY and INTENT_USER_ID required');
   process.exit(1);
@@ -123,12 +128,12 @@ function gateOpen() {
   catch { return false; }
 }
 
-if (gateOpen()) await iak.publishStatus({ status: 'active', currentTask: null });
+if (gateOpen()) await iak.publishStatus({ status: 'active', currentTask: null, heartbeat: true });
 
 const agentTimer = setInterval(() => {
   if (!gateOpen()) return;
-  iak.publishStatus({ status: 'active', currentTask: null }).catch(() => {});
-}, pollIntervalMs);
+  iak.publishStatus({ status: 'active', currentTask: null, heartbeat: true }).catch(() => {});
+}, Math.min(pollIntervalMs, 120000));
 
 console.log(`uik-daemon: device=${deviceId} agent=${agentHandle} interval=${pollIntervalMs}ms`);
 // Which endpoint the model label comes from, so a blank label on the

packages/user-intent-kit/src/adapters/desktop.js

@@ -8,6 +8,7 @@ const IDLE_AFTER_SEC = 300;
 import { collectHostTelemetry } from '../host-telemetry.js';
 import { ServedModelProbe } from '../served-model.js';
 import { ModelAvailabilityProbe } from '../model-availability.js';
+import { StatePublisher } from '../state-publisher.js';
 
 /**
  * Desktop Adapter - detects active window and context on macOS.
@@ -25,6 +26,7 @@ export class DesktopAdapter {
   #modelProbe;
   #availabilityProbe;
   #pollIntervalMs;
+  #publisher;
 
   /**
    * @param {import('../client.js').IntentClient} client
@@ -57,7 +59,12 @@ export class DesktopAdapter {
     this.#modelProbe = modelProbe === undefined
       ? new ServedModelProbe({ guard: id => scan?.couldLoad(id) ?? false })
       : modelProbe;
-    this.#pollIntervalMs = pollIntervalMs;
+    if (!Number.isFinite(pollIntervalMs) || pollIntervalMs <= 0) throw new Error('Invalid pollIntervalMs');
+    // One timer owns both telemetry and liveness. Default server device TTL is 90s.
+    this.#pollIntervalMs = Math.min(pollIntervalMs, 60000);
+    this.#publisher = new StatePublisher(fields => this.#client.patchDevice(fields), {
+      refreshMs: this.#pollIntervalMs,
+    });
     this.#pollTimer = null;
   }
 
@@ -111,7 +118,7 @@ export class DesktopAdapter {
    */
   async publishState() {
     const state = this.#detectState();
-    await this.#client.patchDevice(state);
+    await this.#publisher.publish({ ...state, ttl_sec: 90 });
   }
 
   /**
@@ -121,14 +128,16 @@ export class DesktopAdapter {
     this.stop();
     // Publish immediately
     this.publishState().catch(() => {});
-    this.#client.startHeartbeat();
     // Started AFTER the first publish and never awaited, so the machine
     // appears on the dashboard at once with whatever vitals it has, label or
     // no label. The first probe fills the label in for the next beat.
     this.#modelProbe?.start();
     // Same contract, slower timer: the scan is disk-bound, so it never runs on
     // the heartbeat's path and the first beat goes out without waiting for it.
     this.#availabilityProbe?.start();
+    // this.#client.startHeartbeat() removed: the StatePublisher's own
+    // refreshMs timer now re-sends the last state as the liveness beat, so a
+    // second heartbeat mechanism only doubled the writes (c0ed66d, 18 Sep 2026).
     this.#pollTimer = setInterval(() => {
       this.publishState().catch(() => {});
     }, this.#pollIntervalMs);
@@ -142,7 +151,6 @@ export class DesktopAdapter {
     }
     this.#modelProbe?.stop();
     this.#availabilityProbe?.stop();
-    this.#client.stopHeartbeat();
   }
 
   /**

packages/user-intent-kit/src/adapters/iak.js

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0
 
 import { collectHostTelemetry } from '../host-telemetry.js';
+import { StatePublisher } from '../state-publisher.js';
 
 /**
  * IAK Adapter - integrates user-intent-kit with IDE Agent Kit.
@@ -12,6 +13,7 @@ export class IAKAdapter {
   #client;
   #agentHandle;
   #machine;
+  #publisher;
 
   /**
    * @param {import('../client.js').IntentClient} client
@@ -27,6 +29,8 @@ export class IAKAdapter {
     this.#client = client;
     this.#agentHandle = agentHandle;
     this.#machine = machine ?? client?.deviceId ?? undefined;
+    this.#publisher = new StatePublisher(fields => this.#client.patchAgent(
+      this.#agentHandle.replace(/^@/, ''), { ...fields, host: collectHostTelemetry({ machine: this.#machine }) }), { refreshMs: 120000 });
   }
 
   /**
@@ -37,13 +41,12 @@ export class IAKAdapter {
    * string on some publishers, which gave the dashboard nowhere to put a
    * temperature or a load figure.
    */
-  async publishStatus({ status = 'active', currentTask = null }) {
-    const name = this.#agentHandle.replace(/^@/, '');
-    await this.#client.patchAgent(name, {
+  async publishStatus({ status = 'active', currentTask = null, heartbeat = false }) {
+    await this.#publisher.publish({
       status,
       last_task: currentTask,
-      host: collectHostTelemetry({ machine: this.#machine }),
-    });
+      ttl_sec: 300,
+    }, { force: heartbeat });
   }
 
   /**

packages/user-intent-kit/src/state-publisher.js

@@ -0,0 +1,32 @@
+// SPDX-License-Identifier: AGPL-3.0
+
+/** Change-triggered writes with a bounded refresh and one request in flight.
+ * Only acknowledged writes advance the checkpoint. The latest queued state wins.
+ * A monotonic clock avoids missed refreshes when the system clock moves backward.
+ */
+export class StatePublisher {
+  #send; #refreshMs; #now; #lastKey; #lastAt = -Infinity;
+  #pending; #running;
+  constructor(send, { refreshMs, now = () => performance.now() }) {
+    if (!Number.isFinite(refreshMs) || refreshMs <= 0) throw new Error('Invalid refreshMs');
+    this.#send = send; this.#refreshMs = refreshMs; this.#now = now;
+  }
+  publish(state, { force = false } = {}) {
+    this.#pending = { state: structuredClone(state), force };
+    if (!this.#running) {
+      this.#running = this.#drain().finally(() => { this.#running = undefined; });
+    }
+    return this.#running;
+  }
+  async #drain() {
+    while (this.#pending) {
+      const { state, force } = this.#pending;
+      this.#pending = undefined;
+      const key = JSON.stringify(state);
+      if (!force && key === this.#lastKey && this.#now() - this.#lastAt < this.#refreshMs) continue;
+      await this.#send(state);
+      this.#lastKey = key;
+      this.#lastAt = this.#now();
+    }
+  }
+}

packages/user-intent-kit/test/state-publisher.test.js

@@ -0,0 +1,63 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { StatePublisher } from '../src/state-publisher.js';
+import { DesktopAdapter } from '../src/adapters/desktop.js';
+
+test('unchanged agent status: 30s sampling yields 720 writes/day, not 2880', async () => {
+  let now = 0, count = 0;
+  const p = new StatePublisher(async () => count++, { refreshMs: 120000, now: () => now });
+  for (; now < 86400000; now += 30000) await p.publish({ status: 'active' });
+  assert.equal(count, 720);
+});
+test('task and offline transitions publish immediately', async () => {
+  const sent = [];
+  const p = new StatePublisher(async s => sent.push(s), { refreshMs: 120000, now: () => 0 });
+  await p.publish({ status: 'active', task: null });
+  await p.publish({ status: 'active', task: 'new' });
+  await p.publish({ status: 'offline', task: null });
+  assert.equal(sent.length, 3);
+});
+test('failed refresh is retried without advancing successful checkpoint', async () => {
+  let tries = 0;
+  const p = new StatePublisher(async () => { if (++tries === 1) throw Error('offline'); }, { refreshMs: 60000 });
+  await assert.rejects(p.publish({ a: 1 }));
+  await p.publish({ a: 1 });
+  assert.equal(tries, 2);
+});
+test('in-flight changes coalesce to the latest state without concurrent writes', async () => {
+  const sent = []; let release;
+  const p = new StatePublisher(async s => {
+    sent.push(s);
+    if (sent.length === 1) await new Promise(r => { release = r; });
+  }, { refreshMs: 60000 });
+  const first = p.publish({ n: 1 });
+  p.publish({ n: 2 });
+  const last = p.publish({ n: 3 });
+  release(); await first; await last;
+  assert.deepEqual(sent, [{ n: 1 }, { n: 3 }]);
+});
+test('desktop uses one bounded timer, not an independent heartbeat', async (t) => {
+  t.mock.timers.enable({ apis: ['setInterval'] });
+  let patches = 0, beats = 0;
+  const a = new DesktopAdapter({ patchDevice: async s => { patches++; assert.equal(s.ttl_sec, 90); },
+    startHeartbeat: () => beats++, stopHeartbeat() {} }, { pollIntervalMs: 120000 });
+  let samples = 0;
+  a.publishState = async () => { samples++; };
+  a.start(); await new Promise(r => setImmediate(r));
+  assert.equal(samples, 1); assert.equal(beats, 0);
+  t.mock.timers.tick(60000); await new Promise(r => setImmediate(r));
+  assert.equal(samples, 2); // bounded detection with old 120-second setting
+  a.stop();
+});
+test('invalid refresh intervals are rejected', () => {
+  for (const n of [0, -1, NaN, Infinity]) assert.throws(() => new StatePublisher(async () => {}, { refreshMs: n }));
+});
+
+ test('explicit liveness tick is not skipped because the last request completed late', async () => {
+  let now = 0, sends = 0;
+  const p = new StatePublisher(async () => { sends++; now += 1000; }, { refreshMs: 120000, now: () => now });
+  await p.publish({ status: 'active' });
+  now = 120000;
+  await p.publish({ status: 'active' }, { force: true });
+  assert.equal(sends, 2);
+});