feat(core): Post a budget-stop message on external agent channels (#40767)

packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts

@@ -354,3 +354,89 @@ describe('AgentChatStreamConsumer — suspension routing', () => {
 		expect(handleSuspension).toHaveBeenCalledWith(SUSPENDED_CHUNK, thread, undefined);
 	});
 });
+
+const SESSION_BUDGET_MESSAGE =
+	'⚠️ This session reached its cost cap and stopped. Open the agent in n8n and increase the cap.';
+const MONTHLY_BUDGET_MESSAGE =
+	'⚠️ This agent reached its monthly budget and stopped. Open the agent in n8n and increase the cap.';
+
+describe('AgentChatStreamConsumer — budget stop', () => {
+	it('posts the session message once and clears the status when the run never starts', async () => {
+		const postErrorToThread = vi.fn().mockResolvedValue(undefined);
+		const consumer = makeConsumer(postErrorToThread);
+		const localThread = mock<Thread<unknown, unknown>>();
+		const clearBeforeResponse = vi.fn().mockResolvedValue(undefined);
+
+		await consumer.consume(
+			makeStream([
+				{ type: 'finish', finishReason: 'guardrail', guardrail: { code: 'budget.session' } },
+			]),
+			localThread,
+			{ statusHandle: { clearBeforeResponse } },
+		);
+
+		expect(localThread.post).toHaveBeenCalledTimes(1);
+		expect(localThread.post).toHaveBeenCalledWith({ markdown: SESSION_BUDGET_MESSAGE });
+		expect(clearBeforeResponse).toHaveBeenCalledTimes(1);
+		expect(postErrorToThread).not.toHaveBeenCalled();
+	});
+
+	it.each([
+		['streaming', false],
+		['buffered', true],
+	])('posts partial text and then the monthly message (%s)', async (_name, disableStreaming) => {
+		const { thread: localThread, streamedText, discrete } = makeStreamingThread();
+
+		await makeStreamingConsumer({ disableStreaming }).consume(
+			makeStream([
+				{ type: 'text-delta', id: 't-1', delta: 'Partial answer' },
+				{ type: 'finish', finishReason: 'guardrail', guardrail: { code: 'budget.monthly' } },
+			]),
+			localThread,
+		);
+
+		if (disableStreaming) {
+			expect(discrete).toEqual([
+				{ markdown: 'Partial answer' },
+				{ markdown: MONTHLY_BUDGET_MESSAGE },
+			]);
+			return;
+		}
+		expect(streamedText).toEqual(['Partial answer']);
+		expect(discrete).toEqual([{ markdown: MONTHLY_BUDGET_MESSAGE }]);
+	});
+
+	it('does not post a budget message when the run finishes normally', async () => {
+		const { thread: localThread, discrete } = makeStreamingThread();
+
+		await makeStreamingConsumer({ disableStreaming: true }).consume(
+			makeStream([
+				{ type: 'text-delta', id: 't-1', delta: 'All done' },
+				{ type: 'finish', finishReason: 'stop' },
+			]),
+			localThread,
+		);
+
+		expect(discrete).toEqual([{ markdown: 'All done' }]);
+	});
+
+	it('reports a rejected budget post once and does not post the budget line again', async () => {
+		const postErrorToThread = vi.fn().mockResolvedValue(undefined);
+		const consumer = makeConsumer(postErrorToThread);
+		const localThread = mock<Thread<unknown, unknown>>();
+		const deliveryError = new Error('delivery failed');
+		localThread.post.mockRejectedValue(deliveryError);
+
+		await consumer.consume(
+			makeStream([
+				{ type: 'finish', finishReason: 'guardrail', guardrail: { code: 'budget.session' } },
+			]),
+			localThread,
+		);
+
+		expect(localThread.post).toHaveBeenCalledTimes(1);
+		expect(localThread.post).toHaveBeenCalledWith({ markdown: SESSION_BUDGET_MESSAGE });
+		expect(postErrorToThread).toHaveBeenCalledTimes(1);
+		expect(postErrorToThread).toHaveBeenCalledWith(localThread, deliveryError, undefined);
+	});
+});

packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts

@@ -96,6 +96,18 @@ const createResponseState = (options: ConsumeStreamOptions): ResponseState => ({
 	quietErrors: options.quietErrors === true,
 });
 
+const BUDGET_STOP_TEXT = {
+	'budget.session':
+		'⚠️ This session reached its cost cap and stopped. Open the agent in n8n and increase the cap.',
+	'budget.monthly':
+		'⚠️ This agent reached its monthly budget and stopped. Open the agent in n8n and increase the cap.',
+} as const;
+
+function budgetStopText(code: string | undefined): string | undefined {
+	if (code === 'budget.session' || code === 'budget.monthly') return BUDGET_STOP_TEXT[code];
+	return undefined;
+}
+
 export class AgentChatStreamConsumer {
 	constructor(private readonly options: AgentChatStreamConsumerOptions) {}
 
@@ -222,6 +234,7 @@ export class AgentChatStreamConsumer {
 			endStreamingPost,
 		});
 		const responseState = createResponseState(options);
+		let budgetStop: string | undefined;
 
 		try {
 			for await (const chunk of stream) {
@@ -267,13 +280,20 @@ export class AgentChatStreamConsumer {
 							pendingText = '';
 						}
 						break;
+					case 'finish':
+						budgetStop ??= budgetStopText(chunk.guardrail?.code);
+						break;
 					default:
-						// Ignore non-user-visible chunks (reasoning, finish,
-						// tool-input-*, start-step, finish-step, etc.)
+						// Ignore non-user-visible chunks (reasoning, tool-input-*,
+						// start-step, finish-step, etc.)
 						break;
 				}
 			}
-			await this.postFallbackIfNeeded(responseState, responseLifecycle, thread);
+			if (budgetStop !== undefined) {
+				await this.postBudgetStop(thread, budgetStop, responseState, responseLifecycle);
+			} else {
+				await this.postFallbackIfNeeded(responseState, responseLifecycle, thread);
+			}
 		} finally {
 			await responseLifecycle.finish();
 		}
@@ -327,6 +347,18 @@ export class AgentChatStreamConsumer {
 		}
 	}
 
+	private async postBudgetStop(
+		thread: Thread<unknown, unknown>,
+		text: string,
+		state: ResponseState,
+		lifecycle: ResponseLifecycle,
+		throwOnDeliveryError = false,
+	): Promise<void> {
+		await lifecycle.startDiscreteResponse();
+		await this.postBufferedText(thread, text, state, throwOnDeliveryError);
+		state.hasVisibleResponse = true;
+	}
+
 	private async postError(
 		thread: Thread<unknown, unknown>,
 		error: unknown,
@@ -436,6 +468,7 @@ export class AgentChatStreamConsumer {
 			await this.postBufferedText(thread, text, responseState, options.throwOnDeliveryError);
 			responseState.hasVisibleResponse = true;
 		};
+		let budgetStop: string | undefined;
 
 		try {
 			for await (const chunk of stream) {
@@ -488,17 +521,30 @@ export class AgentChatStreamConsumer {
 							buffer = '';
 						}
 						break;
+					case 'finish':
+						budgetStop ??= budgetStopText(chunk.guardrail?.code);
+						break;
 					default:
 						break;
 				}
 			}
 			await flushBuffer();
-			await this.postFallbackIfNeeded(
-				responseState,
-				responseLifecycle,
-				thread,
-				options.throwOnDeliveryError,
-			);
+			if (budgetStop !== undefined) {
+				await this.postBudgetStop(
+					thread,
+					budgetStop,
+					responseState,
+					responseLifecycle,
+					options.throwOnDeliveryError,
+				);
+			} else {
+				await this.postFallbackIfNeeded(
+					responseState,
+					responseLifecycle,
+					thread,
+					options.throwOnDeliveryError,
+				);
+			}
 		} finally {
 			await flushBuffer();
 			await responseLifecycle.finish();