feat(dashboard): 60s base refresh with event-driven immediate updates

Raise the status dashboard base refresh from 10s to 60s, and refresh
immediately (debounced 1.5s) on events between ticks:
- a real chat message arrives in a registered room (index.ts onMessage)
- an agent starts/finishes a run, i.e. activity moves between rooms
  (GroupQueue.setOnActivityChange fired on activeCount changes)

requestImmediateStatusUpdate() drives updateStatus out-of-band via a coalescing
trigger. The existing re-entrancy guard means the base periodic refresh does not
run while an edit-retry is in flight; a refresh requested during that window is
remembered and runs once afterward. Adds createCoalescingTrigger with tests.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Codex
2026-08-21 12:53:39 +09:00
parent 9108174380
commit 8600213bcc
5 changed files with 122 additions and 4 deletions

View File

@@ -259,7 +259,9 @@ export function loadConfig(): AppConfig {
moa: buildMoaConfig(), moa: buildMoaConfig(),
status: { status: {
channelId: readText('STATUS_CHANNEL_ID') ?? '', 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, usageUpdateInterval: 60000,
showRooms: readBooleanUnlessFalse('STATUS_SHOW_ROOMS', true), showRooms: readBooleanUnlessFalse('STATUS_SHOW_ROOMS', true),
showRoomDetails: readBooleanUnlessFalse('STATUS_SHOW_ROOM_DETAILS', true), showRoomDetails: readBooleanUnlessFalse('STATUS_SHOW_ROOM_DETAILS', true),

View File

@@ -59,12 +59,28 @@ export class GroupQueue {
return state; return state;
} }
private onActivityChange: (() => void) | null = null;
setProcessMessagesFn( setProcessMessagesFn(
fn: (groupJid: string, context: GroupRunContext) => Promise<boolean>, fn: (groupJid: string, context: GroupRunContext) => Promise<boolean>,
): void { ): void {
this.processMessagesFn = fn; 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. */ /** Limit concurrency after restart to avoid API rate-limit storms. */
enterRecoveryMode(): void { enterRecoveryMode(): void {
this.recoveryMode = true; this.recoveryMode = true;
@@ -528,6 +544,7 @@ export class GroupQueue {
transitionRunPhase(state, groupJid, 'running_messages', { runId, reason }); transitionRunPhase(state, groupJid, 'running_messages', { runId, reason });
assertRunPhaseInvariants(state, groupJid); assertRunPhaseInvariants(state, groupJid);
this.activeCount++; this.activeCount++;
this.notifyActivityChange();
logger.info( logger.info(
{ {
@@ -580,6 +597,7 @@ export class GroupQueue {
resetRunState(state, groupJid); resetRunState(state, groupJid);
assertRunPhaseInvariants(state, groupJid); assertRunPhaseInvariants(state, groupJid);
this.activeCount--; this.activeCount--;
this.notifyActivityChange();
this.drainGroup(groupJid); this.drainGroup(groupJid);
} }
} }
@@ -592,6 +610,7 @@ export class GroupQueue {
assertRunPhaseInvariants(state, groupJid); assertRunPhaseInvariants(state, groupJid);
this.activeCount++; this.activeCount++;
this.activeTaskCount++; this.activeTaskCount++;
this.notifyActivityChange();
logger.info( logger.info(
{ {
@@ -628,6 +647,7 @@ export class GroupQueue {
assertRunPhaseInvariants(state, groupJid); assertRunPhaseInvariants(state, groupJid);
this.activeCount--; this.activeCount--;
this.activeTaskCount--; this.activeTaskCount--;
this.notifyActivityChange();
this.drainGroup(groupJid); this.drainGroup(groupJid);
} }
} }

View File

@@ -52,7 +52,10 @@ import {
} from './sender-allowlist.js'; } from './sender-allowlist.js';
import { createMessageRuntime } from './message-runtime.js'; import { createMessageRuntime } from './message-runtime.js';
import { nudgeSchedulerLoop, startSchedulerLoop } from './task-scheduler.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 { startUsagePrimer } from './usage-primer.js';
import { startWebDashboardServer } from './web-dashboard-server.js'; import { startWebDashboardServer } from './web-dashboard-server.js';
import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js';
@@ -395,6 +398,11 @@ async function main(): Promise<void> {
} }
} }
storeMessage(msg); 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: ( onChatMetadata: (
chatJid: string, chatJid: string,
@@ -569,6 +577,9 @@ async function main(): Promise<void> {
}, },
purgeOnStart: true, 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({ webDashboardServer = startWebDashboardServer({
...WEB_DASHBOARD, ...WEB_DASHBOARD,
getRoomBindings: runtimeState.getRoomBindings, getRoomBindings: runtimeState.getRoomBindings,

View File

@@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest';
import type { UsageRow } from './dashboard-usage-rows.js'; import type { UsageRow } from './dashboard-usage-rows.js';
import { import {
buildWebUsageRowsForSnapshot, buildWebUsageRowsForSnapshot,
createCoalescingTrigger,
editStatusMessageWithRetry, editStatusMessageWithRetry,
formatStatusHeader, formatStatusHeader,
getDashboardDuplicateCleanupIntervalMs, getDashboardDuplicateCleanupIntervalMs,
@@ -312,3 +313,30 @@ describe('editStatusMessageWithRetry', () => {
expect(slept).toBe(0); 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<typeof setTimeout>;
},
};
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);
});
});

View File

@@ -111,12 +111,28 @@ let channelMetaLastRefresh = 0;
let dashboardUpdateLogged = false; let dashboardUpdateLogged = false;
/** Guards against overlapping status updates while a slow edit-retry runs. */ /** Guards against overlapping status updates while a slow edit-retry runs. */
let statusUpdateRunning = false; 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 * 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. * STATUS_EDIT_RETRY_DELAY_MS spacing before falling back to a fresh message.
*/ */
const STATUS_EDIT_MAX_RETRIES = 2; const STATUS_EDIT_MAX_RETRIES = 2;
const STATUS_EDIT_RETRY_DELAY_MS = 15_000; 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. */ /** Codex service only: cached usage rows written into the status snapshot. */
let cachedCodexUsageRows: UsageRow[] = []; let cachedCodexUsageRows: UsageRow[] = [];
/** Codex service only: ISO timestamp of last successful usage fetch. */ /** Codex service only: ISO timestamp of last successful usage fetch. */
@@ -801,6 +817,29 @@ export async function editStatusMessageWithRetry(args: {
return false; 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<typeof setTimeout>;
} = { set: (cb, ms) => setTimeout(cb, ms) },
): () => void {
let scheduled: ReturnType<typeof setTimeout> | null = null;
return () => {
if (scheduled) return;
scheduled = timer.set(() => {
scheduled = null;
run();
}, delayMs);
};
}
export async function startUnifiedDashboard( export async function startUnifiedDashboard(
opts: UnifiedDashboardOptions, opts: UnifiedDashboardOptions,
): Promise<void> { ): Promise<void> {
@@ -836,8 +875,13 @@ export async function startUnifiedDashboard(
writeLocalStatusSnapshot(opts); writeLocalStatusSnapshot(opts);
if (!isRenderer) return; if (!isRenderer) return;
// A failed edit now retries with delays, so a single updateStatus can run // 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. // for tens of seconds. Skip overlapping ticks (this is also why the base
if (statusUpdateRunning) return; // 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; statusUpdateRunning = true;
try { try {
@@ -926,9 +970,22 @@ export async function startUnifiedDashboard(
statusMessageId = null; statusMessageId = null;
} finally { } finally {
statusUpdateRunning = false; 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(updateStatus, opts.statusUpdateInterval);
setInterval(() => { setInterval(() => {
if (!isRenderer || !statusMessageId) return; if (!isRenderer || !statusMessageId) return;