ConvAgent: reassemble agent audio sub-chunks into one enqueue (fix pops)

After sub-chunking each ElevenLabs message into N ~0.8s Opus packets, the
client enqueued each sub-chunk separately, re-running the playback state
machine per sub-chunk (spurious Play() restart / pre-buffer race / underrun
at the seam) -> an audible pop at each ~0.8s boundary (2-3 per sentence).
Confirmed via multi-agent analysis + logs (no generation skips, no decode
truncation -> not a codec issue; the discontinuity is on the enqueue side).

- MulticastReceiveAgentAudio gains a bLastSubChunk flag; the server marks the
  final sub-chunk of each message (encode path frame-aligned so the last
  sub-chunk is well-defined; raw-PCM fallback likewise).
- The client accumulates decoded PCM across a message's sub-chunks and calls
  EnqueueAgentAudio (and lip-sync broadcast) ONCE, on the last sub-chunk.
- ClientReassemblyPCM reset in StopAgentAudio so a half-received message can't
  leak across a turn/interruption.

No added latency: a message's sub-chunks are sent back-to-back in one server
tick and arrive together. Each RPC still <=40 frames (Reliable, under the
encode/decode limits).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Unreal/PS_AI_Agent/Plugins/PS_AI_ConvAgent/Source/PS_AI_ConvAgent/Private/PS_AI_ConvAgent_ElevenLabsComponent.cpp

@@ -1248,10 +1248,18 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::HandleAudioReceived(const TArray<uint
 				* static_cast<int32>(sizeof(int16));            // 640 at 16kHz mono
 			static constexpr int32 MaxOpusChunkBytes = 40 * BytesPerFrame; // 25600 = 0.8s
 
+			// Frame-align the message length (Encode drops any sub-frame remainder
+			// anyway). A frame-aligned length guarantees every sub-chunk holds >=1
+			// whole frame and lets us reliably mark the FINAL sub-chunk (bLast), so
+			// the client reassembles one message into a single enqueue (no per-chunk
+			// playback re-arm → no pops).
+			const int32 UsableBytes = (PCMData.Num() / BytesPerFrame) * BytesPerFrame;
+
 			int32 Offset = 0;
-			while (Offset < PCMData.Num())
+			while (Offset < UsableBytes)
 			{
-				const int32 BlockBytes = FMath::Min(MaxOpusChunkBytes, PCMData.Num() - Offset);
+				const int32 BlockBytes = FMath::Min(MaxOpusChunkBytes, UsableBytes - Offset);
+				const bool bLast = (Offset + BlockBytes >= UsableBytes);
 
 				uint32 CompressedSize = static_cast<uint32>(OpusWorkBuffer.Num());
 				OpusEncoder->Encode(PCMData.GetData() + Offset, BlockBytes,
@@ -1266,11 +1274,11 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::HandleAudioReceived(const TArray<uint
 					{
 						const float Ratio = static_cast<float>(BlockBytes) / static_cast<float>(CompressedSize);
 						UE_LOG(LogPS_AI_ConvAgent_ElevenLabs, Log,
-							TEXT("[NET-SRV] Opus audio chunk: %u bytes sent (raw %d → %.1fx compression)."),
+							TEXT("[NET-SRV] Opus audio chunk: %u bytes sent (raw %d → %.1fx compression, last=%d)."),
-							CompressedSize, BlockBytes, Ratio);
+							CompressedSize, BlockBytes, Ratio, bLast ? 1 : 0);
 					}
 
-					MulticastReceiveAgentAudio(CompressedData);
+					MulticastReceiveAgentAudio(CompressedData, bLast);
 				}
 
 				Offset += BlockBytes;
@@ -1293,30 +1301,32 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::HandleAudioReceived(const TArray<uint
 
 			static constexpr int32 MaxChunkBytes = 4000;
 			static constexpr int32 MaxFallbackBytes = 320000; // ~80 reliable chunks/message, well under the 512-bunch buffer
-			int32 Offset = 0;
+
-			while (Offset < PCMData.Num())
+			const int32 SendableBytes = FMath::Min(PCMData.Num(), MaxFallbackBytes);
-			{
+			if (PCMData.Num() > MaxFallbackBytes)
-				if (Offset >= MaxFallbackBytes)
 			{
 				UE_LOG(LogPS_AI_ConvAgent_ElevenLabs, Warning,
 					TEXT("[NET-SRV] Raw-PCM fallback capped at %d bytes — dropping %d-byte tail "
 						 "to protect the reliable buffer."),
-						MaxFallbackBytes, PCMData.Num() - Offset);
+					MaxFallbackBytes, PCMData.Num() - MaxFallbackBytes);
-					break;
 			}
 
-				const int32 ChunkSize = FMath::Min(MaxChunkBytes, PCMData.Num() - Offset);
+			int32 Offset = 0;
+			while (Offset < SendableBytes)
+			{
+				const int32 ChunkSize = FMath::Min(MaxChunkBytes, SendableBytes - Offset);
+				const bool bLast = (Offset + ChunkSize >= SendableBytes);
 				TArray<uint8> Chunk;
 				Chunk.Append(PCMData.GetData() + Offset, ChunkSize);
 
 				if (bNetAudioDebug)
 				{
 					UE_LOG(LogPS_AI_ConvAgent_ElevenLabs, Warning,
-						TEXT("[NET-SRV] RAW PCM chunk: %d bytes sent (NO compression — Opus disabled)."),
+						TEXT("[NET-SRV] RAW PCM chunk: %d bytes sent (NO compression — Opus disabled, last=%d)."),
-						ChunkSize);
+						ChunkSize, bLast ? 1 : 0);
 				}
 
-				MulticastReceiveAgentAudio(Chunk);
+				MulticastReceiveAgentAudio(Chunk, bLast);
 				Offset += ChunkSize;
 			}
 		}
@@ -1882,6 +1892,7 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::StopAgentAudio()
 	bool bWasSpeaking = false;
 	double Now = 0.0;
 	bPreBuffering = false;      // Clear pre-buffer state on stop.
+	ClientReassemblyPCM.Reset(); // Drop any half-received message so it can't leak into the next turn.
 	{
 		FScopeLock Lock(&AudioQueueLock);
 		AudioQueue.Empty();
@@ -2546,7 +2557,7 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::ClientConversationFailed_Implementati
 // Network: Multicast RPCs
 // ─────────────────────────────────────────────────────────────────────────────
 void UPS_AI_ConvAgent_ElevenLabsComponent::MulticastReceiveAgentAudio_Implementation(
-	const TArray<uint8>& AudioData)
+	const TArray<uint8>& AudioData, bool bLastSubChunk)
 {
 	// Server already handled playback in HandleAudioReceived.
 	if (GetOwnerRole() == ROLE_Authority) return;
@@ -2574,8 +2585,7 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::MulticastReceiveAgentAudio_Implementa
 		uint32 DecompressedSize = MaxDecompressedSize;
 		OpusDecoder->Decode(AudioData.GetData(), AudioData.Num(),
 			PCMBuffer.GetData(), DecompressedSize);
-		if (DecompressedSize == 0) return;
+		PCMBuffer.SetNum(DecompressedSize);   // may be 0 — handled by reassembly below
-		PCMBuffer.SetNum(DecompressedSize);
 	}
 	else
 	{
@@ -2583,13 +2593,33 @@ void UPS_AI_ConvAgent_ElevenLabsComponent::MulticastReceiveAgentAudio_Implementa
 		PCMBuffer = AudioData;
 	}
 
-	// Local playback.
+	// Reassemble one ElevenLabs message from its N sub-chunks, then enqueue ONCE.
-	EnqueueAgentAudio(PCMBuffer);
+	// The server splits a message into sub-chunks (to respect the Opus 76-frame /
+	// client decode-buffer limits), sent back-to-back in the same tick. Enqueuing
+	// each sub-chunk separately re-ran the playback state machine → a pop at each
+	// ~0.8s seam. Accumulating and enqueuing once per message plays it contiguously.
+	if (PCMBuffer.Num() > 0)
+	{
+		ClientReassemblyPCM.Append(PCMBuffer);
+	}
+
+	if (!bLastSubChunk)
+	{
+		return;   // wait for the remaining sub-chunks of this message
+	}
+
+	if (ClientReassemblyPCM.Num() > 0)
+	{
+		// Local playback (one contiguous block per message).
+		EnqueueAgentAudio(ClientReassemblyPCM);
 
 		// Feed lip-sync (within LOD or speaker).
 		if (bIsSpeaker || LipSyncLODDistance <= 0.f || Dist <= LipSyncLODDistance)
 		{
-		OnAgentAudioData.Broadcast(PCMBuffer);
+			OnAgentAudioData.Broadcast(ClientReassemblyPCM);
+		}
+
+		ClientReassemblyPCM.Reset();
 	}
 }
 

