[Bug]: JS SDK: filesystem watch handle.stop() always fires onExit with a TimeoutError blaming 'requestTimeoutMs' #1895

/**
 * Repro: WatchHandle.stop() — clean, user-initiated exit, or an error?
 *
 * The Python SDK treats stop() as a clean exit (on_exit(None), added in the
 * same commit de0c4016 that fixed the JS watch handle). The JS SDK has no such
 * handling — this repro drives the real connect transport against a mock envd
 * to see what onExit receives when the user calls stop().
 */
import { describe, expect, it } from 'vitest'
import http from 'node:http'

import { createConnectTransport } from '@connectrpc/connect-web'

import { Filesystem } from '../../src/sandbox/filesystem/index'
import { EnvdApiClient } from '../../src/envd/api'
import { ConnectionConfig } from '../../src/connectionConfig'

function frame(flags: number, payload: string): Buffer {
  const body = Buffer.from(payload)
  const head = Buffer.alloc(5)
  head.writeUInt8(flags, 0)
  head.writeUInt32BE(body.length, 1)
  return Buffer.concat([head, body])
}

type Stream = {
  send: (obj: unknown) => void
  end: () => void
}

function startMockEnvd(routes: Record<string, (res: http.ServerResponse) => Stream>) {
  const requests: Array<{ path: string; headers: http.IncomingHttpHeaders }> = []

  const server = http.createServer((req, res) => {
    const chunks: Buffer[] = []
    req.on('data', (c: Buffer) => chunks.push(c))
    req.on('end', () => {
      requests.push({ path: req.url ?? '', headers: req.headers })
      const handler = routes[(req.url ?? '').split('?')[0]]
      if (!handler) {
        res.writeHead(404)
        res.end()
        return
      }
      handler(res)
    })
  })

  return new Promise<{ port: number; requests: typeof requests; close: () => Promise<void> }>(
    (resolve) => {
      server.listen(0, '127.0.0.1', () => {
        const { port } = server.address() as { port: number }
        resolve({
          port,
          requests,
          close: () => new Promise((done) => server.close(() => done())),
        })
      })
    }
  )
}

function streamRoute(res: http.ServerResponse): Stream {
  res.writeHead(200, { 'content-type': 'application/connect+json' })
  return {
    send: (obj) => res.write(frame(0x00, JSON.stringify(obj))),
    end: () => {
      res.write(frame(0x02, JSON.stringify({})))
      res.end()
    },
  }
}

describe('WatchHandle.stop() repro', () => {
  it('stop() fires onExit cleanly (no error)', async () => {
    let watchStream: Stream | undefined
    const mock = await startMockEnvd({
      '/filesystem.Filesystem/WatchDir': (res) => {
        watchStream = streamRoute(res)
        // Send the start event; keep the stream open afterwards.
        watchStream.send({ start: {} })
        return watchStream
      },
    })

    const transport = createConnectTransport({
      baseUrl: `http://127.0.0.1:${mock.port}`,
      useBinaryFormat: false,
    })
    const envdApi = new EnvdApiClient(
      { apiUrl: `http://127.0.0.1:${mock.port}` },
      { version: '0.6.4' }
    )
    const files = new Filesystem(transport, envdApi, new ConnectionConfig({}))

    const events: string[] = []
    let exitErr: Error | undefined | 'unset' = 'unset'
    let cleanExit = false

    const handle = await files.watchDir('dir', (e) => {
      events.push(e.name)
    }, {
      onExit: (err) => {
        if (err === undefined) cleanExit = true
        exitErr = err ?? undefined
      },
    })

    // Deliver one filesystem event so the watch is established and streaming.
    watchStream!.send({
      filesystem: { name: 'a.txt', type: 'EVENT_TYPE_WRITE' },
    })
    await new Promise((r) => setTimeout(r, 100))
    expect(events).toEqual(['a.txt'])

    // User-initiated stop.
    await handle.stop()
    await new Promise((r) => setTimeout(r, 300))

    console.log('cleanExit:', cleanExit)
    console.log('exitErr:', exitErr === 'unset' ? 'NOT CALLED' : exitErr)
    if (exitErr !== 'unset' && exitErr !== undefined) {
      console.log('exitErr.name:', exitErr.name)
      console.log('exitErr.message:', exitErr.message)
    }

    // Python SDK (same fix commit de0c4016): stop() => on_exit(None).
    // Expect the same here.
    expect(cleanExit).toBe(true)

    await mock.close()
  }, 10_000)
})