deepseek-harness/packages/workflow/workflow-vm/tests/session.spec.ts

import { describe, expect, it, vi } from 'vitest'
import { MessageChannel } from 'node:worker_threads'
import type { MessagePort } from 'node:worker_threads'
import { HostToWorkerType, WorkerToHostType } from '../src/protocol.ts'
import type { HostToWorkerMessage, WorkerToHostMessage } from '../src/protocol.ts'
import { requireParentPort, runWorkerSession } from '../src/session.ts'
import type { ChildResult, WorkerInit } from '../src/types.ts'

/** Default limits for in-process sessions (concurrency pinned; auto is machine-derived). */
function limits(overrides?: Partial<WorkerInit['limits']>): WorkerInit['limits'] {
  return { maxConcurrentAgents: 8, maxTotalAgents: 1000, maxItemsPerCall: 4096, syncTimeoutMs: 5000, ...overrides }
}

/** Wrap a body in the minimal valid meta header (the session receives it pre-extracted). */
function init(body: string, args?: unknown, limitOverrides?: Partial<WorkerInit['limits']>): WorkerInit {
  return {
    meta: { name: 'test-flow', description: 'a test workflow' },
    body,
    ...args !== undefined ? { args } : {},
    limits: limits(limitOverrides),
  }
}

/** One scripted host over the other end of a MessageChannel. */
interface FakeHost {
  port: MessagePort
  messages: WorkerToHostMessage[]
  /** Messages of one type, as they arrive. */
  ofType<T extends WorkerToHostMessage['type']>(type: T): Extract<WorkerToHostMessage, { type: T }>[]
  send(message: HostToWorkerMessage): void
  /** Resolves with the terminal result message. */
  result(): Promise<Extract<WorkerToHostMessage, { type: 'result' }>['result']>
  close(): void
}

interface FakeHostOptions {
  /** Auto-respond to child-start: reply started + settled per child index. Omit a reply to leave the child pending. */
  reply?: (request: { prompt: string; schema?: unknown; model?: string }, index: number) => ChildResult | undefined
  /** Reject the start instead (child-start-error) when returning a string. */
  refuse?: (index: number) => string | undefined
  /** Auto-send `go` on `ready` (default true). */
  go?: boolean
  /** Manual mode: do NOT auto-answer child-start at all (the test scripts the replies). */
  manual?: boolean
}

/**
 * Drive runWorkerSession IN-PROCESS over a MessageChannel: this is where the
 * worker-side files earn their coverage — code inside a real Worker is
 * invisible to main-process coverage. The fake host mirrors the real host's
 * protocol discipline (one started/start-error per start; settled/disposed
 * follow).
 */
function fakeHost(options?: FakeHostOptions): FakeHost {
  const channel = new MessageChannel()
  const messages: WorkerToHostMessage[] = []
  const resultGate = Promise.withResolvers<Extract<WorkerToHostMessage, { type: 'result' }>['result']>()
  let childIndex = 0
  channel.port1.on('message', (message: WorkerToHostMessage) => {
    messages.push(message)
    switch (message.type) {
      case WorkerToHostType.Ready:
        if (options?.go !== false) channel.port1.postMessage({ type: HostToWorkerType.Go } satisfies HostToWorkerMessage)
        break
      case WorkerToHostType.ChildStart: {
        if (options?.manual) break
        const index = childIndex
        childIndex += 1
        const refusal = options?.refuse?.(index)
        if (refusal !== undefined) {
          channel.port1.postMessage(
            { type: HostToWorkerType.ChildStartError, callId: message.callId, rendered: refusal } satisfies HostToWorkerMessage,
          )
          break
        }
        channel.port1.postMessage({ type: HostToWorkerType.ChildStarted, callId: message.callId, childId: `child-${index}` } satisfies HostToWorkerMessage)
        const reply = options?.reply?.(message.request, index)
        if (reply !== undefined) {
          channel.port1.postMessage(
            { type: HostToWorkerType.ChildSettled, callId: message.callId, result: reply } satisfies HostToWorkerMessage,
          )
        }
        break
      }
      case WorkerToHostType.ChildDispose:
        channel.port1.postMessage({ type: HostToWorkerType.ChildDisposed, callId: message.callId } satisfies HostToWorkerMessage)
        break
      case WorkerToHostType.Result:
        resultGate.resolve(message.result)
        break
      default:
        break
    }
  })
  return {
    port: channel.port2,
    messages,
    ofType: type => messages.filter((message): message is never => message.type === type),
    send: (message) => { channel.port1.postMessage(message) },
    result: () => resultGate.promise,
    close: () => { channel.port1.close() },
  }
}

