diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/__tests__/stream-agent-chat.job.spec.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/__tests__/stream-agent-chat.job.spec.ts
index 34e48ac5b1..238d8efafb 100644
--- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/__tests__/stream-agent-chat.job.spec.ts
+++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/__tests__/stream-agent-chat.job.spec.ts
@@ -87,14 +87,14 @@ describe('StreamAgentChatJob', () => {
     streamChatRejection,
     addMessageRejection,
     assistantPersistRejection,
-    assistantMessageExistedAtStart = false,
+    totalsUpdateAffected = 1,
   }: {
     workspaceFound?: boolean;
     chatStream?: ReturnType<typeof createFakeChatStream>;
     streamChatRejection?: Error;
     addMessageRejection?: Error;
     assistantPersistRejection?: Error;
-    assistantMessageExistedAtStart?: boolean;
+    totalsUpdateAffected?: number;
   } = {}) => {
     const publishedEvents: PublishedEvent[] = [];
 
@@ -102,7 +102,14 @@ describe('StreamAgentChatJob', () => {
       findOne: jest
         .fn()
         .mockResolvedValue({ id: 'thread-id', deletedAt: null }),
-      update: jest.fn().mockResolvedValue({ affected: 1 }),
+      update: jest.fn().mockImplementation((_workspaceId, _criteria, values) =>
+        Promise.resolve({
+          affected:
+            values && typeof values.totalInputTokens === 'function'
+              ? totalsUpdateAffected
+              : 1,
+        }),
+      ),
     };
     const workspaceRepository = {
       findOne: jest.fn().mockResolvedValue(workspaceFound ? workspace : null),
@@ -114,9 +121,6 @@ describe('StreamAgentChatJob', () => {
       upsertAssistantMessage: assistantPersistRejection
         ? jest.fn().mockRejectedValue(assistantPersistRejection)
         : jest.fn().mockResolvedValue(undefined),
-      hasMessageById: jest
-        .fn()
-        .mockResolvedValue(assistantMessageExistedAtStart),
       generateTitleIfNeeded: jest.fn().mockResolvedValue(null),
       notifyThreadUsageUpdated: jest.fn().mockResolvedValue(undefined),
     };
@@ -210,7 +214,7 @@ describe('StreamAgentChatJob', () => {
     );
     expect(threadRepository.update).toHaveBeenCalledWith(
       'workspace-id',
-      { id: 'thread-id' },
+      { id: 'thread-id', activeStreamId: 'stream-id' },
       expect.objectContaining({ lastStreamError: null }),
     );
     expect(threadRepository.update).toHaveBeenCalledWith(
@@ -221,9 +225,9 @@ describe('StreamAgentChatJob', () => {
     expect(agentChatStreamingService.flushNextQueuedMessage).toHaveBeenCalled();
   });
 
-  it('persists the assistant message but does not re-apply thread totals when a prior execution already persisted it', async () => {
+  it('gates the thread totals on still owning the stream so a prior completion is not double-counted', async () => {
     const { job, agentChatService, threadRepository } = buildJob({
-      assistantMessageExistedAtStart: true,
+      totalsUpdateAffected: 0,
     });
 
     await job.handle(jobData);
@@ -231,18 +235,23 @@ describe('StreamAgentChatJob', () => {
     expect(agentChatService.upsertAssistantMessage).toHaveBeenCalledWith(
       expect.objectContaining({ turnId: 'turn-id' }),
     );
-    // The thread-totals accumulation must not run twice for the same stream.
-    const totalsUpdate = threadRepository.update.mock.calls.find(
-      ([, criteria]) =>
-        criteria &&
-        typeof criteria === 'object' &&
-        !('activeStreamId' in criteria),
+    expect(threadRepository.update).toHaveBeenCalledWith(
+      'workspace-id',
+      { id: 'thread-id', activeStreamId: 'stream-id' },
+      expect.objectContaining({ lastStreamError: null }),
     );
-
-    expect(totalsUpdate).toBeUndefined();
     expect(agentChatService.notifyThreadUsageUpdated).not.toHaveBeenCalled();
   });
 
+  it('applies thread totals when the claim is still held even if the message already exists from a checkpoint', async () => {
+    const { job, agentChatService } = buildJob({ totalsUpdateAffected: 1 });
+
+    await job.handle(jobData);
+
+    expect(agentChatService.upsertAssistantMessage).toHaveBeenCalled();
+    expect(agentChatService.notifyThreadUsageUpdated).toHaveBeenCalled();
+  });
+
   it('never publishes the opaque error chunk to subscribers', async () => {
     const { job, publishedEvents } = buildJob({
       chatStream: createFakeChatStream({
@@ -382,12 +391,17 @@ describe('StreamAgentChatJob', () => {
   it('resolves without flushing the queue when the stream is cancelled', async () => {
     let triggerCancel: (() => void) | undefined;
 
-    const { job, publishedEvents, agentChatStreamingService, cancelCallbacks } =
-      buildJob({
-        chatStream: createFakeChatStream({
-          onFirstChunk: () => triggerCancel?.(),
-        }),
-      });
+    const {
+      job,
+      publishedEvents,
+      agentChatService,
+      agentChatStreamingService,
+      cancelCallbacks,
+    } = buildJob({
+      chatStream: createFakeChatStream({
+        onFirstChunk: () => triggerCancel?.(),
+      }),
+    });
 
     triggerCancel = () => cancelCallbacks.forEach((callback) => callback());
 
@@ -399,5 +413,6 @@ describe('StreamAgentChatJob', () => {
     expect(
       agentChatStreamingService.flushNextQueuedMessage,
     ).not.toHaveBeenCalled();
+    expect(agentChatService.notifyThreadUsageUpdated).toHaveBeenCalled();
   });
 });
diff --git a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts
index 84aef08ec8..c43188dd72 100644
--- a/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts
+++ b/packages/twenty-server/src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job.ts
@@ -227,12 +227,6 @@ export class StreamAgentChatJob {
       ASSISTANT_MESSAGE_ID_NAMESPACE,
     );
 
-    const assistantMessageExistedAtStart =
-      await this.agentChatService.hasMessageById({
-        id: assistantMessageId,
-        workspaceId: data.workspaceId,
-      });
-
     return new Promise<void>((resolve, reject) => {
       let streamUsage = {
         inputTokens: 0,
@@ -271,7 +265,13 @@ export class StreamAgentChatJob {
         resolveStreamFinished = res;
       });
 
-      abortSignal.addEventListener('abort', () => resolve(), { once: true });
+      abortSignal.addEventListener(
+        'abort',
+        () => {
+          void streamFinishedPromise.then(() => resolve());
+        },
+        { once: true },
+      );
 
       const uiStream = createUIMessageStream<ExtendedUIMessage>({
         execute: async ({ writer }) => {
@@ -352,7 +352,7 @@ export class StreamAgentChatJob {
                   await persistChain;
                   await this.handleStreamFinish({
                     assistantMessageId,
-                    assistantMessageExistedAtStart,
+                    streamId: data.streamId,
                     responseMessage,
                     isAborted,
                     streamError,
@@ -572,7 +572,7 @@ export class StreamAgentChatJob {
 
   private async handleStreamFinish({
     assistantMessageId,
-    assistantMessageExistedAtStart,
+    streamId,
     responseMessage,
     isAborted,
     streamError,
@@ -587,7 +587,7 @@ export class StreamAgentChatJob {
     userMessagePromise,
   }: {
     assistantMessageId: string;
-    assistantMessageExistedAtStart: boolean;
+    streamId: string;
     responseMessage: Omit<ExtendedUIMessage, 'id'>;
     isAborted: boolean;
     streamError: unknown;
@@ -649,7 +649,7 @@ export class StreamAgentChatJob {
         parts: responseMessage.parts,
         workspaceId,
       });
-    } else if (!assistantMessageExistedAtStart) {
+    } else {
       await this.agentChatService.addMessage({
         threadId,
         uiMessage: responseMessage,
@@ -658,13 +658,9 @@ export class StreamAgentChatJob {
       });
     }
 
-    if (assistantMessageExistedAtStart) {
-      return;
-    }
-
-    await this.threadRepository.update(
+    const totalsUpdate = await this.threadRepository.update(
       workspaceId,
-      { id: threadId },
+      { id: threadId, activeStreamId: streamId },
       {
         totalInputTokens: () =>
           `"totalInputTokens" + ${streamUsage.inputTokens}`,
@@ -687,6 +683,10 @@ export class StreamAgentChatJob {
       },
     );
 
+    if (!totalsUpdate.affected) {
+      return;
+    }
+
     await this.agentChatService.notifyThreadUsageUpdated({
       threadId,
       userWorkspaceId,