fix(aggregator): emit 'thinking' only on first reasoning part

Author: grinev

src/summary/aggregator.ts

@@ -117,6 +117,7 @@ class SummaryAggregator {
   private onFileChangeCallback: FileChangeCallback | null = null;
   private onClearedCallback: ClearedCallback | null = null;
   private processedToolStates: Set<string> = new Set();
+  private thinkingFiredForMessages: Set<string> = new Set();
   private bot: Bot | null = null;
   private chatId: number | null = null;
   private typingTimer: ReturnType<typeof setInterval> | null = null;
@@ -271,6 +272,7 @@ class SummaryAggregator {
     this.messages.clear();
     this.partHashes.clear();
     this.processedToolStates.clear();
+    this.thinkingFiredForMessages.clear();
     this.messageCount = 0;
     this.lastUpdated = 0;
 
@@ -303,18 +305,6 @@ class SummaryAggregator {
         this.currentMessageParts.set(messageID, []);
         this.messageCount++;
         this.startTypingIndicator();
-
-        const isSummaryMessage = (info as { summary?: boolean }).summary === true;
-
-        // Notify that agent started thinking
-        if (!isSummaryMessage && this.onThinkingCallback) {
-          const callback = this.onThinkingCallback;
-          setImmediate(() => {
-            if (typeof callback === "function") {
-              callback(info.sessionID);
-            }
-          });
-        }
       }
 
       const pending = this.pendingParts.get(messageID) || [];
@@ -394,7 +384,20 @@ class SummaryAggregator {
     const messageID = part.messageID;
     const messageInfo = this.messages.get(messageID);
 
-    if (part.type === "text" && "text" in part && part.text) {
+    if (part.type === "reasoning") {
+      // Fire the thinking callback once per message on the first reasoning part.
+      // This is the signal that the model is actually doing extended thinking.
+      if (!this.thinkingFiredForMessages.has(messageID) && this.onThinkingCallback) {
+        this.thinkingFiredForMessages.add(messageID);
+        const callback = this.onThinkingCallback;
+        const sessionID = part.sessionID;
+        setImmediate(() => {
+          if (typeof callback === "function") {
+            callback(sessionID);
+          }
+        });
+      }
+    } else if (part.type === "text" && "text" in part && part.text) {
       const partHash = this.hashString(part.text);
 
       if (!this.partHashes.has(messageID)) {

tests/summary/aggregator.test.ts

@@ -137,7 +137,7 @@ describe("summary/aggregator", () => {
     expect(onToolFile).not.toHaveBeenCalled();
   });
 
-  it("passes sessionId to thinking callback", async () => {
+  it("passes sessionId to thinking callback when reasoning part arrives", async () => {
     const onThinking = vi.fn();
     summaryAggregator.setOnThinking(onThinking);
     summaryAggregator.setSession("session-1");
@@ -154,34 +154,101 @@ describe("summary/aggregator", () => {
       },
     } as unknown as Event);
 
+    summaryAggregator.processEvent({
+      type: "message.part.updated",
+      properties: {
+        part: {
+          id: "part-reasoning-1",
+          sessionID: "session-1",
+          messageID: "message-1",
+          type: "reasoning",
+          text: "Let me think about this...",
+          time: { start: Date.now() },
+        },
+      },
+    } as unknown as Event);
+
     await new Promise<void>((resolve) => setImmediate(resolve));
 
     expect(onThinking).toHaveBeenCalledWith("session-1");
   });
 
-  it("does not send thinking callback for summary assistant messages", async () => {
+  it("does not send thinking callback when no reasoning part arrives", async () => {
     const onThinking = vi.fn();
     summaryAggregator.setOnThinking(onThinking);
     summaryAggregator.setSession("session-1");
 
+    // Only a message.updated event without any reasoning part — should NOT trigger thinking
     summaryAggregator.processEvent({
       type: "message.updated",
       properties: {
         info: {
-          id: "message-summary",
+          id: "message-no-reasoning",
           sessionID: "session-1",
           role: "assistant",
-          summary: true,
           time: { created: Date.now() },
         },
       },
     } as unknown as Event);
 
+    summaryAggregator.processEvent({
+      type: "message.part.updated",
+      properties: {
+        part: {
+          id: "part-text-1",
+          sessionID: "session-1",
+          messageID: "message-no-reasoning",
+          type: "text",
+          text: "Here is my answer.",
+          time: { start: Date.now() },
+        },
+      },
+    } as unknown as Event);
+
     await new Promise<void>((resolve) => setImmediate(resolve));
 
     expect(onThinking).not.toHaveBeenCalled();
   });
 
+  it("fires thinking callback only once per message even with multiple reasoning parts", async () => {
+    const onThinking = vi.fn();
+    summaryAggregator.setOnThinking(onThinking);
+    summaryAggregator.setSession("session-1");
+
+    summaryAggregator.processEvent({
+      type: "message.updated",
+      properties: {
+        info: {
+          id: "message-multi-reasoning",
+          sessionID: "session-1",
+          role: "assistant",
+          time: { created: Date.now() },
+        },
+      },
+    } as unknown as Event);
+
+    for (let i = 0; i < 3; i++) {
+      summaryAggregator.processEvent({
+        type: "message.part.updated",
+        properties: {
+          part: {
+            id: `part-reasoning-${i}`,
+            sessionID: "session-1",
+            messageID: "message-multi-reasoning",
+            type: "reasoning",
+            text: `Thinking step ${i}`,
+            time: { start: Date.now() },
+          },
+        },
+      } as unknown as Event);
+    }
+
+    await new Promise<void>((resolve) => setImmediate(resolve));
+
+    expect(onThinking).toHaveBeenCalledTimes(1);
+    expect(onThinking).toHaveBeenCalledWith("session-1");
+  });
+
   it("sends apply_patch payload as tool file", () => {
     const onToolFile = vi.fn();
     summaryAggregator.setOnToolFile(onToolFile);