Read only new transcript bytes on each watch poll

src/Capacitor.Cli.Core/Models.cs

@@ -270,6 +270,10 @@ class WatchState {
     // range it last verified. Only ever set for vendor == "cursor".
     public long CursorByteOffset { get; set; }
 
+    // Where each line of the transcript starts, so a poll seeks to LinesProcessed. Every vendor's
+    // live drain but Cursor's uses it.
+    public TranscriptLineIndex LineOffsets { get; } = new();
+
     // poll counter driving the periodic full-prefix re-hash cadence
     // (WatchCommand.CursorFullPrefixVerifyEveryNPolls). Incremented once per poll for vendor ==
     // "cursor" only; a plain counter (not wall-clock time) so the cadence is exact regardless of

src/Capacitor.Cli.Core/TranscriptLineIndex.cs

@@ -0,0 +1,136 @@
+namespace Capacitor.Cli.Core;
+
+/// <summary>
+/// The byte offset each transcript line starts at, learned as a watcher reads, so a poll seeks to
+/// its line cursor instead of decoding the whole file again. Line numbers are only ever the ones a
+/// full <see cref="StreamReader"/> pass gives: a lone <c>'\r'</c> also ends a line there, so the
+/// index stops learning at one, and a UTF-16 or UTF-32 file is never indexed.
+/// </summary>
+/// <remarks>
+/// Only the bytes just before the indexed end are compared on each poll: a file that shrank below
+/// it or changed there drops the index and is read from byte 0. A rewrite further back goes
+/// unnoticed, which the line cursor already relies on never happening — a transcript is appended
+/// to, never edited. One reader at a time.
+/// </remarks>
+sealed class TranscriptLineIndex {
+    const int TailLength = 4096;
+
+    // _starts[i] is where line i starts; the last entry is the end of the last indexed line.
+    readonly List<long> _starts = [0];
+
+    byte[] _tail    = [];
+    long   _tailEnd;
+    bool   _unsupported;
+
+    long _position;
+    bool _scanning;
+    bool _pendingCr;
+
+    long Frontier => _starts[^1];
+
+    /// <summary>
+    /// Validates the index against <paramref name="stream"/> and returns the line and byte to
+    /// start decoding at for a reader whose cursor is <paramref name="line"/>.
+    /// </summary>
+    public async ValueTask<(int Line, long Offset)> StartAsync(FileStream stream, long length, int line, CancellationToken ct) {
+        if (Frontier > 0 && (Frontier != _tailEnd || length < Frontier || !await TailMatchesAsync(stream, ct))) Reset();
+
+        // An unsupported file is read from byte 0 every time, so its encoding is checked again.
+        var from = _unsupported ? 0 : Math.Min(line, _starts.Count - 1);
+        _unsupported = false;
+        _position    = _starts[from];
+        _scanning    = true;
+        _pendingCr   = false;
+
+        return (from, _position);
+    }
+
+    /// <summary>Learns the line ends in the next bytes read after the offset <see cref="StartAsync"/> returned.</summary>
+    public void Observe(ReadOnlySpan<byte> bytes) {
+        if (!_scanning || bytes.IsEmpty) {
+            _position += bytes.Length;
+            return;
+        }
+
+        if (_position == 0 && bytes[0] is 0xFE or 0xFF or 0x00) {
+            _unsupported = true;
+            _scanning    = false;
+            Reset();
+            return;
+        }
+
+        var skip = (int)Math.Clamp(Frontier - _position, 0, bytes.Length);
+        var span = bytes[skip..];
+        var at   = _position + skip;
+        _position += bytes.Length;
+
+        if (_pendingCr && span.Length > 0 && span[0] != (byte)'\n') {
+            _scanning = false;
+            return;
+        }
+
+        while (span.Length > 0) {
+            var i = span.IndexOfAny((byte)'\r', (byte)'\n');
+            if (i < 0) {
+                _pendingCr = false;
+                return;
+            }
+
+            if (span[i] == (byte)'\n') {
+                _starts.Add(at + i + 1);
+                _pendingCr = false;
+            } else if (i + 1 == span.Length) {
+                _pendingCr = true;
+                return;
+            } else if (span[i + 1] != (byte)'\n') {
+                _scanning = false;
+                return;
+            }
+
+            at   += i + 1;
+            span  = span[(i + 1)..];
+        }
+    }
+
+    /// <summary>Records the bytes before the new indexed end, for the next poll to compare.</summary>
+    public async ValueTask FinishAsync(FileStream stream, CancellationToken ct) {
+        _scanning = false;
+        if (_unsupported || Frontier == _tailEnd) return;
+
+        var tail = new byte[(int)Math.Min(TailLength, Frontier)];
+        if (await ReadAtAsync(stream, Frontier - tail.Length, tail, ct) != tail.Length) {
+            Reset();
+            return;
+        }
+
+        _tail    = tail;
+        _tailEnd = Frontier;
+    }
+
+    async ValueTask<bool> TailMatchesAsync(FileStream stream, CancellationToken ct) {
+        var current = new byte[_tail.Length];
+
+        return await ReadAtAsync(stream, _tailEnd - _tail.Length, current, ct) == current.Length
+            && current.AsSpan().SequenceEqual(_tail);
+    }
+
+    static async ValueTask<int> ReadAtAsync(FileStream stream, long offset, byte[] buffer, CancellationToken ct) {
+        stream.Position = offset;
+        var total = 0;
+
+        while (total < buffer.Length) {
+            var n = await stream.ReadAsync(buffer.AsMemory(total), ct);
+            if (n == 0) break;
+            total += n;
+        }
+
+        return total;
+    }
+
+    void Reset() {
+        _starts.Clear();
+        _starts.Add(0);
+        _tail    = [];
+        _tailEnd = 0;
+    }
+}

src/Capacitor.Cli/Commands/WatchCommand.cs

@@ -199,6 +199,40 @@ internal static bool IsFinalLineComplete(string fileText) {
         return tail.Trim().Length == 0 || IsCompleteJsonRecord(tail);
     }
 
+    static readonly UTF8Encoding Utf8NoPreamble = new(encoderShouldEmitUTF8Identifier: false);
+
+    /// <summary>
+    /// The file's text from its last <c>'\n'</c> on, or all of it when it has none: what
+    /// <see cref="IsFinalLineComplete"/> decides on, read without decoding the lines before it. A
+    /// UTF-16 or UTF-32 file is read whole, since its newline is not a lone <c>0x0A</c> byte.
+    /// </summary>
+    internal static async Task<string> ReadFinalLineSharedAsync(string path) {
+        await using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete);
+        if (stream.Length > 0 && stream.ReadByte() is 0xFE or 0xFF or 0x00) return await File.ReadAllTextSharedAsync(path);
+
+        var end   = stream.Length;
+        var start = end;
+
+        while (start > 0) {
+            var chunk = new byte[(int)Math.Min(64 * 1024, start)];
+            stream.Position = start - chunk.Length;
+            if (await ReadFullyAsync(stream, chunk, CancellationToken.None) != chunk.Length) break;
+
+            var newline = chunk.AsSpan().LastIndexOf((byte)'\n');
+            start -= chunk.Length;
+            if (newline >= 0) {
+                start += newline;
+                break;
+            }
+        }
+
+        var tail = new byte[end - start];
+        stream.Position = start;
+        var text = tail.AsSpan(0, await ReadFullyAsync(stream, tail, CancellationToken.None));
+
+        return Utf8NoPreamble.GetString(start == 0 && text.StartsWith(Encoding.UTF8.Preamble) ? text[3..] : text);
+    }
+
     /// <summary>
     /// Bounded wait (≤2s, 4×500ms) for the shutdown final drain: re-reads <paramref name="path"/>
     /// looking for <see cref="IsFinalLineComplete"/> before the watcher decides whether it's safe to
