deepseek-harness/scripts/python-snapshot-workflow-order.spec.ts

import { getEventListeners } from 'node:events'
import { Context } from '@deepseek-ai/cordis'
import { agentEvents, type Agent, type PreStepDecision } from '@deepseek-ai/dsh-agent'
import { createToolResultMessage, createUserMessage, ToolCallId } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import { WorkflowRunId } from '@deepseek-ai/dsh-workflow'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
import SubagentRuntime from '@deepseek-ai/dsh-subagent'
import * as spawn from '@deepseek-ai/dsh-subagent-spawn-in-process'
import { MockAdapter, textResponse } from '../packages/core/agent-loop/tests/mock-adapter.ts'
import type {} from '@deepseek-ai/dsh-tool-workflow/types'
import type {} from '@deepseek-ai/dsh-session-log-deepseek'
import { afterEach, describe, expect, it, vi } from 'vitest'
// @ts-expect-error Scenario plugins are runtime JavaScript without declaration artifacts.
import * as fixtureModule from './fixtures/python-snapshot-workflow-order.mjs'

const config = { parentSessionId: 'advanced-parent', prompt: 'workflow child prompt' }
const fixture = fixtureModule as unknown as {
  name: string
  apply(ctx: Context, config: { parentSessionId: string; prompt: string; direct?: { prompt: string; callId: string } }): void
}
const cleanups: (() => Promise<unknown>)[] = []
afterEach(async () => { for (const cleanup of cleanups.splice(0).reverse()) await cleanup() })

async function harness(direct = false) {
  const ctx = new Context()
  const store = ctx.plugin(SessionStore)
  await store
  cleanups.push(() => store.dispose())
  const fiber = ctx.plugin(fixture, { ...config, ...direct ? { direct: { prompt: config.prompt, callId: 'direct-child' } } : {} })
  await fiber
  cleanups.push(() => fiber.dispose())
  const parent = ctx.sessions.create(SessionId(config.parentSessionId))
  const other = ctx.sessions.create(SessionId('other-parent'))
  const session = ctx.sessions.create(SessionId('workflow-child'), { meta: { parentSession: parent.id } })
  // The dispatcher needs only the subject identity; the fixture reads its Session.
  const agent = { id: session.id, session } as Agent
  const controller = new AbortController()
  const messages = [createUserMessage({ content: [{ type: 'text', text: config.prompt }], source: { kind: 'user' } })]
  const decision: PreStepDecision = { kind: 'enter', messages }
  const next = vi.fn(async () => decision)
  const start = (owner = parent, childId = agent.id) => owner.append('tool-workflow/agent-start', {
    runId: WorkflowRunId('run'), seq: 1, label: 'workflow-child', childId,
  })
  const step = (overrides = {}) => agentEvents(ctx, agent).waterfall('agent/pre-step', {
    turn: 1, step: 1, messages, signal: controller.signal, ...overrides,
  }, next)
  return { ctx, fiber, parent, other, session, agent, controller, decision, next, start, step }
}

