From 4c736204589dbb3ae8d2705b25326d3f6f183980 Mon Sep 17 00:00:00 2001 From: Ben Date: Thu, 13 Aug 2026 12:34:47 +0200 Subject: [PATCH] Stream Whisper segments so the UI can show live transcript text. Consume faster-whisper SSE instead of blocking on one JSON response, persist growing text and progress over Reverb, and drop the fake percent heartbeat. --- app/Jobs/TranscribeRecording.php | 10 +- app/Models/Recording.php | 12 +- app/Services/TranscriptionService.php | 229 +++++++++++++----- app/Services/WhisperTranscriptionStream.php | 161 ++++++++++++ .../views/livewire/recordings/show.blade.php | 7 + tests/Feature/RecordingUploadTest.php | 115 ++++++++- tests/Feature/Recordings/ShowTest.php | 24 ++ tests/Unit/TranscriptionServiceTest.php | 63 ----- tests/Unit/WhisperTranscriptionStreamTest.php | 91 +++++++ 9 files changed, 574 insertions(+), 138 deletions(-) create mode 100644 app/Services/WhisperTranscriptionStream.php delete mode 100644 tests/Unit/TranscriptionServiceTest.php create mode 100644 tests/Unit/WhisperTranscriptionStreamTest.php diff --git a/app/Jobs/TranscribeRecording.php b/app/Jobs/TranscribeRecording.php index 6b0c4bd..d01fd59 100644 --- a/app/Jobs/TranscribeRecording.php +++ b/app/Jobs/TranscribeRecording.php @@ -76,9 +76,11 @@ class TranscribeRecording implements ShouldQueue try { $text = $transcription->transcribe( $this->recording, - function (string $message, int $percent): void { - $this->reportIfOwned($message, $percent); + function (string $message, int $percent, ?string $partialTranscript = null): void { + $this->reportIfOwned($message, $percent, $partialTranscript); }, + fn (): bool => Recording::query()->find($this->recording->id) + ?->ownsTranscriptionRun($this->runStartedAt) ?? false, ); if (! $this->claimSuccessfulTranscript($text)) { @@ -170,12 +172,12 @@ class TranscribeRecording implements ShouldQueue return false; } - private function reportIfOwned(string $message, int $percent): void + private function reportIfOwned(string $message, int $percent, ?string $partialTranscript = null): void { if (! $this->recording->ownsTranscriptionRun($this->runStartedAt)) { return; } - $this->recording->reportProgress($message, $percent); + $this->recording->reportProgress($message, $percent, partialTranscript: $partialTranscript); } } diff --git a/app/Models/Recording.php b/app/Models/Recording.php index d753624..9c6fcbc 100644 --- a/app/Models/Recording.php +++ b/app/Models/Recording.php @@ -502,14 +502,20 @@ class Recording extends Model /** * Update the live progress fields shown in the UI. */ - public function reportProgress(string $message, int $percent, string $status = 'processing'): void + public function reportProgress(string $message, int $percent, string $status = 'processing', ?string $partialTranscript = null): void { - $this->forceFill([ + $attributes = [ 'transcription_status' => $status, 'transcription_progress' => $message, 'transcription_percent' => max(0, min(100, $percent)), 'transcription_error' => null, - ])->save(); + ]; + + if ($partialTranscript !== null) { + $attributes['transcript'] = $partialTranscript; + } + + $this->forceFill($attributes)->save(); RecordingTranscriptionUpdated::dispatch($this->fresh()); } diff --git a/app/Services/TranscriptionService.php b/app/Services/TranscriptionService.php index 9e183b5..bc98d1a 100644 --- a/app/Services/TranscriptionService.php +++ b/app/Services/TranscriptionService.php @@ -4,9 +4,7 @@ namespace App\Services; use App\Models\Recording; use Closure; -use GuzzleHttp\Promise\PromiseInterface; use Illuminate\Http\Client\PendingRequest; -use Illuminate\Http\Client\Promises\LazyPromise; use Illuminate\Http\Client\Response; use Illuminate\Support\Facades\Http; use Laravel\Ai\Transcription; @@ -14,14 +12,17 @@ use RuntimeException; class TranscriptionService { + public function __construct(private WhisperTranscriptionStream $stream) {} + /** * Transcribe a recording with the local faster-whisper server. * - * @param (Closure(string, int): void)|null $onProgress + * @param (Closure(string, int, ?string): void)|null $onProgress + * @param (Closure(): bool)|null $shouldContinue */ - public function transcribe(Recording $recording, ?Closure $onProgress = null): string + public function transcribe(Recording $recording, ?Closure $onProgress = null, ?Closure $shouldContinue = null): string { - $report = $onProgress ?? static fn (string $message, int $percent) => null; + $report = $onProgress ?? static fn (string $message, int $percent, ?string $partial = null) => null; $model = config('ai.local_whisper_model', 'Systran/faster-whisper-base'); $report('Connecting to local faster-whisper server…', 30); @@ -32,7 +33,7 @@ class TranscriptionService ->timeout((int) config('ai.transcription_timeout', 600)) ->generate('local-whisper', $model); } else { - $transcript = $this->transcribeViaLocalWhisper($recording, $report, $model); + $transcript = $this->transcribeViaLocalWhisper($recording, $report, $model, $shouldContinue); } $report('Received transcript from local Whisper…', 85); @@ -41,11 +42,12 @@ class TranscriptionService } /** - * Call local Whisper and pulse progress while waiting. + * Call local Whisper, streaming SSE when the server supports it. * - * @param Closure(string, int): void $report + * @param Closure(string, int, ?string): void $report + * @param (Closure(): bool)|null $shouldContinue */ - private function transcribeViaLocalWhisper(Recording $recording, Closure $report, string $model): string + private function transcribeViaLocalWhisper(Recording $recording, Closure $report, string $model, ?Closure $shouldContinue): string { $path = $recording->absolutePath(); @@ -55,31 +57,165 @@ class TranscriptionService $message = "Transcribing locally with {$model} (audio stays on this machine)…"; $filename = $recording->original_filename ?: basename($path); - $request = $this->localWhisperRequest($filename, $path); - if ($this->shouldTranscribeSynchronously()) { - return $this->extractTranscriptText($request->post('audio/transcriptions', [ - 'model' => $model, - 'response_format' => 'json', - ])); - } - - /** @var LazyPromise $promise */ - $promise = $request - ->async() + $response = $this->localWhisperRequest($filename, $path) + ->withOptions(['stream' => true]) ->post('audio/transcriptions', [ 'model' => $model, 'response_format' => 'json', + 'stream' => 'true', + 'without_timestamps' => 'false', ]); - $promise->buildPromise(); + if (! $response->successful()) { + throw new RuntimeException( + 'Local transcription failed (HTTP '.$response->status().'): '.$response->body() + ); + } - $this->pulseProgressWhilePending($promise, $report, $message); + $contentType = strtolower((string) $response->header('Content-Type')); - /** @var Response $response */ - $response = $promise->wait(); + if (! str_contains($contentType, 'text/event-stream')) { + return $this->extractTranscriptText($response); + } - return $this->extractTranscriptText($response); + return $this->consumeWhisperStream($response, $report, $message, $recording->duration_seconds, $shouldContinue); + } + + /** + * @param Closure(string, int, ?string): void $report + * @param (Closure(): bool)|null $shouldContinue + */ + private function consumeWhisperStream( + Response $response, + Closure $report, + string $message, + ?int $durationSeconds, + ?Closure $shouldContinue, + ): string { + $body = $response->toPsrResponse()->getBody(); + $buffer = ''; + $accumulated = ''; + $lastPercent = 50; + $eventsWithoutTimestamp = 0; + $lastFlushAt = 0.0; + $pending = false; + + try { + while (! $body->eof()) { + if ($shouldContinue !== null && ! $shouldContinue()) { + $body->close(); + + break; + } + + $chunk = $body->read(8192); + + if ($chunk === '') { + break; + } + + $buffer .= $chunk; + + foreach ($this->stream->extractPayloads($buffer) as $payload) { + [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending] = $this->ingestEvent( + $payload, + $accumulated, + $lastPercent, + $eventsWithoutTimestamp, + $durationSeconds, + $report, + $message, + $lastFlushAt, + $pending, + ); + } + } + + foreach ($this->stream->flushBuffer($buffer) as $payload) { + [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending] = $this->ingestEvent( + $payload, + $accumulated, + $lastPercent, + $eventsWithoutTimestamp, + $durationSeconds, + $report, + $message, + $lastFlushAt, + $pending, + ); + } + + if ($pending && $accumulated !== '') { + $report($message, $lastPercent, $accumulated); + } + } finally { + $response->close(); + } + + if ($accumulated === '') { + throw new RuntimeException('Local transcription returned an empty transcript.'); + } + + return $accumulated; + } + + /** + * @param Closure(string, int, ?string): void $report + * @return array{0: string, 1: int, 2: int, 3: bool} + */ + private function ingestEvent( + string $payload, + string $accumulated, + int $lastPercent, + int $eventsWithoutTimestamp, + ?int $durationSeconds, + Closure $report, + string $message, + float &$lastFlushAt, + bool $pending, + ): array { + $event = $this->stream->parseEvent($payload); + + if ($event === null) { + return [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending]; + } + + $wasEmpty = $accumulated === ''; + $accumulated = $this->stream->applyEvent($accumulated, $event); + + if ($accumulated === '') { + return [$accumulated, $lastPercent, $eventsWithoutTimestamp, $pending]; + } + + $percent = $this->percentForEvent($event, $durationSeconds, $lastPercent, $eventsWithoutTimestamp); + $lastPercent = max($lastPercent, $percent); + + $now = microtime(true); + $shouldFlush = $wasEmpty || $event['done'] || ($now - $lastFlushAt) >= 1.0; + + if ($shouldFlush) { + $report($message, $lastPercent, $accumulated); + $lastFlushAt = $now; + + return [$accumulated, $lastPercent, $eventsWithoutTimestamp, false]; + } + + return [$accumulated, $lastPercent, $eventsWithoutTimestamp, true]; + } + + /** + * @param array{append: ?string, replace: ?string, end: ?float, done: bool, legacy: bool} $event + */ + private function percentForEvent(array $event, ?int $durationSeconds, int $lastPercent, int &$eventsWithoutTimestamp): int + { + if ($event['end'] !== null && $durationSeconds !== null && $durationSeconds > 0) { + return (int) min(99, max(50, round(100 * $event['end'] / $durationSeconds))); + } + + $eventsWithoutTimestamp++; + + return min(84, max($lastPercent, 50 + $eventsWithoutTimestamp)); } private function localWhisperRequest(string $filename, string $path): PendingRequest @@ -91,25 +227,12 @@ class TranscriptionService return Http::baseUrl($baseUrl) ->withHeaders(['Authorization' => 'Bearer '.($config['key'] ?? 'not-needed')]) ->timeout($timeout) + ->connectTimeout(10) ->attach('file', fopen($path, 'r'), $filename); } - /** - * Laravel's HTTP fake does not resolve async promises; use sync in tests. - */ - private function shouldTranscribeSynchronously(): bool - { - return app()->runningUnitTests(); - } - private function extractTranscriptText(Response $response): string { - if (! $response->successful()) { - throw new RuntimeException( - 'Local transcription failed (HTTP '.$response->status().'): '.$response->body() - ); - } - $text = $response->json('text'); if (! is_string($text) || $text === '') { @@ -118,32 +241,4 @@ class TranscriptionService return $text; } - - /** - * Broadcast incremental progress while Whisper is working. - * - * @param Closure(string, int): void $report - */ - private function pulseProgressWhilePending(LazyPromise $promise, Closure $report, string $message, int $floor = 50, int $ceiling = 84): void - { - $lastPercent = $floor; - $lastPulseAt = microtime(true); - $startedAt = microtime(true); - - while ($promise->getState() === PromiseInterface::PENDING) { - $now = microtime(true); - - if ($now - $lastPulseAt >= 2.0) { - $elapsed = $now - $startedAt; - $percent = $floor + (int) floor(($ceiling - $floor) * (1 - exp(-$elapsed / 90))); - $percent = max($lastPercent + 1, min($ceiling, $percent)); - - $report($message, $percent); - $lastPercent = $percent; - $lastPulseAt = $now; - } - - usleep(250_000); - } - } } diff --git a/app/Services/WhisperTranscriptionStream.php b/app/Services/WhisperTranscriptionStream.php new file mode 100644 index 0000000..4d2f0f0 --- /dev/null +++ b/app/Services/WhisperTranscriptionStream.php @@ -0,0 +1,161 @@ + + */ + public function extractPayloads(string &$buffer): array + { + $frames = explode("\n\n", str_replace("\r\n", "\n", $buffer)); + $buffer = array_pop($frames) ?? ''; + $payloads = []; + + foreach ($frames as $frame) { + foreach ($this->dataLines($frame) as $payload) { + if ($payload === '[DONE]' || $payload === '') { + continue; + } + + $payloads[] = $payload; + } + } + + return $payloads; + } + + /** + * Flush a trailing incomplete frame at end-of-stream. + * + * @return list + */ + public function flushBuffer(string &$buffer): array + { + $trimmed = trim($buffer); + + if ($trimmed === '') { + $buffer = ''; + + return []; + } + + $buffer .= "\n\n"; + + return $this->extractPayloads($buffer); + } + + /** + * Parse one SSE JSON payload into append/replace/end/done fields. + * + * @return array{append: ?string, replace: ?string, end: ?float, done: bool, legacy: bool}|null + */ + public function parseEvent(string $json): ?array + { + $data = json_decode($json, true); + + if (! is_array($data)) { + return null; + } + + $type = $data['type'] ?? null; + + if ($type === 'transcript.text.delta') { + $delta = $data['delta'] ?? ''; + + if (! is_string($delta) || $delta === '') { + return null; + } + + return [ + 'append' => $delta, + 'replace' => null, + 'end' => $this->nullableFloat($data['end'] ?? null), + 'done' => false, + 'legacy' => false, + ]; + } + + if ($type === 'transcript.text.done') { + $text = $data['text'] ?? ''; + + return [ + 'append' => null, + 'replace' => is_string($text) ? $text : '', + 'end' => $this->nullableFloat($data['end'] ?? null), + 'done' => true, + 'legacy' => false, + ]; + } + + if ($type === null && isset($data['text']) && is_string($data['text']) && $data['text'] !== '') { + return [ + 'append' => $data['text'], + 'replace' => null, + 'end' => $this->nullableFloat($data['end'] ?? null), + 'done' => false, + 'legacy' => true, + ]; + } + + return null; + } + + /** + * Apply a parsed event to the accumulated transcript. + * + * @param array{append: ?string, replace: ?string, end: ?float, done: bool, legacy: bool} $event + */ + public function applyEvent(string $accumulated, array $event): string + { + if (is_string($event['replace'])) { + return $event['replace']; + } + + $chunk = $event['append'] ?? ''; + + if ($chunk === '') { + return $accumulated; + } + + if ($event['legacy']) { + $chunk = trim($chunk); + + if ($accumulated === '') { + return $chunk; + } + + return rtrim($accumulated).' '.$chunk; + } + + return $accumulated.$chunk; + } + + /** + * @return list + */ + private function dataLines(string $frame): array + { + $payloads = []; + + foreach (explode("\n", $frame) as $line) { + if (str_starts_with($line, 'data:')) { + $payloads[] = ltrim(substr($line, 5)); + } + } + + return $payloads; + } + + private function nullableFloat(mixed $value): ?float + { + if (! is_numeric($value)) { + return null; + } + + return (float) $value; + } +} diff --git a/resources/views/livewire/recordings/show.blade.php b/resources/views/livewire/recordings/show.blade.php index 01b0c34..bd6b54d 100644 --- a/resources/views/livewire/recordings/show.blade.php +++ b/resources/views/livewire/recordings/show.blade.php @@ -178,6 +178,13 @@ +

