From 79c25ea2184d5f20ee719b5e50a700bb60d3e49e Mon Sep 17 00:00:00 2001 From: Jaakko Husso Date: Fri, 16 Oct 2026 21:13:18 +0300 Subject: [PATCH] fix(core): Load earlier files and report failed resumes for system agents (no-changelog) Every system turn, start and resume, now attaches the attachment store. A later turn can then load the files of earlier turns. Only a completed system turn sends `done`, like Preview. `resumeRun` rejects when the resumed turn fails, so the route can send an error. The background `resume` still logs the failure. The host sets `persistence` after the provider run options, and the provider run options type omits it. Tests cover thread creation for a new session and the error event of a refused system turn. Co-Authored-By: Claude Opus 5.5 --- ...ent-message-queue-consumer.service.test.ts | 25 +-- .../system-agent-execution.service.test.ts | 170 +++++++++++++++++- .../system-agent-execution.service.ts | 76 +++++--- .../system-agents/system-agent.types.ts | 7 +- 4 files changed, 237 insertions(+), 41 deletions(-) diff --git a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts index 7adccc5d7b3..7954ce5f77b 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-message-queue-consumer.service.test.ts @@ -524,18 +524,25 @@ describe('AgentMessageQueueConsumer', () => { expect(queue.recordFailure).not.toHaveBeenCalled(); }); + /** A refused system turn fails the item, streams an error and never runs. */ + function expectRefused(item: ReturnType) { + expect(queue.recordFailure).toHaveBeenCalledWith( + item, + expect.objectContaining({ message: 'You can no longer execute this agent' }), + expect.any(AbortSignal), + ); + expect(chatExecutions.register).not.toHaveBeenCalled(); + expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expect(sender.send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); + } + it('records a failure when the provider no longer authorizes the owner', async () => { registerSystemAgent(false); const item = systemClaim('session'); await consumeOne(item); - expect(queue.recordFailure).toHaveBeenCalledWith( - item, - expect.objectContaining({ message: 'You can no longer execute this agent' }), - expect.any(AbortSignal), - ); - expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expectRefused(item); }); it('records a failure when the owner can no longer read the working project', async () => { @@ -545,9 +552,8 @@ describe('AgentMessageQueueConsumer', () => { await consumeOne(item); - expect(queue.recordFailure).toHaveBeenCalled(); + expectRefused(item); expect(provider.authorize).not.toHaveBeenCalled(); - expect(systemAgentExecution.consume).not.toHaveBeenCalled(); }); it('records a failure for a system message without a registered provider', async () => { @@ -555,8 +561,7 @@ describe('AgentMessageQueueConsumer', () => { await consumeOne(item); - expect(queue.recordFailure).toHaveBeenCalled(); - expect(systemAgentExecution.consume).not.toHaveBeenCalled(); + expectRefused(item); }); it('keeps the agent:execute check for Preview messages of a system agent id', async () => { diff --git a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts index 9ddb16cc751..2dec618cf97 100644 --- a/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/system-agents/__tests__/system-agent-execution.service.test.ts @@ -5,6 +5,7 @@ import { mock } from 'vitest-mock-extended'; import { userHasScopes } from '@/permissions.ee/check-access'; +import type { AgentChatAttachmentService } from '../../agent-chat-attachment.service'; import type { AgentChatExecutionService } from '../../agent-chat-execution.service'; import type { AgentExecutionService } from '../../agent-execution.service'; import type { @@ -75,11 +76,16 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { const executionService = mock(); const chatExecutionService = mock(); const agentRepository = mock(); + const attachmentService = mock(); const onChunk = vi.fn(); const onSettled = vi.fn(async () => {}); const handle: SystemAgentTurnHandle = { - agent: { name: 'runtime-agent' } as unknown as SystemAgentTurnHandle['agent'], + agent: { + name: 'runtime-agent', + fileStore: vi.fn(), + snapshot: { model: { provider: 'anthropic' } }, + } as unknown as SystemAgentTurnHandle['agent'], input: 'model input', hostMetadata: { turnOptions: { runId: 'host-run-1' } }, onChunk, @@ -116,7 +122,7 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { chatExecutionService, checkpointStorage, txRunner, - mock(), + attachmentService, ); return { @@ -136,6 +142,7 @@ function setup(chunks: AgentExecutionStreamChunk[] = []) { chatExecutionService, agentRepository, registry, + attachmentService, }; } @@ -305,14 +312,17 @@ describe('SystemAgentExecutionService', () => { const { service, onSettled } = setup([ { type: 'error', error: failure } as AgentExecutionStreamChunk, ]); + const send = vi.fn(); - await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + await service.consume(claimFor(), user, new AbortController().signal, send); expect(onSettled).toHaveBeenCalledWith({ status: 'errored', executionId: 'exec-1', error: failure, }); + expect(send).toHaveBeenCalledWith(expect.objectContaining({ type: 'error' })); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); }); it('settles as cancelled when the recorded execution was cancelled', async () => { @@ -332,12 +342,42 @@ describe('SystemAgentExecutionService', () => { throw failure; }); + const send = vi.fn(); + await expect( - service.consume(claimFor(), user, new AbortController().signal, vi.fn()), + service.consume(claimFor(), user, new AbortController().signal, send), ).rejects.toThrow('stream broke'); expect(onSettled).toHaveBeenCalledWith( expect.objectContaining({ status: 'errored', error: failure }), ); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + }); + + it('attaches the attachment store to a turn without attachments', async () => { + const { service, handle, attachmentService } = setup([textChunk]); + const store = mock>(); + attachmentService.getFileStore.mockReturnValue(store); + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(attachmentService.getFileStore).toHaveBeenCalledWith( + { agentId: AGENT_ID, projectId: 'project-1' }, + 'anthropic', + ); + expect(handle.agent.fileStore).toHaveBeenCalledWith(store); + }); + + it('keeps the host persistence when the provider run options set one', async () => { + const { service, handle, prepared } = setup([textChunk]); + handle.runOptions = { + persistence: { threadId: 'other-thread', resourceId: 'other-resource' }, + } as never; + + await service.consume(claimFor(), user, new AbortController().signal, vi.fn()); + + expect(prepared[0].options).toMatchObject({ + persistence: { threadId: 'thread-1', resourceId: 'system-chat:user-1' }, + }); }); it('runs the turn under the controller of the provider turn signal', async () => { @@ -426,6 +466,39 @@ describe('SystemAgentExecutionService', () => { }); }); + it('attaches the attachment store to a resumed turn', async () => { + const { service, handle, checkpointStorage } = setup([textChunk]); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + + const result = await service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + resumeData: { approved: true }, + }); + await result.done; + + expect(handle.agent.fileStore).toHaveBeenCalled(); + }); + + it('logs a failed background resume instead of rejecting', async () => { + const { service, checkpointStorage, turnExecutionService } = setup(); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + // eslint-disable-next-line require-yield + turnExecutionService.execute.mockImplementation(async function* () { + throw new Error('resume broke'); + }); + + const result = await service.resume({ + agentId: AGENT_ID, + user, + threadId: 'thread-1', + resumeData: { approved: true }, + }); + + await expect(result.done).resolves.toBeUndefined(); + }); + it('runs the resumed turn under the controller of the provider turn signal', async () => { const { service, provider, checkpointStorage, turnExecutionService, prepared } = setup([ textChunk, @@ -510,6 +583,31 @@ describe('SystemAgentExecutionService', () => { ); }); + it('rejects when the resumed turn fails, so the route can report it', async () => { + const { service, checkpointStorage, turnExecutionService } = setup(); + checkpointStorage.load.mockResolvedValue({ + persistence: { threadId: 'thread-1' }, + } as never); + checkpointStorage.findSuspendedForThread.mockResolvedValue(suspendedCheckpoint() as never); + // eslint-disable-next-line require-yield + turnExecutionService.execute.mockImplementation(async function* () { + throw new Error('resume broke'); + }); + const send = vi.fn(); + + await expect( + service.resumeRun({ + agentId: AGENT_ID, + user, + runId: 'run-1', + toolCallId: 'tc-1', + resumeData: {}, + send, + }), + ).rejects.toThrow('resume broke'); + expect(send).not.toHaveBeenCalledWith(expect.objectContaining({ type: 'done' })); + }); + it('refuses a run without a stored checkpoint', async () => { const { service, checkpointStorage } = setup(); checkpointStorage.load.mockResolvedValue(undefined); @@ -867,6 +965,70 @@ describe('SystemAgentExecutionService', () => { ).rejects.toThrow(NotFoundError); }); + it('creates a private session for a new thread id', async () => { + const { service, threadRepository, txRunner } = setup(); + const ctx = {}; + threadRepository.findOneBy.mockResolvedValue(null); + txRunner.run.mockImplementation(async (_ctx, fn) => await fn(ctx)); + threadRepository.findOrCreate.mockResolvedValue({ + thread: { ...thread, id: 'thread-new', projectId: 'project-1' }, + created: true, + } as never); + threadRepository.renameOwned.mockResolvedValue(true); + + const created = await service.createThread({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + threadId: 'thread-new', + title: 'First chat', + }); + + expect(threadRepository.findOrCreate).toHaveBeenCalledWith( + 'thread-new', + AGENT_ID, + 'Test Assistant', + 'project-1', + { accessScope: 'user', ownerId: 'user-1' }, + ctx, + ); + expect(threadRepository.renameOwned).toHaveBeenCalledWith( + AGENT_ID, + 'user-1', + 'thread-new', + 'First chat', + ctx, + ); + expect(created).toMatchObject({ id: 'thread-new', title: 'First chat' }); + }); + + it('queues the first message of a new session in the created thread', async () => { + const { service, threadRepository, txRunner } = setup(); + threadRepository.findOwnedById.mockResolvedValue(null); + threadRepository.findOneBy.mockResolvedValue(null); + txRunner.run.mockImplementation(async (_ctx, fn) => await fn({})); + threadRepository.findOrCreate.mockResolvedValue({ + thread: { ...thread, id: 'thread-new', projectId: 'project-1' }, + created: true, + } as never); + + const input = await service.prepareChatMessage({ + agentId: AGENT_ID, + user, + projectId: 'project-1', + sessionId: 'thread-new', + message: 'Hello', + }); + + expect(input).toMatchObject({ + agentId: AGENT_ID, + projectId: 'project-1', + threadId: 'thread-new', + source: 'system_chat', + payload: { kind: 'system', userId: 'user-1', message: 'Hello' }, + }); + }); + it('refuses to create a session with the id of a thread of another user', async () => { const { service, threadRepository, txRunner } = setup(); threadRepository.findOneBy.mockResolvedValue({ ...thread, ownerId: 'user-2' } as never); diff --git a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts index a5f01638c37..59c2a06193e 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent-execution.service.ts @@ -40,6 +40,22 @@ type SuspendedToolCall = Extract< export type SystemAgentThreadStatus = 'running' | 'suspended' | 'idle'; +interface ResumeParams { + agentId: string; + user: User; + threadId: string; + resumeData: unknown; + runId?: string; + toolCallId?: string; + send?: (event: AgentSseEvent) => void; +} + +interface ResumeHandle { + runId: string; + toolCallId: string; + done: Promise; +} + /** The SDK run options of a handle, without its abort signal. `runTurn` links that signal. */ function runOptionsOf(handle: SystemAgentTurnHandle) { const { abortSignal: _abortSignal, ...runOptions } = handle.runOptions ?? {}; @@ -344,15 +360,7 @@ export class SystemAgentExecutionService { options: payload.options ?? {}, }); const text = handle.input ?? payload.message; - if (attachments.length > 0) { - // Messages keep file references. The runtime loads bytes from binary data per call. - handle.agent.fileStore( - this.attachmentService.getFileStore( - { agentId: thread.agentId, projectId: thread.projectId }, - handle.agent.snapshot.model.provider ?? '', - ), - ); - } + this.attachFileStore(handle, thread); const input = attachments.length > 0 && typeof text === 'string' ? buildInboundUserMessage(text, attachments) @@ -361,8 +369,8 @@ export class SystemAgentExecutionService { type: 'start', input, options: { - persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, ...runOptionsOf(handle), + persistence: { threadId: thread.id, resourceId, hostMetadata: handle.hostMetadata }, // The claim signal stops the turn only until the runtime admits it. abortSignal: signal, }, @@ -373,16 +381,20 @@ export class SystemAgentExecutionService { })); } - /** Resume a suspended tool call. Runs the continuation in the background. */ - async resume(params: { - agentId: string; - user: User; - threadId: string; - resumeData: unknown; - runId?: string; - toolCallId?: string; - send?: (event: AgentSseEvent) => void; - }): Promise<{ runId: string; toolCallId: string; done: Promise }> { + /** + * Resume a suspended tool call. Runs the continuation in the background: + * `done` settles when the turn ends and logs a failure instead of rejecting. + */ + async resume(params: ResumeParams): Promise { + const resumed = await this.startResume(params); + const done = resumed.done.catch((error: unknown) => { + this.logger.warn('System agent resume failed', { threadId: params.threadId, error }); + }); + return { ...resumed, done }; + } + + /** Start a resume. `done` rejects when the turn fails. */ + private async startResume(params: ResumeParams): Promise { const thread = await this.getUsableThread(params.agentId, params.user, params.threadId); const provider = this.getProvider(params.agentId); const checkpoint = await this.checkpointStorage.findSuspendedForThread( @@ -412,6 +424,7 @@ export class SystemAgentExecutionService { checkpointHostMetadata: checkpoint.persistence.hostMetadata ?? {}, resumeData: params.resumeData, }); + this.attachFileStore(handle, thread); const recording: StartExecutionParams = { threadId: thread.id, agentId: thread.agentId, @@ -441,9 +454,7 @@ export class SystemAgentExecutionService { }, recording, }), - ).catch((error: unknown) => { - this.logger.warn('System agent resume failed', { threadId: thread.id, error }); - }); + ); return { runId, toolCallId: pending.toolCallId, done }; } @@ -517,7 +528,8 @@ export class SystemAgentExecutionService { const state = await this.checkpointStorage.load(params.runId, params.agentId); const threadId = state?.persistence?.threadId; if (!threadId) throw new UserError('This action is no longer waiting for input'); - const { done } = await this.resume({ + // A failure rejects, so the route can send an error event. + const { done } = await this.startResume({ agentId: params.agentId, user: params.user, threadId, @@ -549,6 +561,19 @@ export class SystemAgentExecutionService { }); } + /** + * Attach the attachment store on every turn. Memory keeps file references, + * and without the store a later turn cannot load the files of earlier turns. + */ + private attachFileStore(handle: SystemAgentTurnHandle, thread: AgentExecutionThread): void { + handle.agent.fileStore( + this.attachmentService.getFileStore( + { agentId: thread.agentId, projectId: thread.projectId }, + handle.agent.snapshot.model.provider ?? '', + ), + ); + } + /** * Run a prepared turn under `controller`, the controller the provider got * the signal of. A cancel aborts it, and the abort signal in the run @@ -611,7 +636,8 @@ export class SystemAgentExecutionService { error: settleError, }); } - if (status !== 'suspended' && status !== 'cancelled') { + // Like Preview, only a completed turn sends `done`. An error has its own event. + if (status === 'completed') { send({ type: 'done', sessionId: thread.id, executionId: executionId ?? '' }); } } diff --git a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts index fb59aa40efb..75933dc5043 100644 --- a/packages/cli/src/modules/agents/system-agents/system-agent.types.ts +++ b/packages/cli/src/modules/agents/system-agents/system-agent.types.ts @@ -61,8 +61,11 @@ export interface SystemAgentTurnHandle { input?: string | Message[]; /** Stored on the checkpoint. A resume receives it back. */ hostMetadata?: JSONObject; - /** Extra SDK run options, such as telemetry. `abortSignal` stops the turn at any time. */ - runOptions?: Partial & Partial; + /** + * Extra SDK run options, such as telemetry. `abortSignal` stops the turn at + * any time. The host sets `persistence`, so a provider cannot change it. + */ + runOptions?: Omit & Partial, 'persistence'>; /** Hide the user message from the transcript (machine turns). */ hideUserMessage?: boolean; onChunk?: (chunk: AgentExecutionStreamChunk) => void;