From a5d2be6fa8b47d4f130627f19edbbc16eb3c13fa Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 12 Nov 2026 23:58:11 -0800 Subject: [PATCH] fix: idle quiet watch for disconnected-only workers Keep disconnected observations for episode and reminder bookkeeping without admitting recurring sweeps. Existing topology and status changes wake discovery on reconnect. Cover disconnected fleets, reconnect grace and budgets, mixed workers, and teardown through the real watch. Co-authored-by: Taras Kornichuk --- .../reports/worker-quiet-admission.test.ts | 59 ++++++++++++ .../reports/worker-quiet-attachments.test.ts | 95 ++++++++++--------- .../workers/reports/worker-quiet-runtime.ts | 2 +- 3 files changed, 112 insertions(+), 44 deletions(-) diff --git a/src/main/runtime/orchestration/workers/reports/worker-quiet-admission.test.ts b/src/main/runtime/orchestration/workers/reports/worker-quiet-admission.test.ts index 18abc4c5bd8..c6fbe497e7e 100644 --- a/src/main/runtime/orchestration/workers/reports/worker-quiet-admission.test.ts +++ b/src/main/runtime/orchestration/workers/reports/worker-quiet-admission.test.ts @@ -1,6 +1,8 @@ import { expect, it, vi } from 'vitest' import { runtimeFixture } from './worker-quiet-runtime.test-fixture' import { startWorkerQuietWatch } from './worker-quiet-watch' +import { TerminalIdentitySource } from '../../../terminal/identity/terminal-identity-source' +import { PANE } from '../../../../agent-hooks/server/testing/server.test-fixtures' const fixture = runtimeFixture() @@ -64,6 +66,63 @@ it.each([ } ) +it.each([100, 1000])( + 'leaves %i disconnected working workers idle until reconnect wakes discovery', + (population) => { + const { db, chat, changes } = fixture + const { run } = fixture.workers(population) + chat.handles.clear() + const terminals = new TerminalIdentitySource(changes.notify) + for (let n = 0; n < population; n++) { + const paneKey = n === 0 ? PANE : `offline-pane-${n}` + const pty = fixture.terminal(paneKey, `term_worker_${n}`) + terminals.createPty({ ...pty, ptyId: paneKey, connected: false }) + } + fixture.store.ingestRemote( + { + paneKey: PANE, + source: 'codex', + hookEventName: 'UserPromptSubmit', + payload: { state: 'working', prompt: 'work', agentType: 'codex' } + }, + 'ssh-owner' + ) + const inventory = vi.fn(() => terminals.ptys.keys()) + const local = vi.spyOn(db, 'getActiveDispatchForTerminal') + fixture.start({ + readPaneKeys: inventory, + readPty: (key) => terminals.ptys.get(key) ?? null + }) + // One startup pass retains each episode; none can produce a notice while disconnected. + expect(inventory).toHaveBeenCalledTimes(1) + expect(local).toHaveBeenCalledTimes(population) + vi.advanceTimersByTime(3_600_000) + expect(inventory).toHaveBeenCalledTimes(1) + expect(local).toHaveBeenCalledTimes(population) + expect(db.getUnreadRunMailbox(run.id, 100, ['status'])).toHaveLength(0) + const paneKey = PANE + const pty = terminals.ptys.get(paneKey)! + terminals.setPtyConnected(pty, true) + vi.advanceTimersByTime(0) + expect(inventory).toHaveBeenCalledTimes(2) + expect(local).toHaveBeenCalledTimes(population * 2) + vi.advanceTimersByTime(840_000) + expect(db.getUnreadRunMailbox(run.id, 100, ['status'])).toHaveLength(0) + vi.advanceTimersByTime(60_000) + const notices = db.getUnreadRunMailbox(run.id, 100, ['status']) + expect(notices).toHaveLength(1) + expect(JSON.parse(notices[0]!.payload!).quietNotice.observedAt).toBe(4_500_000) + terminals.setPtyConnected(pty, false) + fixture.stop() + inventory.mockClear() + local.mockClear() + terminals.setPtyConnected(pty, true) + vi.advanceTimersByTime(3_600_000) + expect(inventory).not.toHaveBeenCalled() + expect(local).not.toHaveBeenCalled() + } +) + it('admits a newly working chat within one interval without event-driven postponement', async () => { const { db, chat, changes } = fixture const { run } = fixture.workers(1) diff --git a/src/main/runtime/orchestration/workers/reports/worker-quiet-attachments.test.ts b/src/main/runtime/orchestration/workers/reports/worker-quiet-attachments.test.ts index 272f0fd780d..c16686fb7b6 100644 --- a/src/main/runtime/orchestration/workers/reports/worker-quiet-attachments.test.ts +++ b/src/main/runtime/orchestration/workers/reports/worker-quiet-attachments.test.ts @@ -102,46 +102,55 @@ it('recovers acknowledged remote notices only for the current live process after expect(rows[1]!.message_id).not.toBe(first.message_id) }) -it('preserves a disconnected worker episode, excluding offline time and retaining its notice budget', () => { - const { run, workers } = fixture.workers(2) - fixture.chat.handles.delete('term_worker_0') - const pty = fixture.terminal(PANE, 'term_worker_0') - fixture.store.ingestRemote( - { - paneKey: PANE, - source: 'codex', - hookEventName: 'UserPromptSubmit', - payload: { state: 'working', prompt: 'work', agentType: 'codex' } - }, - 'ssh-owner' - ) - expect(fixture.store.getStatusSnapshotForPane(PANE)[0]).toMatchObject({ - state: 'working', - turnStartedAt: 0 - }) - const noticeTimes = () => - fixture.db - .getUnreadRunMailbox(run.id, 100, ['status']) - .map((message) => JSON.parse(message.payload!).quietNotice) - .filter((notice) => notice.dispatchId === workers[0]!.id) - .map((notice) => notice.observedAt) - fixture.start() - vi.advanceTimersByTime(960_000) - expect(noticeTimes()).toEqual([900_000]) - fixture.panes.set(PANE, { ...pty, connected: false }) - fixture.changes.notify(new Set(['status'])) - vi.advanceTimersByTime(1_200_000) - fixture.panes.set(PANE, { ...pty, connected: true }) - fixture.changes.notify(new Set(['status'])) - vi.advanceTimersByTime(840_000) - expect(noticeTimes()).toEqual([900_000]) - vi.advanceTimersByTime(960_000) - expect(noticeTimes()).toEqual([900_000, 3_900_000]) - fixture.panes.set(PANE, { ...pty, connected: false }) - fixture.changes.notify(new Set(['status'])) - vi.advanceTimersByTime(2_040_000) - fixture.panes.set(PANE, { ...pty, connected: true }) - fixture.changes.notify(new Set(['status'])) - vi.advanceTimersByTime(7_200_000) - expect(noticeTimes()).toEqual([900_000, 3_900_000]) -}) +it.each([ + [0, 3_840_000], + [1, 3_900_000] +])( + 'preserves a disconnected episode and budget with %i other working chats', + (chats, reminderAt) => { + const { run, workers } = fixture.workers(1 + chats) + fixture.chat.handles.delete('term_worker_0') + const pty = fixture.terminal(PANE, 'term_worker_0') + fixture.store.ingestRemote( + { + paneKey: PANE, + source: 'codex', + hookEventName: 'UserPromptSubmit', + payload: { state: 'working', prompt: 'work', agentType: 'codex' } + }, + 'ssh-owner' + ) + expect(fixture.store.getStatusSnapshotForPane(PANE)[0]).toMatchObject({ + state: 'working', + turnStartedAt: 0 + }) + const noticeTimes = () => + fixture.db + .getUnreadRunMailbox(run.id, 100, ['status']) + .map((message) => JSON.parse(message.payload!).quietNotice) + .filter((notice) => notice.dispatchId === workers[0]!.id) + .map((notice) => notice.observedAt) + const inventory = vi.fn(() => fixture.panes.keys()) + fixture.start({ readPaneKeys: inventory }) + vi.advanceTimersByTime(960_000) + expect(noticeTimes()).toEqual([900_000]) + fixture.panes.set(PANE, { ...pty, connected: false }) + inventory.mockClear() + fixture.changes.notify(new Set(['status'])) + vi.advanceTimersByTime(1_200_000) + expect(inventory).toHaveBeenCalledTimes(chats ? 20 : 1) + fixture.panes.set(PANE, { ...pty, connected: true }) + fixture.changes.notify(new Set(['status'])) + vi.advanceTimersByTime(840_000) + expect(noticeTimes()).toEqual([900_000]) + vi.advanceTimersByTime(960_000) + expect(noticeTimes()).toEqual([900_000, reminderAt]) + fixture.panes.set(PANE, { ...pty, connected: false }) + fixture.changes.notify(new Set(['status'])) + vi.advanceTimersByTime(2_040_000) + fixture.panes.set(PANE, { ...pty, connected: true }) + fixture.changes.notify(new Set(['status'])) + vi.advanceTimersByTime(7_200_000) + expect(noticeTimes()).toEqual([900_000, reminderAt]) + } +) diff --git a/src/main/runtime/orchestration/workers/reports/worker-quiet-runtime.ts b/src/main/runtime/orchestration/workers/reports/worker-quiet-runtime.ts index e794070ca55..ba589159cc6 100644 --- a/src/main/runtime/orchestration/workers/reports/worker-quiet-runtime.ts +++ b/src/main/runtime/orchestration/workers/reports/worker-quiet-runtime.ts @@ -63,7 +63,7 @@ export function startWorkerQuietSweep( Date.now(), [...candidates.keys()] ) - return workers.length > 0 + return workers.some(({ observation }) => observation.connected !== false) } return startWorkerQuietWatch({ sweep: run,