Method collectBatch

cli/src/utils/MessageQueue2.ts:341–384

* Collect a batch of messages with the same mode, respecting isolation requirements

()

     * Collect a batch of messages with the same mode, respecting isolation requirements
     */
    private collectBatch(): { message: string, mode: T, hash: string, isolate: boolean } | null {
        if (this.queue.length === 0) {
            return null;
        }

        const firstItem = this.queue[0];
        const sameModeMessages: string[] = [];
        const consumedLocalIds: string[] = [];
        let mode = firstItem.mode;
        let isolate = firstItem.isolate ?? false;
        const targetModeHash = firstItem.modeHash;

        // If the first message requires isolation, only process it alone
        if (firstItem.isolate) {
            const item = this.queue.shift()!;
            sameModeMessages.push(item.message);
            if (item.localId) consumedLocalIds.push(item.localId);
            logger.debug(`[MessageQueue2] Collected isolated message with mode hash: ${targetModeHash}`);
        } else {
            // Collect all messages with the same mode until we hit an isolated message
            while (this.queue.length > 0 &&
                this.queue[0].modeHash === targetModeHash &&
                !this.queue[0].isolate) {
                const item = this.queue.shift()!;
                sameModeMessages.push(item.message);
                if (item.localId) consumedLocalIds.push(item.localId);
            }
            logger.debug(`[MessageQueue2] Collected batch of ${sameModeMessages.length} messages with mode hash: ${targetModeHash}`);
        }

        // Join all messages with newlines
        const combinedMessage = sameModeMessages.join('\n');

        if (consumedLocalIds.length > 0) {
            this.onBatchConsumed?.(consumedLocalIds);
        }

        return {
            message: combinedMessage,
            mode,
            hash: targetModeHash,
            isolate
        };
    }

    /**
     * Wait for messages to arrive