boocode/apps/coder/src/services/backends/__tests__/pushable-iterable.test.ts

import { describe, it, expect } from 'vitest';
import { createPushable } from '../pushable-iterable.js';

/**
 * The pushable async-iterable that feeds the Claude SDK's streaming-input query()
 * one message per turn while staying open across turns. Tests cover the ordering
 * contract (push/close/async-iterate) without any SDK shape.
 */
describe('createPushable — push/iterate ordering', () => {
  it('yields buffered values in FIFO order then parks', async () => {
    const p = createPushable<number>();
    const it = p.iterable[Symbol.asyncIterator]();

    p.push(1);
    p.push(2);
    expect(await it.next()).toEqual({ value: 1, done: false });
    expect(await it.next()).toEqual({ value: 2, done: false });

    // No more buffered → next() parks; resolve it by pushing.
    const parked = it.next();
    p.push(3);
    expect(await parked).toEqual({ value: 3, done: false });
  });

  it('hands a value directly to a parked consumer (push after await)', async () => {
    const p = createPushable<string>();
    const it = p.iterable[Symbol.asyncIterator]();
    const pending = it.next(); // parks immediately (empty buffer)
    p.push('hello');
    expect(await pending).toEqual({ value: 'hello', done: false });
  });

  it('close() resolves a parked consumer as done and reports done thereafter', async () => {
    const p = createPushable<number>();
    const it = p.iterable[Symbol.asyncIterator]();
    const pending = it.next();
    p.close();
    expect(await pending).toEqual({ value: undefined, done: true });
    expect(await it.next()).toEqual({ value: undefined, done: true });
    expect(p.closed).toBe(true);
  });

  it('still drains values buffered BEFORE close', async () => {
    const p = createPushable<number>();
    const it = p.iterable[Symbol.asyncIterator]();
    p.push(10);
    p.push(20);
    p.close();
    expect(await it.next()).toEqual({ value: 10, done: false });
    expect(await it.next()).toEqual({ value: 20, done: false });
    expect(await it.next()).toEqual({ value: undefined, done: true });
  });

  it('drops values pushed after close', async () => {
    const p = createPushable<number>();
    const it = p.iterable[Symbol.asyncIterator]();
    p.close();
    p.push(99); // no-op
    expect(await it.next()).toEqual({ value: undefined, done: true });
  });

  it('close() is idempotent', () => {
    const p = createPushable<number>();
    p.close();
    expect(() => p.close()).not.toThrow();
    expect(p.closed).toBe(true);
  });

  it('works with a for-await loop driven by interleaved pushes', async () => {
    const p = createPushable<number>();
    const seen: number[] = [];
    const consumer = (async () => {
      for await (const v of p.iterable) seen.push(v);
    })();

    p.push(1);
    await Promise.resolve();
    p.push(2);
    await Promise.resolve();
    p.close();
    await consumer;
    expect(seen).toEqual([1, 2]);
  });

  it('return() on the iterator closes the queue (for-await break)', async () => {
    const p = createPushable<number>();
    const it = p.iterable[Symbol.asyncIterator]();
    p.push(1);
    expect(await it.next()).toEqual({ value: 1, done: false });
    // Simulate a `break` in for-await: the runtime calls return().
    expect(await it.return!()).toEqual({ value: undefined, done: true });
    expect(p.closed).toBe(true);
    p.push(2); // dropped — queue is closed
    expect(await it.next()).toEqual({ value: undefined, done: true });
  });
});