Auto-send buffered stream output in group scope conversations

packages/apps/src/http/http-stream.spec.ts

@@ -264,6 +264,35 @@ describe('HttpStream', () => {
     );
   });
 
+  test('group-scope stream buffers and sends a single normal message on close', async () => {
+    const groupRef = {
+      ...ref,
+      conversation: { ...ref.conversation, isGroup: true },
+    };
+    const stream = new HttpStream(client, groupRef, logger);
+    mockCreate();
+
+    stream.update('Thinking...');
+    stream.emit('first message');
+    stream.emit('last message');
+    await jest.runAllTimersAsync();
+
+    expect(client.conversations.activities().create).toHaveBeenCalledTimes(0);
+
+    await stream.close();
+
+    expect(client.conversations.activities().create).toHaveBeenCalledTimes(1);
+    expect(client.conversations.activities().create).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: 'message',
+        text: 'first messagelast message',
+      })
+    );
+
+    const sent = client.conversations.activities().create.mock.calls[0][0];
+    expect(sent.channelData?.streamType).toBeUndefined();
+  });
+
   test('close waits for flush to complete before sending final message', async () => {
     mockCreate();
 

packages/apps/src/http/http-stream.ts

@@ -65,6 +65,10 @@ export class HttpStream implements IStreamer {
     this._logger = logger?.child('stream') || new ConsoleLogger('@teams/http-stream');
   }
 
+  protected get isGroupConversation(): boolean {
+    return this.ref.conversation.isGroup === true;
+  }
+
   /**
    * Emit a new activity or text to the stream.
    * @param activity Activity object or string message.
@@ -106,7 +110,13 @@ export class HttpStream implements IStreamer {
    * Waits for all queued activities to flush.
    */
   async close() {
-    if (!this.index && !this.queue.length && !this._flushing) {
+    const hasBufferedContent = this.text !== '' || this.attachments.length > 0 || this.entities.length > 0;
+    if (
+      !this.index
+      && !this.queue.length
+      && !this._flushing
+      && (!hasBufferedContent || !this.isGroupConversation)
+    ) {
       this._logger.debug('closed with no content');
       return;
     }
@@ -124,7 +134,7 @@ export class HttpStream implements IStreamer {
     // Wait until all queued activities are flushed
     const start = Date.now();
 
-    while ((this.queue.length || !this.id || this._flushing) && !this._canceled) {
+    while ((this.queue.length || (!this.id && !this.isGroupConversation) || this._flushing) && !this._canceled) {
       if (Date.now() - start > this._totalTimeout) {
         this._logger.warn('Timeout while waiting for id and queue to flush');
         return;
@@ -138,23 +148,34 @@ export class HttpStream implements IStreamer {
       return;
     }
 
-    if (!this.id) {
-      this._logger.warn('no stream id set, cannot close stream');
-      return;
-    }
-
     if (this.text === '' && !this.attachments.length) {
       this._logger.warn('no text or attachments to send, cannot close stream');
       return;
     }
 
+    const channelData = this.isGroupConversation
+      ? { ...this.channelData }
+      : this.channelData;
+    if (this.isGroupConversation && 'streamType' in channelData) {
+      delete channelData.streamType;
+    }
+
     // Build final message activity
     const activity = new MessageActivity(this.text)
-      .withId(this.id)
       .addAttachments(...this.attachments)
       .addEntities(...this.entities)
-      .withChannelData(this.channelData)
-      .addStreamFinal();
+      .withChannelData(channelData);
+
+    if (!this.isGroupConversation) {
+      const streamId = this.id;
+      if (!streamId) {
+        this._logger.warn('no stream id set, cannot close stream');
+        return;
+      }
+      activity
+        .withId(streamId)
+        .addStreamFinal();
+    }
 
     const res = await promises.retry(() => this.send(activity), {
       logger: this._logger
@@ -227,12 +248,14 @@ export class HttpStream implements IStreamer {
       if (startLength === 0) return;
 
       // Send informative updates immediately
-      for (const informativeUpdate of informativeUpdates) {
-        const activity = new TypingActivity().withText(informativeUpdate.text || '').withChannelData({ streamType: 'informative' });
-        await this.pushStreamChunk(activity);
+      if (!this.isGroupConversation) {
+        for (const informativeUpdate of informativeUpdates) {
+          const activity = new TypingActivity().withText(informativeUpdate.text || '').withChannelData({ streamType: 'informative' });
+          await this.pushStreamChunk(activity);
+        }
       }
 
-      if (this.text) {
+      if (this.text && !this.isGroupConversation) {
         const activity = new TypingActivity().withText(this.text);
         await this.pushStreamChunk(activity);
       }