describe('advanced Python snapshot workflow ordering', () => {
  it('blocks a real activated child before its first model request', async () => {
    const ctx = new Context()
    const entered = Promise.withResolvers<Agent>()
    const order: string[] = []
    const adapter = new MockAdapter([textResponse('child complete')])
    const assembly = ctx.plugin({
      name: 'workflow-order-driver-test',
      async apply(inner: Context) {
        await mountAgentLoopTestDependencies(inner)
        await inner.plugin(AgentLoop, { agents: [] })
        await inner.plugin(SubagentRuntime)
        await inner.plugin(spawn, { providerName: 'spawn' })
        inner.on('agent/pre-step', ({ agent }, next) => {
          if (agent.session.header.parentSession === config.parentSessionId) entered.resolve(agent)
          return next()
        })
        inner.on('session/event', (_session, event) => {
          if (event.type === 'subagent/catalog' || event.type === 'tool-workflow/agent-start') order.push(event.type)
        })
        await inner.plugin(fixture, config)
      },
    })
    cleanups.push(() => assembly.dispose())
    await assembly
    ctx.llm.registerAdapter(['mock'], adapter)
    const parent = await ctx.agentLoop.create(SessionId(config.parentSessionId), { provider: 'mock', model: 'mock' })
    const run = await ctx.subagents.startActivation({
      provider: 'spawn', label: 'workflow-child', delivery: 'caller',
      request: { parent, prompt: [{ type: 'text', text: config.prompt }] }, signal: new AbortController().signal,
    })
    cleanups.push(() => run.dispose())
    const child = await entered.promise
    expect(child.id).toBe(run.childId)
    expect(adapter.requests).toHaveLength(0)
    expect(child.session.snapshotEvents().some(event => event.type === 'subagent/descriptor')).toBe(true)
    parent.session.append('tool-workflow/agent-start', {
      runId: WorkflowRunId('run'), seq: 1, label: 'workflow-child', childId: child.id,
    })
    expect((await run.result).output).toEqual([{ type: 'text', text: 'child complete' }])
    expect(adapter.requests).toHaveLength(1)
    expect(order).toEqual(['subagent/catalog', 'tool-workflow/agent-start'])
  })

  it('holds the child until the exact parent records the exact member', async () => {
    const h = await harness()
    const pending = h.step()
    expect(h.next).not.toHaveBeenCalled()
    expect(getEventListeners(h.controller.signal, 'abort')).toHaveLength(1)
    h.start(h.other)
    h.start(h.parent, SessionId('other-child'))
    h.parent.append('tool-workflow/run-start', { runId: WorkflowRunId('run'), name: 'workflow' })
    await Promise.resolve()
    expect(h.next).not.toHaveBeenCalled()
    h.start()
    expect(await pending).toBe(h.decision)
    expect(h.next).toHaveBeenCalledOnce()
    expect(getEventListeners(h.controller.signal, 'abort')).toHaveLength(0)
  })

  it('waits for the parent to consume the direct start and acknowledge its request log', async () => {
    const h = await harness(true)
    const pending = h.step()
    h.parent.append('turn/start', { turn: 1 })
    h.parent.append('step/start', { turn: 1, step: 1 })
    const callId = ToolCallId('direct-child')
    h.parent.append('tool/call', { turn: 1, step: 1, callId, name: 'subagent', arguments: '{}' })
    const result = h.parent.append('tool/result', {
      turn: 1, step: 1,
      message: createToolResultMessage({ callId, content: [{ type: 'text', text: 'started subagent child' }], isError: false }),
    }, { surfaceOp: 'append' })
    h.parent.append('step/start', { turn: 1, step: 2 })
    await Promise.resolve()
    expect(h.next).not.toHaveBeenCalled()
    h.parent.append('session-log-deepseek/delivery-accepted', {
      sessionId: h.parent.id, sessionFormatVersion: h.parent.header.version, throughSeq: result.seq,
    })
    expect(await pending).toBe(h.decision)
    expect(h.next).toHaveBeenCalledOnce()
  })

  it('retains a start recorded before the child reaches its first step', async () => {
    const h = await harness()
    h.start()
    expect(await h.step()).toBe(h.decision)
    expect(h.next).toHaveBeenCalledOnce()
    expect(getEventListeners(h.controller.signal, 'abort')).toHaveLength(0)
  })

  it.each(['prompt', 'parent', 'turn', 'step'])('does not hold an unrelated %s', async (difference) => {
    const h = await harness()
    const overrides = difference === 'prompt' ? { messages: [] }
      : difference === 'turn' ? { turn: 2 }
        : difference === 'step' ? { step: 2 } : {}
    if (difference === 'parent') {
      const session = h.ctx.sessions.create(SessionId('unrelated-child'), { meta: { parentSession: h.other.id } })
      Object.assign(h.agent, { session })
    }
    expect(await h.step(overrides)).toBe(h.decision)
    expect(h.next).toHaveBeenCalledOnce()
  })

  it.each([false, true])('rejects cancellation and detaches the waiter (already aborted: %s)', async (alreadyAborted) => {
    const h = await harness()
    const reason = new Error('cancelled child')
    if (alreadyAborted) h.controller.abort(reason)
    const pending = h.step()
    const rejected = expect(pending).rejects.toBe(reason)
    h.controller.abort(reason)
    await rejected
    h.start()
    expect(h.next).not.toHaveBeenCalled()
    expect(getEventListeners(h.controller.signal, 'abort')).toHaveLength(0)
  })

  it('does not admit a cancelled child when start and cancellation share a tick', async () => {
    const h = await harness()
    const reason = new Error('cancelled after membership')
    const pending = h.step()
    const rejected = expect(pending).rejects.toBe(reason)
    h.start()
    h.controller.abort(reason)
    await rejected
    expect(h.next).not.toHaveBeenCalled()
    expect(getEventListeners(h.controller.signal, 'abort')).toHaveLength(0)
  })

  it('settles pending waits before disposal completes and removes both listeners', async () => {
    const h = await harness()
    const pending = h.step()
    const rejected = expect(pending).rejects.toThrow('workflow snapshot barrier disposed')
    await h.fiber.dispose()
    await rejected
    h.start()
    expect(h.next).not.toHaveBeenCalled()
    expect(getEventListeners(h.controller.signal, 'abort')).toHaveLength(0)
    expect(await h.step()).toBe(h.decision)
  })
})