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 <noreply@anthropic.com>

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<typeof systemClaim>) {
+			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 () => {

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<AgentExecutionService>();
 	const chatExecutionService = mock<AgentChatExecutionService>();
 	const agentRepository = mock<AgentRepository>();
+	const attachmentService = mock<AgentChatAttachmentService>();
 
 	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<ReturnType<AgentChatAttachmentService['getFileStore']>>();
+			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);

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<void>;
+}
+
 /** 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<void> }> {
+	/**
+	 * 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<ResumeHandle> {
+		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<ResumeHandle> {
 		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 ?? '' });
 			}
 		}

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<RunOptions & ExecutionOptions> & Partial<ResumeOptions>;
+	/**
+	 * 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<RunOptions & ExecutionOptions> & Partial<ResumeOptions>, 'persistence'>;
 	/** Hide the user message from the transcript (machine turns). */
 	hideUserMessage?: boolean;
 	onChunk?: (chunk: AgentExecutionStreamChunk) => void;