@@ -212,7 +246,7 @@ internal static async Task<bool> WaitForFinalLineCompletionAsync(
             string path, TimeProvider time, int attempts = 4, int delayMs = 500) {
         for (var i = 0; i < attempts; i++) {
             try {
-                if (IsFinalLineComplete(await File.ReadAllTextSharedAsync(path))) return true;
+                if (IsFinalLineComplete(await ReadFinalLineSharedAsync(path))) return true;
             } catch {
                 // Transient read failure (e.g. concurrent write) — try again on the next iteration.
             }
@@ -1033,7 +1067,7 @@ await ScanCodexSubagents(sessionId, transcriptPath, seenSubagents, codexRuledOut
             // instead of silently dropping it.
             await SpoolUndeliveredTranscriptTailAsync(
                 transcriptSpool, transcriptPath, sessionId, agentId, vendor, state.LinesProcessed, state.Commits, CancellationToken.None,
-                ShutdownTailRedactionDeadline - time.GetElapsedTime(shutdownStarted));
+                ShutdownTailRedactionDeadline - time.GetElapsedTime(shutdownStarted), state.LineOffsets);
 
             // One last subagent-link scan on the way out — the parent may have emitted an
             // INVOKE_SUBAGENT step after the main loop's last tick but before exit, and this is
@@ -2156,7 +2190,8 @@ internal async Task<IReadOnlyList<string>> DrainNewLines(
                     isFinalDrain ? IncompleteFinalLinePolicy.ConsumeIfComplete : IncompleteFinalLinePolicy.Hold,
                     ct, captureRawBytes: vendor == "cursor",
                     rawBytesReadFrom: cursorBoundedReadFrom,
-                    newRangeByteOffset: vendor == "cursor" ? cursorGuardOldOffset : null);
+                    newRangeByteOffset: vendor == "cursor" ? cursorGuardOldOffset : null,
+                    lineOffsets: state.LineOffsets);
             }
 
             // commit the cadence counter now that the guarded