+
  • Engine: diff --git a/tests/Feature/RecordingUploadTest.php b/tests/Feature/RecordingUploadTest.php index a8a9d03..0e1686d 100644 --- a/tests/Feature/RecordingUploadTest.php +++ b/tests/Feature/RecordingUploadTest.php @@ -228,10 +228,123 @@ class RecordingUploadTest extends TestCase $this->assertSame('Hello from whisper.', $text); Http::assertSent(function ($request): bool { - return str_contains($request->url(), '/audio/transcriptions'); + $body = $request->body(); + + return str_contains($request->url(), '/audio/transcriptions') + && str_contains($body, 'name="stream"') + && str_contains($body, 'true'); }); } + public function test_transcription_service_streams_sse_progress_and_text(): void + { + Storage::fake('local'); + Storage::disk('local')->put('recordings/long.mp3', 'fake-audio-bytes'); + + $sse = "data: {\"type\":\"transcript.text.delta\",\"delta\":\"Hello \"}\n\n" + ."data: {\"type\":\"transcript.text.delta\",\"delta\":\"from whisper.\"}\n\n" + ."data: {\"type\":\"transcript.text.done\",\"text\":\"Hello from whisper.\"}\n\n"; + + Http::fake([ + '*' => Http::response($sse, 200, ['Content-Type' => 'text/event-stream']), + ]); + + $recording = Recording::query()->create([ + 'user_id' => $this->user->id, + 'title' => 'Streamed whisper', + 'original_filename' => 'long.mp3', + 'file_path' => 'recordings/long.mp3', + 'file_size_bytes' => 100, + 'duration_seconds' => 40, + 'transcription_status' => 'processing', + 'transcription_driver' => 'local', + ]); + + $partials = []; + + $text = app(TranscriptionService::class)->transcribe( + $recording, + function (string $message, int $percent, ?string $partial = null) use (&$partials): void { + $partials[] = [ + 'percent' => $percent, + 'partial' => $partial, + ]; + }, + ); + + $this->assertSame('Hello from whisper.', $text); + $this->assertNotEmpty(array_filter($partials, fn (array $row): bool => $row['partial'] === 'Hello ')); + $this->assertSame('Hello from whisper.', $partials[array_key_last($partials)]['partial'] ?? $text); + $this->assertGreaterThanOrEqual(50, max(array_column($partials, 'percent'))); + } + + public function test_transcription_service_streams_legacy_segments_with_timestamp_percent(): void + { + Storage::fake('local'); + Storage::disk('local')->put('recordings/long.mp3', 'fake-audio-bytes'); + + $sse = "data: {\"text\":\"Hello from\",\"start\":0,\"end\":20}\n\n" + ."data: {\"text\":\"whisper.\",\"start\":20,\"end\":40}\n\n"; + + Http::fake([ + '*' => Http::response($sse, 200, ['Content-Type' => 'text/event-stream']), + ]); + + $recording = Recording::query()->create([ + 'user_id' => $this->user->id, + 'title' => 'Legacy stream', + 'original_filename' => 'long.mp3', + 'file_path' => 'recordings/long.mp3', + 'file_size_bytes' => 100, + 'duration_seconds' => 40, + 'transcription_status' => 'processing', + 'transcription_driver' => 'local', + ]); + + $percents = []; + $partials = []; + + $text = app(TranscriptionService::class)->transcribe( + $recording, + function (string $message, int $percent, ?string $partial = null) use (&$percents, &$partials): void { + $percents[] = $percent; + if ($partial !== null) { + $partials[] = $partial; + } + }, + ); + + $this->assertSame('Hello from whisper.', $text); + $this->assertContains('Hello from', $partials); + $this->assertContains(50, $percents); + $this->assertContains(99, $percents); + } + + public function test_report_progress_persists_partial_transcript_without_completing(): void + { + $recording = Recording::query()->create([ + 'user_id' => $this->user->id, + 'title' => 'Partial', + 'original_filename' => 'partial.mp3', + 'file_path' => 'recordings/partial.mp3', + 'file_size_bytes' => 100, + 'transcription_status' => 'processing', + 'transcription_driver' => 'local', + 'transcription_started_at' => now(), + 'transcript' => 'Previous transcript', + ]); + + $recording->reportProgress('Transcribing locally…', 62, partialTranscript: 'Hello from the '); + + $recording->refresh(); + $this->assertSame('processing', $recording->transcription_status); + $this->assertSame(62, $recording->transcription_percent); + $this->assertSame('Hello from the ', $recording->transcript); + $this->assertNull($recording->transcribed_at); + $this->assertTrue($recording->transcriptionStatusPayload()['has_transcript']); + $this->assertSame('Hello from the ', $recording->transcriptionStatusPayload()['transcript']); + } + public function test_transcription_job_stores_transcript(): void { Storage::fake('local'); diff --git a/tests/Feature/Recordings/ShowTest.php b/tests/Feature/Recordings/ShowTest.php index 914e829..ab0e97a 100644 --- a/tests/Feature/Recordings/ShowTest.php +++ b/tests/Feature/Recordings/ShowTest.php @@ -83,6 +83,30 @@ class ShowTest extends TestCase ->assertSee('Queued — waiting to start…'); } + public function test_show_page_includes_partial_transcript_while_processing(): void + { + $user = User::factory()->create(); + $this->actingAs($user); + + $recording = Recording::query()->create([ + 'user_id' => $user->id, + 'title' => 'Live words', + 'original_filename' => 'active.mp3', + 'file_path' => 'recordings/active.mp3', + 'file_size_bytes' => 100, + 'transcription_status' => 'processing', + 'transcription_progress' => 'Transcribing locally…', + 'transcription_percent' => 62, + 'transcription_driver' => 'local', + 'transcription_started_at' => now(), + 'transcript' => 'Live partial sentence from whisper', + ]); + + Livewire::test(Show::class, ['recording' => $recording]) + ->assertSee('Live partial sentence from whisper') + ->assertSee('status.transcript', false); + } + public function test_show_does_not_poll_when_transcription_is_idle(): void { $user = User::factory()->create(); diff --git a/tests/Unit/TranscriptionServiceTest.php b/tests/Unit/TranscriptionServiceTest.php deleted file mode 100644 index bb7c62f..0000000 --- a/tests/Unit/TranscriptionServiceTest.php +++ /dev/null @@ -1,63 +0,0 @@ -checks < 12 ? self::PENDING : self::FULFILLED; - } - - public function resolve($value): void {} - - public function reject($reason): void {} - - public function cancel(): void {} - - public function wait(bool $unwrap = true): mixed - { - return null; - } - }; - }); - - $promise->buildPromise(); - - $percents = []; - $method = new ReflectionMethod(TranscriptionService::class, 'pulseProgressWhilePending'); - $method->invoke($service, $promise, function (string $message, int $percent) use (&$percents): void { - $percents[] = $percent; - }, 'Transcribing…'); - - $this->assertNotEmpty($percents); - $this->assertGreaterThan(50, max($percents)); - $this->assertLessThanOrEqual(84, max($percents)); - } -} diff --git a/tests/Unit/WhisperTranscriptionStreamTest.php b/tests/Unit/WhisperTranscriptionStreamTest.php new file mode 100644 index 0000000..3828c6a --- /dev/null +++ b/tests/Unit/WhisperTranscriptionStreamTest.php @@ -0,0 +1,91 @@ +stream = new WhisperTranscriptionStream; + } + + public function test_extracts_sse_data_payloads_from_buffer(): void + { + $buffer = "data: {\"type\":\"transcript.text.delta\",\"delta\":\"Hello\"}\n\npartial"; + + $payloads = $this->stream->extractPayloads($buffer); + + $this->assertSame(['{"type":"transcript.text.delta","delta":"Hello"}'], $payloads); + $this->assertSame('partial', $buffer); + } + + public function test_parses_openai_delta_and_done_events(): void + { + $delta = $this->stream->parseEvent('{"type":"transcript.text.delta","delta":"Hello "}'); + $done = $this->stream->parseEvent('{"type":"transcript.text.done","text":"Hello world"}'); + + $this->assertSame('Hello ', $delta['append']); + $this->assertFalse($delta['done']); + $this->assertSame('Hello world', $done['replace']); + $this->assertTrue($done['done']); + } + + public function test_parses_legacy_segment_events(): void + { + $event = $this->stream->parseEvent('{"text":"First segment","start":0,"end":12.5}'); + + $this->assertTrue($event['legacy']); + $this->assertSame('First segment', $event['append']); + $this->assertSame(12.5, $event['end']); + } + + public function test_applies_delta_then_done_replace(): void + { + $delta = $this->stream->parseEvent('{"type":"transcript.text.delta","delta":"Hel"}'); + $text = $this->stream->applyEvent('', $delta); + $delta2 = $this->stream->parseEvent('{"type":"transcript.text.delta","delta":"lo"}'); + $text = $this->stream->applyEvent($text, $delta2); + $done = $this->stream->parseEvent('{"type":"transcript.text.done","text":"Hello world"}'); + $text = $this->stream->applyEvent($text, $done); + + $this->assertSame('Hello world', $text); + } + + public function test_joins_legacy_segments_with_spaces(): void + { + $first = $this->stream->parseEvent('{"text":"Hello from","end":10}'); + $second = $this->stream->parseEvent('{"text":"whisper.","end":20}'); + + $text = $this->stream->applyEvent('', $first); + $text = $this->stream->applyEvent($text, $second); + + $this->assertSame('Hello from whisper.', $text); + } + + public function test_ignores_done_sentinel_and_non_json(): void + { + $buffer = "data: [DONE]\n\ndata: not-json\n\n"; + + $payloads = $this->stream->extractPayloads($buffer); + + $this->assertSame(['not-json'], $payloads); + $this->assertNull($this->stream->parseEvent('not-json')); + } + + public function test_flush_buffer_emits_trailing_frame(): void + { + $buffer = 'data: {"text":"Last"}'; + + $payloads = $this->stream->flushBuffer($buffer); + + $this->assertSame(['{"text":"Last"}'], $payloads); + $this->assertSame('', $buffer); + } +}