Merge goal tool fixes and align same-session blockers

This commit is contained in:
Tianyi Cui
2026-07-20 17:23:55 +08:00
55 changed files with 860 additions and 891 deletions
@@ -1,138 +0,0 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'
import { mkdtemp, readFile, readdir, rm } 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 { foldGoal } from '@deepseek-ai/dsh-goal'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
const binScript = fileURLToPath(new URL('../../../examples/stdio-demo/src/bin.ts', import.meta.url))
const configPath = fileURLToPath(new URL(
'../../../../examples/echo-agent/tests/fixtures/goal/goal-session/cordis.yml',
import.meta.url,
))
const repoTsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url))
const PROCESS_TIMEOUT_MS = 30_000
const TEST_TIMEOUT_MS = PROCESS_TIMEOUT_MS + 15_000
let child: ChildProcessWithoutNullStreams | undefined
let workdir: string | undefined
afterEach(async () => {
if (child !== undefined && child.exitCode === null) child.kill('SIGKILL')
child = undefined
if (workdir !== undefined) await rm(workdir, { recursive: true, force: true })
workdir = undefined
})
/** Recursively locate persistence JSONL files in one temporary root. */
async function jsonlFiles(dir: string): Promise<string[]> {
const entries = await readdir(dir, { withFileTypes: true })
const paths = await Promise.all(entries.map(async (entry) => {
const path = join(dir, entry.name)
if (entry.isDirectory()) return jsonlFiles(path)
return entry.isFile() && entry.name.endsWith('.jsonl') ? [path] : []
}))
return paths.flat()
}
/** Run the complete deterministic human-turn plus two-round composition. */
async function runComposition(): Promise<{ stdout: string; stderr: string }> {
workdir = await mkdtemp(join(tmpdir(), 'goal-session-e2e-'))
const cwd = workdir
return new Promise((resolve, reject) => {
const launch = resolveExampleLaunch({
srcBin: binScript,
configArgs: [configPath],
tsconfigPath: repoTsconfig,
exposeInternals: true,
env: {
DSH_HOME: join(cwd, '.dsh'),
DSH_AGENTS_HOME: join(cwd, '.agents'),
},
})
const proc = spawn(launch.command, launch.args, {
cwd,
env: { ...process.env, ...launch.env },
stdio: ['pipe', 'pipe', 'pipe'],
})
child = proc
let stdout = ''
let stderr = ''
let inputClosed = false
proc.stdout.setEncoding('utf8')
proc.stdout.on('data', (chunk: string) => {
stdout += chunk
if (!inputClosed && stdout.includes('ROUND TWO COMPLETE') && stdout.includes('\n> ')) {
inputClosed = true
proc.stdin.end()
}
})
proc.stderr.setEncoding('utf8')
proc.stderr.on('data', (chunk: string) => { stderr += chunk })
const timer = setTimeout(() => {
proc.kill('SIGKILL')
reject(new Error(
`goal-session e2e did not exit within ${PROCESS_TIMEOUT_MS / 1_000}s. stdout:\n${stdout}\nstderr:\n${stderr}`,
))
}, PROCESS_TIMEOUT_MS)
proc.on('exit', (code) => {
clearTimeout(timer)
if (code === 0) resolve({ stdout, stderr })
else reject(new Error(`goal-session e2e exited ${String(code)}. stdout:\n${stdout}\nstderr:\n${stderr}`))
})
proc.on('error', (error) => { clearTimeout(timer); reject(error) })
proc.stdin.write('start\n')
})
}
describe('same-session goal rounds through a real Loader, app, and stdio process', () => {
it('persists two exact rounds and stops the completion turn without another request', async () => {
const { stdout, stderr } = await runComposition()
expect(stderr).not.toContain('UNHANDLED')
expect(stdout).toContain('goal-session e2e ready.')
expect(stdout).toContain('GOAL CREATED')
expect(stdout).toContain('ROUND ONE')
expect(stdout).toContain('ROUND TWO COMPLETE')
const logs = await jsonlFiles(join(workdir as string, '.sessions'))
expect(logs).toHaveLength(1)
const lines = (await readFile(logs[0] as string, 'utf8')).trimEnd().split('\n')
const events = lines.slice(1).map(line => JSON.parse(line) as SessionEvent)
const calls = events.filter(event => event.type === 'tool/call')
expect(calls.map(event => event.data.name)).toEqual(['create_goal', 'get_goal', 'update_goal'])
expect(events.filter(event => event.type === 'step/start')).toHaveLength(5)
expect(events.filter(event => event.type === 'tool/result').every(event => !event.data.isError)).toBe(true)
const rounds = events.filter(event => event.type === 'user/message'
&& event.data.source.kind === 'goal')
expect(rounds).toHaveLength(2)
const roundNumbers: number[] = []
const revisions: number[] = []
const prompts: string[] = []
for (const event of events) {
if (event.type !== 'user/message' || event.data.source.kind !== 'goal') continue
roundNumbers.push(event.data.source.round)
revisions.push(event.data.source.revision)
prompts.push(event.data.content.find(block => block.type === 'text')?.text ?? '')
}
expect(roundNumbers).toEqual([1, 2])
expect(revisions).toEqual([1, 1])
expect(prompts[0]).toContain('Round: 1/2')
expect(prompts[1]).toContain('Round: 2/2')
expect(foldGoal(events)).toMatchObject({
goal: {
objective: 'Complete two deterministic same-session rounds',
phase: 'complete',
revision: 2,
maxGoalRounds: 2,
},
roundsStarted: 2,
})
}, TEST_TIMEOUT_MS)
})
@@ -126,18 +126,18 @@ describe('goal-round outcome policy', () => {
[{ kind: 'aborted', reason: 'operator stopped' }, true, { kind: 'pause', reason: 'operator stopped' }],
[{ kind: 'aborted' }, true, { kind: 'pause', reason: 'cancelled' }],
[{ kind: 'error', step: 1, message: 'slow down', code: 'RATE_LIMIT' }, true,
{ kind: 'usage-limited', message: 'slow down' }],
{ kind: 'blocked', code: 'usage-limited', message: 'slow down' }],
[{ kind: 'error', step: 1, message: 'broken' }, true,
{ kind: 'blocked', reason: 'error', detail: 'broken' }],
{ kind: 'blocked', code: 'turn-error', message: 'broken' }],
[{ kind: 'max-tokens' }, true,
{ kind: 'blocked', reason: 'max-tokens', detail: 'model output reached max tokens' }],
{ kind: 'blocked', code: 'max-tokens', message: 'model output reached max tokens' }],
[{ kind: 'rejected', reason: 'policy' }, true,
{ kind: 'blocked', reason: 'rejected', detail: 'policy' }],
{ kind: 'blocked', code: 'prompt-rejected', message: 'policy' }],
[{ kind: 'disposed' }, true, { kind: 'disarm', reason: 'disposed' }],
[{ kind: 'interrupted' }, true, { kind: 'disarm', reason: 'interrupted' }],
[{ kind: 'completed' }, false, { kind: 'disarm', reason: 'durability-failed' }],
[{ kind: 'future-outcome' } as unknown as TurnEndReason, true,
{ kind: 'blocked', reason: 'unknown', detail: 'unknown turn outcome: future-outcome' }],
{ kind: 'blocked', code: 'unknown-turn-outcome', message: 'unknown turn outcome: future-outcome' }],
] as const)('maps %j without abnormal automatic retry', (reason, durable, expected) => {
expect(goalSession.classifyGoalRound(reason, durable)).toEqual(expected)
})
@@ -187,9 +187,13 @@ describe('same-session goal driving', () => {
const test = await harness([textResponse('round one'), textResponse('round two')])
const created = test.ctx.goals.create(test.agent, { objective: 'finish twice', maxGoalRounds: 2 })
const final = await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
const final = await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
expect(final).toMatchObject({ id: created.id, roundsStarted: 2, activation: 'disarmed' })
expect(final?.blockedReason).toEqual({
code: 'round-limit',
message: 'Goal reached its configured limit of 2 rounds.',
})
expect(test.adapter.requests).toHaveLength(2)
const rounds: number[] = []
for (const event of test.agent.session.events) {
@@ -219,21 +223,22 @@ describe('same-session goal driving', () => {
expect(adapter.requests).toHaveLength(0)
ctx.goals.resume(agent, created)
await waitForGoal(ctx, agent, goal => goal?.phase === 'budget-limited')
await waitForGoal(ctx, agent, goal => goal?.phase === 'blocked')
expect(adapter.requests).toHaveLength(1)
})
it.each([
['rate limit', Object.assign(new Error('slow down'), { code: 'RATE_LIMIT' }), 'usage-limited'],
['request error', new Error('provider broke'), 'blocked'],
['max tokens', maxTokensResponse('unfinished'), 'blocked'],
] as const)('stops after a %s without an automatic retry', async (_label, response, phase) => {
['request error', new Error('provider broke'), 'turn-error'],
['max tokens', maxTokensResponse('unfinished'), 'max-tokens'],
] as const)('stops after a %s without an automatic retry', async (_label, response, code) => {
const test = await harness([response])
test.ctx.goals.create(test.agent, { objective: 'stop safely', maxGoalRounds: 8 })
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === phase)
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' })
expect(goal?.blockedReason?.code).toBe(code)
expect(test.adapter.requests).toHaveLength(1)
})
@@ -247,6 +252,7 @@ describe('same-session goal driving', () => {
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
expect(goal?.roundsStarted).toBe(0)
expect(goal?.blockedReason).toEqual({ code: 'prompt-rejected', message: 'deployment policy' })
expect(test.adapter.requests).toHaveLength(0)
expect(test.agent.session.events.some(event => event.type === 'prompt/blocked'
&& event.data.reason === 'deployment policy')).toBe(true)
@@ -305,7 +311,7 @@ describe('same-session goal driving', () => {
test.ctx.goals.create(test.agent, { objective: 'continue after the human', maxGoalRounds: 1 })
test.agent.send([{ type: 'text', text: 'human goes first' }])
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
expect(test.adapter.requests).toHaveLength(2)
expect(requestText(test.adapter.requests[0]!)).toContain('human goes first')
@@ -323,7 +329,7 @@ describe('same-session goal driving', () => {
})
test.ctx.goals.create(test.agent, { objective: 'yield to nested human input', maxGoalRounds: 1 })
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
expect(test.adapter.requests).toHaveLength(2)
expect(requestText(test.adapter.requests[0]!)).toContain('human joined the pending batch')
@@ -343,7 +349,7 @@ describe('same-session goal driving', () => {
})
test.ctx.goals.create(test.agent, { objective: 'old objective', maxGoalRounds: 1 })
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'budget-limited')
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
expect(goal).toMatchObject({ revision: 3, objective: 'new objective', roundsStarted: 1 })
const blocked = test.agent.session.events.find(event => event.type === 'prompt/blocked')
@@ -370,7 +376,7 @@ describe('same-session goal driving', () => {
})
test.ctx.goals.create(test.agent, { objective: 'edit during admission', maxGoalRounds: 1 })
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'budget-limited')
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
expect(goal).toMatchObject({ objective: 'edited downstream', roundsStarted: 1 })
expect(test.adapter.requests).toHaveLength(1)
@@ -442,6 +448,10 @@ describe('same-session goal driving', () => {
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
expect(goal).toMatchObject({ roundsStarted: 0, activation: 'disarmed' })
expect(goal?.blockedReason).toEqual({
code: 'queue-failed',
message: 'Could not queue goal round 1: queue rejected',
})
expect(test.adapter.requests).toHaveLength(0)
})
@@ -519,7 +529,7 @@ describe('same-session goal driving', () => {
})
test.ctx.goals.create(test.agent, { objective: 'retry stale admission', maxGoalRounds: 1 })
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
expect(test.adapter.requests).toHaveLength(1)
expect(test.agent.session.events.some(event => event.type === 'prompt/blocked'
@@ -669,7 +679,7 @@ describe('same-session goal driving', () => {
expect(test.adapter.requests).toHaveLength(0)
test.ctx.goals.resume(test.agent, created)
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'budget-limited')
await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
expect(test.adapter.requests).toHaveLength(1)
})