diff --git a/packages/context/workspace-context/src/index.ts b/packages/context/workspace-context/src/index.ts index 86e7388f46..9b24dac297 100644 --- a/packages/context/workspace-context/src/index.ts +++ b/packages/context/workspace-context/src/index.ts @@ -2,9 +2,9 @@ * Workspace instruction loader for AGENTS.md-compatible files. * * Baseline instructions enter durable context before the first request; successful fs - * tool touches mark nested, changed, and removed instructions for reconciliation - * at the next pre-step. Plugin lifecycle reads use - * the optional `ctx.fs` provider, so providerless products mount it as a no-op. + * tool touches project nested, changed, and removed instructions into the inbox. + * Plugin lifecycle reads use the optional `ctx.fs` provider, so providerless products + * mount it as a no-op. * * @module @deepseek-ai/dsh-workspace-context */ @@ -14,7 +14,7 @@ import { isDeepStrictEqual } from 'node:util' import type { Agent, PreStepDecision } from '@deepseek-ai/dsh-agent' import { createUserMessage } from '@deepseek-ai/dsh-llm' import type { UserMessage } from '@deepseek-ai/dsh-session' -import type { ToolExecution, ToolExecutionResult, ToolExecutionToken } from '@deepseek-ai/dsh-tools' +import type { ToolExecution, ToolExecutionResult } from '@deepseek-ai/dsh-tools' import { Config, resolveConfig, type ResolvedConfig } from './config.ts' import { loadBaselineInstructionSet } from './files.ts' import { @@ -70,8 +70,9 @@ function filePathFromExecution(exec: ToolExecution): string | undefined { export function apply(ctx: Context, config: Config): void { const resolved: ResolvedConfig = resolveConfig(config) const instructionVersions: InstructionVersionCache = new WeakMap() - const pendingTouches = new Map }>() - const touchedPaths = new WeakMap>() + // Emit listeners are not awaited, so each projection must compose against the + // inbox produced by earlier file results for the same agent. + const projectionTails = new WeakMap>() const compose = async ( agent: Agent, @@ -79,19 +80,14 @@ export function apply(ctx: Context, config: Config): void { claimed: readonly UserMessage[], pending: readonly UserMessage[], touchedPaths: readonly string[] = [], - ): Promise<{ - desired?: UserMessage - versions: Map - }> => { + ): Promise => { signal.throwIfAborted() - const candidateVersions: InstructionVersionCache = new WeakMap() - const candidateVersionStates = new Map(instructionVersions.get(agent.session) ?? []) - candidateVersions.set(agent.session, candidateVersionStates) if (resolved.maxBytes <= 0 || !Number.isFinite(resolved.maxBytes)) { - return { versions: new Map() } + return undefined } const fileSystem = ctx.get('fs') - if (fileSystem === undefined) return { versions: new Map() } + if (fileSystem === undefined) return undefined + if (touchedPaths.length === 0 && pending.length > 0) return pending[0] const content: UserMessage['content'][number][] = [] const changes: WorkspaceInstructionChange[] = [] let desiredBaseline = false @@ -112,7 +108,12 @@ export function apply(ctx: Context, config: Config): void { signal, }, fileSystem) const baseline = baselineInstructionState(instructions?.included ?? []) - for (const [scope, state] of baseline.versions) candidateVersionStates.set(scope, state) + let versionStates = instructionVersions.get(agent.session) + if (versionStates === undefined && baseline.versions.size > 0) { + versionStates = new Map() + instructionVersions.set(agent.session, versionStates) + } + for (const [scope, state] of baseline.versions) versionStates?.set(scope, state) if (instructions !== undefined && instructions.rendered.text.length > 0) { content.push(...workspaceContextMessage(instructions.rendered.text).content) changes.push(...baseline.changes.values()) @@ -122,7 +123,7 @@ export function apply(ctx: Context, config: Config): void { const update = await reconcileInstructionContext( agent, resolved, - candidateVersions, + instructionVersions, fileSystem, { authorityMessages, scopeMessages: pending, includeBaselineScopes: baselinePresent, touchedPaths, signal }, ) @@ -132,22 +133,17 @@ export function apply(ctx: Context, config: Config): void { if (update.context.source.kind === 'workspace-instructions') { changes.push(...update.context.source.changes) } - applyInstructionVersionUpdates(agent.session, update.versionUpdates, candidateVersions) + applyInstructionVersionUpdates(agent.session, update.versionUpdates, instructionVersions) } - const versions = new Map(candidateVersions.get(agent.session) ?? []) - return content.length === 0 - ? { versions } - : { - desired: createUserMessage({ - content, - source: { - kind: 'workspace-instructions', - ...desiredBaseline ? { baseline: true } : {}, - changes, - }, - }), - versions, - } + if (content.length === 0) return undefined + return createUserMessage({ + content, + source: { + kind: 'workspace-instructions', + ...desiredBaseline ? { baseline: true } : {}, + changes, + }, + }) } const syncInbox = (agent: Agent, claimed: readonly UserMessage[], desired: UserMessage | undefined): void => { @@ -176,22 +172,41 @@ export function apply(ctx: Context, config: Config): void { for (const message of pending.slice(1)) agent.inbox.remove('next-step', message.id) } - const commitSync = ( + const composeAndSync = async ( agent: Agent, + signal: AbortSignal, claimed: readonly UserMessage[], - desired: UserMessage | undefined, - versions: Map, - ): void => { + touchedPaths: readonly string[] = [], + ): Promise => { + const pending = agent.inbox.nextStep.filter(isWorkspaceContext) + const desired = await compose(agent, signal, claimed, pending, touchedPaths) + signal.throwIfAborted() syncInbox(agent, claimed, desired) - if (versions.size === 0) instructionVersions.delete(agent.session) - else instructionVersions.set(agent.session, versions) } - const restoreTouchedPaths = (agent: Agent, paths: Set | undefined): void => { - if (paths === undefined || paths.size === 0) return - const current = touchedPaths.get(agent) - if (current === undefined) touchedPaths.set(agent, paths) - else for (const path of paths) current.add(path) + const queueProjection = ( + agent: Agent, + signal: AbortSignal, + touchedPath: string, + ): void => { + const previous = projectionTails.get(agent) ?? Promise.resolve() + const current = previous.then(() => composeAndSync(agent, signal, [], [touchedPath])) + .catch((error: unknown) => { + if (!signal.aborted) ctx.logger.warn('workspace instruction refresh failed: %o', error) + }) + projectionTails.set(agent, current) + void current.then(() => { + if (projectionTails.get(agent) === current) projectionTails.delete(agent) + }) + } + + const waitForProjections = async (agent: Agent): Promise => { + while (true) { + const projection = projectionTails.get(agent) + if (projection === undefined) return + await projection + if (projectionTails.get(agent) === projection) return + } } ctx.on('agent/pre-step', async ( @@ -201,45 +216,15 @@ export function apply(ctx: Context, config: Config): void { next, ): Promise => { const decision = await next() - const pending = agent.inbox.nextStep.filter(isWorkspaceContext) - const paths = touchedPaths.get(agent) - touchedPaths.delete(agent) - try { - const composed = await compose(agent, signal, messages, pending, [...paths ?? []]) - /* v8 ignore next 4 -- every awaited filesystem operation checks this signal before settling. */ - if (signal.aborted) { - restoreTouchedPaths(agent, paths) - return decision - } - commitSync(agent, messages, composed.desired, composed.versions) - return decision - } catch (error: unknown) { - restoreTouchedPaths(agent, paths) - throw error - } + await waitForProjections(agent) + await composeAndSync(agent, signal, messages) + return decision }) ctx.on('tools/result', (exec: ToolExecution, result: ToolExecutionResult) => { - const staged = pendingTouches.get(exec.token) - pendingTouches.delete(exec.token) - if (exec.parent !== undefined) { - const paths = new Set(staged?.paths ?? []) - const ownPath = result.isError ? undefined : filePathFromExecution(exec) - if (ownPath !== undefined) paths.add(ownPath) - if (!result.isError && exec.agent !== undefined && paths.size > 0) { - const parent = pendingTouches.get(exec.parent) - if (parent === undefined) pendingTouches.set(exec.parent, { agent: exec.agent, paths }) - else for (const path of paths) parent.paths.add(path) - } - return - } - if (result.isError || exec.agent === undefined) return - const paths = new Set(staged?.paths ?? []) + if (result.isError || exec.agent === undefined || exec.signal.aborted) return const ownPath = filePathFromExecution(exec) - if (ownPath !== undefined) paths.add(ownPath) - if (paths.size === 0) return - const pending = touchedPaths.get(exec.agent) - if (pending === undefined) touchedPaths.set(exec.agent, paths) - else for (const path of paths) pending.add(path) + if (ownPath === undefined) return + queueProjection(exec.agent, exec.signal, ownPath) }) } diff --git a/packages/context/workspace-context/src/state.ts b/packages/context/workspace-context/src/state.ts index 200b145d21..960ee48ccb 100644 --- a/packages/context/workspace-context/src/state.ts +++ b/packages/context/workspace-context/src/state.ts @@ -62,13 +62,13 @@ export interface InstructionVersionState { /** Session-isolated fast-path state keyed by logical instruction scope. */ export type InstructionVersionCache = WeakMap> -/** A cache transition coupled to the model-visible change that authorizes it. */ +/** A metadata-cache transition associated with one rendered instruction change. */ export interface InstructionVersionUpdate { change: WorkspaceInstructionChange state?: InstructionVersionState } -/** Rendered reconciliation plus cache transitions awaiting final policy. */ +/** Rendered reconciliation plus its metadata-cache transitions. */ export interface ReconciledInstructionContext { context: UserMessage versionUpdates: InstructionVersionUpdate[] @@ -193,20 +193,20 @@ function versionStatesFor(session: Session, cache: InstructionVersionCache): Map } /** - * Keep only cache updates whose model-visible changes survived final policy. + * Keep only cache updates represented by rendered changes. * @param updates - proposed updates from one or more reconciliations. - * @param committedChanges - transitions retained on the authoritative result. - * @returns updates authorized by an exact retained transition. + * @param renderedChanges - transitions retained by the renderer. + * @returns updates represented by an exact retained transition. */ export function retainedInstructionVersionUpdates( updates: readonly InstructionVersionUpdate[], - committedChanges: readonly WorkspaceInstructionChange[], + renderedChanges: readonly WorkspaceInstructionChange[], ): InstructionVersionUpdate[] { - return updates.filter(update => committedChanges.some(change => sameInstructionChange(update.change, change))) + return updates.filter(update => renderedChanges.some(change => sameInstructionChange(update.change, change))) } /** - * Apply authorized metadata-cache transitions without retaining instruction prose. + * Apply metadata-cache transitions without retaining instruction prose. * @param session - owning session. * @param updates - ordered set/delete transitions. * @param cache - session-isolated metadata cache. diff --git a/packages/context/workspace-context/tests/workspace-context.spec.ts b/packages/context/workspace-context/tests/workspace-context.spec.ts index 6e4f185f70..116a85b818 100644 --- a/packages/context/workspace-context/tests/workspace-context.spec.ts +++ b/packages/context/workspace-context/tests/workspace-context.spec.ts @@ -177,13 +177,10 @@ function stubAgent(cwd?: string, seed: SessionEvent[] = []): Agent { session, inbox: new Inbox(session, { inserted: () => {}, discarded: () => {} }), status: 'idle', - acceptsNextStep: false, send: () => {}, followup: () => {}, steer: () => {}, inject: () => { throw new Error('workspace-context must append directly to the open step') }, - updateInbox: () => 'not-found', - reserveTurnAdmission: () => undefined, cancel() {}, whenIdle: () => Promise.resolve(), } @@ -229,16 +226,6 @@ function baselineEvents(agent: Agent): SessionEvent[] { && event.data.source.baseline === true) } -function workspaceChangeContext(scope: string, digest: string): UserMessage { - return createUserMessage({ - content: [{ type: 'text', text: `instructions for ${scope}` }], - source: { - kind: 'workspace-instructions', - changes: [{ action: 'set', scope, path: `${scope}/AGENTS.md`, digest }], - }, - }) -} - async function appendAdditionalContexts(ctx: Context, agent: Agent): Promise { await syncedWorkspaceContext(ctx, agent) let lastSeq: number | undefined @@ -2052,7 +2039,7 @@ describe('workspace context request injection', () => { }) describe('dynamic nested workspace context injection', () => { - it('commits a buffered instruction change before a later tool abort closes the step', async () => { + it('projects a successful file result even when a later sibling aborts the step', async () => { const root = await tempRepo() const home = await tempRepo() const ctx = new Context() @@ -2105,7 +2092,7 @@ describe('dynamic nested workspace context injection', () => { // Cancellation discards the aborted step's pending context. The next // successful read discovers and durably injects it once. expect(contexts).toHaveLength(1) - expect(adapter.requests).toHaveLength(3) + expect(adapter.requests).toHaveLength(4) expect(adapter.requests.at(-1)?.messages.map(blocks => blocksText(blocks.content)).join('\n')) .toContain('nested rule survives an aborted tool batch') } finally { @@ -3353,6 +3340,7 @@ describe('dynamic nested workspace context injection', () => { await write(join(root, 'pkg/deep/file.txt'), 'hello') const ctx = new Context() await mountFileToolsAndWorkspaceContext(ctx, { dshHome: home, maxBytes: 65536 }) + const agent = stubAgent(root) ctx.on('tools/post-execute', async () => ({ kind: 'block' as const, feedback: [{ type: 'text' as const, text: 'blocked downstream' }], @@ -3363,7 +3351,7 @@ describe('dynamic nested workspace context injection', () => { callId: CallId('read-blocked-downstream'), name: 'read', arguments: { file_path: join('pkg', 'deep', 'file.txt') }, - agent: stubAgent(root), + agent, }) // The pipeline rejected this touch, so no workspace instructions from it @@ -3371,13 +3359,15 @@ describe('dynamic nested workspace context injection', () => { expect(result.isError).toBe(true) expect(blocksText(result.content)).toBe('blocked downstream') expect(result.additionalContexts).toBeUndefined() + await syncWorkspaceContext(ctx, agent) + expect(agent.inbox.nextStep).toEqual([]) } finally { await rm(root, { recursive: true, force: true }) await rm(home, { recursive: true, force: true }) } }) - it('does not commit pending state when an outer post-execute listener blocks the final result', async () => { + it('does not project a file touch when an outer post-execute listener blocks the final result', async () => { const root = await tempRepo() const home = await tempRepo() const ctx = new Context() @@ -3406,6 +3396,9 @@ describe('dynamic nested workspace context injection', () => { arguments: { file_path: join('pkg', 'deep', 'file.txt') }, agent, }) + await syncWorkspaceContext(ctx, agent) + expect(agent.inbox.nextStep).toEqual([]) + shouldBlock = false const accepted = await ctx.tools.execute({ signal: testToolSignal, @@ -3426,7 +3419,7 @@ describe('dynamic nested workspace context injection', () => { } }) - it('rolls back parent-token pending state when a composite result is blocked', async () => { + it('projects a successful nested file result independently of a blocked composite result', async () => { const root = await tempRepo() const home = await tempRepo() const ctx = new Context() @@ -3456,10 +3449,9 @@ describe('dynamic nested workspace context injection', () => { return nested.content }, })) - let shouldBlock = true ctx.on('tools/post-execute', async (exec, _result, next) => { const downstream = await next() - return exec.name === 'composite-read' && shouldBlock + return exec.name === 'composite-read' ? { kind: 'block' as const, feedback: [{ type: 'text' as const, text: 'outer composite block' }] } : downstream }) @@ -3470,15 +3462,9 @@ describe('dynamic nested workspace context injection', () => { signal: testToolSignal, callId: CallId('composite-first'), name: 'composite-read', arguments: {}, agent, }) - shouldBlock = false - const accepted = await ctx.tools.execute({ - signal: testToolSignal, - callId: CallId('composite-retry'), name: 'composite-read', arguments: {}, agent, - }) expect(blocked.isError).toBe(true) expect(blocked.additionalContexts).toBeUndefined() - expect(accepted.isError).toBe(false) expect(blocksText(((await syncedWorkspaceContext(ctx, agent))).content)).toContain('nested package rule') } finally { await ctx.fiber.dispose() @@ -3487,40 +3473,26 @@ describe('dynamic nested workspace context injection', () => { } }) - it('handles defensive tools/result observer branches without retaining staged state', async () => { + it('ignores failed, aborted, agentless, and non-file final results', async () => { const ctx = new Context() try { + await ctx.plugin(RecordingFileSystem) await ctx.plugin(workspaceContext, { maxBytes: 65536 }) + const fs = ctx.fs as RecordingFileSystem const agent = stubAgent('/') - const parent = Symbol('parent') as ToolExecutionToken const plainResult = { callId: CallId('plain'), content: [], isError: false as const, value: null } + const aborted = new AbortController() + aborted.abort(new Error('cancelled')) ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('agentless-child'), name: 'read', arguments: {}, parent, + signal: testToolSignal, callId: CallId('agentless'), name: 'read', arguments: { file_path: 'file.txt' }, }), plainResult) ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('contextless-child'), name: 'read', arguments: {}, agent, parent, - }), { ...plainResult, additionalContexts: [createUserMessage({ - content: [], source: { kind: 'plugin', plugin: 'workspace-context' }, - })] }) - ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('failed-child'), name: 'read', arguments: { file_path: 'failed/file.txt' }, agent, parent, + signal: testToolSignal, callId: CallId('failed'), name: 'read', arguments: { file_path: 'failed/file.txt' }, agent, }), { content: [], isError: true, error: { message: 'failed' } }) ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('first-child'), name: 'read', arguments: { file_path: 'first/file.txt' }, agent, parent, - }), { ...plainResult, additionalContexts: [workspaceChangeContext('first', 'one')] }) - ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('second-child'), name: 'read', arguments: { file_path: 'second/file.txt' }, agent, parent, - }), { ...plainResult, additionalContexts: [workspaceChangeContext('second', 'two')] }) - ctx.emit('tools/result', { - ...stubToolExecution({ signal: testToolSignal, callId: CallId('agentless-parent'), name: 'composite', arguments: {} }), - token: parent, - }, plainResult) + signal: aborted.signal, callId: CallId('aborted'), name: 'read', arguments: { file_path: 'aborted/file.txt' }, agent, + }), plainResult) ctx.emit('tools/result', stubToolExecution({ signal: testToolSignal, callId: CallId('null-arguments'), name: 'read', arguments: null, agent, @@ -3534,53 +3506,14 @@ describe('dynamic nested workspace context injection', () => { callId: CallId('non-fs'), name: 'composite', arguments: {}, agent, }), plainResult) - expect(agent.session.deriveMessages()).toEqual([]) + await Promise.resolve() + expect(fs.signals).toEqual([]) + expect(agent.inbox.nextStep).toEqual([]) } finally { await ctx.fiber.dispose() } }) - it('ignores post-execute events that are not successful structured file touches', async () => { - const root = await tempRepo() - const home = await tempRepo() - try { - await mkdir(join(root, '.git'), { recursive: true }) - await write(join(root, 'pkg/AGENTS.md'), 'nested package rule') - await write(join(root, 'pkg/deep/file.txt'), 'hello') - const ctx = new Context() - await mountFileToolsAndWorkspaceContext(ctx, { dshHome: home, maxBytes: 65536 }) - const agent = stubAgent(root) - const result = { - callId: CallId('manual'), - content: [{ type: 'text' as const, text: 'manual result' }], - isError: false as const, - value: null, - } - const cases = [ - { name: 'read', arguments: { file_path: join('pkg', 'deep', 'file.txt') }, agent: undefined }, - { name: 'bash', arguments: { file_path: join('pkg', 'deep', 'file.txt') }, agent }, - { name: 'read', arguments: null, agent }, - { name: 'read', arguments: {}, agent }, - { name: 'read', arguments: { file_path: 1 }, agent }, - { name: 'read', arguments: { file_path: ' ' }, agent }, - ] - - for (const item of cases) { - const decision = await ctx.waterfall('tools/post-execute', stubToolExecution({ - signal: testToolSignal, - callId: CallId(`manual-${item.name}-${cases.indexOf(item)}`), - name: item.name, - arguments: item.arguments, - ...item.agent === undefined ? {} : { agent: item.agent }, - }), result, async () => ({ kind: 'accept' as const })) - expect(decision).toEqual({ kind: 'accept' }) - } - } finally { - await rm(root, { recursive: true, force: true }) - await rm(home, { recursive: true, force: true }) - } - }) - it('does not attach nested instructions when the byte budget is disabled', async () => { const root = await tempRepo() const home = await tempRepo() @@ -3632,7 +3565,7 @@ describe('dynamic nested workspace context injection', () => { } }) - it('cleans up its tools/post-execute listener when the plugin fiber is disposed', async () => { + it('cleans up its tools/result listener when the plugin fiber is disposed', async () => { const root = await tempRepo() const home = await tempRepo() try { @@ -3642,17 +3575,20 @@ describe('dynamic nested workspace context injection', () => { const ctx = new Context() const fiber = await mountFileToolsAndWorkspaceContext(ctx, { dshHome: home, maxBytes: 65536 }) await fiber.dispose() + const agent = stubAgent(root) const result = await ctx.tools.execute({ signal: testToolSignal, callId: CallId('read-after-dispose'), name: 'read', arguments: { file_path: join('pkg', 'deep', 'file.txt') }, - agent: stubAgent(root), + agent, }) expect(result.isError).toBe(false) expect(result.additionalContexts).toBeUndefined() + await Promise.resolve() + expect(agent.inbox.nextStep).toEqual([]) } finally { await rm(root, { recursive: true, force: true }) await rm(home, { recursive: true, force: true }) @@ -3757,7 +3693,7 @@ describe('workspace context inbox synchronization', () => { } }) - it('restores drained dirty paths when pre-step reconciliation aborts', async () => { + it('keeps a completed tool projection when a later pre-step aborts', async () => { const root = await tempRepo() const home = await tempRepo() const ctx = new Context() @@ -3769,23 +3705,23 @@ describe('workspace context inbox synchronization', () => { fs.entries.set(join(root, 'b/AGENTS.md'), { type: 'file', content: 'restored B' }) await ctx.plugin(workspaceContext, { dshHome: home, maxBytes: 65536 }) const agent = stubAgent(root) - const dirtyA = stubToolExecution({ + const first = stubToolExecution({ signal: testToolSignal, - callId: CallId('dirty-before-abort'), name: 'read', arguments: { file_path: join('a', 'file.txt') }, agent, + callId: CallId('projected-before-abort'), name: 'read', arguments: { file_path: join('a', 'file.txt') }, agent, }) - ctx.emit('tools/result', dirtyA, acceptedResult) + ctx.emit('tools/result', first, acceptedResult) const controller = new AbortController() - controller.abort(new Error('abort dirty reconciliation')) + controller.abort(new Error('abort pre-step reconciliation')) await expect(agentEvents(ctx, agent).waterfall( 'agent/pre-step', [], { turn: 1, step: 1, signal: controller.signal }, async () => ({ kind: 'enter' as const, messages: [] }), - )).rejects.toThrow('abort dirty reconciliation') + )).rejects.toThrow('abort pre-step reconciliation') ctx.emit('tools/result', stubToolExecution({ signal: testToolSignal, - callId: CallId('dirty-after-abort'), name: 'read', arguments: { file_path: join('b', 'file.txt') }, agent, + callId: CallId('projected-after-abort'), name: 'read', arguments: { file_path: join('b', 'file.txt') }, agent, }), acceptedResult) await syncWorkspaceContext(ctx, agent) const text = blocksText(agent.inbox.nextStep[0]?.content) @@ -3798,43 +3734,6 @@ describe('workspace context inbox synchronization', () => { } }) - it('merges a final tool touch back into dirty paths while a pre-step aborts', async () => { - const root = await tempRepo() - const home = await tempRepo() - const ctx = new Context() - try { - await ctx.plugin(BlockingReadFileSystem) - const fs = ctx.fs as BlockingReadFileSystem - fs.entries.set(join(root, '.git'), { type: 'directory' }) - fs.entries.set(join(root, 'a/AGENTS.md'), { type: 'file', content: 'blocked A' }) - await ctx.plugin(workspaceContext, { dshHome: home, maxBytes: 65536 }) - const agent = stubAgent(root) - ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('dirty-blocked-a'), name: 'read', arguments: { file_path: join('a', 'file.txt') }, agent, - }), acceptedResult) - const controller = new AbortController() - const preparing = agentEvents(ctx, agent).waterfall( - 'agent/pre-step', [], - { turn: 1, step: 1, signal: controller.signal }, - async () => ({ kind: 'enter' as const, messages: [] }), - ) - await fs.started.promise - ctx.emit('tools/result', stubToolExecution({ - signal: testToolSignal, - callId: CallId('dirty-concurrent-b'), name: 'read', arguments: { file_path: join('b', 'file.txt') }, agent, - }), acceptedResult) - controller.abort(new Error('abort blocked reconciliation')) - - await expect(preparing).rejects.toThrow('abort blocked reconciliation') - expect(agent.inbox.nextStep).toEqual([]) - } finally { - await ctx.fiber.dispose() - await rm(root, { recursive: true, force: true }) - await rm(home, { recursive: true, force: true }) - } - }) - it('serializes concurrent final results and merges both touched scopes into one pending context', async () => { const root = await tempRepo() const home = await tempRepo()