Chunking, bubbles, and total duration limit

The policy implementation follows.

export async function* publishReplyEvents(
  request: { text: string; delivery?: ReplyDeliveryOptions },
  context: ExecutionContext,
): AsyncIterable<AgentOutputEvent> {
  breakWhile(
    !request || typeof request !== 'object',
    () => new TypeError('Use a named reply delivery request'),
  );
  const { text, delivery } = request,
    { signal } = context;
  breakWhile(
    typeof text !== 'string' || !signal,
    () => new TypeError('Reply delivery requires text and ExecutionContext'),
  );
  breakWhile(
    delivery !== undefined && (!delivery || typeof delivery !== 'object'),
    () => new TypeError('Reply delivery requires named options'),
  );
  const {
    chunkSize = 240,
    intervalMs = 0,
    bubblePauseMs = 0,
    maxDurationMs = 3000,
  } = delivery ?? {};
  breakWhile(
    !Number.isSafeInteger(chunkSize) || chunkSize < 1,
    () => new TypeError('Invalid output chunk size'),
  );
  for (const [name, value] of Object.entries({ intervalMs, bubblePauseMs, maxDurationMs }))
    breakWhile(!Number.isFinite(value) || value < 0, () => new TypeError(`Invalid ${name}`));
  signal.throwIfAborted();
  const bubbles = text.split(REPLY_BUBBLE_BOUNDARY).map((bubble) => Array.from(bubble));
  const chunkCount = bubbles.reduce((sum, bubble) => sum + Math.ceil(bubble.length / chunkSize), 0);
  const pause = Math.min(bubblePauseMs, maxDurationMs / Math.max(1, bubbles.length - 1));
  const interval = Math.min(
    intervalMs,
    Math.max(0, maxDurationMs - pause * (bubbles.length - 1)) /
      Math.max(1, chunkCount - bubbles.filter((bubble) => bubble.length).length),
  );
  const deadline = performance.now() + maxDurationMs;
  for (let bubble = 0; bubble < bubbles.length; bubble++) {
    if (bubble) {
      await wait(Math.min(pause, Math.max(0, deadline - performance.now())), signal);
      signal.throwIfAborted();
      yield { type: 'bubble' };
    }
    for (let i = 0; i < bubbles[bubble].length; i += chunkSize) {
      signal.throwIfAborted();
      if (i) await wait(Math.min(interval, Math.max(0, deadline - performance.now())), signal);
      signal.throwIfAborted();
      yield { type: 'text', text: bubbles[bubble].slice(i, i + chunkSize).join('') };
    }
  }
  signal.throwIfAborted();
}

export async function* serializeOutputEvents(
  events: AsyncIterable<AgentOutputEvent>,
): AsyncIterable<string> {
  for await (const event of events) {
    if (event.type === 'preview') continue;
    yield event.type === 'bubble' ? REPLY_BUBBLE_BOUNDARY : event.text;
  }
}

export async function* textOutputEvents(
  chunks: AsyncIterable<string>,
): AsyncIterable<AgentOutputEvent> {
  for await (const text of chunks) yield { type: 'text', text };
}