hermes-ui/tests/server/agent-bridge-worker-readiness.test.ts

import { createServer, type Server } from 'net'
import { afterEach, describe, expect, it } from 'vitest'
import { AgentBridgeClient } from '../../packages/server/src/services/hermes/agent-bridge/client'

const servers: Server[] = []
afterEach(async () => {
  await Promise.all(servers.splice(0).map(server => new Promise<void>(resolve => server.close(() => resolve()))))
})

async function broker(reply: (request: Record<string, unknown>) => Record<string, unknown>, retryMs = 1000) {
  const server = createServer(socket => {
    let buffer = ''
    socket.on('data', data => {
      buffer += data.toString()
      if (!buffer.includes('\n')) return
      socket.end(`${JSON.stringify(reply(JSON.parse(buffer)))}\n`)
    })
  })
  servers.push(server)
  await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve))
  const address = server.address()
  if (!address || typeof address === 'string') throw new Error('Missing test broker address')
  return new AgentBridgeClient({ endpoint: `tcp://127.0.0.1:${address.port}`, connectRetryMs: retryMs, timeoutMs: 1000 })
}

const refused = { ok: false, error: '[Errno 111] Connection refused', error_type: 'ConnectionRefusedError' }

describe('Agent Bridge worker readiness', () => {
  it('waits for a cold worker and for a worker restarting after idle without losing the message', async () => {
    const requests: Record<string, unknown>[] = []
    let ready = false
    const client = await broker(request => {
      requests.push(request)
      if (!ready) {
        ready = true
        return refused
      }
      return { ok: true, run_id: `run-${requests.length}`, session_id: request.session_id }
    })
    await expect(client.chat('session', 'first message', [], undefined, 'default')).resolves.toMatchObject({ ok: true })
    ready = false
    await expect(client.chat('session', 'after idle', [{ role: 'user', content: 'first message' }])).resolves.toMatchObject({ ok: true })
    expect(requests).toHaveLength(4)
    expect(requests[0]).toEqual(requests[1])
    expect(requests[2]).toEqual(requests[3])
  })

  it('retries a worker socket that has not been created yet', async () => {
    let attempts = 0
    const client = await broker(() => ++attempts === 1
      ? { ok: false, error: '[Errno 2] No such file or directory', error_type: 'FileNotFoundError' }
      : { ok: true })
    await expect(client.request({ action: 'worker_ping', profile: 'default' })).resolves.toEqual({ ok: true })
    expect(attempts).toBe(2)
  })

  it.each([
    { ok: false, error: '[Errno 111] Connection refused' },
    { ok: false, error: '[Errno 111] Connection refused', error_type: 'RuntimeError' },
    { ok: false, error: 'worker closed without a response', error_type: 'RuntimeError' },
    { ok: false, error: 'timed out', error_type: 'TimeoutError' },
  ])('does not replay ambiguous failures: $error_type $error', async response => {
    let attempts = 0
    const client = await broker(() => { attempts++; return response })
    await expect(client.chat('session', 'message')).rejects.toThrow(response.error)
    expect(attempts).toBe(1)
  })

  it('stops retrying persistent worker refusal within the configured budget', async () => {
    let attempts = 0
    const client = await broker(() => { attempts++; return refused }, 150)
    await expect(client.chat('session', 'message')).rejects.toThrow('Connection refused')
    expect(attempts).toBeGreaterThan(1)
    expect(attempts).toBeLessThanOrEqual(4)
  })

  it('honors disabled retries', async () => {
    let attempts = 0
    const client = await broker(() => { attempts++; return refused }, 0)
    await expect(client.chat('session', 'message')).rejects.toThrow('Connection refused')
    expect(attempts).toBe(1)
  })
})