deepseek-harness/apps/web/tests/subagent-conversation.e2e.ts

import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { fileURLToPath } from 'node:url'
import { join } from 'node:path'
import type { Browser, Page } from 'playwright'
import { chromium } from 'playwright'
import { afterAll, beforeAll, describe, expect, it, onTestFailed } from 'vitest'
import {
  SESSION_FORMAT_VERSION, SessionId as sessionId, type SessionEvent, type SessionId,
} from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-agent'
import { snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent'
import {
  acknowledgeReloadConnectionLoss, captureStableAria, compareOrRefreshGolden,
  launchWebScaffold, watchConsole,
  webSnapshotMode, type WebScaffold,
} from './scaffold.ts'
import { connectFreshWorkspace, saveFailureShot } from './support.ts'

const BASE_FIXTURE = fileURLToPath(new URL('./snapshots/live-interactions/session.jsonl', import.meta.url))
const AVAILABLE_CHILD_EXPECTED = fileURLToPath(new URL('./snapshots/subagent-conversation/ui.expected.md', import.meta.url))
const TREE_EXPECTED = fileURLToPath(new URL('./snapshots/subagent-conversation/tree.expected.md', import.meta.url))
const SIDEBAR_EXPECTED = fileURLToPath(new URL('./snapshots/subagent-conversation/sidebar.expected.md', import.meta.url))
const UNAVAILABLE_GRANDCHILD_EXPECTED = fileURLToPath(new URL('./snapshots/subagent-conversation/nested.expected.md', import.meta.url))
const FORK_EXPECTED = fileURLToPath(new URL('./snapshots/subagent-conversation/fork.expected.md', import.meta.url))
const MODE = webSnapshotMode()
const LABEL = 'event-sourcing researcher'
const ONE_SHOT_LABEL = 'event-sourcing reviewer'
const NESTED_LABEL = 'example editor'
const PARENT_PROMPT = 'Ask a research subagent to explain event sourcing.'
const INITIAL_PROMPT = 'Explain event sourcing in one sentence.'
const FOLLOWUP = 'Now give the same explanation to a human reader.'
const POST_FORK_FOLLOWUP = 'Continue the original conversation after the fork.'

function childFixture(source: string, fixtureId: string, withContinuation: boolean): string {
  const [header, ...eventLines] = source.trimEnd().split('\n')
  if (header === undefined) throw new Error('base replay fixture has no header')
  const childHeader = header
    .replace('"id":"{{sessionId}}"', `"id":"${fixtureId}"`)
    .replace(/"createdAt":\d+/, '"createdAt":1784998084442')
  if (!withContinuation) return [childHeader, ...eventLines, ''].join('\n')
  const continued = eventLines.map(line => line
    .replace(/"seq":(\d+)/g, (_match, seq: string) => `"seq":${String(Number(seq) + 100)}`)
    .replace(/"seq0":(\d+)/g, (_match, seq: string) => `"seq0":${String(Number(seq) + 100)}`)
    .replaceAll('"turn":1', '"turn":2'))
  return [childHeader, ...eventLines, ...continued, ''].join('\n')
}

async function waitForAgentToSettle(scaffold: WebScaffold, id: SessionId): Promise<void> {
  const deadline = Date.now() + 30_000
  while (scaffold.ctx.agents.get(id) !== undefined) {
    if (Date.now() >= deadline) throw new Error(`subagent ${id} did not settle`)
    await new Promise<void>(resolve => setTimeout(resolve, 10))
  }
}

describe('web e2e: persisted subagent conversation and human continuation', () => {
  let scaffold: WebScaffold
  let browser: Browser
  let page: Page
  let sidecarRoot: string
  let childId: SessionId
  let oneShotId: SessionId
  let grandchildId: SessionId
  let tripwire: ReturnType<typeof watchConsole>
  const apiCalls: string[] = []

  beforeAll(async () => {
    if (MODE === 'record') throw new Error('subagent conversation is a keyless assembled snapshot')
    const baseFixture = await readFile(BASE_FIXTURE, 'utf8')
    sidecarRoot = await mkdtemp(join(tmpdir(), 'dsh-web-subagent-'))
    const childFixturePath = join(sidecarRoot, 'child.jsonl')
    await writeFile(childFixturePath, childFixture(baseFixture, 'recorded-subagent', true))
    scaffold = await launchWebScaffold({
      replayFixture: BASE_FIXTURE,
      replayChildFixtures: [childFixturePath],
      paceMs: 25,
    })
    browser = await chromium.launch()
    page = await browser.newPage({ viewport: { width: 1680, height: 1000 } })
    page.on('request', (request) => {
      const path = new URL(request.url()).pathname
      if (path.startsWith('/api/')) apiCalls.push(path)
    })
    tripwire = watchConsole(page)
    await page.goto(scaffold.baseUrl, { waitUntil: 'load' })
    await page.waitForSelector('[class*="frame"]', { timeout: 30_000 })
    await connectFreshWorkspace(page, scaffold.workspaceCwd)

    const parent = scaffold.ctx.agents.roots()[0]
    if (parent === undefined) throw new Error('fresh workspace did not publish its parent Agent')
    const parentSettled = scaffold.whenTurnSettled()
    const parentInput = page.locator('textarea:enabled').first()
    await parentInput.fill(PARENT_PROMPT)
    await parentInput.press('Enter')
    expect(await parentSettled).toBe(parent.id)

    const started = await scaffold.ctx.subagents.startContinuable({
      provider: 'spawn',
      label: LABEL,
      signal: new AbortController().signal,
      request: {
        prompt: [{ type: 'text', text: INITIAL_PROMPT }],
        parent,
      },
    })
    childId = started.childId
    await waitForAgentToSettle(scaffold, childId)
    oneShotId = sessionId('recorded-one-shot')
    const oneShotAt = Date.now()
    await scaffold.ctx.sessionPersistence.create({
      version: SESSION_FORMAT_VERSION,
      id: oneShotId,
      createdAt: oneShotAt,
      cwd: scaffold.workspaceCwd,
      parentSession: parent.id,
      origin: 'subagent',
      delegationDepth: 1,
    })
    await scaffold.ctx.sessionPersistence.append(oneShotId, [
      {
        type: 'turn/start',
        seq: 0,
        time: oneShotAt,
        data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
      },
      {
        type: 'user/message',
        seq: 1,
        time: oneShotAt + 1,
        data: {
          content: [{ type: 'text', text: 'Review the event sourcing explanation.' }],
          source: { kind: 'user' },
        },
        surfaceOp: 'append',
      },
      {
        type: 'subagent/descriptor',
        seq: 2,
        time: oneShotAt + 2,
        data: snapshotSubagentDescriptor({
          mode: 'one-shot', provider: 'spawn', label: ONE_SHOT_LABEL,
        }),
      },
      {
        type: 'turn/end',
        seq: 3,
        time: oneShotAt + 3,
        data: { turn: 1, reason: { kind: 'completed' } },
      },
    ] as SessionEvent[])
    grandchildId = sessionId('recorded-grandchild')
    const authoredAt = Date.now()
    await scaffold.ctx.sessionPersistence.create({
      version: SESSION_FORMAT_VERSION,
      id: grandchildId,
      createdAt: authoredAt,
      cwd: scaffold.workspaceCwd,
      parentSession: childId,
      origin: 'subagent',
      delegationDepth: 2,
    })
    await scaffold.ctx.sessionPersistence.append(grandchildId, [
      {
        type: 'turn/start',
        seq: 0,
        time: authoredAt,
        data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
      },
      {
        type: 'user/message',
        seq: 1,
        time: authoredAt + 1,
        data: {
          content: [{ type: 'text', text: 'Give one concrete event sourcing example.' }],
          source: { kind: 'user' },
        },
        surfaceOp: 'append',
      },
      {
        type: 'subagent/descriptor',
        seq: 2,
        time: authoredAt + 2,
        data: snapshotSubagentDescriptor({
          mode: 'continuable', provider: 'spawn', label: NESTED_LABEL,
        }),
      },
      {
        type: 'turn/end',
        seq: 3,
        time: authoredAt + 3,
        data: { turn: 1, reason: { kind: 'completed' } },
      },
    ] as SessionEvent[])
    expect(scaffold.ctx.agents.get(childId)).toBeUndefined()
    expect(scaffold.ctx.agents.get(oneShotId)).toBeUndefined()
    expect(scaffold.ctx.agents.get(grandchildId)).toBeUndefined()
    await expect(scaffold.ctx.subagents.listChildren(parent.id)).resolves.toMatchObject([
      {
        kind: 'child', id: childId, mode: 'continuable', label: LABEL,
        activity: 'inactive', hasChildren: true,
      },
      {
        kind: 'child', id: oneShotId, mode: 'one-shot',
        label: ONE_SHOT_LABEL, activity: 'inactive', hasChildren: false,
      },
    ])
    await expect(scaffold.ctx.subagents.listChildren(childId)).resolves.toMatchObject([
      {
        kind: 'child', id: grandchildId, mode: 'continuable',
        label: NESTED_LABEL, activity: 'inactive', hasChildren: false,
      },
    ])
    // These two cold fixtures were authored after the page's initial
    // session.list and intentionally emitted no session-added frame. Reload
    // to exercise the restart baseline that discovers their full lineage.
    const warningStart = tripwire.warnings.length
    await page.reload({ waitUntil: 'load' })
    await page.waitForSelector('[class*="frame"]', { timeout: 30_000 })
    const catalogButton = page.getByRole('button', { name: /个子代理/ })
    await catalogButton.waitFor({ timeout: 15_000 })
    await catalogButton.click()
    const catalogTree = page.getByRole('tree', { name: '子代理会话' })
    await catalogTree.getByRole('treeitem').nth(1).waitFor({ timeout: 15_000 })
    await catalogTree.press('Escape')
    await page.getByRole('button', { name: '3 个子代理' }).waitFor({ timeout: 15_000 })
    acknowledgeReloadConnectionLoss(tripwire, warningStart)
  }, 120_000)

  afterAll(async () => {
    const failures: unknown[] = []
    await browser?.close().catch((error: unknown) => failures.push(error))
    await scaffold?.close().catch((error: unknown) => failures.push(error))
    if (sidecarRoot !== undefined) {
      await rm(sidecarRoot, { recursive: true, force: true })
        .catch((error: unknown) => failures.push(error))
    }
    if (failures.length === 1) throw failures[0]
    if (failures.length > 1) throw new AggregateError(failures, 'subagent Web teardown failed')
  })

  it('expands a persisted grandchild progressively without activating either level', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-tree'))
    await page.getByRole('button', { name: '3 个子代理' }).click()
    expect(await page.getByRole('button', {
      name: `展开 ${ONE_SHOT_LABEL} 的下级子代理`,
    }).count()).toBe(0)
    await page.getByRole('button', { name: `展开 ${LABEL} 的下级子代理` }).click()
    await page.getByRole('treeitem', { name: new RegExp(NESTED_LABEL) }).waitFor({ timeout: 15_000 })
    expect(scaffold.ctx.agents.get(childId)).toBeUndefined()
    expect(scaffold.ctx.agents.get(grandchildId)).toBeUndefined()
    const snapshot = await captureStableAria(
      page,
      '[role="tree"][aria-label="子代理会话"]',
      scaffold.workspaceCwd,
    )
    await compareOrRefreshGolden(TREE_EXPECTED, snapshot, MODE)
    await page.getByRole('tree', { name: '子代理会话' }).press('Escape')
  })

  it('opens the completed child from persistence without activating it', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-open'))
    await page.getByRole('button', { name: '3 个子代理' }).click()
    await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
    await expect.poll(
      () => page.getByText(INITIAL_PROMPT, { exact: true }).count(),
      { timeout: 15_000 },
    ).toBe(1)
    if (scaffold.ctx.agents.get(childId) !== undefined) {
      throw new Error(`viewing the child activated it; API calls: ${apiCalls.join(', ')}`)
    }
    const hierarchy = page.getByRole('navigation', { name: 'Session hierarchy' })
    await hierarchy.getByRole('button', { name: LABEL, disabled: true }).waitFor()
    const sidebar = await captureStableAria(
      page,
      '[role="tree"][aria-label="Sessions"]',
      scaffold.workspaceCwd,
    )
    await compareOrRefreshGolden(SIDEBAR_EXPECTED, sidebar, MODE)
  })

  it('continues through FIFO follow-up admission and receives the child mux events', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-followup'))
    const ended = new Promise<void>((resolveEnded, reject) => {
      const timer = setTimeout(() => {
        off()
        reject(new Error('subagent follow-up did not reach turn/end'))
      }, 30_000)
      const off = scaffold.ctx.on('session/event', (session: { id: SessionId }, event: SessionEvent) => {
        if (session.id !== childId || event.type !== 'turn/end') return
        clearTimeout(timer)
        off()
        resolveEnded()
      })
    })
    const input = page.getByRole('textbox', { name: 'Message the agent' })
    await input.fill(FOLLOWUP)
    await input.press('Enter')
    await expect.poll(
      () => scaffold.ctx.agents.get(childId)?.status,
      { timeout: 10_000 },
    ).toBe('running')
    const hierarchy = page.getByRole('navigation', { name: 'Session hierarchy' })
    await hierarchy.getByRole('button').first().click()
    const runningTrigger = page.getByRole('button', { name: '3 个子代理，正在运行' })
    await runningTrigger.waitFor({ timeout: 10_000 })
    expect(await runningTrigger.locator('[data-state="ongoing"]').count()).toBe(1)
    await runningTrigger.click()
    await page.getByRole('treeitem', {
      name: new RegExp(`${LABEL}.*正在运行`),
    }).waitFor({ timeout: 10_000 })
    await ended
    await page.getByRole('treeitem', {
      name: new RegExp(`${LABEL}.*当前未运行`),
    }).waitFor({ timeout: 10_000 })
    expect(await page.getByRole('button', { name: '3 个子代理' })
      .locator('[data-state="ongoing"]').count()).toBe(0)
    await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
    await expect.poll(() => page.getByText(FOLLOWUP, { exact: true }).count(), { timeout: 10_000 }).toBe(1)
    await expect.poll(() => scaffold.ctx.agents.get(childId), { timeout: 10_000 }).toBeUndefined()
    expect(await page.getByRole('button', { name: 'Stop generating' }).count()).toBe(0)
  })

  it('matches the settled addressed-conversation aria golden and stays clean', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-aria'))
    const snapshot = await captureStableAria(page, '[class*="centerCol"]', scaffold.workspaceCwd)
    await compareOrRefreshGolden(AVAILABLE_CHILD_EXPECTED, snapshot, MODE)
    expect(tripwire.pageErrors).toEqual([])
    expect(tripwire.warnings).toEqual([])
  })

  it('opens an unavailable persisted grandchild after recording the available child', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-grandchild'))
    await page.getByRole('button', { name: '1 个子代理' }).click()
    await page.getByRole('treeitem', { name: new RegExp(NESTED_LABEL) }).click()
    await page.getByText('父会话当前不在线，重新打开父会话后即可继续发送消息。').waitFor()
    const hierarchy = page.getByRole('navigation', { name: 'Session hierarchy' })
    const crumbs = await hierarchy.getByRole('button').allTextContents()
    expect(crumbs.slice(-2)).toEqual([LABEL, NESTED_LABEL])
    expect(scaffold.ctx.agents.get(childId)).toBeUndefined()
    expect(scaffold.ctx.agents.get(grandchildId)).toBeUndefined()
    await compareOrRefreshGolden(
      UNAVAILABLE_GRANDCHILD_EXPECTED,
      await captureStableAria(page, '[class*="centerCol"]', scaffold.workspaceCwd),
      MODE,
    )
  })

  it('opens a one-shot child as permanently read-only history', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-one-shot'))
    const parentSession = page.getByRole('tree', { name: 'Sessions' })
      .getByRole('treeitem')
      .last()
    await parentSession.click()
    await page.getByRole('button', { name: '3 个子代理' }).click()
    await page.getByRole('treeitem', { name: new RegExp(ONE_SHOT_LABEL) }).click()
    await page.getByText('一次性任务不支持后续消息，可在这里查看完整执行记录。').waitFor()
    expect(scaffold.ctx.agents.get(oneShotId)).toBeUndefined()
  })

  it('places an ordinary fork from a subagent beside its workspace-owning ancestor', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-fork'))
    await page.getByRole('tree', { name: 'Sessions' })
      .getByRole('treeitem', { name: /Ask a research subagent to/ })
      .click()
    await page.getByRole('button', { name: '3 个子代理' }).click()
    await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
    await page.getByRole('textbox', { name: 'Message the agent' }).waitFor()
    const forkResponse = page.waitForResponse(response =>
      new URL(response.url()).pathname === '/api/session.fork')
    await page.getByRole('button', { name: 'Branch into a new conversation' }).last().click()
    const forkReceipt = await (await forkResponse).json() as { result: { ok: boolean } }
    expect(forkReceipt.result).toMatchObject({ ok: true })
    await expect.poll(
      () => page.getByRole('tree', { name: 'Sessions' }).getByRole('treeitem').count(),
      { timeout: 15_000 },
    ).toBe(3)
    expect(await page.getByText('Ungrouped', { exact: true }).count()).toBe(0)
    const hierarchy = page.getByRole('navigation', { name: 'Session hierarchy' })
    expect(await hierarchy.getByRole('button').count()).toBe(1)
    await compareOrRefreshGolden(
      FORK_EXPECTED,
      await captureStableAria(page, '[role="tree"][aria-label="Sessions"]', scaffold.workspaceCwd),
      MODE,
    )
  })

  it('cold-resumes the original subagent while its ordinary fork stays active', async () => {
    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-post-fork-followup'))
    const sessions = page.getByRole('tree', { name: 'Sessions' })
    await sessions.getByRole('treeitem', { name: /Ask a research subagent to/ }).click()
    await page.getByRole('button', { name: '3 个子代理' }).click()
    await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
    await page.locator('textarea:enabled').first().waitFor()
    expect(scaffold.ctx.agents.get(childId)).toBeUndefined()

    const forkResponse = page.waitForResponse(response =>
      new URL(response.url()).pathname === '/api/session.fork')
    await page.getByRole('button', { name: 'Branch into a new conversation' }).last().click()
    const forkReceipt = await (await forkResponse).json() as {
      result: { ok: true; value: { sessionId: string } } | { ok: false }
    }
    expect(forkReceipt.result).toMatchObject({ ok: true })
    if (!forkReceipt.result.ok) return
    const forkId = sessionId(forkReceipt.result.value.sessionId)
    await expect.poll(() => scaffold.ctx.agents.get(forkId)).not.toBeUndefined()

    await sessions.getByRole('treeitem', { name: /Ask a research subagent to/ }).click()
    await page.getByRole('button', { name: '3 个子代理' }).click()
    await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
    const input = page.locator('textarea:enabled').first()
    await input.waitFor()
    const promptResponse = page.waitForResponse(response =>
      new URL(response.url()).pathname === '/api/subagent.prompt')
    await input.fill(POST_FORK_FOLLOWUP)
    await input.press('Enter')
    const promptReceipt = await (await promptResponse).json() as {
      result: { ok: true } | { ok: false; error: { code: string; message: string } }
    }
    if (!promptReceipt.result.ok) {
      throw new Error(`post-fork follow-up rejected: ${JSON.stringify(promptReceipt.result.error)}`)
    }
    await expect.poll(async () => {
      const loaded = await scaffold.ctx.sessionPersistence.load(childId)
      const messageIndex = loaded.events.findIndex(event => event.type === 'user/message'
        && event.data.content.some(block => block.type === 'text' && block.text === POST_FORK_FOLLOWUP))
      return messageIndex >= 0 && loaded.events.slice(messageIndex + 1).some(event => event.type === 'turn/end')
    }, { timeout: 30_000 }).toBe(true)
    expect(scaffold.ctx.agents.get(forkId)).not.toBeUndefined()
    await expect.poll(() => scaffold.ctx.agents.get(childId), { timeout: 10_000 }).toBeUndefined()
  })
})