From ea2424400dec093b7989b79b90572c46091e6170 Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Fri, 18 Sep 2026 00:48:19 +0200 Subject: [PATCH 1/2] Coalesce intent publishing and remove duplicate desktop heartbeat --- packages/user-intent-kit/README.md | 21 ++++++++ packages/user-intent-kit/bin/uik-daemon.js | 7 ++- .../user-intent-kit/src/adapters/desktop.js | 16 ++++-- packages/user-intent-kit/src/adapters/iak.js | 9 ++-- .../user-intent-kit/src/state-publisher.js | 32 +++++++++++ .../test/state-publisher.test.js | 54 +++++++++++++++++++ 6 files changed, 131 insertions(+), 8 deletions(-) create mode 100644 packages/user-intent-kit/src/state-publisher.js create mode 100644 packages/user-intent-kit/test/state-publisher.test.js diff --git a/packages/user-intent-kit/README.md b/packages/user-intent-kit/README.md index 4c55191..594d830 100644 --- a/packages/user-intent-kit/README.md +++ b/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. diff --git a/packages/user-intent-kit/bin/uik-daemon.js b/packages/user-intent-kit/bin/uik-daemon.js index 02f7efe..86ea6b1 100755 --- a/packages/user-intent-kit/bin/uik-daemon.js +++ b/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); @@ -128,7 +133,7 @@ if (gateOpen()) await iak.publishStatus({ status: 'active', currentTask: null }) const agentTimer = setInterval(() => { if (!gateOpen()) return; iak.publishStatus({ status: 'active', currentTask: null }).catch(() => {}); -}, pollIntervalMs); +}, 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 diff --git a/packages/user-intent-kit/src/adapters/desktop.js b/packages/user-intent-kit/src/adapters/desktop.js index d54f02f..71caf78 100644 --- a/packages/user-intent-kit/src/adapters/desktop.js +++ b/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,7 +128,6 @@ 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. @@ -129,6 +135,9 @@ export class DesktopAdapter { // 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(); } /** diff --git a/packages/user-intent-kit/src/adapters/iak.js b/packages/user-intent-kit/src/adapters/iak.js index 729b5e5..049f3bd 100644 --- a/packages/user-intent-kit/src/adapters/iak.js +++ b/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 }); } /** @@ -38,11 +42,10 @@ export class IAKAdapter { * temperature or a load figure. */ async publishStatus({ status = 'active', currentTask = null }) { - const name = this.#agentHandle.replace(/^@/, ''); - await this.#client.patchAgent(name, { + await this.#publisher.publish({ status, last_task: currentTask, - host: collectHostTelemetry({ machine: this.#machine }), + ttl_sec: 300, }); } diff --git a/packages/user-intent-kit/src/state-publisher.js b/packages/user-intent-kit/src/state-publisher.js new file mode 100644 index 0000000..4715ddb --- /dev/null +++ b/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) { + this.#pending = structuredClone(state); + if (!this.#running) { + this.#running = this.#drain().finally(() => { this.#running = undefined; }); + } + return this.#running; + } + async #drain() { + while (this.#pending) { + const state = this.#pending; + this.#pending = undefined; + const key = JSON.stringify(state); + if (key === this.#lastKey && this.#now() - this.#lastAt < this.#refreshMs) continue; + await this.#send(state); + this.#lastKey = key; + this.#lastAt = this.#now(); + } + } +} diff --git a/packages/user-intent-kit/test/state-publisher.test.js b/packages/user-intent-kit/test/state-publisher.test.js new file mode 100644 index 0000000..d0ecd82 --- /dev/null +++ b/packages/user-intent-kit/test/state-publisher.test.js @@ -0,0 +1,54 @@ +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 })); +}); From 1a8482f9dc07d655b6270a7cc72cfa897e6bd9a8 Mon Sep 17 00:00:00 2001 From: Petrus Pennanen Date: Fri, 18 Sep 2026 01:04:40 +0200 Subject: [PATCH 2/2] Keep scheduled liveness ticks independent of request latency --- packages/user-intent-kit/bin/uik-daemon.js | 4 ++-- packages/user-intent-kit/src/adapters/iak.js | 4 ++-- packages/user-intent-kit/src/state-publisher.js | 8 ++++---- packages/user-intent-kit/test/state-publisher.test.js | 9 +++++++++ 4 files changed, 17 insertions(+), 8 deletions(-) diff --git a/packages/user-intent-kit/bin/uik-daemon.js b/packages/user-intent-kit/bin/uik-daemon.js index 86ea6b1..025cdc9 100755 --- a/packages/user-intent-kit/bin/uik-daemon.js +++ b/packages/user-intent-kit/bin/uik-daemon.js @@ -128,11 +128,11 @@ 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(() => {}); + iak.publishStatus({ status: 'active', currentTask: null, heartbeat: true }).catch(() => {}); }, Math.min(pollIntervalMs, 120000)); console.log(`uik-daemon: device=${deviceId} agent=${agentHandle} interval=${pollIntervalMs}ms`); diff --git a/packages/user-intent-kit/src/adapters/iak.js b/packages/user-intent-kit/src/adapters/iak.js index 049f3bd..1092aab 100644 --- a/packages/user-intent-kit/src/adapters/iak.js +++ b/packages/user-intent-kit/src/adapters/iak.js @@ -41,12 +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 }) { + async publishStatus({ status = 'active', currentTask = null, heartbeat = false }) { await this.#publisher.publish({ status, last_task: currentTask, ttl_sec: 300, - }); + }, { force: heartbeat }); } /** diff --git a/packages/user-intent-kit/src/state-publisher.js b/packages/user-intent-kit/src/state-publisher.js index 4715ddb..84cc7cc 100644 --- a/packages/user-intent-kit/src/state-publisher.js +++ b/packages/user-intent-kit/src/state-publisher.js @@ -11,8 +11,8 @@ export class StatePublisher { if (!Number.isFinite(refreshMs) || refreshMs <= 0) throw new Error('Invalid refreshMs'); this.#send = send; this.#refreshMs = refreshMs; this.#now = now; } - publish(state) { - this.#pending = structuredClone(state); + publish(state, { force = false } = {}) { + this.#pending = { state: structuredClone(state), force }; if (!this.#running) { this.#running = this.#drain().finally(() => { this.#running = undefined; }); } @@ -20,10 +20,10 @@ export class StatePublisher { } async #drain() { while (this.#pending) { - const state = this.#pending; + const { state, force } = this.#pending; this.#pending = undefined; const key = JSON.stringify(state); - if (key === this.#lastKey && this.#now() - this.#lastAt < this.#refreshMs) continue; + if (!force && key === this.#lastKey && this.#now() - this.#lastAt < this.#refreshMs) continue; await this.#send(state); this.#lastKey = key; this.#lastAt = this.#now(); diff --git a/packages/user-intent-kit/test/state-publisher.test.js b/packages/user-intent-kit/test/state-publisher.test.js index d0ecd82..310793c 100644 --- a/packages/user-intent-kit/test/state-publisher.test.js +++ b/packages/user-intent-kit/test/state-publisher.test.js @@ -52,3 +52,12 @@ test('desktop uses one bounded timer, not an independent heartbeat', async (t) = 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); +});