fix(subagent): complete product provider lifecycle

2026-08-04 22:18:35 +08:00
parent d780902679
commit 32f829c4e6
21 changed files with 216 additions and 301 deletions
--- a/packages/subagent/subagent-claude-code/README.i18n.yaml
+++ b/packages/subagent/subagent-claude-code/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/subagent/subagent-claude-code/README.md
-README.md: facaae300eeb8907076182a129aa216863dec8ec
-README.zh.md: 75627cb54032edde07ec1ba5e21a058768515e29
+README.md: e19b119e355953a388ec4fca6a2db511e7eb43da
+README.zh.md: 6d3b8719329691681582abd91b82abd167e89f6d
--- a/packages/subagent/subagent-claude-code/README.md
+++ b/packages/subagent/subagent-claude-code/README.md
@@ -27,7 +27,7 @@ The provider advertises no optional start-time capabilities and reports `inherit
 | Key | Default | Meaning |
 |---|---|---|
 | `env` | `{}` | Explicit SDK/CLI environment layered over the shared credential-scrubbed parent environment. |
-| `disposeGraceMs` | `3000` | Positive finite process-tree termination grace in milliseconds; the final exit proof is bounded at twice this value. |
+| `disposeGraceMs` | `3000` | Positive finite grace in milliseconds between the shared process-tree owner's termination tiers; disposal then waits for whole-tree exit. |

 Production uses the Claude Code CLI supplied by `@anthropic-ai/claude-agent-sdk` and the host's native settings and authentication. The plugin does not install another CLI, select a model, create a product home, log in, or probe an account. Credential-shaped ambient variables are removed before the explicit `env` overlay is applied, so an API key or endpoint intended for the child must be supplied there; ordinary ambient values such as `PATH` and `HOME` remain available unless overridden.

--- a/packages/subagent/subagent-claude-code/README.zh.md
+++ b/packages/subagent/subagent-claude-code/README.zh.md
@@ -27,7 +27,7 @@ SDK 接收由文本块原样拼接成的任务。提供方会完整迭代 SDK
 | 配置键 | 默认值 | 含义 |
 |---|---|---|
 | `env` | `{}` | 显式指定的 SDK/CLI 环境，叠加在由共享机制清除凭证后的父环境之上。 |
-| `disposeGraceMs` | `3000` | 进程树终止宽限期，须为正有限值，单位为毫秒；最终退出确认的等待时间上限为该值的两倍。 |
+| `disposeGraceMs` | `3000` | 共享进程树责任方各终止层级之间的宽限期，单位为毫秒且须为正有限值；随后资源释放会等待整棵进程树退出。 |

 生产环境使用 `@anthropic-ai/claude-agent-sdk` 提供的 Claude Code CLI，以及宿主机原生设置与身份验证。本插件不安装另一份 CLI、不选择模型、不创建产品主目录、不执行登录，也不探测账户。具有凭证特征的环境变量会在显式 `env` 覆盖生效前被清除，因此供子进程使用的 API 密钥或端点必须在该配置中显式提供；除非被覆盖，`PATH` 和 `HOME` 等普通环境变量仍然可用。

--- a/packages/subagent/subagent-claude-code/src/run.ts
+++ b/packages/subagent/subagent-claude-code/src/run.ts
@@ -18,9 +18,9 @@ import {
 import type { ContentBlock } from '@deepseek-ai/dsh-llm'
 import { SessionId } from '@deepseek-ai/dsh-session'
 import {
-  doubledGraceWindow,
  settleRunResult,
  subprocessRunHandle,
+  thrownError,
  type SubagentResult,
  type SubagentRun,
  type SubagentStartRequest,
@@ -39,32 +39,20 @@ import {
 /** Default POSIX grace between subprocess termination tiers. */
 export const DEFAULT_DISPOSE_GRACE_MS = 3_000

-type QueryFactory = (params: {
-  prompt: string
-  options: Options
-}) => Query
-
 /** Fully resolved inputs for one official Claude Agent SDK query. */
 export interface ClaudeCodeRunSpec {
  /** Parent Session workspace supplied to the SDK and real CLI. */
  readonly cwd: string
  /** Explicit deployment/test environment layered after shared scrubbing. */
  readonly env: Record<string, string>
-  /** Subprocess termination grace and final tree-exit bound. */
+  /** Subprocess termination grace passed to the shared process-tree owner. */
  readonly disposeGraceMs: number
  /** Shared subprocess service spawn operation. */
  readonly spawn: (spec: SubprocessSpawnSpec) => SubprocessHandle
-  /** Official query entrypoint; replaced only by package-local unit tests. */
-  readonly query?: QueryFactory
  /** Diagnostic sink for a post-publication error flattened into a result. */
  readonly onError?: (error: Error, stopReason: SubagentStopReason) => void
 }

-function thrown(value: unknown): Error {
-  /* v8 ignore next -- SDK and subprocess failures reject with Error. */
-  return value instanceof Error ? value : new Error(String(value))
-}
-
 /**
 * Validate and preserve the one-shot task before crossing the SDK boundary.
 * @param prompt - task content accepted from the shared subagent service.
@@ -110,18 +98,15 @@ export function successfulResult(message: SDKResultMessage): string {
 * Consume the complete SDK stream and require one strict success plus normal
 * iterator completion.
 * @param query - published official SDK query.
- * @param setOutput - captures the candidate result for error diagnostics.
 * @returns the completed shared result.
 */
 export async function consumeClaudeQuery(
  query: AsyncIterable<SDKMessage>,
-  setOutput: (output: ContentBlock[]) => void,
 ): Promise<SubagentResult> {
  let answer: string | undefined
  for await (const message of query) {
    if (message.type !== 'result') continue
    answer = successfulResult(message)
-    setOutput([{ type: 'text', text: answer }])
  }
  if (answer === undefined) {
    throw new Error('subagent-claude-code: Claude Code ended without a result')
@@ -137,48 +122,30 @@ export async function consumeClaudeQuery(
 * the subprocess owner to prove it is gone.
 * @param query - official SDK query, when creation reached that point.
 * @param child - shared-service handle that owns the CLI process tree.
- * @param graceMs - termination grace used to bound final exit observation.
 */
 export async function disposeClaudeCodeChild(
  query: Pick<Query, 'close'> | undefined,
  child: SubprocessHandle,
-  graceMs: number,
 ): Promise<void> {
  const failures: Error[] = []
-  let treeExited = child.pid <= 0
  try {
    query?.close()
  } catch (error: unknown) {
-    failures.push(thrown(error))
+    failures.push(thrownError(error))
  }

  if (child.pid > 0) {
    child.terminate()
-    const exitWindow = doubledGraceWindow(graceMs)
    try {
-      treeExited = await child.waitForExit(exitWindow.signal)
-      if (!treeExited) {
-        failures.push(new Error(
-          'subagent-claude-code: Claude Code process tree did not exit within its dispose window',
-        ))
-      }
+      await child.waitForExit()
    } catch (error: unknown) {
-      failures.push(thrown(error))
-    } finally {
-      exitWindow.cancel()
+      failures.push(thrownError(error))
    }
  }
-  if (treeExited) {
-    try {
-      await child.done
-    } catch (error: unknown) {
-      failures.push(thrown(error))
-    }
-  } else {
-    // The bounded tree observation owns teardown completion. Keep a later
-    // direct-child spawn failure observed without turning that bound into an
-    // unbounded wait.
-    void child.done.catch(() => {})
+  try {
+    await child.done
+  } catch (error: unknown) {
+    failures.push(thrownError(error))
  }

  const firstFailure = failures[0]
@@ -244,7 +211,7 @@ export async function startClaudeCodeRun(
  let child: SubprocessHandle | undefined
  let query: Query | undefined
  try {
-    query = (spec.query ?? officialQuery)({
+    query = officialQuery({
      prompt,
      options: claudeQueryOptions(spec, controller, (captured) => {
        child = captured
@@ -264,10 +231,10 @@ export async function startClaudeCodeRun(
    requestCancel()
    if (child !== undefined) {
      try {
-        await disposeClaudeCodeChild(query, child, spec.disposeGraceMs)
+        await disposeClaudeCodeChild(query, child)
      } catch (disposeError: unknown) {
        throw new AggregateError(
-          [thrown(error), thrown(disposeError)],
+          [thrownError(error), thrownError(disposeError)],
          'subagent-claude-code: startup failed and CLI cleanup also failed',
        )
      }
@@ -276,7 +243,7 @@ export async function startClaudeCodeRun(
        query.close()
      } catch (disposeError: unknown) {
        throw new AggregateError(
-          [thrown(error), thrown(disposeError)],
+          [thrownError(error), thrownError(disposeError)],
          'subagent-claude-code: startup failed and query cleanup also failed',
        )
      }
@@ -285,17 +252,14 @@ export async function startClaudeCodeRun(
    if (cancelledBeforeCleanup || request.signal.aborted) {
      throw new Error('subagent-claude-code: request was aborted before SDK startup')
    }
-    throw thrown(error)
+    throw thrownError(error)
  }

-  let output: ContentBlock[] = []
  const publishedQuery = query
  const publishedChild = child
  const result = settleRunResult({
-    attempt: () => consumeClaudeQuery(publishedQuery, (value) => {
-      output = value
-    }),
-    collectOutput: () => output,
+    attempt: () => consumeClaudeQuery(publishedQuery),
+    collectOutput: () => [],
    cancelled: () => controller.signal.aborted,
    onError: spec.onError,
    signal: request.signal,
@@ -311,7 +275,6 @@ export async function startClaudeCodeRun(
    teardown: () => disposeClaudeCodeChild(
      publishedQuery,
      publishedChild,
-      spec.disposeGraceMs,
    ),
  })
 }
--- a/packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts
+++ b/packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts
@@ -1,5 +1,6 @@
 import { PassThrough } from 'node:stream'
 import type {
+  Options,
  Query,
  SDKMessage,
  SDKResultMessage,
@@ -7,7 +8,15 @@ import type {
 } from '@anthropic-ai/claude-agent-sdk'
 import { Context } from 'cordis'
 import Loader from '@cordisjs/plugin-loader'
-import { afterEach, describe, expect, it, type Mock, vi } from 'vitest'
+import {
+  afterEach,
+  beforeEach,
+  describe,
+  expect,
+  it,
+  type Mock,
+  vi,
+} from 'vitest'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
 import type { ContentBlock } from '@deepseek-ai/dsh-llm'
@@ -35,6 +44,18 @@ import {
  type ClaudeCodeRunSpec,
 } from '../src/run.ts'

+type QueryFactory = (params: {
+  prompt: string
+  options: Options
+}) => Query
+
+const queryMock = vi.hoisted(() => vi.fn<QueryFactory>())
+
+vi.mock('@anthropic-ai/claude-agent-sdk', async importOriginal => ({
+  ...await importOriginal<typeof import('@anthropic-ai/claude-agent-sdk')>(),
+  query: queryMock,
+}))
+
 const fakeParent = {
  id: 'parent',
  session: { header: { cwd: process.cwd() } },
@@ -56,7 +77,6 @@ interface FakeChildOptions {
  readonly stdin?: PassThrough | undefined
  readonly stdout?: PassThrough | undefined
  readonly exitOnTerminate?: boolean
-  readonly waitForExitResult?: boolean
  readonly waitForExitError?: Error
  readonly doneError?: Error
 }
@@ -103,9 +123,6 @@ function fakeChild(options: FakeChildOptions = {}): FakeChild {
    if (options.waitForExitError !== undefined) {
      throw options.waitForExitError
    }
-    if (options.waitForExitResult !== undefined) {
-      return options.waitForExitResult
-    }
    if (exited) return true
    if (signal === undefined) {
      await done.catch(() => {})
@@ -218,7 +235,7 @@ interface FakeRun {
  readonly query: Query
  readonly close: ReturnType<typeof vi.fn>
  readonly spawnSpecs: SubprocessSpawnSpec[]
-  readonly options: Array<Parameters<NonNullable<ClaudeCodeRunSpec['query']>>[0]['options']>
+  readonly options: Options[]
  readonly spec: ClaudeCodeRunSpec
 }

@@ -239,16 +256,28 @@ function fakeRun(
      spawnSpecs.push(spawnSpec)
      return child.handle
    },
-    query: (params) => {
-      options.push(params.options)
-      params.options.spawnClaudeCodeProcess!(sdkSpawnOptions())
-      return query
-    },
  }
+  queryMock.mockImplementation((params) => {
+    options.push(params.options)
+    params.options.spawnClaudeCodeProcess!(sdkSpawnOptions())
+    return query
+  })
  return { child, query, close, spawnSpecs, options, spec }
 }

+beforeEach(() => {
+  queryMock.mockImplementation(({ options }) => {
+    options.spawnClaudeCodeProcess!(sdkSpawnOptions({
+      cwd: options.cwd!,
+      env: options.env!,
+      signal: options.abortController!.signal,
+    }))
+    return queryFrom([])
+  })
+})
+
 afterEach(() => {
+  queryMock.mockReset()
  vi.restoreAllMocks()
  vi.unstubAllEnvs()
 })
@@ -523,25 +552,17 @@ describe('query options and result mapping', () => {
  })

  it('consumes the complete stream and keeps the latest strict success', async () => {
-    const outputs: ContentBlock[][] = []
    const query = queryFrom([
      { type: 'system', subtype: 'init' } as SDKMessage,
      success('first'),
      success('last'),
    ])
-    await expect(consumeClaudeQuery(query, (output) => {
-      outputs.push(output)
-    })).resolves.toEqual({
+    await expect(consumeClaudeQuery(query)).resolves.toEqual({
      output: [{ type: 'text', text: 'last' }],
      stopReason: 'completed',
    })
-    expect(outputs).toEqual([
-      [{ type: 'text', text: 'first' }],
-      [{ type: 'text', text: 'last' }],
-    ])
    await expect(consumeClaudeQuery(
      queryFrom([{ type: 'system', subtype: 'init' } as SDKMessage]),
-      () => {},
    )).rejects.toThrow('ended without a result')
  })
 })
@@ -596,14 +617,14 @@ describe('run publication, cancellation, and settlement', () => {
    }
  })

-  it('preserves candidate output when iteration fails after a result', async () => {
+  it('fails closed when iteration rejects after a result', async () => {
    const fixture = fakeRun(
      [success('partial final')],
      new Error('iterator boom'),
    )
    const run = await startClaudeCodeRun(request(), fixture.spec)
    await expect(run.result).resolves.toEqual({
-      output: [{ type: 'text', text: 'partial final' }],
+      output: [],
      stopReason: 'error',
    })
    await run.dispose()
@@ -635,14 +656,14 @@ describe('run publication, cancellation, and settlement', () => {
      env: {},
      disposeGraceMs: 5,
      spawn: () => children[index++]!.handle,
-      query: ({ prompt, options }) => {
-        controllers.push(options.abortController!)
-        options.spawnClaudeCodeProcess!(sdkSpawnOptions())
-        return prompt === 'wait'
-          ? waitingQuery(options.abortController!.signal)
-          : queryFrom([success('second answer')])
-      },
    }
+    queryMock.mockImplementation(({ prompt, options }) => {
+      controllers.push(options.abortController!)
+      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
+      return prompt === 'wait'
+        ? waitingQuery(options.abortController!.signal)
+        : queryFrom([success('second answer')])
+    })
    const firstAbort = new AbortController()
    const first = await startClaudeCodeRun(
      request([{ type: 'text', text: 'wait' }], firstAbort.signal),
@@ -667,6 +688,33 @@ describe('run publication, cancellation, and settlement', () => {
    await Promise.all([first.dispose(), second.dispose()])
  })

+  it('keeps local cancellation authoritative when the SDK iterator ends normally', async () => {
+    const parentAbort = new AbortController()
+    const child = fakeChild()
+    async function* stream(): AsyncGenerator<SDKMessage, void> {
+      yield success('candidate answer')
+      parentAbort.abort(new Error('parent cancelled at iterator completion'))
+    }
+    queryMock.mockImplementation(({ options }) => {
+      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
+      return Object.assign(stream(), { close: vi.fn() }) as unknown as Query
+    })
+    const run = await startClaudeCodeRun(
+      request(undefined, parentAbort.signal),
+      {
+        cwd: '/workspace',
+        env: {},
+        disposeGraceMs: 5,
+        spawn: () => child.handle,
+      },
+    )
+    await expect(run.result).resolves.toEqual({
+      output: [],
+      stopReason: 'aborted',
+    })
+    await run.dispose()
+  })
+
  it('rejects pre-abort and every incomplete startup transaction', async () => {
    const preAborted = new AbortController()
    preAborted.abort()
@@ -678,32 +726,36 @@ describe('run publication, cancellation, and settlement', () => {
    expect(unused.options).toEqual([])

    const noChildClose = vi.fn()
+    queryMock.mockImplementationOnce(
+      () => queryFrom([], undefined, noChildClose),
+    )
    await expect(startClaudeCodeRun(request(), {
      ...unused.spec,
-      query: () => queryFrom([], undefined, noChildClose),
    })).rejects.toThrow('did not publish a controllable')
    expect(noChildClose).toHaveBeenCalledOnce()

    const closeFailure = vi.fn(() => { throw new Error('close boom') })
+    queryMock.mockImplementationOnce(
+      () => queryFrom([], undefined, closeFailure),
+    )
    const noChild = startClaudeCodeRun(request(), {
      ...unused.spec,
-      query: () => queryFrom([], undefined, closeFailure),
    })
    await expect(noChild).rejects.toBeInstanceOf(AggregateError)

    const startupAbort = new AbortController()
    const abortedChild = fakeChild()
    const abortedClose = vi.fn()
+    queryMock.mockImplementationOnce(({ options }) => {
+      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
+      startupAbort.abort(new Error('startup cancelled'))
+      return queryFrom([], undefined, abortedClose)
+    })
    const abortedDuringStartup = startClaudeCodeRun(
      request(undefined, startupAbort.signal),
      {
        ...unused.spec,
        spawn: () => abortedChild.handle,
-        query: ({ options }) => {
-          options.spawnClaudeCodeProcess!(sdkSpawnOptions())
-          startupAbort.abort(new Error('startup cancelled'))
-          return queryFrom([], undefined, abortedClose)
-        },
      },
    )
    await expect(abortedDuringStartup)
@@ -711,27 +763,27 @@ describe('run publication, cancellation, and settlement', () => {
    expect(abortedClose).toHaveBeenCalledOnce()
    expect(abortedChild.terminate).toHaveBeenCalledOnce()

+    queryMock.mockImplementationOnce(() => {
+      throw new Error('query failed before resource creation')
+    })
    await expect(startClaudeCodeRun(request(), {
      ...unused.spec,
-      query: () => {
-        throw new Error('query failed before resource creation')
-      },
    })).rejects.toThrow('query failed before resource creation')

    const spawned = fakeChild()
    const spawnSpecs: SubprocessSpawnSpec[] = []
    let factoryController: AbortController | undefined
+    queryMock.mockImplementationOnce(({ options }) => {
+      factoryController = options.abortController
+      options.spawnClaudeCodeProcess!(sdkSpawnOptions())
+      throw new Error('query construction failed')
+    })
    const factoryFailure = startClaudeCodeRun(request(), {
      ...unused.spec,
      spawn: (spawnSpec) => {
        spawnSpecs.push(spawnSpec)
        return spawned.handle
      },
-      query: ({ options }) => {
-        factoryController = options.abortController
-        options.spawnClaudeCodeProcess!(sdkSpawnOptions())
-        throw new Error('query construction failed')
-      },
    })
    await expect(factoryFailure).rejects.toThrow('query construction failed')
    expect(spawnSpecs).toHaveLength(1)
@@ -749,76 +801,45 @@ describe('run publication, cancellation, and settlement', () => {
  })
 })

-describe('bounded query and process disposal', () => {
+describe('query and process disposal', () => {
  it('closes the query, terminates the tree, and waits for direct-child outcome', async () => {
    const child = fakeChild()
    const close = vi.fn()
-    await disposeClaudeCodeChild({ close }, child.handle, 5)
+    await disposeClaudeCodeChild({ close }, child.handle)
    expect(close).toHaveBeenCalledOnce()
    expect(child.terminate).toHaveBeenCalledOnce()
    expect(child.waitForExit).toHaveBeenCalledOnce()
+    expect(child.waitForExit).toHaveBeenCalledWith()
    await expect(child.handle.done).resolves.toEqual({
      exitCode: 0,
      signal: null,
    })
  })

-  it('accepts fractional and larger-than-Node grace windows', async () => {
-    for (const graceMs of [0.25, Number.MAX_VALUE]) {
-      const child = fakeChild()
-      await expect(disposeClaudeCodeChild(
-        { close: vi.fn() },
-        child.handle,
-        graceMs,
-      )).resolves.toBeUndefined()
-      const signal = child.waitForExit.mock.calls[0]?.[0]
-      expect(signal?.aborted).toBe(false)
-    }
-  })
-
-  it('chains a doubled grace window beyond one Node timer segment', async () => {
-    vi.useFakeTimers()
-    try {
-      const child = fakeChild({ exitOnTerminate: false })
-      const disposal = disposeClaudeCodeChild(
-        { close: vi.fn() },
-        child.handle,
-        1_073_741_823.75,
-      )
-      const rejected = expect(disposal)
-        .rejects.toThrow('did not exit within its dispose window')
-      await vi.advanceTimersByTimeAsync(2_147_483_647)
-      await vi.advanceTimersByTimeAsync(1)
-      await rejected
-    } finally {
-      vi.useRealTimers()
-    }
-  })
-
-  it('does not turn a missed tree-exit bound into an unbounded done wait', async () => {
-    const child = fakeChild({
-      exitOnTerminate: false,
-      waitForExitResult: false,
-    })
-    await expect(disposeClaudeCodeChild(
+  it('does not finish disposal before the managed tree exits', async () => {
+    const child = fakeChild({ exitOnTerminate: false })
+    let disposed = false
+    const disposal = disposeClaudeCodeChild(
      { close: vi.fn() },
      child.handle,
-      5,
-    )).rejects.toThrow('did not exit within its dispose window')
-    child.fail(new Error('late direct-child failure'))
+    ).then(() => {
+      disposed = true
+    })
    await nextTask()
+    expect(disposed).toBe(false)
+    child.settle()
+    await disposal
+    expect(disposed).toBe(true)
  })

  it('reports wait, close, and direct-child failures without skipping cleanup', async () => {
    const waitFailure = fakeChild({
-      exitOnTerminate: false,
      waitForExitError: new Error('wait boom'),
    })
    const closeFailure = vi.fn(() => { throw new Error('close boom') })
    await expect(disposeClaudeCodeChild(
      { close: closeFailure },
      waitFailure.handle,
-      5,
    )).rejects.toBeInstanceOf(AggregateError)
    expect(waitFailure.terminate).toHaveBeenCalledOnce()

@@ -829,7 +850,6 @@ describe('bounded query and process disposal', () => {
    await expect(disposeClaudeCodeChild(
      { close: vi.fn() },
      doneFailure.handle,
-      5,
    )).rejects.toThrow('spawn boom')

    const both = fakeChild({
@@ -839,7 +859,6 @@ describe('bounded query and process disposal', () => {
    await expect(disposeClaudeCodeChild(
      { close: () => { throw new Error('close boom') } },
      both.handle,
-      5,
    )).rejects.toBeInstanceOf(AggregateError)
  })
 })
--- a/packages/subagent/subagent-claude-code/tsconfig.json
+++ b/packages/subagent/subagent-claude-code/tsconfig.json
@@ -9,6 +9,12 @@
    "src/**/*.ts"
  ],
  "references": [
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../../vendor/schemastery"
+    },
    {
      "path": "../../llm/llm"
    },