refactor(jobs-local): extract the output ring and the registry-owned pump

packages/jobs/jobs-local/src/pump.ts

@@ -0,0 +1,85 @@
+/**
+ * The registry-owned pull pump: copies a job's {@link JobOutputSource}s into
+ * its ring at a bounded cadence and drains them once more after the
+ * producer settles, so the ring holds every byte before settlement trims and
+ * closes it. Pure utility — no cordis, no timers retained past settlement.
+ * @module @deepseek-ai/dsh-jobs-local/pump
+ */
+
+import type { JobAppendOptions, JobOutputSource } from '@deepseek-ai/dsh-jobs'
+
+/** One pump run; `done` resolves after the final post-settlement drain. */
+export interface PumpHandle {
+  done: Promise<void>
+}
+
+/**
+ * Drain every source in array order, sleep `pollMs` or until `until`
+ * settles, repeat, then drain one final time. A lossy source read appends
+ * its surviving tail with `gapBefore`, so the discontinuity stays visible to
+ * observers. Sources drain in array order each round, so bytes two sources
+ * produced inside one poll window land in that order, not in the order they
+ * were written: the ring is a best-effort live view whose cross-source
+ * reordering is bounded by `pollMs`.
+ *
+ * The wait holds constant resources however long the job runs: one
+ * subscription on `until` for the whole run and one pending timer at a time.
+ * @param sources - producer streams, each pumped at its own offset.
+ * @param append - the ring append; receives each copied chunk.
+ * @param pollMs - poll interval in milliseconds; a positive finite number.
+ * @param until - settles (or rejects, which counts as settlement) when the producer finished.
+ * @returns the handle whose `done` resolves after the final drain.
+ */
+export function startPump(
+  sources: readonly JobOutputSource[],
+  append: (text: string, options?: JobAppendOptions) => void,
+  pollMs: number,
+  until: Promise<unknown>,
+): PumpHandle {
+  if (!Number.isFinite(pollMs) || pollMs <= 0) {
+    throw new Error(`invalid pump pollMs: expected a positive finite number of milliseconds, got ${JSON.stringify(pollMs)}`)
+  }
+  const states = sources.map(source => ({ source, cursor: 0 }))
+  const drain = (): void => {
+    for (const state of states) {
+      const { source } = state
+      const read = source.read(state.cursor)
+      state.cursor = read.nextOffset
+      if (read.text.length === 0) continue
+      if (source.channel === undefined && !read.lossy) {
+        append(read.text)
+      } else {
+        append(read.text, {
+          ...source.channel !== undefined ? { channel: source.channel } : {},
+          ...read.lossy ? { gapBefore: true as const } : {},
+        })
+      }
+    }
+  }
+  const done = (async (): Promise<void> => {
+    let settled = false
+    let timer: ReturnType<typeof setTimeout> | undefined
+    let wake: (() => void) | undefined
+    const finish = (): void => {
+      settled = true
+      // Clear the pending poll so a finished pump holds no timer for up to one
+      // interval, and release the current wait so the final drain runs at once.
+      // `finish` always runs from a microtask, after the loop parked on its
+      // first timer, so the timer is never absent here.
+      clearTimeout(timer)
+      wake?.()
+    }
+    void until.then(finish, finish)
+    while (!settled) {
+      drain()
+      await new Promise<void>((resolve) => {
+        wake = resolve
+        timer = setTimeout(resolve, pollMs)
+      })
+      wake = undefined
+      timer = undefined
+    }
+    drain()
+  })()
+  return { done }
+}

packages/jobs/jobs-local/src/ring.ts