Unreal/PS_AI_Agent/Plugins/PS_AI_ConvAgent/Source/PS_AI_ConvAgent/Public/PS_AI_ConvAgent_ElevenLabsComponent.h

@@ -494,9 +494,12 @@ public:
 	 *  sub-chunk is ~2 KB (2-3 partial bunches, below the 8-bunch reliable-
 	 *  promotion threshold) at ~1.25 RPC/s, so in-flight bunches stay in single
 	 *  digits vs the 512-bunch buffer. Sub-chunking in HandleAudioReceived is what
-	 *  keeps packets this small — it must stay. */
+	 *  keeps packets this small — it must stay.
+	 *  bLastSubChunk marks the final sub-chunk of one ElevenLabs message so the
+	 *  client can reassemble all sub-chunks and enqueue ONCE per message (enqueuing
+	 *  each sub-chunk separately re-ran the playback state machine → pops). */
 	UFUNCTION(NetMulticast, Reliable)
-	void MulticastReceiveAgentAudio(const TArray<uint8>& AudioData);
+	void MulticastReceiveAgentAudio(const TArray<uint8>& AudioData, bool bLastSubChunk);
 
 	/** Notify all clients that the agent started speaking (first audio chunk). */
 	UFUNCTION(NetMulticast, Reliable)
@@ -804,6 +807,11 @@ private:
 	int32 AudioQueueReadOffset = 0;
 	FCriticalSection AudioQueueLock;
 
+	// Client-only: accumulates decoded PCM across the N sub-chunks of one
+	// ElevenLabs message; flushed to EnqueueAgentAudio() once, on the last
+	// sub-chunk (bLastSubChunk). Game-thread only. Reset in StopAgentAudio().
+	TArray<uint8> ClientReassemblyPCM;
+
 	// Pre-buffer state: delay playback start to absorb TTS inter-chunk gaps.
 	bool bPreBuffering = false;
 	double PreBufferStartTime = 0.0;