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 <dread9ko@gmail.com>
This commit is contained in:
Brennan Benson
2026-11-12 23:58:11 -08:00
co-authored by Taras Kornichuk
parent 5ba2809643
commit a5d2be6fa8
3 changed files with 112 additions and 44 deletions
@@ -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)
@@ -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])
}
)
@@ -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,