kevin/deepseek-harness
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into feat/tui-package

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-19 02:10:33 +08:00
parent a3c05e44d6 8518dd31c7
commit 64e920fad7
14 changed files with 931 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/rfc/implemented/testing/2026-07-08-shared-acp-snapshot-package.md
+6 -4
View File
@@ -6,13 +6,15 @@ Status: implemented
The ACP snapshot tier ([snapshot RFC](2026-06-19-acp-snapshot-tests.md)) was built from three modules living inside one example's test directory: `snapshot-harness.ts` (boot the real bin subprocess, drive it over ACP JSON-RPC, harvest the persisted logs), `snapshot-normalize.ts` (the pure golden normalizers), and the ~150-line scenario body plus fixture guards in `acp.snapshot.ts` (record/replay modes, the stdout-golden and log compares, the pinned-header uniformity guard, the orphan/required-file/single-pin meta-tests).
A second ACP example could only copy record, normalization, and harvest logic that must stay consistent. Code under `examples/` also sat outside the package coverage gate, and the original harness could only cancel permission requests. The shared package makes the machinery measured and lets scenarios script approval answers.
A second ACP example wanting snapshot coverage — the sandbox/approval composition is the immediate consumer — could only copy those modules, forking exactly the logic that must not drift: record write-back, header scrubbing, child-session harvest ordering. The spawn/client glue was also triplicated across `acp.e2e.ts`, `hooks.e2e.ts`, and the harness. Location decided test rigor: the per-file 100% coverage gate measures `packages/*/*/src` only, so none of this machinery was measured — the same gap that had moved `dsh-llm-replay` out of `examples/` into [packages/support](../../../../packages/support/README.md). And the harness's ACP client hardcoded `requestPermission → cancelled`, so an approval round-trip — the headline behavior of the sandbox composition — could not be expressed at the snapshot tier at all.
## Decision
The machinery lives in [`packages/support/acp-snapshot`](../../../../packages/support/acp-snapshot/README.md) (`@deepseek-ai/dsh-acp-snapshot`); an example's `*.snapshot.ts` is its scenario table, its agent paths, and one factory call, over its own `snapshots/` fixtures and `cordis.snapshot.yml` overlay ([single-source replay config](2026-07-04-single-source-acp-replay-config.md)). Reading `DSH_SNAPSHOT` stays at that edge — the library takes a resolved `mode`.
**`src/harness.ts`** provides `runScenario` and its script/result types, parameterized by the agent's bin and config paths. Permission answers form a FIFO queue keyed by stable option kind rather than random option id. Missing answers cancel the request; an unavailable kind cancels the agent request and fails the scenario.
**`src/launcher.ts`** — `launchAcpTestAgent` owns the common unbuilt-process boundary: absolute tsx loader resolution, `TSX_TSCONFIG_PATH`, isolated harness homes, stdio wiring, a raw-byte stdout tee, stderr and update capture, fail-closed permission fallback, update waiters, and graceful or signalled shutdown. Snapshot scenarios and ordinary e2e suites supply the same `AgentUnderTest` (`binScript`, `configPath`, `tsconfigPath`); a test that plays a user supplies only its permission handler. The ACP and hook e2e suites plus the sandbox/approval e2e suite use this launcher instead of rebuilding the SDK client boundary.
**`src/harness.ts`** — `runScenario` and the input-script/result types layer deterministic steps, temp workspaces, snapshot environment, and persisted-log harvest over the launcher. Its `session/request_permission` handler consumes an optional `InputScript.permissionAnswers` FIFO queue, each entry selecting by option **kind** (ids are agent-issued randoms a committed script cannot know; kinds are the ACP-stable vocabulary, mapped to the offered `optionId` at answer time); an absent or exhausted queue answers `cancelled`, and a kind the request never offered rejects the run — the agent itself is answered `cancelled`, so the scenario bug fails the harness rather than being absorbed as an agent-side denial. This is what lets an approval suite drive allow/reject round-trips deterministically from `input.json`.
**`src/normalize.ts`** — the pure normalizers, hook-free by policy: when a future event carries a new volatile field (an approval duration, say), the shared normalizer learns it in the same change, keeping one home for what "normalized" means rather than per-suite scrub extensions.
@@ -29,8 +31,8 @@ The machinery lives in [`packages/support/acp-snapshot`](../../../../packages/su
## Testing
Extraction preserved every existing ACP golden byte. The package's `src/` has per-file 100% coverage through a scripted ACP subprocess: harness tests cover every step operation, both expected-error branches, permission selection/fallback/impossible choice, environment forwarding, workspace seeding, and harvest ordering/noise/fallback; suite tests execute replay against committed synthetic fixtures and record against a temporary copy, plus the pure helpers. Two structurally unreachable guards retain reasoned coverage exclusions. The fake agent substitutes the `session/new` cwd into logs, including Darwin's `/var` realpath behavior, matching the real bin.
Extraction parity was proven mechanically: after the move, `pnpm run test:snapshot` matched the base commit's result with zero byte changes under `examples/acp-agent/tests/snapshots/`. The package's `src/` holds per-file 100% statements/branches/functions/lines under the gating unit run, driven through the real launcher by a scripted fake ACP bin (`tests/fixtures/fake-acp-agent.ts`, behavior scripted per scenario via a `behavior.json` beside the fixture): `harness.spec.ts` directly covers launcher defaults, captures, update waiting, shutdown, and environment/config variants, then covers every scenario step op, both expect-error arms, the permission queue (selection, fallback, impossible-click), workspace seeding, and the harvest ordering/noise/fallback branches; `suite.spec.ts` runs the factory for real at collection time — a replay suite over committed synthetic fixtures and a record suite over a temp copy (write-back never touches the committed tree; `ACP_SNAPSHOT_SPEC_BOOTSTRAP=1` re-bootstraps it) — plus direct cases for the pure helpers. The fake bin substitutes the `session/new` cwd, not `process.cwd()`, into scripted logs, matching what the real bin's header carries (darwin realpaths `/var/folders/…` to `/private/var/folders/…`).
## Consequences
A new example gets the whole snapshot tier from a scenario table plus fixtures — the sandbox branch merges master down and adds its own suite (own pin scenario, own overlay, fixtures via `test:snapshot:record`, approvals via `permissionAnswers`). The costs: `suite.ts` imports vitest, so the package is importable only inside a vitest run — a shape no other package has, stated in its README; each suite pins its own ~8 KB header fixture (a genuinely distinct composition deserves its own pin; an identical one would be caught by that suite's uniformity guard); and the e2e launcher duplication remains (`TODO(acp-test-harness)`) — the harness is the extraction target when that migration lands.
A new example gets the whole snapshot tier from a scenario table plus fixtures, while an ordinary ACP e2e gets the same tested process/client boundary from one launcher call. The costs: `suite.ts` imports vitest, so the package entry is importable only inside a vitest run — a shape no other package has, stated in its README; and each suite pins its own ~8 KB header fixture (a genuinely distinct composition deserves its own pin; an identical one would be caught by that suite's uniformity guard).
examples/acp-agent/tests/acp.e2e.ts
+66 -158
View File
@@ -1,173 +1,62 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { Readable, Writable } from 'node:stream'
import { mkdtemp, rm, readFile } from 'node:fs/promises'
import { mkdtemp, readFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { fileURLToPath } from 'node:url'
import { afterEach, describe, expect, it } from 'vitest'
import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
import {
ClientSideConnection,
ndJsonStream,
PROTOCOL_VERSION,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
} from '@agentclientprotocol/sdk'
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
launchAcpTestAgent,
type AgentUnderTest,
type LaunchedAcpTestAgent,
} from '@deepseek-ai/dsh-acp-snapshot'
import { cleanupAcpExampleTest } from './cleanup.ts'
/**
* Boots examples/acp-agent as an ACP subprocess. The key-gated prompt leg
* verifies its filesystem effect; a keyless initialize leg verifies that stdout
* contains only framed JSON-RPC. Each subprocess is disposed in `afterEach`.
* End-to-end: boot examples/acp-agent as a real subprocess speaking ACP over
* its stdio, drive it with a real ClientSideConnection, send a real prompt, and
* verify the WORLD (a file the agent wrote), not the agent's self-report. Owns
* and disposes the subprocess in afterEach. Key-gated.
*
* Also asserts stdout purity (only framed JSON-RPC on stdout) — that one runs
* WITHOUT a key, since it only needs the server to boot and answer initialize.
*/
// The child runs from a temp cwd, so its bin and config path are absolute.
const binScript = fileURLToPath(new URL('../../../packages/examples/acp-demo/src/bin.ts', import.meta.url))
const configPath = fileURLToPath(new URL('../cordis.yml', import.meta.url))
// The root tsconfig supplies unbuilt workspace `paths`; making it explicit
// avoids accidental resolution through stale built output.
const repoTsconfig = fileURLToPath(new URL('../../../tsconfig.json', import.meta.url))
interface Spawned {
child: ChildProcessWithoutNullStreams
client: ClientSideConnection
updates: SessionNotification['update'][]
stderr: string[]
const AGENT: AgentUnderTest = {
binScript: fileURLToPath(new URL('../../../packages/examples/acp-demo/src/bin.ts', import.meta.url)),
configPath: fileURLToPath(new URL('../cordis.yml', import.meta.url)),
tsconfigPath: fileURLToPath(new URL('../../../tsconfig.json', import.meta.url)),
}
const DANGER_FULL_ACCESS_ENV = { DSH_PERMISSION_MODE: 'danger-full-access' }
function spawnAcpAgent(cwd: string, env: NodeJS.ProcessEnv = process.env): Spawned {
const launch = resolveExampleLaunch({
srcBin: binScript,
configArgs: ['--config', configPath],
tsconfigPath: repoTsconfig,
env: {
DSH_PERMISSION_MODE: 'danger-full-access',
DSH_HOME: join(cwd, '.dsh'),
DSH_AGENTS_HOME: join(cwd, '.agents'),
},
})
const child = spawn(
launch.command,
launch.args,
{ cwd, env: { ...env, ...launch.env }, stdio: ['pipe', 'pipe', 'pipe'] },
)
const stderr: string[] = []
child.stderr.setEncoding('utf8')
child.stderr.on('data', (chunk: string) => stderr.push(chunk))
const updates: SessionNotification['update'][] = []
const stream = ndJsonStream(
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>,
)
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
updates.push(params.update)
return Promise.resolve()
},
requestPermission(_params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
// This suite selects danger-full-access (approval never), so the bridge
// never prompts here; answer cancelled if an unexpected ask arrives.
return Promise.resolve({ outcome: { outcome: 'cancelled' } })
},
})
const client = new ClientSideConnection(makeClient, stream)
return { child, client, updates, stderr }
}
let spawned: Spawned | undefined
let spawned: LaunchedAcpTestAgent | undefined
let workdir: string | undefined
function hasStdoutLine(out: string[]): boolean {
return out.join('').split('\n').some(line => line.trim().length > 0)
}
async function waitForStdoutLine(child: ChildProcessWithoutNullStreams, out: string[], stderr: string[], timeoutMs: number): Promise<void> {
await new Promise<void>((resolve, reject) => {
const cleanup = () => {
clearTimeout(timeout)
child.stdout.off('data', onData)
child.off('exit', onExit)
child.off('error', onError)
}
const pass = () => {
cleanup()
resolve()
}
const fail = (reason: string) => {
cleanup()
reject(new Error(`${reason}; stderr: ${stderr.join('')}`))
}
const onData = () => {
if (hasStdoutLine(out)) pass()
}
const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
fail(`ACP child exited before emitting a stdout frame (code ${code ?? 'null'}, signal ${signal ?? 'null'})`)
}
const onError = (error: Error) => {
fail(`ACP child failed before emitting a stdout frame: ${error.message}`)
}
const timeout = setTimeout(() => {
fail(`ACP child did not emit a stdout frame within ${timeoutMs}ms`)
}, timeoutMs)
child.stdout.on('data', onData)
child.on('exit', onExit)
child.on('error', onError)
onData()
})
}
afterEach(async () => {
if (spawned) {
spawned.child.kill('SIGKILL')
spawned = undefined
}
if (workdir !== undefined) await rm(workdir, { recursive: true, force: true })
const ownedSpawned = spawned
const ownedWorkdir = workdir
spawned = undefined
workdir = undefined
await cleanupAcpExampleTest(ownedSpawned, ownedWorkdir)
})
describe('acp-agent over real stdio (no key required)', () => {
it('emits only framed JSON-RPC on stdout', async () => {
workdir = await mkdtemp(join(tmpdir(), 'acp-e2e-'))
// Collect raw stdout bytes directly (bypass the SDK framing) to inspect.
// A dummy key boots the adapter; this purity test sends no prompt and makes no model call.
const launch = resolveExampleLaunch({
srcBin: binScript,
configArgs: ['--config', configPath],
tsconfigPath: repoTsconfig,
// Inspect the launcher's raw-byte tee in addition to driving its SDK client.
// A dummy key lets the deepseek adapter APPLY (it only checks the key is
// present at boot, not valid — the key is used only on a real model call,
// which this purity test never triggers). So this runs WITHOUT real creds.
spawned = launchAcpTestAgent({
agent: AGENT,
cwd: workdir,
env: {
DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot',
DSH_PERMISSION_MODE: 'danger-full-access',
DSH_HOME: join(workdir, '.dsh'),
DSH_AGENTS_HOME: join(workdir, '.agents'),
...DANGER_FULL_ACCESS_ENV,
},
})
const child = spawn(launch.command, launch.args, {
cwd: workdir,
env: { ...process.env, ...launch.env },
stdio: ['pipe', 'pipe', 'pipe'],
})
const out: string[] = []
const stderr: string[] = []
child.stdout.setEncoding('utf8')
child.stderr.setEncoding('utf8')
child.stdout.on('data', (c: string) => out.push(c))
child.stderr.on('data', (c: string) => stderr.push(c))
await spawned.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
// Send a single initialize request as a newline-delimited JSON-RPC frame.
const req = JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} } })
child.stdin.write(req + '\n')
try {
await waitForStdoutLine(child, out, stderr, 15_000)
} finally {
child.kill('SIGKILL')
}
const lines = out.join('').split('\n').filter(l => l.trim().length > 0)
const lines = spawned.rawStdout().split('\n').filter(line => line.trim().length > 0)
expect(lines.length).toBeGreaterThan(0)
for (const line of lines) {
// Every stdout line MUST parse as JSON (a JSON-RPC frame). A non-JSON
@@ -177,14 +66,28 @@ describe('acp-agent over real stdio (no key required)', () => {
}, 30_000)
it('session/new succeeds over real stdio (no model call)', async () => {
// Regression guard (this exact RPC crashed a real Zed session with "cannot get property
// \"agents\" without inject"): `session/new` drives the full bridge →
// `ctx.agents.create({sessionId, meta:{cwd}})` → AgentLoop → registry/persistence path, ALL
// of which run from the JSON-RPC read loop outside the bridge plugin's injection scope.
// REGRESSION GUARD (this exact RPC crashed a real Zed session with
// "cannot get property \"agents\" without inject"): `session/new` drives the
// full bridge → `ctx.agents.create({sessionId, meta:{cwd}})` → AgentLoop →
// registry/persistence path, ALL of which run from the JSON-RPC read loop
// OUTSIDE the bridge plugin's injection scope. A lazy `ctx.<service>` read
// on that path throws and the RPC fails with an Internal error — yet the
// call never touches the model, so this reproduces WITHOUT a key. The
// key-gated prompt test below never caught it (it needs real creds); the
// initialize-only purity test never caught it (initialize does not reach
// the factory). This closes that gap: boot the real subprocess and create a
// session, asserting the RPC RESOLVES (not rejects with an inject error).
workdir = await mkdtemp(join(tmpdir(), 'acp-e2e-'))
// A dummy key lets the deepseek adapter boot (it only checks presence, not
// validity, at apply time); no model call is made, so the key is never used.
spawned = spawnAcpAgent(workdir, { ...process.env, DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot' })
spawned = launchAcpTestAgent({
agent: AGENT,
cwd: workdir,
env: {
DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot',
...DANGER_FULL_ACCESS_ENV,
},
})
const { client } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
@@ -197,7 +100,7 @@ describe('acp-agent over real stdio (no key required)', () => {
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('acp-agent e2e: real prompt over ACP', () => {
it('runs a real turn and the agent writes the requested file (verified on disk)', async () => {
workdir = await mkdtemp(join(tmpdir(), 'acp-e2e-'))
spawned = spawnAcpAgent(workdir)
spawned = launchAcpTestAgent({ agent: AGENT, cwd: workdir, env: DANGER_FULL_ACCESS_ENV })
const { client, updates } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
@@ -211,29 +114,34 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('acp-agent e2e: real prompt over
})
expect(['end_turn', 'max_tokens']).toContain(res.stopReason)
// Verify the filesystem effect rather than the agent's report.
// Verify the WORLD, not the agent's self-report: read the file from disk.
const proof = await readFile(join(workdir, 'proof.txt'), 'utf8')
expect(proof).toContain('ACP_OK')
// And the client saw tool-call activity stream through.
const toolCalls = updates.filter(u => u.sessionUpdate === 'tool_call')
expect(toolCalls.length).toBeGreaterThan(0)
// Bash execute cards hide rawInput, so `presentCall` uses the exact command
// as the title rather than the bare tool name "bash".
// Tool-call UI quality (the tool owns its presentation): the bash tool's
// `presentCall` sets the title to the exact command (an execute card hides
// rawInput, so the command IS the title) — NOT the bare tool name "bash".
// A `bash` call must therefore carry an execute kind, a non-"bash" title,
// and a string rawInput (the command). `toolCalls` is already narrowed to
// the `tool_call` shape by the filter above, so these fields are reachable.
const bashCall = toolCalls.find(u => u.kind === 'execute')
expect(bashCall).toBeDefined()
if (bashCall === undefined) throw new Error('expected an execute tool_call')
expect(typeof bashCall.title).toBe('string')
expect(bashCall.title.length).toBeGreaterThan(0)
expect(bashCall.title).not.toBe('bash')
expect(typeof bashCall.rawInput).toBe('string')
// Without the terminal capability, output uses the console-text path.
expect(bashCall.title).not.toBe('bash') // the old, unhelpful title
expect(typeof bashCall.rawInput).toBe('string') // the exact command
// Capability OFF: no terminal _meta — the ```console text path renders.
expect((bashCall as { _meta?: unknown })._meta).toBeUndefined()
}, 180_000)
it('with the terminal_output capability, a real bash call renders as a terminal card (content + _meta + exit)', async () => {
workdir = await mkdtemp(join(tmpdir(), 'acp-e2e-'))
spawned = spawnAcpAgent(workdir)
spawned = launchAcpTestAgent({ agent: AGENT, cwd: workdir, env: DANGER_FULL_ACCESS_ENV })
const { client, updates } = spawned
// Advertise the Zed `_meta.terminal_output` capability so the bridge emits
examples/acp-agent/tests/cleanup.e2e.ts
+38
View File
@@ -0,0 +1,38 @@
/** Regression coverage for ACP example teardown. */
import { access, mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { cleanupAcpExampleTest } from './cleanup.ts'
let fallbackWorkdir: string | undefined
afterEach(async () => {
if (fallbackWorkdir !== undefined) await rm(fallbackWorkdir, { recursive: true, force: true })
fallbackWorkdir = undefined
})
describe('cleanupAcpExampleTest', () => {
it('removes the workspace after process shutdown fails', async () => {
fallbackWorkdir = await mkdtemp(join(tmpdir(), 'acp-cleanup-'))
const closeFailure = new Error('close failed')
const spawned = { close: vi.fn().mockRejectedValue(closeFailure) }
await expect(cleanupAcpExampleTest(spawned, fallbackWorkdir))
.rejects.toMatchObject({ errors: [closeFailure] })
await expect(access(fallbackWorkdir)).rejects.toThrow()
fallbackWorkdir = undefined
})
it('reports process and workspace failures together', async () => {
const closeFailure = new Error('close failed')
const spawned = { close: vi.fn().mockRejectedValue(closeFailure) }
const failure = await cleanupAcpExampleTest(spawned, '\0').catch((error: unknown) => error)
expect(failure).toBeInstanceOf(AggregateError)
expect((failure as AggregateError).errors).toHaveLength(2)
expect((failure as AggregateError).errors[0]).toBe(closeFailure)
})
})
examples/acp-agent/tests/cleanup.ts
+23
View File
@@ -0,0 +1,23 @@
/** Shared teardown for ACP example tests. */
import { rm } from 'node:fs/promises'
import type { LaunchedAcpTestAgent } from '@deepseek-ai/dsh-acp-snapshot'
/**
* Close the test agent, then remove its workspace, attempting both operations
* and reporting every failure instead of allowing the later one to mask the
* earlier one.
*/
export async function cleanupAcpExampleTest(
spawned: Pick<LaunchedAcpTestAgent, 'close'> | undefined,
workdir: string | undefined,
): Promise<void> {
const results: PromiseSettledResult<unknown>[] = []
if (spawned !== undefined) results.push(...await Promise.allSettled([spawned.close('SIGKILL')]))
if (workdir !== undefined) results.push(...await Promise.allSettled([rm(workdir, { recursive: true, force: true })]))
const failures = results
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
.map(result => result.reason as unknown)
if (failures.length > 0) throw new AggregateError(failures, 'ACP example cleanup failed')
}
examples/acp-agent/tests/escalation.e2e.ts
+65 -71
View File
@@ -1,40 +1,49 @@
import { spawn, spawnSync, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { Readable, Writable } from 'node:stream'
import { mkdtemp, readFile, rm } from 'node:fs/promises'
import { spawnSync } from 'node:child_process'
import { mkdtemp, readFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { fileURLToPath } from 'node:url'
import { afterEach, describe, expect, it } from 'vitest'
import {
ClientSideConnection,
ndJsonStream,
PROTOCOL_VERSION,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
} from '@agentclientprotocol/sdk'
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
import {
launchAcpTestAgent,
type AgentUnderTest,
type LaunchedAcpTestAgent,
} from '@deepseek-ai/dsh-acp-snapshot'
import { cleanupAcpExampleTest } from './cleanup.ts'
/**
* Exercises the default ACP composition through the real bin and Loader. The
* keyless leg boots sandbox, approval, permission, and bridge services, then
* initializes and opens a session without a model call or runner probe. With a
* key and usable runner, the prompt asserts a prior denial; the model requests
* a wider retry with justification, and a scripted client grants or rejects it.
* The filesystem must show that only the granted retry ran. Missing credentials
* or runner support self-skip; real denial markers remain on sandbox e2e tiers.
* The default ACP composition (`cordis.yml`) end to end.
*
* Keyless smoke: boot the REAL `cordis.yml` through the `dsh-acp-agent` bin as
* an ACP subprocess and drive initialize + session/new — the real-Loader-path
* guard (postmortem 0001) for THIS tree's export shapes, which now include the
* sandbox executor AND the approval service. No prompt is sent, so neither the
* model nor a sandbox runner is ever exercised.
*
* With-key escalation flow (self-skips without DEEPSEEK_API_KEY or a usable
* platform runner): a scripted ACP client plays the human. The prompt asserts
* a prior denial (the organic denial→marker path lives on the sandbox e2e
* legs and unit tiers), the real model escalates with `sandbox_permissions` +
* `justification`, the bridge prompts THIS client over
* `session/request_permission`, the client answers `allow-once`, and the
* retried write must land ON DISK (world-verified) — under the granted mode,
* a temp-dir session cwd is writable either way.
*/
const binScript = fileURLToPath(new URL('../../../packages/examples/acp-demo/src/bin.ts', import.meta.url))
const configPath = fileURLToPath(new URL('../cordis.yml', import.meta.url))
// The subprocess runs from a temp cwd outside the repo; point tsx at the repo
// tsconfig so the unbuilt `paths` map resolves in src mode (see examples/AGENTS.md).
const repoTsconfig = fileURLToPath(new URL('../../../tsconfig.json', import.meta.url))
const AGENT: AgentUnderTest = {
binScript: fileURLToPath(new URL('../../../packages/examples/acp-demo/src/bin.ts', import.meta.url)),
configPath: fileURLToPath(new URL('../cordis.yml', import.meta.url)),
tsconfigPath: fileURLToPath(new URL('../../../tsconfig.json', import.meta.url)),
}
// Without a usable bwrap/Seatbelt runner, the strict attempt fails closed with
// SANDBOX_UNAVAILABLE instead of producing the denial this flow requires.
// A usable confining runner, probed the same way the executor suites do:
// bwrap on Linux, Seatbelt's sandbox-exec on macOS. Without one the strict
// attempt would fail closed (SANDBOX_UNAVAILABLE) instead of producing the
// denial this flow starts from.
const hasBwrap = spawnSync('bwrap', ['--ro-bind', '/', '/', '--dev', '/dev', '--proc', '/proc', '--die-with-parent', '--', 'true'], {
timeout: 5_000,
stdio: 'ignore',
@@ -45,72 +54,49 @@ const hasSeatbelt = process.platform === 'darwin' && spawnSync('sandbox-exec', [
}).status === 0
const hasRunner = hasBwrap || hasSeatbelt
interface Spawned {
child: ChildProcessWithoutNullStreams
client: ClientSideConnection
updates: SessionNotification['update'][]
interface Spawned extends LaunchedAcpTestAgent {
permissionRequests: RequestPermissionRequest[]
stderr: string[]
}
/** Boot the example as an ACP subprocess; the scripted client answers every permission prompt with `answer`. */
function spawnAcpAgent(cwd: string, answer: 'allow-once' | 'reject-once'): Spawned {
const launch = resolveExampleLaunch({
srcBin: binScript,
configArgs: ['--config', configPath],
tsconfigPath: repoTsconfig,
// A dummy key lets the deepseek adapter boot keyless (presence-checked at
// apply, used only on a real model call); the with-key tests carry the
// real key, so the fallback is inert there.
env: { DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY || 'sk-dummy-for-boot' },
})
const child = spawn(
launch.command,
launch.args,
{ cwd, env: { ...process.env, ...launch.env }, stdio: ['pipe', 'pipe', 'pipe'] },
)
const stderr: string[] = []
child.stderr.setEncoding('utf8')
child.stderr.on('data', (chunk: string) => stderr.push(chunk))
const updates: SessionNotification['update'][] = []
function launchExampleAcpAgent(cwd: string, answer: 'allow-once' | 'reject-once'): Spawned {
const permissionRequests: RequestPermissionRequest[] = []
const stream = ndJsonStream(
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>,
)
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
updates.push(params.update)
return Promise.resolve()
},
requestPermission(params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
const launched = launchAcpTestAgent({
agent: AGENT,
cwd,
// A dummy key lets the adapter boot keylessly; live tests carry the real key.
env: { DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY ?? 'sk-dummy-for-boot' },
requestPermission(params) {
permissionRequests.push(params)
const option = params.options.find(o => o.optionId === answer)
// An unexpected prompt shape cancels without granting.
// The scripted human: pick the requested option when the prompt offers
// it; an unexpected prompt shape cancels (fail closed, never grants).
if (option === undefined) return Promise.resolve({ outcome: { outcome: 'cancelled' } })
return Promise.resolve({ outcome: { outcome: 'selected', optionId: option.optionId } })
},
})
const client = new ClientSideConnection(makeClient, stream)
return { child, client, updates, permissionRequests, stderr }
return Object.assign(launched, { permissionRequests })
}
let spawned: Spawned | undefined
let workdir: string | undefined
afterEach(async () => {
if (spawned !== undefined && spawned.child.exitCode === null) spawned.child.kill('SIGKILL')
const ownedSpawned = spawned
const ownedWorkdir = workdir
spawned = undefined
if (workdir !== undefined) await rm(workdir, { recursive: true, force: true })
workdir = undefined
await cleanupAcpExampleTest(ownedSpawned, ownedWorkdir)
})
describe('default sandbox composition keyless smoke (real cordis.yml via the Loader)', () => {
it('boots the tree — sandbox executor + approval service + bridge — and opens a session', async () => {
workdir = await mkdtemp(join(tmpdir(), 'sandbox-acp-smoke-'))
spawned = spawnAcpAgent(workdir, 'reject-once')
spawned = launchExampleAcpAgent(workdir, 'reject-once')
const { client } = spawned
// A dummy key boots the adapter; no prompt is ever sent, so no model call
// and no sandbox runner probe happen. This drives the fiber tree the same
// way an editor would, which is what catches a broken export/inject shape.
const init = await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
expect(init.protocolVersion).toBe(PROTOCOL_VERSION)
const { sessionId } = await client.newSession({ cwd: workdir, mcpServers: [] })
@@ -119,14 +105,18 @@ describe('default sandbox composition keyless smoke (real cordis.yml via the Loa
it('advertises model and Permissions selects and honors a permission switch without a model call', async () => {
workdir = await mkdtemp(join(tmpdir(), 'sandbox-acp-config-'))
spawned = spawnAcpAgent(workdir, 'reject-once')
spawned = launchExampleAcpAgent(workdir, 'reject-once')
const { client } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
// This tree composes the permission presets over bash-sandbox + approval →
// ONE select advertises, current from the configured default preset.
const created = await client.newSession({ cwd: workdir, mcpServers: [] })
const advertised = created.configOptions ?? []
const modelValue = JSON.stringify(['deepseek', 'deepseek-v4-flash'])
expect(advertised.map(option => [option.id, 'currentValue' in option ? option.currentValue : undefined]))
.toEqual([['model', modelValue], ['permission', 'workspace-write']])
// A switch responds with the COMPLETE refreshed state (the spec contract),
// and the new current survives in the response of a second switch.
const afterFullAccess = await client.setSessionConfigOption({
sessionId: created.sessionId, configId: 'permission', value: 'danger-full-access',
})
@@ -137,6 +127,7 @@ describe('default sandbox composition keyless smoke (real cordis.yml via the Loa
})
expect((again.configOptions ?? []).map(option => [option.id, 'currentValue' in option ? option.currentValue : undefined]))
.toEqual([['model', modelValue], ['permission', 'danger-full-access']])
// An out-of-vocabulary value is a protocol error, never a silent default.
await expect(client.setSessionConfigOption({
sessionId: created.sessionId, configId: 'permission', value: 'plan',
})).rejects.toThrow(/unknown permission value/)
@@ -146,7 +137,7 @@ describe('default sandbox composition keyless smoke (real cordis.yml via the Loa
describe.skipIf(!process.env.DEEPSEEK_API_KEY || !hasRunner)('default sandbox composition e2e: the live approval loop', () => {
it('denial → model escalation → editor prompt → allow-once → the retried write lands on disk', async () => {
workdir = await mkdtemp(join(tmpdir(), 'sandbox-acp-e2e-'))
spawned = spawnAcpAgent(workdir, 'allow-once')
spawned = launchExampleAcpAgent(workdir, 'allow-once')
const { client, permissionRequests } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
@@ -158,11 +149,13 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY || !hasRunner)('default sandbox co
})
expect(['end_turn', 'max_tokens']).toContain(res.stopReason)
// Verify the filesystem, not the model's report.
// The WORLD: the approved escalated retry landed the write.
const proof = await readFile(join(workdir, 'escalated.txt'), 'utf8')
expect(proof).toContain('ACP_ESCALATION_OK')
// Verify that ACP carried the grant with only one-shot choices.
// The CHANNEL: the grant came through a real session/request_permission
// prompt attached to the escalating tool call, offering exactly the
// one-shot options.
expect(permissionRequests.length).toBeGreaterThan(0)
const prompt = permissionRequests[0]
if (prompt === undefined) throw new Error('expected a permission request')
@@ -173,7 +166,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY || !hasRunner)('default sandbox co
it('a rejected escalation stays denied: no write lands, the turn still ends', async () => {
workdir = await mkdtemp(join(tmpdir(), 'sandbox-acp-e2e-'))
spawned = spawnAcpAgent(workdir, 'reject-once')
spawned = launchExampleAcpAgent(workdir, 'reject-once')
const { client, permissionRequests } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
@@ -185,8 +178,9 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY || !hasRunner)('default sandbox co
})
expect(['end_turn', 'max_tokens']).toContain(res.stopReason)
// The WORLD: rejected means the file never appeared.
await expect(readFile(join(workdir, 'refused.txt'), 'utf8')).rejects.toThrow()
// Distinguish a user rejection from a missing approval channel.
// And the rejection really flowed through a prompt (not a missing channel).
expect(permissionRequests.length).toBeGreaterThan(0)
}, 240_000)
})
examples/acp-agent/tests/hooks.e2e.ts
+21 -63
View File
@@ -1,21 +1,15 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { Readable, Writable } from 'node:stream'
import { mkdtemp, rm, writeFile, access } from 'node:fs/promises'
import { mkdtemp, writeFile, access } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { fileURLToPath } from 'node:url'
import { afterEach, describe, expect, it } from 'vitest'
import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
import {
ClientSideConnection,
ndJsonStream,
PROTOCOL_VERSION,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
} from '@agentclientprotocol/sdk'
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
launchAcpTestAgent,
type AgentUnderTest,
type LaunchedAcpTestAgent,
} from '@deepseek-ai/dsh-acp-snapshot'
import { cleanupAcpExampleTest } from './cleanup.ts'
/**
* With-key e2e for the Claude hook bridge. The process-level `./hooks.json` is
@@ -24,61 +18,21 @@ import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
* The test owns and disposes the ACP subprocess.
*/
const binScript = fileURLToPath(new URL('../../../packages/examples/acp-demo/src/bin.ts', import.meta.url))
const configPath = fileURLToPath(new URL('../cordis.yml', import.meta.url))
const repoTsconfig = fileURLToPath(new URL('../../../tsconfig.json', import.meta.url))
interface Spawned {
child: ChildProcessWithoutNullStreams
client: ClientSideConnection
updates: SessionNotification['update'][]
stderr: string[]
const AGENT: AgentUnderTest = {
binScript: fileURLToPath(new URL('../../../packages/examples/acp-demo/src/bin.ts', import.meta.url)),
configPath: fileURLToPath(new URL('../cordis.yml', import.meta.url)),
tsconfigPath: fileURLToPath(new URL('../../../tsconfig.json', import.meta.url)),
}
function spawnAcpAgent(cwd: string): Spawned {
const launch = resolveExampleLaunch({
srcBin: binScript,
configArgs: ['--config', configPath],
tsconfigPath: repoTsconfig,
env: { DSH_PERMISSION_MODE: 'danger-full-access' },
})
const child = spawn(
launch.command,
launch.args,
{ cwd, env: { ...process.env, ...launch.env }, stdio: ['pipe', 'pipe', 'pipe'] },
)
const stderr: string[] = []
child.stderr.setEncoding('utf8')
child.stderr.on('data', (chunk: string) => stderr.push(chunk))
const updates: SessionNotification['update'][] = []
const stream = ndJsonStream(
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
Readable.toWeb(child.stdout) as ReadableStream<Uint8Array>,
)
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
updates.push(params.update)
return Promise.resolve()
},
requestPermission(_params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
return Promise.resolve({ outcome: { outcome: 'cancelled' } })
},
})
const client = new ClientSideConnection(makeClient, stream)
return { child, client, updates, stderr }
}
let spawned: Spawned | undefined
let spawned: LaunchedAcpTestAgent | undefined
let workdir: string | undefined
afterEach(async () => {
if (spawned) {
spawned.child.kill('SIGKILL')
spawned = undefined
}
if (workdir !== undefined) await rm(workdir, { recursive: true, force: true })
const ownedSpawned = spawned
const ownedWorkdir = workdir
spawned = undefined
workdir = undefined
await cleanupAcpExampleTest(ownedSpawned, ownedWorkdir)
})
describe.skipIf(!process.env.DEEPSEEK_API_KEY)('acp-agent e2e: a PreToolUse hook blocks bash (real model)', () => {
@@ -90,7 +44,11 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('acp-agent e2e: a PreToolUse hook
hooks: { PreToolUse: [{ hooks: [{ type: 'command', command: 'echo "bash blocked by policy" >&2; exit 2' }] }] },
}))
spawned = spawnAcpAgent(workdir)
spawned = launchAcpTestAgent({
agent: AGENT,
cwd: workdir,
env: { DSH_PERMISSION_MODE: 'danger-full-access' },
})
const { client, updates } = spawned
await client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
packages/support/README.md
+2 -2
View File
@@ -4,11 +4,11 @@ Packages that exist to serve development, testing, and the examples rather than
| Package | Role | ctx key |
|---|---|---|
| `acp-snapshot/` | ACP snapshot suite kit: subprocess scenario harness + golden normalizers + the `defineAcpSnapshotSuite` factory | (library — imported by example `*.snapshot.ts` suites) |
| `acp-snapshot/` | ACP test kit: shared subprocess/client launcher + snapshot harness, normalizers, and suite factory | (library — imported by ACP e2e and `*.snapshot.ts` suites) |
| `agent-loop-testkit/` | Shared prerequisite mounting for tests that exercise the concrete agent loop | (library — imported by AgentLoop integration tests) |
| `invariants/` | Runtime event-contract assertions for development diagnostics | (listens on `session/*`, `agent/*`) |
| `loader-smoke/` | Shared real-Loader subprocess harness for keyless example smokes | (library — imported by example e2e suites) |
| `llm-replay/` | Record/replay adapter: short-circuits `llm/stream` from a recorded session JSONL (keyless snapshot tests) | (listens on `llm/stream`) |
| `subagent-mock/` | Scripted `SubagentProvider` for deterministic seam/tool tests | (registers on `ctx.subagents`) |
`invariants` is development support but has no environment guard: it runs wherever registered, and the default `dsh-agent-spine-demo` bundle mounts it unconditionally. `agent-loop-testkit` centralizes the mandatory service spine for hand-built AgentLoop tests without owning their loop or scenario. `llm-replay` backs the demos and the snapshot test tier under the per-file coverage gate. `acp-snapshot` carries the snapshot tier's harness/normalizer/suite machinery, while `loader-smoke` owns the parallel stdio/Loader process boundary used by keyless example e2e suites. `subagent-mock` exercises the real `ctx.subagents` load path without a model or child agent. A package graduates OUT of `support/` into a product group only when it gains documented product consumers.
`invariants` is development support but has no environment guard: it runs wherever registered, and the default `dsh-agent-spine-demo` bundle mounts it unconditionally. `agent-loop-testkit` centralizes the mandatory service spine for hand-built AgentLoop tests without owning their loop or scenario. `llm-replay` backs the demos and the snapshot test tier under the per-file coverage gate. `acp-snapshot` carries the ACP subprocess/client boundary plus the snapshot harness, normalizers, and suite machinery, while `loader-smoke` owns the parallel stdio/Loader process boundary used by keyless example e2e suites. `subagent-mock` exercises the real `ctx.subagents` load path without a model or child agent. A package graduates OUT of `support/` into a product group only when it gains documented product consumers.
packages/support/acp-snapshot/README.md
+4 -3
View File
@@ -2,8 +2,9 @@
The ACP snapshot suite kit: the shared machinery behind the keyless snapshot tier (`pnpm run test:snapshot`, [testing policy](../../../docs/testing.md)). An example gets a full snapshot suite from a scenario table plus a fixtures directory; every compare/guard mechanic lives here, under the per-file coverage gate, instead of being copied per example.
Three layers, importable separately:
Four layers, importable separately:
- **`launchAcpTestAgent` (launcher)** — boots an unbuilt ACP agent from a temp cwd, pins tsx to the repo tsconfig, connects the SDK client over a raw-byte stdout tee, collects session updates and stderr, surfaces asynchronous spawn failures through its startup lifecycle, fails closed on unhandled permission requests, and owns graceful or signalled shutdown. Shutdown waits for process exit, inherited stdio closure, and ACP parser exhaustion before resolving or propagating a child error, so captures are complete and callers can remove owned paths after either outcome. Snapshot and ordinary e2e suites share this process boundary; a test supplies only agent paths, cwd, environment overrides, and any permission policy.
- **`runScenario` (harness)** — boots the real agent bin as a subprocess via tsx (unbuilt, Loader path), drives it over ACP JSON-RPC stdio from a deterministic `input.json` script, tees raw stdout for the golden + purity check, and harvests every persisted session JSONL (parent + subagent children, primary-first) after a graceful stdin-EOF shutdown. Parameterized by `AgentUnderTest` (`binScript`, `configPath`, `tsconfigPath` — absolute paths; the subprocess cwd is a temp dir outside the repo). Startup failures preserve captured agent stderr in the rejected diagnostic.
- **Normalizers** — pure functions turning the two captured surfaces into stable text: `normalizeStdout` (JSON-RPC ids → first-seen sequence; UUIDs/cwd → tokens; doubles as the stdout-purity check), `normalizeSessionLog` (times zeroed, `seq` kept), `scrubSystemPrompts` (prompt text → `{{system}}`), `scrubToolSchemas` (schema bulk → `{{tools}}`), and `scrubRequestHeaders` (all header bulk → `{{system}}`/`{{tools}}`/`{{messagePrefix}}` outside each pin, structure kept — [pinned-header RFC](../../../docs/rfc/implemented/testing/2026-07-06-pin-request-header-content-in-one-scenario.md)).
- **`defineAcpSnapshotSuite` (factory)** — registers the whole describe/it tree for a scenario table: per-scenario golden + re-persisted-log compares, record/refresh fixture write-back, rejection of structured `UNKNOWN_TOOL` results, the per-header-class pin (`system-prompt.golden.md` plus `tool-schemas.golden.json`) with its live uniformity guard, and the fixture guard block (no orphan scenario dirs, required files present, exactly one pin per class, every JSONL prompt/schema-scrubbed, non-pinning fixtures fully header-scrubbed). Each scenario directory's `session.jsonl` plus contiguous `session.<n>.jsonl` siblings are the ordered primary/child inventory; the scenario table does not duplicate their count. Must be called at vitest collection time.
@@ -37,9 +38,9 @@ defineAcpSnapshotSuite({
A scenario booting a differently-composed tree sets its own `configPath` (an overlay whose basename still ends in `cordis.yml`, so the bin's replay swap finds the sibling `*cordis.snapshot.yml`) and, when that composition changes the request header, its own `headerClass` with its own pinning scenario — the acp-agent example's Code Mode and filesystem scenarios are templates. Each pinning directory stores the normalized full prompt sequence in generated `system-prompt.golden.md` and the corresponding full tool-schema sequence in generated `tool-schemas.golden.json`; `session.jsonl` stores `"system":"{{system}}","tools":"{{tools}}"` while retaining config, reason, and any model-visible prefix. A pin with legitimate mid-run header changes declares `expectedHeaderChanges`, which fixes the length of both sidecar sequences.
Examples use a `cordis.snapshot.yml` overlay with [`dsh-llm-replay`](../llm-replay/README.md). Recording calls the live model and updates model fixtures; keyless refresh replays those fixtures and updates derived stdout, session-log, prompt, and tool-schema snapshots. See the [snapshot RFC](../../../docs/rfc/implemented/testing/2026-06-19-acp-snapshot-tests.md).
The example also ships a `cordis.snapshot.yml` replay overlay next to its `cordis.yml` (the bin swaps them under `DSH_SNAPSHOT=replay` — [single-source replay config RFC](../../../docs/rfc/implemented/testing/2026-07-04-single-source-acp-replay-config.md)); replay fixtures are served by [`dsh-llm-replay`](../llm-replay/README.md), which this package points at via the `DSH_SNAPSHOT_*` env vars it sets on the child. `pnpm run test:snapshot:record` calls the live LLM and rewrites the recorded scenarios' model fixtures; `pnpm run test:snapshot:refresh` stays keyless, runs the replay overlay, and rewrites stdout, comparable session-log goldens, and each pin's prompt and tool-schema sidecars from the committed model scripts. Fixture roles, record/replay/refresh semantics, and scenario-table fields are documented on `Scenario` and in the [snapshot RFC](../../../docs/rfc/implemented/testing/2026-06-19-acp-snapshot-tests.md).
`suite.ts` imports Vitest, so use this package only inside a Vitest run. The ACP-specific script queues permission answers by stable option kind and maps them to current option ids; a missing answer cancels, while an unavailable kind fails the scenario after cancelling the agent request. It can also set session config options or assert that unknown ids and values are rejected in the transcript.
Constraints: `suite.ts` imports vitest, so the package entry is importable only inside a vitest run (the launcher, harness, and normalizers have no such dependency but ship from the same entry). ACP-specific by design — the launcher speaks the SDK's `ClientSideConnection`. Permission round-trips are scriptable: `InputScript.permissionAnswers` is a FIFO queue of option-kind selections (`allow_once`, `reject_once`, …) the client maps to the agent-issued `optionId` at answer time; an absent or exhausted queue answers `cancelled`, and a kind the request never offered rejects the run (the agent is answered `cancelled`, so a tolerant agent cannot absorb the scenario bug). Session config options are scriptable too: the `setConfigOption` step switches a knob over `session/set_config_option`, and `setConfigOptionExpectError` asserts the bridge rejects an unknown id or out-of-vocabulary value (the error frame stays in the transcript).
## Model Experience
packages/support/acp-snapshot/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@deepseek-ai/dsh-acp-snapshot",
"description": "ACP snapshot suite kit: real-subprocess scenario harness, golden normalizers, and the suite factory behind the keyless snapshot tier",
"description": "ACP test kit: shared subprocess launcher, snapshot scenario harness, golden normalizers, and suite factory",
"version": "0.0.1",
"private": true,
"type": "module",
packages/support/acp-snapshot/src/harness.ts
+140 -162
View File
@@ -1,59 +1,47 @@
/**
* Shared ACP snapshot subprocess harness. It boots the real agent bin through the Cordis
* loader, drives deterministic ACP JSON-RPC over stdio, captures protocol-pure stdout, and
* harvests persisted session logs after graceful shutdown. Normalization stays in
* `normalize.ts`; suite registration stays in `suite.ts`.
* Shared subprocess harness for ACP snapshot suites. A library module driven by
* the suite factory in ./suite.ts (and directly by harness-level specs); each
* example's `*.snapshot.ts` names its own agent-under-test paths.
*
* It boots the REAL agent bin subprocess via the cordis Loader (so the
* export-shape bug class stays guarded — see docs/postmortem/0001), drives it
* over real ACP JSON-RPC stdio with a deterministic input script, tees raw
* stdout (for the golden + a purity check) into an SDK `ClientSideConnection`,
* and — in record mode — harvests the persisted session JSONL after a graceful
* shutdown flush. The pure normalizers in ./normalize.ts turn the captured
* stdout frames and the session-log events into stable, snapshot-able text.
*
* See docs/rfc/implemented/testing/2026-06-19-acp-snapshot-tests.md.
*
* @module @deepseek-ai/dsh-acp-snapshot/harness
*/
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { cp, mkdtemp, readFile, readdir, rm } from 'node:fs/promises'
import { existsSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join, delimiter } from 'node:path'
import { Readable, Writable } from 'node:stream'
import {
ClientSideConnection,
ndJsonStream,
PROTOCOL_VERSION,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
} from '@agentclientprotocol/sdk'
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
import { launchAcpTestAgent, type AgentUnderTest, type LaunchedAcpTestAgent } from './launcher.ts'
export type { AgentUnderTest } from './launcher.ts'
/**
* The agent composition a scenario runs against: which bin to boot and which
* leaf config it loads. All paths are ABSOLUTE — the subprocess cwd is a temp
* dir outside the repo, so relative resolution would miss; a suite resolves
* them from its own `import.meta.url`.
*/
export interface AgentUnderTest {
/** The agent bin's SOURCE entry (e.g. `packages/examples/acp-demo/src/bin.ts`); the `lib` bin is derived from it. */
binScript: string
/** Explicit plain-Node entry for `lib` mode; intended for test fixtures outside a package `src/` tree. */
libBinScript?: string | undefined
/**
* The example's live `cordis.yml`. Under `DSH_SNAPSHOT=replay` the bin swaps
* it for the sibling `cordis.snapshot.yml` (the keyless replay overlay), so
* one path serves both modes.
*/
configPath: string
/**
* The repo-root tsconfig whose `paths` map resolves the unbuilt workspace
* imports in `src` mode (passed to the child as `TSX_TSCONFIG_PATH`). Ignored
* in `lib` mode, where the example resolves plugins through real `exports`.
*/
tsconfigPath: string
}
/**
* One step of a scenario's deterministic input script (`input.json`). The harness interprets
* these in order. `newSession` captures the server-issued (random) session id into a
* `{{sessionId}}` variable that later steps reference. `promptAndCancel` sends without awaiting,
* waits for the first streamed message, then cancels, making transcript order deterministic.
* One step of a scenario's deterministic input script (`input.json`). The
* harness interprets these in order. `newSession` captures the server-issued
* (random) session id into a `{{sessionId}}` variable that later steps
* reference, since a committed file cannot know the id in advance.
*
* `promptAndCancel` sends a prompt WITHOUT awaiting its response, waits until
* the client observes the first streamed `agent_message_chunk` (so the emitted
* frames deterministically precede the cancellation), then cancels the turn —
* the only way to exercise a cancel deterministically (a plain `prompt` step
* awaits the response, which a cancel/hang scenario would block on forever).
*/
export type InputStep =
| { op: 'initialize'; terminalOutput?: boolean }
@@ -70,9 +58,16 @@ export type InputStep =
export interface InputScript {
steps: InputStep[]
/**
* FIFO permission answers selected by stable option kind; the harness maps each kind to the
* agent-issued option id. Exhaustion cancels, while a kind the agent did not offer fails the
* scenario.
* Ordered answers for the agent's `session/request_permission` round-trips,
* consumed FIFO — the Nth request gets the Nth answer. Each answer selects
* by option KIND: option ids are agent-issued randoms a committed script
* cannot know, while kinds are the ACP-stable vocabulary, so the client maps
* kind → the offered `optionId` at answer time. A request beyond the queue
* (or with no queue at all) is answered `cancelled` — the stub behavior a
* scenario without approvals relies on. A scripted kind the request does
* not offer REJECTS the run: the scenario scripted an impossible click,
* and {@link runScenario} throws once the in-flight step settles (the
* agent itself just sees `cancelled`, so it cannot absorb the bug).
*/
permissionAnswers?: PermissionAnswer[]
}
@@ -165,93 +160,47 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
// Fixed path length: spill-policy budgets the preview against the REAL path
// before stdout normalization, so tmpdir() length differences churn goldens.
const spillRoot = '/tmp/dsh-acp-snapshot-spill'
// Everything past the temp-dir creation runs under a try/finally that always
// removes both dirs — so a failure in workspace seeding, spawn, or any step
// never leaks them (the "e2e tests own their resources" rule).
let child: ChildProcessWithoutNullStreams | undefined
// Everything past the temp-dir creation is followed by failure-safe cleanup,
// so a failure in workspace seeding, spawn, or any step never leaks resources.
let launched: LaunchedAcpTestAgent | undefined
let sessionId: string | undefined
let sessionLogs: HarvestedLog[] = []
const rawBuffers: Buffer[] = []
const stderrChunks: string[] = []
try {
const outcome = await (async (): Promise<RunResult> => {
// Seed the workspace if the scenario ships one (a file the agent reads/edits).
// Copied into the temp cwd so the agent's bash tools see it; the goldens
// normalize the cwd, so the seeded paths stay stable across runs.
if (opts.workspaceDir !== undefined && existsSync(opts.workspaceDir)) {
await cp(opts.workspaceDir, cwd, { recursive: true })
}
// Boot the agent in the environment's mode (DSH_EXAMPLE_MODE): `src` runs the
// source bin under tsx with the paths map; `lib` runs the built bin under plain
// Node, resolving plugins through the example's workspace node_modules → lib.
const launch = resolveExampleLaunch({
srcBin: opts.agent.binScript,
libBin: opts.agent.libBinScript,
configArgs: ['--config', opts.configPath ?? opts.agent.configPath],
tsconfigPath: opts.agent.tsconfigPath,
env: {
DSH_SNAPSHOT: opts.mode,
DSH_SNAPSHOT_FILE: opts.fixtureFile,
DSH_SNAPSHOT_SESSIONS_ROOT: sessionsRoot,
DSH_SNAPSHOT_SPILL_ROOT: spillRoot,
DSH_HOME: join(cwd, '.dsh'),
DSH_AGENTS_HOME: join(cwd, '.agents'),
...opts.overrideFile !== undefined ? { DSH_SNAPSHOT_OVERRIDE: opts.overrideFile } : {},
...opts.childFiles !== undefined && opts.childFiles.length > 0
? { DSH_SNAPSHOT_CHILD_FILES: opts.childFiles.join(delimiter) }
: {},
},
})
child = spawn(
launch.command,
launch.args,
{ cwd, env: { ...process.env, ...launch.env }, stdio: ['pipe', 'pipe', 'pipe'] },
)
child.stderr.setEncoding('utf8')
child.stderr.on('data', (c: string) => stderrChunks.push(c))
// Tee the same raw bytes to the golden and SDK client. Decode once at the end so a UTF-8
// sequence split across stream chunks cannot corrupt the transcript.
const passthrough = new Readable({ read() {} })
child.stdout.on('data', (buf: Buffer) => {
rawBuffers.push(buf)
passthrough.push(buf)
})
child.stdout.on('end', () => passthrough.push(null))
const stream = ndJsonStream(
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
Readable.toWeb(passthrough) as ReadableStream<Uint8Array>,
)
// Watcher so a step can block until the client OBSERVES a particular
// session/update — used by promptAndCancel to pin frame order (send cancel
// only after the streamed agent_message_chunk has arrived, so those frames
// deterministically precede the cancelled prompt response).
const updateWaiters: { match: (u: SessionNotification['update']) => boolean; resolve: () => void }[] = []
const waitForUpdate = (match: (u: SessionNotification['update']) => boolean): Promise<void> =>
new Promise<void>(resolve => updateWaiters.push({ match, resolve }))
const env: NodeJS.ProcessEnv = {
DSH_SNAPSHOT: opts.mode,
DSH_SNAPSHOT_FILE: opts.fixtureFile,
DSH_SNAPSHOT_SESSIONS_ROOT: sessionsRoot,
DSH_SNAPSHOT_SPILL_ROOT: spillRoot,
DSH_HOME: join(cwd, '.dsh'),
DSH_AGENTS_HOME: join(cwd, '.agents'),
...opts.overrideFile !== undefined ? { DSH_SNAPSHOT_OVERRIDE: opts.overrideFile } : {},
...opts.childFiles !== undefined && opts.childFiles.length > 0
? { DSH_SNAPSHOT_CHILD_FILES: opts.childFiles.join(delimiter) }
: {},
}
// Permission answers are consumed FIFO across the whole run; exhaustion
// falls back to `cancelled` so approval-free scenarios keep the plain stub.
const permissionQueue = [...input.permissionAnswers ?? []]
// A callback throw would become only an RPC error the agent could absorb. Record an
// impossible permission choice here, answer cancelled, and fail the outer scenario.
// A scenario bug detected inside a client callback (a scripted permission
// kind the agent never offered). It cannot fail the run from in there: a
// callback throw only becomes a JSON-RPC error RESPONSE to the agent, and
// a tolerant agent treats that as a denial and carries on — the run (or
// worse, a record) would absorb the impossible click silently. So the
// callback answers `cancelled` (a well-defined path for the agent),
// captures the error here, and the step loop fails the run on it.
let scriptError: Error | undefined
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
for (let i = updateWaiters.length - 1; i >= 0; i--) {
const waiter = updateWaiters[i]
// The index is always in-bounds (i only decreases; splice removes at
// i, so lower entries stay valid); the guard satisfies
// noUncheckedIndexedAccess.
/* v8 ignore next 1 -- unreachable in-bounds guard, see above */
if (waiter === undefined) continue
if (waiter.match(params.update)) {
updateWaiters.splice(i, 1)
waiter.resolve()
}
}
return Promise.resolve()
},
launched = launchAcpTestAgent({
agent: opts.agent,
cwd,
...opts.configPath !== undefined ? { configPath: opts.configPath } : {},
env,
requestPermission(params: RequestPermissionRequest): Promise<RequestPermissionResponse> {
const answer = permissionQueue.shift()
if (answer === undefined) return Promise.resolve({ outcome: { outcome: 'cancelled' } })
@@ -269,10 +218,12 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
return Promise.resolve({ outcome: { outcome: 'selected', optionId: option.optionId } })
},
})
const client = new ClientSideConnection(makeClient, stream)
const active = launched
await active.spawned
const { client } = active
for (const step of input.steps) {
await runStep(client, step, cwd, waitForUpdate, () => sessionId, (id) => { sessionId = id })
await runStep(client, step, cwd, match => active.waitForUpdate(match), () => sessionId, (id) => { sessionId = id })
// A permission exchange happens while a step's request is in flight, so
// by the time the step settles any script bug it exposed is captured —
// fail the run HERE, as a harness error, rather than hoping the agent's
@@ -281,35 +232,57 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
}
// Done driving: close stdin so the server disposes gracefully (flushing
// persistence) and exits. Then await exit so the harvested log is complete.
child.stdin.end()
await waitForExit(child)
await active.close()
// Harvest EVERY persisted log (parent + any subagent children) while the
// temp dirs still exist, ordered primary-first.
sessionLogs = await harvestSessionLogs(sessionsRoot)
} catch (error: unknown) {
const stderr = stderrChunks.join('')
if (stderr === '') throw error
throw new Error(`snapshot-harness: scenario failed: ${String(error)}\nagent stderr:\n${stderr}`, { cause: error })
} finally {
// Failure-safe teardown: kill a still-running child and drop the temp dirs
// even if seeding/spawn/a step/harvest threw, so a flaky run never leaks a
// process or dir. `child` is undefined only if spawn itself threw.
if (child !== undefined && child.exitCode === null && child.signalCode === null) {
child.kill('SIGKILL')
await waitForExit(child)
return {
rawStdout: launched.rawStdout(),
stderr: launched.stderr(),
cwd,
...sessionId !== undefined ? { sessionId } : {},
sessionLogs,
}
await rm(cwd, { recursive: true, force: true })
await rm(sessionsRoot, { recursive: true, force: true })
await rm(spillRoot, { recursive: true, force: true })
}
})().then(
value => ({ status: 'fulfilled', value } as const),
(error: unknown) => {
const stderr = launched?.stderr() ?? ''
return {
status: 'rejected',
error: stderr === ''
? error
: new Error(`snapshot-harness: scenario failed: ${String(error)}\nagent stderr:\n${stderr}`, { cause: error }),
} as const
},
)
return {
rawStdout: Buffer.concat(rawBuffers).toString('utf8'),
stderr: stderrChunks.join(''),
cwd,
...sessionId !== undefined ? { sessionId } : {},
sessionLogs,
// Failure-safe teardown: wait for a still-running child, then attempt every
// owned-path removal even when an earlier cleanup rejects. Report every
// teardown failure alongside a scenario failure so neither orthogonal
// outcome hides the other.
const cleanupResults: PromiseSettledResult<unknown>[] = []
const cleanup = async (action: () => Promise<unknown>): Promise<void> => {
cleanupResults.push(...await Promise.allSettled([action()]))
}
/* v8 ignore next 1 -- launch itself can only throw on a defensive synchronous spawn API failure */
await cleanup(() => launched?.close('SIGKILL') ?? Promise.resolve())
await cleanup(() => rm(cwd, { recursive: true, force: true }))
await cleanup(() => rm(sessionsRoot, { recursive: true, force: true }))
await cleanup(() => rm(spillRoot, { recursive: true, force: true }))
const cleanupFailures = cleanupResults
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
.map(result => result.reason as unknown)
if (cleanupFailures.length > 0) {
throw new AggregateError(
outcome.status === 'rejected' ? [outcome.error, ...cleanupFailures] : cleanupFailures,
outcome.status === 'rejected'
? 'snapshot scenario and cleanup failed'
: 'snapshot cleanup failed',
)
}
if (outcome.status === 'rejected') throw outcome.error
return outcome.value
}
/** Drive one input step over the client connection. */
@@ -317,7 +290,7 @@ async function runStep(
client: ClientSideConnection,
step: InputStep,
cwd: string,
waitForUpdate: (match: (u: SessionNotification['update']) => boolean) => Promise<void>,
waitForUpdate: (match: (u: SessionNotification['update']) => boolean) => Promise<SessionNotification['update']>,
getSessionId: () => string | undefined,
setSessionId: (id: string) => void,
): Promise<void> {
@@ -334,8 +307,10 @@ async function runStep(
return
}
case 'newSessionExpectError': {
// The bridge rejects a session/new that widens the workspace scope (non-empty
// additionalDirectories / mcpServers — unimplemented).
// The bridge rejects a session/new that widens the workspace scope
// (non-empty additionalDirectories / mcpServers — unimplemented). The SDK
// surfaces that as a rejected RPC; swallow it so the run completes and the
// error frame is captured in the transcript.
await client.newSession({
cwd,
mcpServers: [],
@@ -355,8 +330,10 @@ async function runStep(
case 'promptExpectError': {
const sessionId = getSessionId()
if (sessionId === undefined) throw new Error('snapshot-harness: promptExpectError before newSession')
// The model fails this turn (a recorded provider error), so the bridge answers the prompt
// with a JSON-RPC error and the SDK rejects.
// The model fails this turn (a recorded provider error), so the bridge
// answers the prompt with a JSON-RPC error and the SDK rejects. That
// rejection IS the expected editor experience — swallow it so the run
// completes and the stdout transcript (the error frame) is captured.
await client.prompt({ sessionId, prompt: [{ type: 'text', text: step.text }] })
.then(() => { throw new Error('snapshot-harness: expected the prompt to fail but it succeeded') },
() => { /* expected: the turn failed and the bridge returned an error */ })
@@ -365,8 +342,13 @@ async function runStep(
case 'promptAndCancel': {
const sessionId = getSessionId()
if (sessionId === undefined) throw new Error('snapshot-harness: promptAndCancel before newSession')
// A hang fixture never resolves alone. Wait for its streamed chunk before cancellation
// so updates deterministically precede the cancelled prompt response.
// Dispatch the prompt WITHOUT awaiting (a hang fixture never resolves on
// its own). To pin frame order deterministically, wait until the client
// has OBSERVED the hang's streamed agent_message_chunk before cancelling —
// so those update frames always precede the cancelled prompt response in
// the transcript (without this, the late chunk and the response race).
// Then cancel and await the prompt, which the bridge settles as
// `cancelled` once the abort propagates.
const promptDone = client.prompt({ sessionId, prompt: [{ type: 'text', text: step.text }] })
await waitForUpdate(u => u.sessionUpdate === 'agent_message_chunk')
await client.cancel({ sessionId })
@@ -402,16 +384,6 @@ async function runStep(
}
}
/** Resolve once the child process exits (any code/signal). */
function waitForExit(child: ChildProcessWithoutNullStreams): Promise<void> {
// Race guard: both call sites run within one synchronous frame of
// stdin.end()/kill(), so the exit event cannot have been delivered yet;
// kept for any future caller that awaits in between.
/* v8 ignore next 1 -- unreachable race guard, see above */
if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve()
return new Promise<void>(resolve => child.once('exit', () => { resolve() }))
}
/**
* Harvest EVERY persisted `.jsonl` session log under a sessions root, parse each
* header line, and return them ordered primary-first: the top-level session (no
@@ -452,8 +424,14 @@ async function harvestSessionLogs(root: string): Promise<HarvestedLog[]> {
})
}
}
// Match replay fixture assignment: primary first, then children by creation time, with id as
// a deterministic collision tiebreaker.
// Primary (no parentSession) first, then children by ascending createdAt. A
// scenario has exactly one top-level session. In the synchronous cut sibling
// children are created strictly sequentially, so their createdAt values are
// strictly ordered; the recordedId tiebreak only keeps a degenerate
// same-millisecond collision (unreachable here) deterministic. This harvest
// order must match the replay load order in dsh-llm-replay's loadSessionScripts
// so session.<n>.jsonl maps to the same child on record and replay — replay
// re-sorts childFiles by the same key, so the two stay consistent.
logs.sort((a, b) => {
const ap = a.parentSession === undefined ? 0 : 1
const bp = b.parentSession === undefined ? 0 : 1
packages/support/acp-snapshot/src/index.ts
+20 -4
View File
@@ -1,13 +1,23 @@
/**
* ACP snapshot suite kit: subprocess scenario harness, pure golden normalizers, and the Vitest
* suite factory behind `pnpm run test:snapshot`. Because this entry exports `suite.ts`, importing
* it requires a Vitest run.
* ACP snapshot suite kit — the shared machinery behind the keyless snapshot
* tier (`pnpm run test:snapshot`). Four layers, composable per example: the
* shared subprocess/client launcher ({@link launchAcpTestAgent}), the scripted
* scenario harness ({@link runScenario}), the pure golden normalizers
* ({@link normalizeStdout} / {@link normalizeSessionLog} /
* {@link scrubRequestHeaders} / {@link scrubSystemPrompts}), and the suite
* factory ({@link defineAcpSnapshotSuite}) that registers a scenario table as a
* full describe/it tree. Ordinary ACP e2e tests can use the launcher directly;
* an example's `*.snapshot.ts` supplies only its {@link AgentUnderTest} paths,
* snapshots directory, and {@link Scenario} table.
*
* NOTE: ./suite.ts imports vitest, so this package is importable only inside a
* vitest run — a support-tier constraint stated in the README.
*
* @module @deepseek-ai/dsh-acp-snapshot
*/
export {
runScenario,
type AgentUnderTest,
type HarvestedLog,
type InputScript,
type InputStep,
@@ -15,6 +25,12 @@ export {
type RunOptions,
type RunResult,
} from './harness.ts'
export {
launchAcpTestAgent,
type AcpTestLaunchOptions,
type AgentUnderTest,
type LaunchedAcpTestAgent,
} from './launcher.ts'
export {
normalizeSessionLog,
normalizeStdout,
packages/support/acp-snapshot/src/launcher.ts
+276
View File
@@ -0,0 +1,276 @@
/**
* Shared launcher for ACP tests that drive an agent subprocess over JSON-RPC
* stdio. It owns source-or-built launch resolution, workspace environment,
* stdout tee, SDK client, update collection, permission fallback, and process
* shutdown so e2e and snapshot suites do not each reconstruct that boundary.
*
* @module @deepseek-ai/dsh-acp-snapshot/launcher
*/
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { join } from 'node:path'
import { Readable, Writable } from 'node:stream'
import {
ClientSideConnection,
ndJsonStream,
type Agent as AcpAgent,
type Client,
type RequestPermissionRequest,
type RequestPermissionResponse,
type SessionNotification,
} from '@agentclientprotocol/sdk'
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
/** The source/built agent entry, leaf config, and workspace tsconfig an ACP test boots. */
export interface AgentUnderTest {
/** The agent source bin entry (for example `packages/examples/acp-demo/src/bin.ts`). */
binScript: string
/** Explicit built-mode entry for fixtures whose source path is not under `src/`. */
libBinScript?: string | undefined
/** The leaf `cordis.yml` loaded by the bin. */
configPath: string
/** The repo tsconfig whose paths resolve unbuilt workspace imports. */
tsconfigPath: string
}
/** Options for one ACP test subprocess. */
export interface AcpTestLaunchOptions {
/** The agent composition to boot. */
agent: AgentUnderTest
/** Process cwd and default session-home root. */
cwd: string
/** Alternate leaf config for this launch. */
configPath?: string
/** Extra environment values layered over the parent environment. */
env?: NodeJS.ProcessEnv
/** Permission handler; omitted requests fail closed as `cancelled`. */
requestPermission?: (params: RequestPermissionRequest) => Promise<RequestPermissionResponse>
}
/** A running ACP test process and its captured client-side surfaces. */
export interface LaunchedAcpTestAgent {
/** The child process, exposed for process-level assertions. */
child: ChildProcessWithoutNullStreams
/** Resolve when the OS spawns the child; reject with its asynchronous spawn failure. */
spawned: Promise<void>
/** The SDK connection backed by the child's stdio. */
client: ClientSideConnection
/** Session updates in receive order. */
updates: SessionNotification['update'][]
/** Decode all stdout bytes captured so far. */
rawStdout(): string
/** Decode all stderr chunks captured so far. */
stderr(): string
/** Resolve when a future session update matches the predicate. */
waitForUpdate(match: (update: SessionNotification['update']) => boolean): Promise<SessionNotification['update']>
/** Close the process and drain its streams and callbacks; rejects promptly if fallback termination is refused. */
close(signal?: NodeJS.Signals): Promise<void>
}
/**
* Boot an ACP agent subprocess and connect an SDK client to its stdio.
*
* @param options Agent paths, cwd, environment, and optional permission handler.
* @returns The running process, connected client, captures, and shutdown handle.
*/
export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTestAgent {
const { agent, cwd } = options
const launch = resolveExampleLaunch({
srcBin: agent.binScript,
libBin: agent.libBinScript,
configArgs: ['--config', options.configPath ?? agent.configPath],
tsconfigPath: agent.tsconfigPath,
env: {
...options.env,
DSH_HOME: join(cwd, '.dsh'),
DSH_AGENTS_HOME: join(cwd, '.agents'),
},
})
const child = spawn(
launch.command,
launch.args,
{
cwd,
env: { ...process.env, ...launch.env },
stdio: ['pipe', 'pipe', 'pipe'],
},
)
// A spawn-level failure is an asynchronous `error` event. Observe it in the
// same tick as spawn so a missing cwd or OS rejection cannot crash the test
// runner, then make startup and shutdown surface the original error.
// Keep observing after the first error: a fallback kill attempted during
// shutdown may itself report another process error, which must not become an
// unhandled EventEmitter error after the promise has already settled.
const childFailure = new Promise<Error>(resolve => child.on('error', resolve))
const spawned = Promise.race([
new Promise<void>(resolve => child.once('spawn', resolve)),
childFailure.then((error): never => { throw error }),
])
// `spawned` is public and close() also awaits it, but a caller may ignore both.
// Keep that misuse from turning the already-observed child error into an
// unhandled promise rejection.
void spawned.catch(() => undefined)
const stderrChunks: string[] = []
child.stderr.setEncoding('utf8')
child.stderr.on('data', (chunk: string) => stderrChunks.push(chunk))
const rawBuffers: Buffer[] = []
const passthrough = new Readable({ read() {} })
const updates: SessionNotification['update'][] = []
const updateWaiters: {
match: (update: SessionNotification['update']) => boolean
resolve: (update: SessionNotification['update']) => void
reject: (reason: unknown) => void
}[] = []
let updateStreamFailure: Error | undefined
const closeUpdateStream = (): void => {
if (updateStreamFailure !== undefined) return
updateStreamFailure = new Error('ACP test agent update stream closed before a matching session update arrived')
for (const waiter of updateWaiters.splice(0)) waiter.reject(updateStreamFailure)
}
child.stdout.on('data', (buffer: Buffer) => {
rawBuffers.push(buffer)
passthrough.push(buffer)
})
child.stdout.on('end', () => {
passthrough.push(null)
})
const stream = ndJsonStream(
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
Readable.toWeb(passthrough) as ReadableStream<Uint8Array>,
)
const inFlightClientCallbacks = new Set<Promise<unknown>>()
const trackClientCallback = <T>(callback: () => T | PromiseLike<T>): Promise<T> => {
const pending = Promise.resolve().then(callback)
inFlightClientCallbacks.add(pending)
const untrack = (): void => { inFlightClientCallbacks.delete(pending) }
void pending.then(untrack, untrack)
return pending
}
const requestPermission = options.requestPermission
?? (() => Promise.resolve({ outcome: { outcome: 'cancelled' as const } }))
const makeClient = (_agent: AcpAgent): Client => ({
sessionUpdate(params: SessionNotification): Promise<void> {
return trackClientCallback(() => {
updates.push(params.update)
for (let index = updateWaiters.length - 1; index >= 0; index--) {
const waiter = updateWaiters[index]
/* v8 ignore next 1 -- index is bounded by the array length */
if (waiter === undefined) continue
let matches: boolean
try {
matches = waiter.match(params.update)
} catch (error: unknown) {
updateWaiters.splice(index, 1)
waiter.reject(error)
continue
}
if (!matches) continue
updateWaiters.splice(index, 1)
waiter.resolve(params.update)
}
})
},
requestPermission: params => trackClientCallback(() => requestPermission(params)),
})
const client = new ClientSideConnection(makeClient, stream)
// `exit` only reports the parent process's status. Descendants may retain
// inherited stdout/stderr handles and buffered ACP frames may still be
// crossing the SDK parser. Node's `close` follows stdio closure; the SDK's
// `closed` follows parser exhaustion. Capture both eagerly so a caller that
// invokes close after process exit still joins the complete drain boundary.
const stdioClosed = new Promise<void>(resolve => child.once('close', () => { resolve() }))
const drained = Promise.all([stdioClosed, client.closed]).then(async () => {
// The ACP SDK's readable loop dispatches client callbacks without awaiting
// them. Once `closed` settles no new callbacks can start, but callbacks
// already in flight still belong to this launch's teardown boundary.
while (inFlightClientCallbacks.size > 0) {
await Promise.allSettled([...inFlightClientCallbacks])
}
})
// A caller may await a pending update without calling close(). Make natural
// stream exhaustion terminal for those waiters too, but only after the
// parser has dispatched every buffered frame.
void client.closed.then(closeUpdateStream)
return {
child,
spawned,
client,
updates,
rawStdout: () => Buffer.concat(rawBuffers).toString('utf8'),
stderr: () => stderrChunks.join(''),
waitForUpdate(match): Promise<SessionNotification['update']> {
if (updateStreamFailure !== undefined) return Promise.reject(updateStreamFailure)
return new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject }))
},
async close(signal?: NodeJS.Signals): Promise<void> {
try {
await spawned
} catch (error: unknown) {
await drained
closeUpdateStream()
throw error
}
if (!isRunning(child)) {
await drained
closeUpdateStream()
return
}
const exited = waitForExit(child)
if (signal === undefined) child.stdin.end()
else child.kill(signal)
const failure = await Promise.race([
exited.then((): undefined => undefined),
childFailure,
])
if (failure === undefined) {
await drained
closeUpdateStream()
return
}
// An `error` after spawn is not an exit edge: in particular, a failed
// signal can leave the subprocess live. Force termination, await the
// already-observed exit edge, and only then propagate the child error so
// callers may safely remove cwd/session resources after close rejects.
const fallbackError = Promise.withResolvers<Error>()
const observeFallbackError = (error: Error): void => { fallbackError.resolve(error) }
child.once('error', observeFallbackError)
if (!child.kill('SIGKILL')) {
child.off('error', observeFallbackError)
closeUpdateStream()
throw new AggregateError(
[failure, new Error('Fallback SIGKILL was not accepted by the child process')],
'ACP test agent failed and fallback termination was refused',
)
}
const fallbackFailure = await Promise.race([
exited.then((): undefined => undefined),
fallbackError.promise,
])
child.off('error', observeFallbackError)
if (fallbackFailure !== undefined) {
closeUpdateStream()
throw new AggregateError(
[failure, fallbackFailure],
'ACP test agent failed and fallback termination was refused',
)
}
await drained
closeUpdateStream()
throw failure
},
}
}
/** Resolve once a running child exits. */
function waitForExit(child: ChildProcessWithoutNullStreams): Promise<void> {
return new Promise<void>(resolve => child.once('exit', () => { resolve() }))
}
/** Whether the child still lacks either OS termination marker. */
function isRunning(child: ChildProcessWithoutNullStreams): boolean {
return child.exitCode === null && child.signalCode === null
}
packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts
Vendored
+21
View File
@@ -6,6 +6,7 @@
import { mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
import { readdirSync } from 'node:fs'
import { spawn } from 'node:child_process'
import { dirname, join } from 'node:path'
import { randomUUID } from 'node:crypto'
import { createInterface } from 'node:readline'
@@ -40,6 +41,8 @@ interface Behavior {
echoWorkspace?: boolean
/** Write a line to stderr on boot (spec-side stderr-capture assertions). */
stderrNote?: string
/** Let a short-lived descendant retain stdio and emit one final ACP update plus stderr line after this parent exits. */
lateInheritedOutput?: boolean
/** Session logs to persist on stdin EOF. */
logs?: ScriptedLog[]
/** Leave a stray FILE directly under the sessions root (harvest must skip it). */
@@ -247,6 +250,24 @@ function flushLogsAndExit(): void {
writeFileSync(join(sessionsRoot, 'bucket-noise', 'notes.txt'), 'not a session log\n')
}
if (behavior.deleteSessionsRoot === true) rmSync(sessionsRoot, { recursive: true, force: true })
if (behavior.lateInheritedOutput === true) {
const frame = JSON.stringify({
jsonrpc: '2.0',
method: 'session/update',
params: {
sessionId,
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'late inherited stdout' },
},
},
})
const code = [
`setTimeout(() => process.stdout.write(${JSON.stringify(`${frame}\n`)}), 50)`,
`setTimeout(() => process.stderr.write(${JSON.stringify('late inherited stderr\n')}), 75)`,
].join(';')
spawn(process.execPath, ['-e', code], { stdio: ['ignore', 1, 2] }).unref()
}
process.exit(0)
}
packages/support/acp-snapshot/tests/harness.spec.ts
+248 -2
View File
@@ -1,13 +1,34 @@
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
import { once } from 'node:events'
import { tmpdir } from 'node:os'
import { delimiter, join } from 'node:path'
import { fileURLToPath } from 'node:url'
import { afterAll, describe, expect, it } from 'vitest'
import { afterAll, describe, expect, it, vi } from 'vitest'
import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
import { runScenario, type AgentUnderTest, type InputStep } from '../src/harness.ts'
import { launchAcpTestAgent } from '../src/launcher.ts'
const fsControl = vi.hoisted(() => ({ cleanupFailure: undefined as Error | undefined }))
vi.mock('node:fs/promises', async (importOriginal) => {
const actual = await importOriginal<typeof import('node:fs/promises')>()
return {
...actual,
async rm(...args: Parameters<typeof actual.rm>): Promise<void> {
if (String(args[0]).includes('acp-snap-cwd-') && fsControl.cleanupFailure !== undefined) {
const failure = fsControl.cleanupFailure
fsControl.cleanupFailure = undefined
await actual.rm(...args)
throw failure
}
await actual.rm(...args)
},
}
})
/**
* Unit tests for the subprocess harness, driven through the REAL spawn path
* (tsx loader, temp cwd, env plumbing) against the scripted fake ACP bin in
* (mode-aware launcher, temp cwd, env plumbing) against the scripted fake ACP bin in
* ./fixtures/fake-acp-agent.ts. Each case writes a `behavior.json` next to a
* throwaway fixture path; the fake bin echoes observable facts (env, seeded
* workspace, permission outcomes) into `agent_message_chunk` text, so the
@@ -40,6 +61,181 @@ async function scenario(behavior: object): Promise<{ dir: string; fixtureFile: s
const boot: InputStep[] = [{ op: 'initialize' }, { op: 'newSession' }]
describe('runScenario', () => {
it('surfaces an asynchronous child spawn failure through startup and close', async () => {
const { dir } = await scenario({})
const launched = launchAcpTestAgent({ agent: AGENT, cwd: join(dir, 'missing') })
let stdioClosed = false
let clientClosed = false
launched.child.once('close', () => { stdioClosed = true })
void launched.client.closed.then(
() => { clientClosed = true },
() => { clientClosed = true },
)
await expect(launched.spawned).rejects.toMatchObject({ code: 'ENOENT' })
await expect(launched.close()).rejects.toMatchObject({ code: 'ENOENT' })
expect(stdioClosed).toBe(true)
expect(clientClosed).toBe(true)
})
it('centralizes ACP boot, captures, updates, fail-closed permissions, and shutdown', { timeout: 20_000 }, async () => {
const { dir, fixtureFile } = await scenario({ permissionProbe: true, echoEnv: true, stderrNote: 'launcher stderr' })
const sessionsRoot = await mkdtemp(join(tmpdir(), 'acp-launcher-sessions-'))
tempDirs.push(sessionsRoot)
const launched = launchAcpTestAgent({
agent: AGENT,
cwd: dir,
configPath: AGENT.configPath,
env: {
DSH_SNAPSHOT: 'replay',
DSH_SNAPSHOT_FILE: fixtureFile,
DSH_SNAPSHOT_SESSIONS_ROOT: sessionsRoot,
},
})
await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await launched.client.newSession({ cwd: dir, mcpServers: [] })
const nextChunk = launched.waitForUpdate(update => update.sessionUpdate === 'agent_message_chunk')
const predicateFailure = new Error('predicate failed')
const failedPredicate = launched.waitForUpdate(() => { throw predicateFailure })
.catch((error: unknown): unknown => error)
await launched.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
expect(await failedPredicate).toBe(predicateFailure)
expect((await nextChunk).sessionUpdate).toBe('agent_message_chunk')
expect(launched.updates.some(update => update.sessionUpdate === 'agent_message_chunk')).toBe(true)
expect(launched.rawStdout()).toContain('permission:{\\"outcome\\":\\"cancelled\\"}')
expect(launched.stderr()).toContain('launcher stderr')
const unmatched = expect(launched.waitForUpdate(() => false)).rejects.toThrow(/update stream closed/)
await launched.close()
await unmatched
await expect(launched.waitForUpdate(() => true)).rejects.toThrow(/update stream closed/)
await launched.close('SIGKILL')
// The minimal shape needs no environment or config override.
const minimal = launchAcpTestAgent({ agent: AGENT, cwd: dir })
await minimal.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const childFailure = new Error('child process failed')
let exited = false
minimal.child.once('exit', () => { exited = true })
minimal.child.emit('error', childFailure)
await expect(minimal.close('SIGTERM')).rejects.toBe(childFailure)
// close rejects only after the fallback SIGKILL has produced an exit edge.
expect(exited).toBe(true)
})
it('waits for inherited stdio and buffered ACP parsing after the parent exits', { timeout: 20_000 }, async () => {
const { dir, fixtureFile } = await scenario({ lateInheritedOutput: true })
const launched = launchAcpTestAgent({
agent: AGENT,
cwd: dir,
env: { DSH_SNAPSHOT_FILE: fixtureFile },
})
await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
await launched.client.newSession({ cwd: dir, mcpServers: [] })
const lateUpdate = launched.waitForUpdate(update =>
update.sessionUpdate === 'agent_message_chunk'
&& update.content.type === 'text'
&& update.content.text === 'late inherited stdout')
await launched.close()
await expect(lateUpdate).resolves.toMatchObject({ sessionUpdate: 'agent_message_chunk' })
expect(launched.rawStdout()).toContain('late inherited stdout')
expect(launched.stderr()).toContain('late inherited stderr')
})
it('rejects promptly when fallback termination is refused', async () => {
const { dir } = await scenario({})
const launched = launchAcpTestAgent({ agent: AGENT, cwd: dir })
await launched.spawned
const childFailure = Object.assign(new Error('signal refused'), { code: 'EPERM' })
const originalKill = launched.child.kill.bind(launched.child)
const kill = vi.spyOn(launched.child, 'kill').mockReturnValue(false)
const closed = new Promise<void>(resolve => launched.child.once('close', () => { resolve() }))
try {
launched.child.emit('error', childFailure)
const rejection = await launched.close('SIGTERM').catch((error: unknown): unknown => error)
expect(rejection).toBeInstanceOf(AggregateError)
expect(rejection).toMatchObject({
message: 'ACP test agent failed and fallback termination was refused',
errors: [
childFailure,
expect.objectContaining({ message: 'Fallback SIGKILL was not accepted by the child process' }),
],
})
expect(kill).toHaveBeenNthCalledWith(1, 'SIGTERM')
expect(kill).toHaveBeenNthCalledWith(2, 'SIGKILL')
} finally {
kill.mockRestore()
originalKill('SIGKILL')
await closed
}
})
it('rejects promptly when fallback termination emits an error', async () => {
const { dir } = await scenario({})
const launched = launchAcpTestAgent({ agent: AGENT, cwd: dir })
await launched.spawned
const childFailure = Object.assign(new Error('signal refused'), { code: 'EPERM' })
const fallbackFailure = Object.assign(new Error('fallback signal refused'), { code: 'EPERM' })
const originalKill = launched.child.kill.bind(launched.child)
const kill = vi.spyOn(launched.child, 'kill').mockImplementation((signal) => {
if (signal === 'SIGKILL') queueMicrotask(() => launched.child.emit('error', fallbackFailure))
return signal === 'SIGKILL'
})
const closed = new Promise<void>(resolve => launched.child.once('close', () => { resolve() }))
try {
launched.child.emit('error', childFailure)
const rejection = await launched.close('SIGTERM').catch((error: unknown): unknown => error)
expect(rejection).toBeInstanceOf(AggregateError)
expect(rejection).toMatchObject({
message: 'ACP test agent failed and fallback termination was refused',
errors: [childFailure, fallbackFailure],
})
expect(kill).toHaveBeenNthCalledWith(1, 'SIGTERM')
expect(kill).toHaveBeenNthCalledWith(2, 'SIGKILL')
} finally {
kill.mockRestore()
originalKill('SIGKILL')
await closed
}
})
it('waits for in-flight client callbacks after the ACP stream closes', { timeout: 20_000 }, async () => {
const { dir, fixtureFile } = await scenario({ permissionProbe: true })
let releasePermission: (() => void) | undefined
const permissionReleased = new Promise<void>((resolve) => { releasePermission = resolve })
let markPermissionStarted: (() => void) | undefined
const permissionStarted = new Promise<void>((resolve) => { markPermissionStarted = resolve })
let permissionFinished = false
const launched = launchAcpTestAgent({
agent: AGENT,
cwd: dir,
env: { DSH_SNAPSHOT_FILE: fixtureFile },
async requestPermission() {
markPermissionStarted?.()
await permissionReleased
permissionFinished = true
return { outcome: { outcome: 'cancelled' } }
},
})
await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
const { sessionId } = await launched.client.newSession({ cwd: dir, mcpServers: [] })
void launched.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => undefined)
await permissionStarted
const childClosed = once(launched.child, 'close')
let closeSettled = false
const closing = launched.close('SIGKILL').then(() => { closeSettled = true })
await childClosed
await launched.client.closed
expect(closeSettled).toBe(false)
releasePermission?.()
await closing
expect(permissionFinished).toBe(true)
})
it('includes agent stderr when the ACP connection closes during startup', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({ failOnBoot: true, stderrNote: 'fake agent requested startup failure' })
await expect(runScenario(
@@ -48,6 +244,23 @@ describe('runScenario', () => {
)).rejects.toThrow(/agent stderr:\nfake agent requested startup failure/)
})
it('preserves launch-resolution errors when no child process exists', async () => {
const { dir, fixtureFile } = await scenario({})
vi.stubEnv('DSH_EXAMPLE_MODE', 'lib')
try {
await expect(runScenario(
{ steps: [] },
{
agent: { ...AGENT, binScript: join(dir, 'outside-src.ts'), libBinScript: undefined },
mode: 'replay',
fixtureFile,
},
)).rejects.toThrow(/expected a "\/src\/" segment/)
} finally {
vi.unstubAllEnvs()
}
})
it('drives a full turn: initialize (terminal caps), session, prompt, permission stub, harvest', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({
permissionProbe: true,
@@ -138,6 +351,39 @@ describe('runScenario', () => {
)).rejects.toThrow(/expected the prompt to fail/)
})
it('reports scenario and cleanup failures together', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({ prompt: 'respond' })
const cleanupFailure = new Error('cleanup failed')
fsControl.cleanupFailure = cleanupFailure
const failure = await runScenario(
{ steps: [...boot, { op: 'promptExpectError', text: 'fine' }] },
{ agent: AGENT, mode: 'replay', fixtureFile },
).catch((error: unknown): unknown => error)
expect(failure).toBeInstanceOf(AggregateError)
const failures = (failure as AggregateError).errors as unknown[]
expect(failures).toHaveLength(2)
expect(failures[0]).toBeInstanceOf(Error)
expect((failures[0] as Error).message).toMatch(/expected the prompt to fail/)
expect(failures[1]).toBe(cleanupFailure)
})
it('reports cleanup failure after an otherwise successful scenario', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({})
const cleanupFailure = new Error('cleanup failed')
fsControl.cleanupFailure = cleanupFailure
const failure = await runScenario(
{ steps: boot },
{ agent: AGENT, mode: 'replay', fixtureFile },
).catch((error: unknown): unknown => error)
expect(failure).toBeInstanceOf(AggregateError)
expect((failure as AggregateError).message).toBe('snapshot cleanup failed')
expect((failure as AggregateError).errors as unknown[]).toEqual([cleanupFailure])
})
it('newSessionExpectError swallows the rejection, with and without extra dirs', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({ rejectExtraDirs: true })
const result = await runScenario(