diff --git a/src/config/load-config.ts b/src/config/load-config.ts index ac9ab41..3cc50bf 100644 --- a/src/config/load-config.ts +++ b/src/config/load-config.ts @@ -259,7 +259,9 @@ export function loadConfig(): AppConfig { moa: buildMoaConfig(), status: { channelId: readText('STATUS_CHANNEL_ID') ?? '', - updateInterval: 10000, + // Base periodic refresh. Event-driven updates (new message, agent + // activity change) refresh immediately between these ticks. + updateInterval: 60000, usageUpdateInterval: 60000, showRooms: readBooleanUnlessFalse('STATUS_SHOW_ROOMS', true), showRoomDetails: readBooleanUnlessFalse('STATUS_SHOW_ROOM_DETAILS', true), diff --git a/src/group-queue.ts b/src/group-queue.ts index c92d576..e6933ee 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -59,12 +59,28 @@ export class GroupQueue { return state; } + private onActivityChange: (() => void) | null = null; + setProcessMessagesFn( fn: (groupJid: string, context: GroupRunContext) => Promise, ): void { this.processMessagesFn = fn; } + /** Notified when a group starts or finishes a run (activeCount changes). */ + setOnActivityChange(fn: () => void): void { + this.onActivityChange = fn; + } + + private notifyActivityChange(): void { + if (!this.onActivityChange) return; + try { + this.onActivityChange(); + } catch (err) { + logger.debug({ err }, 'onActivityChange callback threw'); + } + } + /** Limit concurrency after restart to avoid API rate-limit storms. */ enterRecoveryMode(): void { this.recoveryMode = true; @@ -528,6 +544,7 @@ export class GroupQueue { transitionRunPhase(state, groupJid, 'running_messages', { runId, reason }); assertRunPhaseInvariants(state, groupJid); this.activeCount++; + this.notifyActivityChange(); logger.info( { @@ -580,6 +597,7 @@ export class GroupQueue { resetRunState(state, groupJid); assertRunPhaseInvariants(state, groupJid); this.activeCount--; + this.notifyActivityChange(); this.drainGroup(groupJid); } } @@ -592,6 +610,7 @@ export class GroupQueue { assertRunPhaseInvariants(state, groupJid); this.activeCount++; this.activeTaskCount++; + this.notifyActivityChange(); logger.info( { @@ -628,6 +647,7 @@ export class GroupQueue { assertRunPhaseInvariants(state, groupJid); this.activeCount--; this.activeTaskCount--; + this.notifyActivityChange(); this.drainGroup(groupJid); } } diff --git a/src/index.ts b/src/index.ts index a30d750..c058e66 100644 --- a/src/index.ts +++ b/src/index.ts @@ -52,7 +52,10 @@ import { } from './sender-allowlist.js'; import { createMessageRuntime } from './message-runtime.js'; import { nudgeSchedulerLoop, startSchedulerLoop } from './task-scheduler.js'; -import { startUnifiedDashboard } from './unified-dashboard.js'; +import { + requestImmediateStatusUpdate, + startUnifiedDashboard, +} from './unified-dashboard.js'; import { startUsagePrimer } from './usage-primer.js'; import { startWebDashboardServer } from './web-dashboard-server.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; @@ -395,6 +398,11 @@ async function main(): Promise { } } storeMessage(msg); + // A real chat message in a registered room: refresh the status dashboard + // immediately instead of waiting for the next periodic tick. + if (!msg.is_from_me && !msg.is_bot_message && roomBindings[chatJid]) { + requestImmediateStatusUpdate(); + } }, onChatMetadata: ( chatJid: string, @@ -569,6 +577,9 @@ async function main(): Promise { }, purgeOnStart: true, }); + // Refresh the status dashboard immediately when an agent starts/finishes a run + // (activity moves between rooms) rather than waiting for the periodic tick. + queue.setOnActivityChange(requestImmediateStatusUpdate); webDashboardServer = startWebDashboardServer({ ...WEB_DASHBOARD, getRoomBindings: runtimeState.getRoomBindings, diff --git a/src/unified-dashboard.test.ts b/src/unified-dashboard.test.ts index 5e27d74..e1a5358 100644 --- a/src/unified-dashboard.test.ts +++ b/src/unified-dashboard.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest'; import type { UsageRow } from './dashboard-usage-rows.js'; import { buildWebUsageRowsForSnapshot, + createCoalescingTrigger, editStatusMessageWithRetry, formatStatusHeader, getDashboardDuplicateCleanupIntervalMs, @@ -312,3 +313,30 @@ describe('editStatusMessageWithRetry', () => { expect(slept).toBe(0); }); }); + +describe('createCoalescingTrigger', () => { + it('coalesces a burst into one run and re-arms after firing', () => { + let runs = 0; + let fire: (() => void) | null = null; + const timer = { + set: (cb: () => void) => { + fire = cb; + return 0 as unknown as ReturnType; + }, + }; + const trigger = createCoalescingTrigger(() => runs++, 1500, timer); + + trigger(); + trigger(); + trigger(); + expect(runs).toBe(0); // nothing runs until the debounce window elapses + + fire!(); + expect(runs).toBe(1); // three calls collapsed into a single run + + // A new request after the run schedules and fires again. + trigger(); + fire!(); + expect(runs).toBe(2); + }); +}); diff --git a/src/unified-dashboard.ts b/src/unified-dashboard.ts index 8f9b3cd..df28687 100644 --- a/src/unified-dashboard.ts +++ b/src/unified-dashboard.ts @@ -111,12 +111,28 @@ let channelMetaLastRefresh = 0; let dashboardUpdateLogged = false; /** Guards against overlapping status updates while a slow edit-retry runs. */ let statusUpdateRunning = false; +/** A refresh was requested while one was already running; run once more after. */ +let statusUpdatePending = false; /** * On a failed status-message edit, retry the edit this many extra times at * STATUS_EDIT_RETRY_DELAY_MS spacing before falling back to a fresh message. */ const STATUS_EDIT_MAX_RETRIES = 2; const STATUS_EDIT_RETRY_DELAY_MS = 15_000; +/** Coalesce bursts of event-driven refresh requests into one update. */ +const IMMEDIATE_UPDATE_DEBOUNCE_MS = 1_500; +/** Set by the renderer's startUnifiedDashboard so external events can nudge it. */ +let immediateUpdateTrigger: (() => void) | null = null; + +/** + * Request an out-of-band status refresh (e.g. a new chat message arrived or an + * agent's activity changed) so the dashboard updates immediately instead of + * waiting for the next periodic tick. No-op until the renderer dashboard starts; + * bursts are debounced. + */ +export function requestImmediateStatusUpdate(): void { + immediateUpdateTrigger?.(); +} /** Codex service only: cached usage rows written into the status snapshot. */ let cachedCodexUsageRows: UsageRow[] = []; /** Codex service only: ISO timestamp of last successful usage fetch. */ @@ -801,6 +817,29 @@ export async function editStatusMessageWithRetry(args: { return false; } +/** + * Returns a trigger that runs `run` at most once per `delayMs` window: the + * first call schedules a run after `delayMs`, and further calls within that + * window are folded into the same pending run. Timer fns are injectable for + * tests. + */ +export function createCoalescingTrigger( + run: () => void, + delayMs: number, + timer: { + set: (cb: () => void, ms: number) => ReturnType; + } = { set: (cb, ms) => setTimeout(cb, ms) }, +): () => void { + let scheduled: ReturnType | null = null; + return () => { + if (scheduled) return; + scheduled = timer.set(() => { + scheduled = null; + run(); + }, delayMs); + }; +} + export async function startUnifiedDashboard( opts: UnifiedDashboardOptions, ): Promise { @@ -836,8 +875,13 @@ export async function startUnifiedDashboard( writeLocalStatusSnapshot(opts); if (!isRenderer) return; // A failed edit now retries with delays, so a single updateStatus can run - // for tens of seconds. Skip overlapping interval ticks to avoid double posts. - if (statusUpdateRunning) return; + // for tens of seconds. Skip overlapping ticks (this is also why the base + // periodic refresh does not fire mid-retry); remember the request so it runs + // once the in-flight update finishes. + if (statusUpdateRunning) { + statusUpdatePending = true; + return; + } statusUpdateRunning = true; try { @@ -926,9 +970,22 @@ export async function startUnifiedDashboard( statusMessageId = null; } finally { statusUpdateRunning = false; + if (statusUpdatePending) { + statusUpdatePending = false; + immediateUpdateTrigger?.(); + } } }; + if (isRenderer) { + // Event-driven refresh: external callers (new chat message, agent activity + // change) call requestImmediateStatusUpdate(); bursts are coalesced. + immediateUpdateTrigger = createCoalescingTrigger( + () => void updateStatus(), + IMMEDIATE_UPDATE_DEBOUNCE_MS, + ); + } + setInterval(updateStatus, opts.statusUpdateInterval); setInterval(() => { if (!isRenderer || !statusMessageId) return;