@@ -2714,15 +2749,16 @@ internal static string BuildTranscriptSpoolBatch(
 
     /// <summary>Spool only records beyond the acknowledged source frontier.</summary>
     internal async Task<TranscriptSpool.AppendResult?> SpoolUndeliveredTranscriptTailAsync(
-            TranscriptSpool   transcriptSpool,
-            string            transcriptPath,
-            string            sessionId,
-            string?           agentId,
-            string            vendor,
-            int               linesProcessed,
-            CommitObservation commits,
-            CancellationToken ct,
-            TimeSpan?         redactionLimit = null
+            TranscriptSpool      transcriptSpool,
+            string               transcriptPath,
+            string               sessionId,
+            string?              agentId,
+            string               vendor,
+            int                  linesProcessed,
+            CommitObservation    commits,
+            CancellationToken    ct,
+            TimeSpan?            redactionLimit = null,
+            TranscriptLineIndex? lineOffsets    = null
         ) {
         if (!File.Exists(transcriptPath)) return null;
 
@@ -2739,7 +2775,7 @@ internal static string BuildTranscriptSpoolBatch(
             await using var stream = new FileStream(
                 transcriptPath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
             tail = await ReadNewCompleteLinesAsync(
-                stream, linesProcessed, IncompleteFinalLinePolicy.ConsumeIfComplete, ct);
+                stream, linesProcessed, IncompleteFinalLinePolicy.ConsumeIfComplete, ct, lineOffsets: lineOffsets);
         } catch (IOException ex) {
             Log(time, $"Shutdown-tail spool: failed to read transcript for {sessionId}: {ex.Message}");
 
@@ -4392,14 +4428,20 @@ internal async Task<IReadOnlyList<string>> GatedDrainNewLinesAsync(
     /// lands exactly on a line boundary (true by construction for the Cursor watcher's own
     /// <c>CursorByteOffset</c> — see <see cref="ByteOffsetForAckedLines"/>).
     /// </param>
+    /// <param name="lineOffsets">
+    /// Ignored when <paramref name="captureRawBytes"/> is true. Otherwise lets the read start at
+    /// <paramref name="linesProcessed"/>'s byte offset rather than byte 0, and learns the offsets of
+    /// the lines it reads; the result is the same as without it.
+    /// </param>
     internal static async Task<NewTranscriptLines> ReadNewCompleteLinesAsync(
             FileStream                stream,
             int                       linesProcessed,
             IncompleteFinalLinePolicy policy,
             CancellationToken         ct,
             bool                      captureRawBytes    = false,
             long                      rawBytesReadFrom   = 0,
-            long?                     newRangeByteOffset = null) {
+            long?                     newRangeByteOffset = null,
+            TranscriptLineIndex?      lineOffsets        = null) {
         var length = stream.Length;
 
         bool endsWithNewline;
@@ -4482,7 +4524,19 @@ internal static async Task<NewTranscriptLines> ReadNewCompleteLinesAsync(
                 }
             }
         } else {
-            using var reader = new StreamReader(new LengthLimitedReadStream(stream, length), leaveOpen: true);
+            var startOffset = 0L;
+            if (lineOffsets is not null) {
+                (lineIndex, startOffset) = await lineOffsets.StartAsync(stream, length, linesProcessed, ct);
+                stream.Position          = startOffset;
+            }
+
+            // Mid-file there is no byte-order mark to detect, and a line that happens to begin with
+            // the UTF-8 one must keep it, as it does when decoded from byte 0.
+            using var reader = startOffset == 0
+                ? new StreamReader(new LengthLimitedReadStream(stream, length, lineOffsets), leaveOpen: true)
+                : new StreamReader(new LengthLimitedReadStream(stream, length - startOffset, lineOffsets),
+                    Utf8NoPreamble, detectEncodingFromByteOrderMarks: false, leaveOpen: true);
+
             while (await reader.ReadLineAsync(ct) is { } line) {
                 if (lineIndex >= linesProcessed && !string.IsNullOrWhiteSpace(line)) {
                     newLines.Add(line);
@@ -4491,6 +4545,8 @@ internal static async Task<NewTranscriptLines> ReadNewCompleteLinesAsync(
 
                 lineIndex++;
             }
+
+            if (lineOffsets is not null) await lineOffsets.FinishAsync(stream, ct);
         }
 
         // `length` is the exact byte boundary this read was capped at
@@ -4578,14 +4634,15 @@ static bool IsCompleteJsonRecord(string line) {
 
     /// <summary>Read-only view of <c>inner</c> that reports EOF after <c>limit</c> bytes, so a
     /// StreamReader over it can't read past a length sampled before a concurrent append.</summary>
-    internal sealed class LengthLimitedReadStream(Stream inner, long limit) : Stream {
+    internal sealed class LengthLimitedReadStream(Stream inner, long limit, TranscriptLineIndex? observer = null) : Stream {
         long _consumed;
 
         public override int Read(byte[] buffer, int offset, int count) {
             var allowed = (int)Math.Min(count, limit - _consumed);
             if (allowed <= 0) return 0;
             var n = inner.Read(buffer, offset, allowed);
             _consumed += n;
+            observer?.Observe(buffer.AsSpan(offset, n));
             return n;
         }
 
@@ -4594,6 +4651,7 @@ public override async ValueTask<int> ReadAsync(Memory<byte> buffer, Cancellation
             if (allowed <= 0) return 0;
             var n = await inner.ReadAsync(buffer[..allowed], ct);
             _consumed += n;
+            observer?.Observe(buffer.Span[..n]);
             return n;
         }