forked from admin/VolcanicAgenticCore
80 lines
3.0 KiB
C#
80 lines
3.0 KiB
C#
using System.Runtime.CompilerServices;
|
|
using System.Text.Json;
|
|
|
|
namespace Volcanic.AgenticCore.Extensions.ClaudeCode;
|
|
|
|
/// <summary>Strict JSONL framing over a byte stream: records split on LF only, one optional CR before it, UTF-8, bounded size.</summary>
|
|
public static class JsonlReader
|
|
{
|
|
public const int DefaultMaxRecordBytes = 8 * 1024 * 1024;
|
|
|
|
public static async IAsyncEnumerable<JsonElement> ReadAsync(
|
|
Stream stream,
|
|
int maxRecordBytes = DefaultMaxRecordBytes,
|
|
[EnumeratorCancellation] CancellationToken cancellationToken = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(stream);
|
|
var buffer = new byte[64 * 1024];
|
|
using var pending = new MemoryStream();
|
|
while (true)
|
|
{
|
|
var read = await stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false);
|
|
if (read == 0) break;
|
|
var start = 0;
|
|
for (var i = 0; i < read; i++)
|
|
{
|
|
if (buffer[i] != (byte)'\n') continue;
|
|
pending.Write(buffer, start, i - start);
|
|
start = i + 1;
|
|
if (TryParse(pending, maxRecordBytes, out var element)) yield return element;
|
|
}
|
|
pending.Write(buffer, start, read - start);
|
|
if (pending.Length > maxRecordBytes + 1) throw Oversized();
|
|
}
|
|
if (pending.Length > 0 && TryParse(pending, maxRecordBytes, out var last)) yield return last;
|
|
}
|
|
|
|
/// <summary>Parses one framed line. Returns false for a blank line. The pending buffer is reset either way.</summary>
|
|
private static bool TryParse(MemoryStream pending, int maxRecordBytes, out JsonElement element)
|
|
{
|
|
var length = (int)pending.Length;
|
|
var bytes = pending.GetBuffer().AsMemory(0, length);
|
|
pending.SetLength(0);
|
|
if (length > 0 && bytes.Span[length - 1] == (byte)'\r') bytes = bytes[..(length - 1)];
|
|
if (bytes.Length > maxRecordBytes) throw Oversized();
|
|
if (IsBlank(bytes.Span))
|
|
{
|
|
element = default;
|
|
return false;
|
|
}
|
|
element = Parse(bytes);
|
|
return true;
|
|
}
|
|
|
|
/// <summary>Parses one record (no framing). Throws <c>protocol_stream</c> for malformed JSON.</summary>
|
|
public static JsonElement Parse(ReadOnlyMemory<byte> line)
|
|
{
|
|
try
|
|
{
|
|
using var document = JsonDocument.Parse(line);
|
|
return document.RootElement.Clone();
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.ProtocolStream, "Claude Code emitted malformed JSONL");
|
|
}
|
|
}
|
|
|
|
private static bool IsBlank(ReadOnlySpan<byte> bytes)
|
|
{
|
|
foreach (var b in bytes)
|
|
{
|
|
if (b is not ((byte)' ' or (byte)'\t' or (byte)'\r')) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
private static ClaudeCodeException Oversized() =>
|
|
new(ClaudeCodeException.Codes.ProtocolStream, "Claude Code emitted an oversized JSONL record");
|
|
}
|