diff --git a/lib/builtin/claude/src/transcript_reader.dart b/lib/builtin/claude/src/transcript_reader.dart index 86dc2854..f97093b5 100644 --- a/lib/builtin/claude/src/transcript_reader.dart +++ b/lib/builtin/claude/src/transcript_reader.dart @@ -152,6 +152,11 @@ String _shortId(String uuid) => uuid.length >= 8 ? uuid.substring(0, 8) : uuid; /// stream incrementally. const _defaultInitialTailBytes = 256 * 1024; +/// Chunks at least this large are parsed in a background isolate; smaller +/// ones parse inline. Streaming appends are small, so this keeps the +/// off-thread parse to the initial-tail case that actually janks a frame. +const _isolateParseThreshold = 64 * 1024; + /// Known major transcript versions. const _knownMajorVersions = {1, 2}; @@ -321,9 +326,12 @@ class TranscriptReader { } if (controller.isClosed) return; - // Parse off the UI isolate — the initial chunk can be sizeable and - // JSON-decoding it on the main thread would jank the frame. - final parsed = await Isolate.run(() => parseTranscriptChunk(chunk)); + // Parse off the UI isolate only when the chunk is big enough to jank a + // frame — the initial tail read (up to [_initialTailBytes]) is the case + // that froze the app. Streaming appends are small (a message at a time); + // parsing those inline avoids spawning a one-shot isolate every poll + // tick, which is pure overhead and adds latency under load. + final parsed = chunk.length >= _isolateParseThreshold ? await Isolate.run(() => parseTranscriptChunk(chunk)) : parseTranscriptChunk(chunk); if (controller.isClosed) return; for (final w in parsed.warnings) { _onWarn(w); diff --git a/test/builtin/claude/transcript_reader_test.dart b/test/builtin/claude/transcript_reader_test.dart index 1c9017b8..abe5794a 100644 --- a/test/builtin/claude/transcript_reader_test.dart +++ b/test/builtin/claude/transcript_reader_test.dart @@ -30,6 +30,19 @@ void appendLines(File file, List> lines) { ); } +/// Poll [ready] until it returns true or [timeout] elapses. Streaming +/// assertions use this instead of a fixed delay so they don't flake under +/// load (the reader polls on a timer and may parse off-isolate). +Future pumpUntil( + bool Function() ready, { + Duration timeout = const Duration(seconds: 5), +}) async { + final deadline = DateTime.now().add(timeout); + while (!ready() && DateTime.now().isBefore(deadline)) { + await Future.delayed(const Duration(milliseconds: 10)); + } +} + /// JSONL envelope skeleton with default sentinel values. Map envelope({ required String type, @@ -559,8 +572,7 @@ void main() { final collected = []; final sub = reader.stream.listen(collected.add); - // Allow a few poll cycles. - await Future.delayed(const Duration(milliseconds: 150)); + await pumpUntil(() => collected.whereType().isNotEmpty && collected.whereType().isNotEmpty); await sub.cancel(); await reader.dispose(); @@ -585,14 +597,14 @@ void main() { final sub = reader.stream.listen(collected.add); // Let the reader consume the initial lines. - await Future.delayed(const Duration(milliseconds: 100)); + await pumpUntil(() => collected.whereType().isNotEmpty); final countAfterInit = collected.length; // Append new lines. appendLines(sessionFile, [assistantText('a1', 'appended reply')]); // Let the reader pick up the append. - await Future.delayed(const Duration(milliseconds: 100)); + await pumpUntil(() => collected.whereType().isNotEmpty); await sub.cancel(); await reader.dispose(); @@ -619,7 +631,7 @@ void main() { final collected = []; final sub = reader.stream.listen(collected.add); - await Future.delayed(const Duration(milliseconds: 100)); + await pumpUntil(() => collected.whereType().any((m) => m.text == 'old session')); // Create a newer session file (ensure mtime difference with touch-like approach). final newerFile = File('${projectDir.path}/session-xyz.jsonl'); @@ -631,7 +643,7 @@ void main() { final now = DateTime.now(); await newerFile.setLastModified(now); - await Future.delayed(const Duration(milliseconds: 200)); + await pumpUntil(() => collected.whereType().any((m) => m.text == 'new session')); await sub.cancel(); await reader.dispose(); @@ -661,7 +673,7 @@ void main() { final collected = []; final sub = reader.stream.listen(collected.add); - await Future.delayed(const Duration(milliseconds: 150)); + await pumpUntil(() => collected.isNotEmpty); await sub.cancel(); await reader.dispose(); @@ -692,7 +704,7 @@ void main() { final collected = []; final sub = reader.stream.listen(collected.add); - await Future.delayed(const Duration(milliseconds: 200)); + await pumpUntil(() => collected.isNotEmpty); await sub.cancel(); await reader.dispose();