deepseek-harness/packages/credentials/credentials-local/tests/watcher.spec.ts

import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { chmod, mkdtemp, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { credentialRef } from '@deepseek-ai/dsh-credentials'
import { CredentialsLocal } from '../src/index.ts'

/** Credential documents are seeded owner-only, exactly as the provider creates them. */
function writeCredentials(file: string, text: string): Promise<void> {
  return writeFile(file, text, { mode: 0o600 })
}

// chokidar is the nondeterministic OS boundary: faking it lets these tests
// drive the event pipeline (error events, races with unreadable files)
// deterministically. Real end-to-end watching stays covered by local.spec.ts.
vi.mock('chokidar', async () => {
  const { EventEmitter } = await import('node:events')
  class FakeWatcher extends EventEmitter {
    close = vi.fn(() => Promise.resolve())
  }
  const instances: Array<{ path: string; options: unknown; watcher: InstanceType<typeof FakeWatcher> }> = []
  return {
    watch: vi.fn((path: string, options: unknown) => {
      const watcher = new FakeWatcher()
      instances.push({ path, options, watcher })
      return watcher
    }),
    __instances: instances,
  }
})

interface FakeChokidar {
  __instances: Array<{
    path: string
    options: { awaitWriteFinish: { stabilityThreshold: number; pollInterval: number } }
    watcher: import('node:events').EventEmitter
  }>
}

async function fakeInstances(): Promise<FakeChokidar['__instances']> {
  const chokidar = await import('chokidar') as unknown as FakeChokidar
  return chokidar.__instances
}

const KEY = credentialRef('DSH_CRED_PIPE')

const cleanups: Array<() => Promise<void>> = []

afterEach(async () => {
  while (cleanups.length > 0) await cleanups.pop()!()
  ;(await fakeInstances()).length = 0
})

async function tempDir(): Promise<string> {
  const dir = await mkdtemp(join(tmpdir(), 'dsh-credentials-watch-'))
  cleanups.push(() => rm(dir, { recursive: true, force: true }))
  return dir
}

async function boot(config: ConstructorParameters<typeof CredentialsLocal>[1]): Promise<Context> {
  const ctx = new Context()
  const fiber = ctx.plugin(CredentialsLocal, config)
  cleanups.push(async () => {
    await fiber.dispose()
  })
  await fiber
  return ctx
}

describe('watcher pipeline', () => {
  it('clamps the write-settle poll interval for a zero debounce', async () => {
    const dir = await tempDir()
    await boot({ path: join(dir, '.credentials.yaml'), debounceMs: 0 })
    const [instance] = await fakeInstances()
    expect(instance!.options.awaitWriteFinish).toEqual({ stabilityThreshold: 0, pollInterval: 1 })
  })

  it('survives a watcher error and keeps publishing later edits', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    const ctx = await boot({ path, debounceMs: 5 })
    const [instance] = await fakeInstances()

    instance!.watcher.emit('error', new Error('watch backend failure'))
    expect(await ctx.credentials.resolve(KEY)).toBeUndefined()

    await writeCredentials(path, 'DSH_CRED_PIPE: arrived\n')
    instance!.watcher.emit('all', 'change', path)
    await vi.waitFor(async () => {
      expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'arrived', source: 'file' })
    })
  })

  it('keeps the last good snapshot when the file turns unreadable at runtime', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    await writeCredentials(path, 'DSH_CRED_PIPE: good\n')
    const ctx = await boot({ path, debounceMs: 5 })

    await chmod(path, 0o000)
    cleanups.push(() => chmod(path, 0o600))
    const [instance] = await fakeInstances()
    instance!.watcher.emit('all', 'change', path)
    // The warn-and-keep path is asynchronous; give the serialized refresh a turn.
    await new Promise(resolve => setTimeout(resolve, 50))
    expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'good', source: 'file' })
  })

  it('keeps the reload queue alive after an invariant violation escapes the fan-out', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    const ctx = await boot({ path, debounceMs: 5 })
    let arm = true
    ctx.on('credentials/updated', () => {
      if (!arm) return
      throw Object.assign(new Error('forged relation'), { code: 'INVARIANT' })
    })
    const [instance] = await fakeInstances()

    await writeCredentials(path, 'DSH_CRED_PIPE: first\n')
    instance!.watcher.emit('all', 'change', path)
    // The snapshot commits before the fan-out, so the value lands even though
    // the listener threw out of the refresh.
    await vi.waitFor(async () => {
      expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'first', source: 'file' })
    })

    arm = false
    await writeCredentials(path, 'DSH_CRED_PIPE: second\n')
    instance!.watcher.emit('all', 'change', path)
    await vi.waitFor(async () => {
      expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'second', source: 'file' })
    })
  })

  it('quiesces the refresh pipeline before dispose completes', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    await writeCredentials(path, 'DSH_CRED_PIPE: initial\n')
    const ctx = new Context()
    const fiber = ctx.plugin(CredentialsLocal, { path, debounceMs: 5 })
    await fiber
    let disposed = false
    let postDisposeCommits = 0
    ctx.on('credentials/updated', () => {
      if (disposed) postDisposeCommits += 1
    })

    await writeCredentials(path, 'DSH_CRED_PIPE: changed\n')
    const [instance] = await fakeInstances()
    // Two queued refreshes: dispose interrupts one mid-flight and the other
    // before it starts, so both closed guards must hold.
    instance!.watcher.emit('all', 'change', path)
    instance!.watcher.emit('all', 'change', path)
    await fiber.dispose()
    disposed = true
    instance!.watcher.emit('all', 'change', path)
    instance!.watcher.emit('ready')
    await new Promise(resolve => setTimeout(resolve, 100))
    expect(postDisposeCommits).toBe(0)
  })

  it('empties the snapshot when the document is deleted and emits the removals', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    await writeCredentials(path, 'DSH_CRED_PIPE: doomed\n')
    const ctx = await boot({ path, debounceMs: 5 })
    const seen: string[] = []
    ctx.on('credentials/updated', (ref) => {
      seen.push(ref)
    })

    await rm(path)
    const [instance] = await fakeInstances()
    instance!.watcher.emit('all', 'unlink', path)
    await vi.waitFor(async () => {
      expect(await ctx.credentials.resolve(KEY)).toBeUndefined()
    })
    expect(seen).toEqual([KEY])
  })

  it('keeps the last good snapshot when an external edit makes the document invalid', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    await writeCredentials(path, 'DSH_CRED_PIPE: a\n')
    const ctx = await boot({ path, debounceMs: 5 })
    const seen: string[] = []
    ctx.on('credentials/updated', (ref) => {
      seen.push(ref)
    })

    // A key the seam cannot address is a rejection, not preserved content:
    // this document holds nothing but credentials. A live reload must warn
    // and keep serving the last good snapshot rather than take the process
    // down or silently drop the entry it could not validate.
    await writeCredentials(path, 'BAD-KEY: 2\nDSH_CRED_PIPE: b\n')
    const [instance] = await fakeInstances()
    instance!.watcher.emit('all', 'change', path)
    await new Promise(resolve => setTimeout(resolve, 50))
    expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'a', source: 'file' })
    expect(seen).toEqual([])

    // Repairing the document resumes publishing.
    await writeCredentials(path, 'DSH_CRED_PIPE: b\n')
    instance!.watcher.emit('all', 'change', path)
    await vi.waitFor(async () => {
      expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'b', source: 'file' })
    })
    expect(seen).toEqual([KEY])
  })

  it('treats an event for a still-absent file as a no-op', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    const ctx = await boot({ path, debounceMs: 5 })
    const [instance] = await fakeInstances()
    instance!.watcher.emit('all', 'add', path)
    await new Promise(resolve => setTimeout(resolve, 50))
    expect(await ctx.credentials.resolve(KEY)).toBeUndefined()
  })

  it('reconciles at watcher ready so a change during setup is not missed', async () => {
    const dir = await tempDir()
    const path = join(dir, '.credentials.yaml')
    await writeCredentials(path, `${KEY}: a\n`)
    const ctx = await boot({ path, debounceMs: 5 })
    // Written after the initial load but before the watcher became active:
    // no 'all' event will ever fire for it.
    await writeCredentials(path, `${KEY}: written-before-ready\n`)
    const [instance] = await fakeInstances()
    instance!.watcher.emit('ready')
    await vi.waitFor(async () => {
      expect(await ctx.credentials.resolve(KEY)).toEqual({ value: 'written-before-ready', source: 'file' })
    })
  })
})