@@ -0,0 +1,113 @@
+/**
+ * The bounded output ring behind one job: chunks at absolute byte offsets,
+ * head eviction that never moves an assigned offset, and non-consuming reads
+ * from any offset.
+ * @module @deepseek-ai/dsh-jobs-local/ring
+ */
+
+import type { JobAppendOptions, JobChannel, JobOutputRead } from '@deepseek-ai/dsh-jobs'
+
+/** One retained ring entry; `bytes` caches the chunk's UTF-8 length. */
+interface RingChunk {
+  at: number
+  text: string
+  bytes: number
+  channel?: JobChannel
+  gapBefore?: true
+}
+
+/**
+ * The UTF-8-safe tail of `text` no longer than `maxBytes`: the byte cut
+ * advances past continuation bytes so the surviving text never starts inside
+ * a code point.
+ * @param text - the oversized chunk text.
+ * @param maxBytes - positive byte budget for the surviving tail.
+ * @returns the surviving tail and its exact byte length.
+ */
+function utf8Tail(text: string, maxBytes: number): { text: string; bytes: number } {
+  const raw = Buffer.from(text, 'utf8')
+  let start = raw.length - maxBytes
+  // The loop bound proves the index is in range; the assertion only
+  // discharges noUncheckedIndexedAccess.
+  while (start < raw.length && ((raw[start] as number) & 0xC0) === 0x80) start += 1
+  const tail = raw.subarray(start)
+  return { text: tail.toString('utf8'), bytes: tail.length }
+}
+
+/** Offsets stay absolute across eviction: `earliest` only ever advances. */
+export class OutputRing {
+  /** Retained chunks in offset order. */
+  private readonly chunks: RingChunk[] = []
+  /** Sum of the retained chunks' byte lengths. */
+  retainedBytes = 0
+  /** Total UTF-8 bytes ever appended — the offset the next chunk starts at. */
+  total = 0
+  /** Offset of the oldest retained byte (equals {@link total} when nothing is retained). */
+  earliest = 0
+
+  /**
+   * Append one chunk and trim the head to `cap`.
+   * @param text - the chunk text, exactly as produced.
+   * @param options - stream label and gap marker.
+   * @param cap - retention cap in UTF-8 bytes after this append.
+   * @returns false when the chunk was empty and nothing changed.
+   */
+  append(text: string, options: JobAppendOptions | undefined, cap: number): boolean {
+    if (text.length === 0) return false
+    const bytes = Buffer.byteLength(text, 'utf8')
+    this.chunks.push({
+      at: this.total,
+      text,
+      bytes,
+      ...options?.channel !== undefined ? { channel: options.channel } : {},
+      ...options?.gapBefore !== undefined ? { gapBefore: options.gapBefore } : {},
+    })
+    this.total += bytes
+    this.retainedBytes += bytes
+    this.trim(cap)
+    return true
+  }
+
+  /**
+   * Drop retained head chunks until the ring fits `cap`; a single oversized
+   * chunk keeps only its UTF-8-safe tail with a `gapBefore` marker.
+   * @param cap - retention cap in UTF-8 bytes.
+   */
+  trim(cap: number): void {
+    while (this.retainedBytes > cap && this.chunks.length > 1) {
+      const dropped = this.chunks.shift()
+      /* v8 ignore next -- the length guard proves shift() returns a chunk; the check only satisfies noUncheckedIndexedAccess. */
+      if (dropped === undefined) break
+      this.retainedBytes -= dropped.bytes
+    }
+    const single = this.chunks.length === 1 ? this.chunks[0] : undefined
+    if (single !== undefined && single.bytes > cap) {
+      const tail = utf8Tail(single.text, cap)
+      single.at += single.bytes - tail.bytes
+      single.text = tail.text
+      single.bytes = tail.bytes
+      single.gapBefore = true
+      this.retainedBytes = tail.bytes
+    }
+    this.earliest = this.chunks[0]?.at ?? this.total
+  }
+
+  /**
+   * Retained chunks overlapping `[from, total)` as fresh wire chunks.
+   * @param from - absolute byte offset to read from.
+   * @returns the chunks, the resume offset, and whether bytes before them were evicted.
+   */
+  readFrom(from: number): JobOutputRead {
+    const chunks = []
+    for (const chunk of this.chunks) {
+      if (chunk.at + chunk.bytes <= from) continue
+      chunks.push({
+        at: chunk.at,
+        text: chunk.text,
+        ...chunk.channel !== undefined ? { channel: chunk.channel } : {},
+        ...chunk.gapBefore !== undefined ? { gapBefore: chunk.gapBefore } : {},
+      })
+    }
+    return { chunks, next: this.total, lossy: from < this.earliest }
+  }
+}

