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

Merge pull request #1059 from deepseek-harness/perf/snapshot-file-parallelism

Browse Source 
perf(test): run replay snapshot files in parallel
This commit is contained in:
Turtle
2026-08-04 13:49:57 +08:00
committed by GitHub
parent 3a060c5688 bc26a366dd
commit 4b591c0abc
4 changed files with 103 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/acp-agent/tests/goal-snapshots/goal-session/input.json
+3 -6
View File
@@ -2,13 +2,10 @@
"steps": [
{ "op": "initialize" },
{ "op": "newSession" },
{
"op": "promptAndWaitForAgentMessage",
"text": "Create a durable two-round goal for the ACP snapshot, inspect it, then report readiness.",
"waitForText": "GOAL ROUND ONE"
},
{ "op": "promptAndWaitForAgentMessage", "text": "Create a durable two-round goal for the ACP snapshot, inspect it, then report readiness.", "waitForText": "GOAL ROUND ONE" },
{ "op": "waitForTurnStart", "minimumTurn": 3 },
{ "op": "cancel", "waitForFile": { "path": ".dsh-snapshot-goal-cancel-ready" } },
{ "op": "waitForTurnEnd" }
{ "op": "waitForTurnEnd" },
{ "op": "waitForEventAfterTurnEnd", "type": "user/message" }
]
}
packages/support/acp-snapshot/src/harness.ts
+41 -4
View File
@@ -43,16 +43,22 @@ const WAIT_POLL_INTERVAL_MS = 10
* reference, since a committed file cannot know the id in advance.
*
* `promptAndCancel` starts a prompt without awaiting completion, waits for a
* readiness condition, then cancels and awaits completion. `waitForFile`
* observes a cwd-relative marker; the default observes the durable turn start.
* readiness condition, then cancels and awaits completion. Its optional
* `waitForFile` observes a cwd-relative marker; otherwise it waits for the
* durable turn start. The standalone `waitForFile` holds the next script step
* behind the same marker.
* `promptAndWaitForAgentMessage` arms an exact text-chunk waiter before sending
* the prompt, then keeps the application live until that later update arrives.
* `waitForTurnStart` waits for an open durable turn, optionally at or beyond a
* specified turn number. `waitForTurnEnd` holds the subprocess open until the
* selected session's latest complete raw-JSONL turn boundary is `turn/end`.
* `waitForSubagentTurnEnd` waits until one background child has persisted a
* closed model-work turn after its own descriptor; child progress has no ACP
* update to wait on.
* `waitForTitleAfterTurnEnd` additionally waits for a later durable title.
* `waitForSubagentTurnEnd` applies the same work-turn boundary to one
* background child, whose progress has no ACP update to wait on.
* `waitForEventAfterTurnEnd` waits until a complete record of the given event
* type follows the latest closed turn — for scenarios whose asserted state
* (e.g. a goal pause) is appended only after cancellation reaches idle.
* A standalone `cancel` may also wait for a cwd-relative readiness marker.
* All wait timeouts default to 10s.
*/
@@ -73,6 +79,7 @@ export type InputStep =
| { op: 'waitForTurnEnd'; timeoutMs?: number }
| { op: 'waitForSubagentTurnEnd'; child?: number; timeoutMs?: number }
| { op: 'waitForTitleAfterTurnEnd'; timeoutMs?: number }
| { op: 'waitForEventAfterTurnEnd'; type: string; timeoutMs?: number }
| { op: 'cancel'; waitForFile?: { path: string; timeoutMs?: number } }
/** A scenario's `input.json`: an ordered list of input steps. */
@@ -296,6 +303,7 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
(id, timeoutMs) => waitForPersistedTurnEnd(sessionsRoot, id, timeoutMs),
(child, timeoutMs) => waitForPersistedChildTurnEnd(sessionsRoot, child, timeoutMs),
(id, timeoutMs) => waitForPersistedTitleAfterTurnEnd(sessionsRoot, id, timeoutMs),
(id, type, timeoutMs) => waitForPersistedEventAfterTurnEnd(sessionsRoot, id, type, timeoutMs),
)
// 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 —
@@ -371,6 +379,7 @@ async function runStep(
waitForTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
waitForChildTurnEnd: (child: number, timeoutMs?: number) => Promise<void>,
waitForTitleAfterTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
waitForEventAfterTurnEnd: (sessionId: string, type: string, timeoutMs?: number) => Promise<void>,
): Promise<void> {
switch (step.op) {
case 'initialize':
@@ -460,6 +469,12 @@ async function runStep(
await waitForTitleAfterTurnEnd(sessionId, step.timeoutMs)
return
}
case 'waitForEventAfterTurnEnd': {
const sessionId = getSessionId()
if (sessionId === undefined) throw new Error('snapshot-harness: waitForEventAfterTurnEnd before newSession')
await waitForEventAfterTurnEnd(sessionId, step.type, step.timeoutMs)
return
}
case 'waitForTurnStart': {
const sessionId = getSessionId()
if (sessionId === undefined) throw new Error('snapshot-harness: waitForTurnStart before newSession')
@@ -576,6 +591,21 @@ async function waitForPersistedTitleAfterTurnEnd(
}, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
}
/** Wait until a complete record of `type` follows the latest closed turn. */
async function waitForPersistedEventAfterTurnEnd(
root: string,
sessionId: string,
type: string,
timeoutMs = DEFAULT_WAIT_TIMEOUT_MS,
): Promise<void> {
await vi.waitFor(async () => {
const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
if (log === undefined || !latestEventFollowsTurnEnd(log.content, type)) {
throw new Error(`snapshot-harness: session "${sessionId}" did not persist ${type} after turn/end within ${timeoutMs}ms`)
}
}, { interval: WAIT_POLL_INTERVAL_MS, timeout: timeoutMs })
}
/** Wait for a cwd-relative marker proving an external action reached readiness. */
async function waitForWorkspaceFile(
cwd: string,
@@ -604,6 +634,13 @@ function latestTitleFollowsTurnEnd(content: string): boolean {
return turnEnd >= 0 && complete.lastIndexOf('\n{"type":"session/title",') > turnEnd
}
/** Return whether a complete record of `type` occurs after the last complete turn end. */
function latestEventFollowsTurnEnd(content: string, type: string): boolean {
const complete = content.slice(0, content.lastIndexOf('\n') + 1)
const turnEnd = complete.lastIndexOf('\n{"type":"turn/end",')
return turnEnd >= 0 && complete.lastIndexOf(`\n{"type":"${type}",`) > turnEnd
}
/** Return the latest open turn number, validating the persisted boundary record. */
function latestOpenTurn(content: string): number | undefined {
const complete = content.slice(0, content.lastIndexOf('\n') + 1)
packages/support/acp-snapshot/tests/harness.spec.ts
+50
View File
@@ -844,6 +844,55 @@ describe('runScenario', () => {
)).rejects.toThrow(/did not persist session\/title after turn\/end within 20ms/)
})
it('waitForEventAfterTurnEnd holds the app for a typed post-boundary record and times out otherwise', { timeout: 20_000 }, async () => {
const late = await scenario({
prompt: 'hang-until-cancel',
persistLogsOnCancel: true,
logs: [{
file: 'project/main/session.jsonl',
lines: [
{ type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
{ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'aborted' } } },
{ type: 'user/message', seq: 2, time: 3, data: { content: [{ type: 'text', text: 'late goal state' }], source: { kind: 'user' } } },
],
}],
})
const result = await runScenario(
{
steps: [
...boot,
{ op: 'promptAndCancel', text: 'hang' },
{ op: 'waitForEventAfterTurnEnd', type: 'user/message' },
],
},
{ agent: AGENT, mode: 'replay', fixtureFile: late.fixtureFile },
)
expect(result.sessionLogs[0]?.content).toMatch(/"turn\/end"[\s\S]*"user\/message"/)
const early = await scenario({
prompt: 'hang-until-cancel',
persistLogsOnCancel: true,
logs: [{
file: 'project/main/session.jsonl',
lines: [
{ type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
{ type: 'user/message', seq: 1, time: 1, data: { content: [{ type: 'text', text: 'early' }], source: { kind: 'user' } } },
{ type: 'turn/end', seq: 2, time: 2, data: { turn: 1, reason: { kind: 'aborted' } } },
],
}],
})
await expect(runScenario(
{
steps: [
...boot,
{ op: 'promptAndCancel', text: 'hang' },
{ op: 'waitForEventAfterTurnEnd', type: 'user/message', timeoutMs: 20 },
],
},
{ agent: AGENT, mode: 'replay', fixtureFile: early.fixtureFile },
)).rejects.toThrow(/did not persist user\/message after turn\/end within 20ms/)
})
it('promptExpectError swallows a model-error response as the expected outcome', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({ prompt: 'error' })
const result = await runScenario(
@@ -963,6 +1012,7 @@ describe('runScenario', () => {
[{ op: 'waitForTurnStart' }, /waitForTurnStart before newSession/],
[{ op: 'waitForTurnEnd' }, /waitForTurnEnd before newSession/],
[{ op: 'waitForTitleAfterTurnEnd' }, /waitForTitleAfterTurnEnd before newSession/],
[{ op: 'waitForEventAfterTurnEnd', type: 'user/message' }, /waitForEventAfterTurnEnd before newSession/],
[{ op: 'cancel' }, /cancel before newSession/],
] as [InputStep, RegExp][])('rejects %j before newSession', { timeout: 20_000 }, async (step, message) => {
const { fixtureFile } = await scenario({})
vitest.snapshot.config.ts
+9 -4
View File
@@ -55,12 +55,17 @@ export default defineConfig({
'packages/sdk/*/tests/**/*.snapshot.ts',
'packages/ui/tui/tests/**/*.snapshot.ts',
],
// Each test boots a subprocess; give it room and keep the worker file singular. Replay tests
// opt into bounded in-file concurrency, while record/refresh stay serial because they write
// fixtures. The environment knob restores serial replay with value 1 on constrained machines.
// Replay never writes committed outputs and every scenario owns its
// mutable runtime state (the subprocess suites use a unique temp dir and
// fixture set per scenario), so replay runs the snapshot files in
// parallel and bounds in-file concurrency with the environment knob
// (value 1 restores fully serial replay on constrained machines). Record
// and refresh stay serial: record spends real API quota per scenario, and
// refresh write-back harvests volatile values from fixtures already on
// disk, so concurrent writers would corrupt goldens.
testTimeout: 120_000,
hookTimeout: 30_000,
fileParallelism: false,
fileParallelism: (process.env.DSH_SNAPSHOT || 'replay') === 'replay' && snapshotMaxConcurrency > 1,
maxConcurrency: snapshotMaxConcurrency,
},
})