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,