packages/jobs/jobs-local/tests/pump.spec.ts

@@ -0,0 +1,111 @@
+import { describe, expect, it, vi } from 'vitest'
+import type { JobAppendOptions, JobOutputSource } from '@deepseek-ai/dsh-jobs'
+import { startPump } from '../src/pump.ts'
+
+/** Record every append so pump behavior is observable without a registry. */
+function sink() {
+  const appends: { text: string; options?: JobAppendOptions }[] = []
+  return {
+    appends,
+    append: (text: string, options?: JobAppendOptions) => { appends.push({ text, ...options !== undefined ? { options } : {} }) },
+  }
+}
+
+/** A scripted source: each read() shifts the next scripted result. */
+function scriptedSource(
+  reads: { text: string; lossy?: boolean }[],
+  channel?: 'stdout' | 'stderr',
+): { source: JobOutputSource; offsets: number[] } {
+  const offsets: number[] = []
+  let offset = 0
+  const source: JobOutputSource = {
+    ...channel !== undefined ? { channel } : {},
+    read(fromByte) {
+      offsets.push(fromByte)
+      const next = reads.shift() ?? { text: '' }
+      offset += Buffer.byteLength(next.text, 'utf8')
+      return { text: next.text, nextOffset: offset, lossy: next.lossy ?? false }
+    },
+  }
+  return { source, offsets }
+}
+
+describe('startPump', () => {
+  it('copies labeled deltas at the poll cadence and resumes each source at its own offset', async () => {
+    vi.useFakeTimers()
+    try {
+      const { appends, append } = sink()
+      let settle!: () => void
+      const until = new Promise<void>((resolve) => { settle = resolve })
+      const out = scriptedSource([{ text: 'a' }, { text: 'bc' }], 'stdout')
+      const err = scriptedSource([{ text: '' }, { text: 'E' }], 'stderr')
+      const pump = startPump([out.source, err.source], append, 50, until)
+
+      await vi.advanceTimersByTimeAsync(50)
+      settle()
+      await pump.done
+      expect(appends).toEqual([
+        { text: 'a', options: { channel: 'stdout' } },
+        { text: 'bc', options: { channel: 'stdout' } },
+        { text: 'E', options: { channel: 'stderr' } },
+      ])
+      // Every read resumed from the previous nextOffset, never from a shared cursor.
+      expect(out.offsets).toEqual([0, 1, 3])
+      expect(err.offsets).toEqual([0, 0, 1])
+      // Settlement cleared the pending poll timer instead of leaving it pending.
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('marks a lossy source read as a gap so observers see the discontinuity', async () => {
+    const { appends, append } = sink()
+    const { source } = scriptedSource([{ text: 'tail', lossy: true }])
+    await startPump([source], append, 1, Promise.resolve()).done
+    expect(appends).toEqual([{ text: 'tail', options: { gapBefore: true } }])
+  })
+
+  it('treats a rejected settlement as settlement and still drains the final bytes', async () => {
+    const { appends, append } = sink()
+    const { source } = scriptedSource([{ text: 'last' }])
+    await startPump([source], append, 1, Promise.reject(new Error('producer broke'))).done
+    expect(appends).toEqual([{ text: 'last' }])
+  })
+
+  it('subscribes to the settlement once and holds one poll timer however long the job runs', async () => {
+    vi.useFakeTimers()
+    try {
+      const { append } = sink()
+      let subscriptions = 0
+      let settle!: () => void
+      const settled = new Promise<void>((resolve) => { settle = resolve })
+      const counting = (promise: Promise<unknown>): Promise<unknown> => ({
+        then(onFulfilled?: (value: unknown) => unknown, onRejected?: (reason: unknown) => unknown) {
+          subscriptions += 1
+          return counting(promise.then(onFulfilled, onRejected))
+        },
+      }) as unknown as Promise<unknown>
+      const { source, offsets } = scriptedSource([])
+      const pump = startPump([source], append, 50, counting(settled))
+
+      const rounds = 10_000
+      await vi.advanceTimersByTimeAsync(50 * rounds)
+      expect(offsets).toHaveLength(rounds + 1)
+      expect(subscriptions).toBeLessThanOrEqual(2)
+      expect(vi.getTimerCount()).toBe(1)
+
+      settle()
+      await pump.done
+      expect(offsets).toHaveLength(rounds + 2)
+      expect(vi.getTimerCount()).toBe(0)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
+  it('rejects a non-positive poll interval before touching any source', () => {
+    const { append } = sink()
+    expect(() => startPump([], append, 0, Promise.resolve())).toThrow(/invalid pump pollMs/)
+  })
+})

packages/jobs/jobs-local/tests/ring.spec.ts

@@ -0,0 +1,65 @@
+import { describe, expect, it } from 'vitest'
+import { OutputRing } from '../src/ring.ts'
+
+describe('OutputRing', () => {
+  it('assigns absolute offsets by UTF-8 byte length and drops empty chunks silently', () => {
+    const ring = new OutputRing()
+    expect(ring.append('', undefined, 1024)).toBe(false)
+    expect(ring.append('héllo', { channel: 'stdout' }, 1024)).toBe(true)
+    expect(ring.append('wörld', { channel: 'stderr', gapBefore: true }, 1024)).toBe(true)
+    expect(ring.total).toBe(12)
+    expect(ring.earliest).toBe(0)
+    expect(ring.retainedBytes).toBe(12)
+    expect(ring.readFrom(0)).toEqual({
+      chunks: [
+        { at: 0, text: 'héllo', channel: 'stdout' },
+        { at: 6, text: 'wörld', channel: 'stderr', gapBefore: true },
+      ],
+      next: 12,
+      lossy: false,
+    })
+  })
+
+  it('reads from a foreign offset by returning the whole overlapping chunk', () => {
+    const ring = new OutputRing()
+    ring.append('abc', undefined, 1024)
+    ring.append('def', undefined, 1024)
+    expect(ring.readFrom(4).chunks).toEqual([{ at: 3, text: 'def' }])
+    expect(ring.readFrom(6)).toEqual({ chunks: [], next: 6, lossy: false })
+  })
+
+  it('trims whole head chunks past the cap and reports the evicted span as lossy', () => {
+    const ring = new OutputRing()
+    ring.append('aaaa', undefined, 8)
+    ring.append('bbbb', undefined, 8)
+    ring.append('cc', undefined, 8)
+    // 10 bytes retained > cap 8: the oldest whole chunk goes.
+    expect(ring.earliest).toBe(4)
+    expect(ring.retainedBytes).toBe(6)
+    expect(ring.readFrom(0)).toEqual({ chunks: [{ at: 4, text: 'bbbb' }, { at: 8, text: 'cc' }], next: 10, lossy: true })
+    expect(ring.readFrom(4).lossy).toBe(false)
+  })
+
+  it('keeps a lone oversized chunk as a UTF-8-safe tail marked gapBefore', () => {
+    const ring = new OutputRing()
+    // 3 two-byte code points = 6 bytes; a 3-byte cap must not start inside a code point.
+    ring.append('ééé', undefined, 3)
+    expect(ring.retainedBytes).toBe(2)
+    expect(ring.earliest).toBe(4)
+    expect(ring.readFrom(0)).toEqual({ chunks: [{ at: 4, text: 'é', gapBefore: true }], next: 6, lossy: true })
+  })
+
+  it('re-trims to a smaller cap on demand and leaves total untouched', () => {
+    const ring = new OutputRing()
+    ring.append('12345678', undefined, 64)
+    ring.append('abcdefgh', undefined, 64)
+    ring.trim(8)
+    expect(ring.total).toBe(16)
+    expect(ring.earliest).toBe(8)
+    expect(ring.readFrom(8)).toEqual({ chunks: [{ at: 8, text: 'abcdefgh' }], next: 16, lossy: false })
+    // Trimming an empty ring is a no-op that keeps earliest at total.
+    const empty = new OutputRing()
+    empty.trim(1)
+    expect(empty.earliest).toBe(0)
+  })
+})