/** A completed text child result. */
function text(reply: string): ChildResult {
  return { output: [{ type: 'text', text: reply }], stopReason: 'completed' }
}

describe('runWorkerSession over an in-process MessageChannel', () => {
  it('runs a script end to end: ready/go handshake, phases, log, agents, result', async () => {
    const host = fakeHost({ reply: (_request, index) => text(`answer-${index}`) })
    const session = runWorkerSession(host.port, init(`
      phase('Scan')
      log('starting with ' + args.files.length + ' files')
      const answers = await pipeline(args.files, (prev, item) => agent('read ' + item))
      return { answers }
    `, { files: ['a.ts', 'b.ts'] }))
    const result = await host.result()
    await session
    expect(result.stopReason).toBe('completed')
    expect(result.agentsStarted).toBe(2)
    expect(result.value).toEqual({ answers: ['answer-0', 'answer-1'] })
    expect(host.messages[0]!.type).toBe('ready')
    expect(host.ofType(WorkerToHostType.Phase).map(m => m.title)).toEqual(['Scan'])
    expect(host.ofType(WorkerToHostType.Log).map(m => m.message)).toEqual(['starting with 2 files'])
    expect(host.ofType(WorkerToHostType.AgentStart).map(m => m.info.childId)).toEqual(['child-0', 'child-1'])
    expect(host.ofType(WorkerToHostType.AgentEnd).every(m => m.info.outcome === 'completed')).toBe(true)
    host.close()
  })

  it('agent({schema}) forwards the schema on the start request and returns the structured value', async () => {
    const host = fakeHost({ reply: () => ({ output: [], structured: { files: ['x.ts'] }, stopReason: 'completed' }) })
    void runWorkerSession(host.port, init(`
      const found = await agent('list files', { schema: { type: 'object', properties: { files: { type: 'array', items: { type: 'string' } } } }, model: 'deepseek-v4-pro' })
      return { first: found.files[0] }
    `))
    const result = await host.result()
    expect(result.value).toEqual({ first: 'x.ts' })
    const start = host.ofType(WorkerToHostType.ChildStart)[0]!
    expect(start.request.schema).toEqual({ type: 'object', properties: { files: { type: 'array', items: { type: 'string' } } } })
    expect(start.request.model).toBe('deepseek-v4-pro')
    host.close()
  })

  it('a schema child completing WITHOUT a structured value resolves null with a failed outcome', async () => {
    const host = fakeHost({ reply: () => text('prose, no structure') })
    void runWorkerSession(host.port, init("return await agent('p', { schema: { type: 'object' } })"))
    const result = await host.result()
    expect(result.value).toBeNull()
    expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('failed')
    host.close()
  })

  it('a child settling non-completed resolves null (scripts filter), never throwing into the script', async () => {
    const host = fakeHost({ reply: (_request, index) => index === 0 ? { output: [], stopReason: 'error' } : text('ok') })
    void runWorkerSession(host.port, init("return await parallel([() => agent('one'), () => agent('two')])"))
    const result = await host.result()
    expect(result.value).toEqual([null, 'ok'])
    expect(host.ofType(WorkerToHostType.AgentEnd).map(m => m.info.outcome)).toEqual(expect.arrayContaining(['failed', 'completed']))
    host.close()
  })

  it('a start refusal (child-start-error) is a fatal AGENT_START that kills the script through a combinator', async () => {
    const host = fakeHost({ refuse: () => 'no provider here' })
    void runWorkerSession(host.port, init("return await pipeline([1], () => agent('p'))"))
    const result = await host.result()
    expect(result.stopReason).toBe('error')
    expect(result.error).toContain('agent() could not start a child')
    expect(result.error).toContain('no provider here')
    host.close()
  })

  it('a child-failed message (infrastructure rejection) is fatal AGENT_RESULT with the paired failed outcome', async () => {
    const host = fakeHost()
    void runWorkerSession(host.port, init(`
      try { await agent('p'); return 'unreachable' } catch (e) { return { name: e.name, code: e.code, fatal: e.fatal } }
    `))
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
    const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
    host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
    host.send({ type: HostToWorkerType.ChildFailed, callId, rendered: 'backend exploded' })
    const result = await host.result()
    expect(result.value).toMatchObject({ name: 'WorkflowError', code: 'AGENT_RESULT', fatal: true })
    expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('failed')
    host.close()
  })

  it('cancel before go: the body never runs at all and the result is cancelled (a second cancel is a no-op)', async () => {
    const host = fakeHost({ go: false })
    const session = runWorkerSession(host.port, init("log('ran')\nreturn 123"))
    await vi.waitFor(() => { expect(host.messages.some(m => m.type === WorkerToHostType.Ready)).toBe(true) })
    host.send({ type: HostToWorkerType.Cancel, reason: 'aborted before start' })
    // Idempotence: the first reason wins; a duplicate cancel changes nothing.
    host.send({ type: HostToWorkerType.Cancel, reason: 'a later reason that must lose' })
    const result = await host.result()
    await session
    expect(result.stopReason).toBe('cancelled')
    expect(result.error).toContain('aborted before start')
    expect(result.error).not.toContain('must lose')
    expect(result.value).toBeNull()
    expect(host.ofType(WorkerToHostType.Log)).toEqual([])
    host.close()
  })

  it('a script with no return value resolves value: null', async () => {
    const host = fakeHost({ reply: () => text('ok') })
    void runWorkerSession(host.port, init("await agent('p')"))
    const result = await host.result()
    expect(result.stopReason).toBe('completed')
    expect(result.value).toBeNull()
    host.close()
  })

  it('cancel mid-run: in-flight children get cancel RPCs, hooks throw at entry, the run reports cancelled', async () => {
    const host = fakeHost()
    void runWorkerSession(host.port, init(`
      phase('before')
      try { await agent('x') } catch (e) {}
      try { phase('after') } catch (e) {}
      try { log('after') } catch (e) {}
      try { await parallel([() => 'ran']) } catch (e) {}
      try { await pipeline(['item'], p => p) } catch (e) {}
      return 'survived by catching'
    `))
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
    const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
    host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
    host.send({ type: HostToWorkerType.Cancel, reason: 'stop everything' })
    // The real host settles the aborted child; mirror it.
    host.send({ type: HostToWorkerType.ChildSettled, callId, result: { output: [], stopReason: 'aborted' } })
    const result = await host.result()
    expect(result.stopReason).toBe('cancelled')
    expect(result.error).toContain('stop everything')
    expect(host.ofType(WorkerToHostType.ChildCancel).map(m => m.callId)).toContain(callId)
    expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('cancelled')
    // No post-cancel narration left the runtime (the hooks threw at entry).
    expect(host.ofType(WorkerToHostType.Phase).map(m => m.title)).toEqual(['before'])
    expect(host.ofType(WorkerToHostType.Log)).toEqual([])
    host.close()
  })

  it('cancellation between a queued waiter and its slot: the waiter rejects without a child-start', async () => {
    const host = fakeHost({ go: true })
    void runWorkerSession(host.port, init(
      "return await parallel([() => agent('a'), () => agent('b')])",
      undefined,
      { maxConcurrentAgents: 1 },
    ))
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
    host.send({ type: HostToWorkerType.Cancel, reason: 'raced' })
    const result = await host.result()
    expect(result.stopReason).toBe('cancelled')
    // Only the first agent ever reached the host.
    expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1)
    host.close()
  })

  it('a stray (never-awaited) agent is reaped after settlement: cancel + dispose RPCs flow, no unhandled rejection', async () => {
    const unhandled: unknown[] = []
    const onUnhandled = (reason: unknown): void => { unhandled.push(reason) }
    process.on('unhandledRejection', onUnhandled)
    try {
      const host = fakeHost()
      void runWorkerSession(host.port, init(`
        agent('stray, never awaited')
        return 'done without awaiting'
      `))
      const result = await host.result()
      expect(result.stopReason).toBe('completed')
      await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
      const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
      host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
      host.send({ type: HostToWorkerType.ChildSettled, callId, result: { output: [], stopReason: 'aborted' } })
      await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildDispose).map(m => m.callId)).toContain(callId) })
      await new Promise(resolve => setTimeout(resolve, 20))
      expect(unhandled).toEqual([])
      host.close()
    } finally {
      process.off('unhandledRejection', onUnhandled)
    }
  })

  it('an unparseable body settles an error result instead of dying without one (host pre-parse skew guard)', async () => {
    const host = fakeHost()
    await runWorkerSession(host.port, init('return ((('))
    const result = await host.result()
    expect(result.stopReason).toBe('error')
    expect(result.error).toContain('does not parse')
    expect(result.agentsStarted).toBe(0)
    host.close()
  })

  it('a synchronous spin in the initial slice dies by the in-worker vm timeout', async () => {
    const host = fakeHost()
    void runWorkerSession(host.port, init('while (true) {}', undefined, { syncTimeoutMs: 50 }))
    const result = await host.result()
    expect(result.stopReason).toBe('error')
    expect(result.error?.toLowerCase()).toContain('timed out')
    host.close()
  })

  it('a non-JSON return value fails loud as RESULT_UNSERIALIZABLE', async () => {
    const host = fakeHost()
    void runWorkerSession(host.port, init('return { when: new Date(0) }'))
    const result = await host.result()
    expect(result.stopReason).toBe('error')
    expect(result.error).toContain('not plain JSON data')
    host.close()
  })

  it('tolerates replies for unknown callIds (a teardown race): nothing crashes, the run completes', async () => {
    const host = fakeHost({ reply: () => text('fine') })
    void runWorkerSession(host.port, init("return await agent('p')"))
    host.send({ type: HostToWorkerType.ChildStarted, callId: 999, childId: 'ghost' })
    host.send({ type: HostToWorkerType.ChildStartError, callId: 999, rendered: 'ghost' })
    host.send({ type: HostToWorkerType.ChildSettled, callId: 999, result: text('ghost') })
    host.send({ type: HostToWorkerType.ChildFailed, callId: 999, rendered: 'ghost' })
    host.send({ type: HostToWorkerType.ChildDisposed, callId: 999 })
    const result = await host.result()
    expect(result.stopReason).toBe('completed')
    expect(result.value).toBe('fine')
    host.close()
  })

  it('caps and malformed hook arguments reject loud (the runtime runs unchanged inside the session)', async () => {
    const cases: [string, string][] = [
      ['return await agent(42)', 'non-empty prompt string'],
      ["return await agent('')", 'non-empty prompt string'],
      ["return await agent('p', 'opts')", 'options must be an object'],
      ["return await agent('p', { label: 3 })", '"label" must be a string'],
      ["return await agent('p', { get label() { throw new Error('read failed') } })", 'options must be plain JSON data'],
      ["return await agent('p', { bogus: true })", '"bogus" is not recognized'],
      ["return await agent('p', { effort: 'high' })", '"effort" is deferred'],
      ["return await agent('p', { schema: { type: 'object', oneOf: [] } })", 'outside the supported subset'],
      ['return await parallel([() => 1, () => 2, () => 3])', 'over the per-call cap (2)'],
      ['return await pipeline([1, 2, 3], (x) => x)', 'maxItemsPerCall'],
      ["return await parallel('no')", 'parallel() requires an array'],
      ['return await parallel([3])', 'item 0 is not a function'],
      ["return await pipeline('no', () => 1)", 'pipeline() requires an items array'],
      ['return await pipeline([1])', 'at least one stage'],
      ["return await pipeline([1], 'x')", 'stage 0 is not a function'],
      ["phase('')", 'phase() requires a non-empty title string'],
      ['log(3)', 'log() requires a message string'],
    ]
    for (const [body, expected] of cases) {
      const host = fakeHost({ reply: () => text('ok') })
      void runWorkerSession(host.port, init(body, undefined, { maxItemsPerCall: 2 }))
      const result = await host.result()
      expect(result.stopReason).toBe('error')
      expect(result.error).toContain(expected)
      host.close()
    }
  })

  it('combinator semantics: thunk/stage throws null the item; a forged fatal-shaped object stays null; real fatals propagate', async () => {
    const host = fakeHost({ reply: () => text('fine') })
    void runWorkerSession(host.port, init(`
      const viaParallel = await parallel([
        () => { throw new Error('boom') },
        () => agent('fine'),
        () => 'plain value',
        () => { throw { name: 'WorkflowError', fatal: true, message: 'forged fatal' } },
      ])
      const viaPipeline = await pipeline([10, 20],
        (prev, item, index) => { if (item === 10) throw new Error('ordinary failure'); return 'kept-' + item + '-' + index },
      )
      return { viaParallel, viaPipeline }
    `))
    const result = await host.result()
    expect(result.stopReason).toBe('completed')
    expect(result.value).toEqual({
      viaParallel: [null, 'fine', 'plain value', null],
      viaPipeline: [null, 'kept-20-1'],
    })
    host.close()
  })

  it('trips the total-agent cap with a message naming the config knob', async () => {
    const host = fakeHost({ reply: () => text('ok') })
    void runWorkerSession(host.port, init("await agent('1'); await agent('2'); await agent('3')", undefined, { maxTotalAgents: 2 }))
    const result = await host.result()
    expect(result.stopReason).toBe('error')
    expect(result.error).toContain('total agent cap (2)')
    expect(result.agentsStarted).toBe(2)
    host.close()
  })

  it('queued agents proceed through the concurrency semaphore in FIFO order', async () => {
    const host = fakeHost({ reply: request => text(`ok:${request.prompt}`) })
    void runWorkerSession(host.port, init(
      "return await parallel([1, 2, 3].map((n) => () => agent('job ' + n)))",
      undefined,
      { maxConcurrentAgents: 1 },
    ))
    const result = await host.result()
    expect(result.value).toEqual(['ok:job 1', 'ok:job 2', 'ok:job 3'])
    host.close()
  })

  it('labels default from the prompt first line, truncated; explicit label/phase options win', async () => {
    const host = fakeHost({ reply: () => text('ok') })
    void runWorkerSession(host.port, init(`
      phase('Find')
      await agent('a prompt that is quite long and will surely get truncated down to a display label\\n'
        + 'with a second line the label must not include')
      await agent('short', { label: 'named', phase: 'Custom' })
      return null
    `))
    await host.result()
    const starts = host.ofType(WorkerToHostType.AgentStart).map(m => m.info)
    expect(starts[0]).toMatchObject({ seq: 1, phase: 'Find' })
    expect(starts[0]!.label.length).toBeLessThanOrEqual(48)
    expect(starts[0]!.label).not.toContain('second line')
    expect(starts[1]).toMatchObject({ seq: 2, label: 'named', phase: 'Custom' })
    host.close()
  })

  it('non-text output blocks are filtered out of the text result', async () => {
    const host = fakeHost({
      reply: () => ({
        output: [
          { type: 'text', text: 'first ' },
          { type: 'tool_call', id: 'c1', name: 'x', arguments: {} } as never,
          { type: 'text', text: 'second' },
        ],
        stopReason: 'completed',
      }),
    })
    void runWorkerSession(host.port, init("return await agent('p')"))
    const result = await host.result()
    expect(result.value).toBe('first second')
    host.close()
  })

  it('a cancel landing DURING the start round-trip winds the fresh child down (cancel + dispose) and dies cancelled', async () => {
    const host = fakeHost({ manual: true })
    void runWorkerSession(host.port, init("return await agent('p')"))
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
    const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
    // Cancel FIRST, then the (stale) started reply: the worker processes them
    // in order, so the agent() continuation resumes already-cancelled — the
    // window the real host cannot produce (it refuses starts once cancelled)
    // but a teardown race can.
    host.send({ type: HostToWorkerType.Cancel, reason: 'raced the start' })
    host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
    const result = await host.result()
    expect(result.stopReason).toBe('cancelled')
    await vi.waitFor(() => {
      expect(host.ofType(WorkerToHostType.ChildCancel).map(m => m.callId)).toContain(callId)
      expect(host.ofType(WorkerToHostType.ChildDispose).map(m => m.callId)).toContain(callId)
    })
    // The child never became an agent-start: it was wound down pre-lifecycle.
    expect(host.ofType(WorkerToHostType.AgentStart)).toEqual([])
    host.close()
  })

  it('a start refusal arriving after a cancel reads as the cancellation, not a broken seam', async () => {
    const host = fakeHost({ manual: true })
    void runWorkerSession(host.port, init(`
      try { await agent('p'); return 'unreachable' } catch (e) { return { code: e.code } }
    `))
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
    const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
    host.send({ type: HostToWorkerType.Cancel, reason: 'stopping' })
    host.send({ type: HostToWorkerType.ChildStartError, callId, rendered: 'workflow run cancelled: stopping' })
    const result = await host.result()
    // The run reports cancelled (the script died of CANCELLED, not AGENT_START).
    expect(result.stopReason).toBe('cancelled')
    host.close()
  })

  it('a child result rejection while cancelled pairs a cancelled agent-end, and the run reports cancelled', async () => {
    const host = fakeHost({ manual: true })
    void runWorkerSession(host.port, init("return await agent('doomed')"))
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.ChildStart).length).toBe(1) })
    const callId = host.ofType(WorkerToHostType.ChildStart)[0]!.callId
    host.send({ type: HostToWorkerType.ChildStarted, callId, childId: 'child-0' })
    await vi.waitFor(() => { expect(host.ofType(WorkerToHostType.AgentStart).length).toBe(1) })
    host.send({ type: HostToWorkerType.Cancel, reason: 'user aborted' })
    host.send({ type: HostToWorkerType.ChildFailed, callId, rendered: 'backend crashed on abort' })
    const result = await host.result()
    expect(result.stopReason).toBe('cancelled')
    expect(host.ofType(WorkerToHostType.AgentEnd)[0]!.info.outcome).toBe('cancelled')
    host.close()
  })

})

describe('the worker bootstrap', () => {
  it('requireParentPort narrows a real port and throws on the main thread', () => {
    const channel = new MessageChannel()
    expect(requireParentPort(channel.port1)).toBe(channel.port1)
    channel.port1.close()
    expect(() => requireParentPort(null)).toThrow(/inside a worker thread/)
  })

  it('the entry module itself throws when loaded on the main thread (no parentPort)', async () => {
    // This import EXECUTES ../src/worker.ts on the main thread, which is what
    // covers the bootstrap file: requireParentPort throws before
    // runWorkerSession is reached.
    await expect(import('../src/worker.ts')).rejects.toThrow(/inside a worker thread/)
  })
})