forked from admin/VolcanicAgenticCore
751 lines
38 KiB
C#
751 lines
38 KiB
C#
using System.Globalization;
|
|
using System.Text;
|
|
using System.Text.Json;
|
|
using System.Text.RegularExpressions;
|
|
using Volcanic.AgenticCore.Abstractions.Messages;
|
|
using Volcanic.AgenticCore.Abstractions.Providers;
|
|
|
|
namespace Volcanic.AgenticCore.Extensions.ClaudeCode;
|
|
|
|
/// <summary>
|
|
/// Maps Claude Code's stream-json stdout records to <see cref="AssistantMessageEvent"/>s through one
|
|
/// <see cref="AssistantMessageBuilder"/>. Validates the <c>system/init</c> record, latches a handoff on a
|
|
/// <c>tool_use</c> or length stop (after which every later stream event belongs to a turn the core never asked for
|
|
/// and is ignored), and reports Claude Code's mid-response recoveries as a retryable <c>stream_interrupted</c> failure.
|
|
/// </summary>
|
|
public sealed partial class StreamEventMapper
|
|
{
|
|
// api_retry categories that describe a transient transport or server failure. Every other category
|
|
// (billing_error, authentication_failed, invalid_request, ...) cannot be cleared by repeating the turn.
|
|
private static readonly HashSet<string> RetryableInterruptions = new(StringComparer.Ordinal) { "overloaded", "server_error", "unknown" };
|
|
private const string InterruptedPhrase = "Claude Code API stream ended before message_stop";
|
|
private const long EpochMillisecondsThreshold = 100_000_000_000;
|
|
|
|
private enum BlockKind { None, Text, Thinking, ToolCall }
|
|
|
|
private readonly ToolCatalog _catalog;
|
|
private readonly List<string> _notices = [];
|
|
private readonly StringBuilder _partialJson = new();
|
|
private bool _messageStarted;
|
|
private bool _messageStopped;
|
|
private bool _successfulResult;
|
|
private bool _finished;
|
|
private string? _streamMessageId;
|
|
private string? _stopReason;
|
|
private string? _deferredToolArguments;
|
|
private string? _rateLimitFailure;
|
|
private string? _assistantDiagnostic;
|
|
private string? _openToolName;
|
|
private int? _openIndex;
|
|
private BlockKind _openKind;
|
|
private JsonElement? _initialInput;
|
|
private int _input, _output, _cacheRead, _cacheWrite, _reasoning;
|
|
|
|
public StreamEventMapper(Model model, ToolCatalog catalog, AssistantMessageBuilder? builder = null)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(model);
|
|
ArgumentNullException.ThrowIfNull(catalog);
|
|
Model = model;
|
|
_catalog = catalog;
|
|
Builder = builder ?? new AssistantMessageBuilder(model);
|
|
}
|
|
|
|
public Model Model { get; }
|
|
|
|
/// <summary>The builder behind every event; the provider uses it to <c>Fail</c> the stream.</summary>
|
|
public AssistantMessageBuilder Builder { get; }
|
|
|
|
public bool SawInit { get; private set; }
|
|
public bool SawResult { get; private set; }
|
|
|
|
/// <summary>A tool-use or length stop was reported: the provider terminates Claude and later stream events are ignored.</summary>
|
|
public bool HandoffLatched { get; private set; }
|
|
|
|
/// <summary>The stop reason a completed response will carry: ToolUse or Length after a latch, Stop after a successful result.</summary>
|
|
public StopReason? PendingStop { get; private set; }
|
|
|
|
public string? ResponseModel { get; private set; }
|
|
public Usage Usage => Builder.Usage;
|
|
public bool MessageStarted => _messageStarted;
|
|
public string? RawStopReason => _stopReason;
|
|
public bool HasSuccessfulResult => _successfulResult;
|
|
public bool IsComplete => _finished;
|
|
|
|
/// <summary>A held tool-argument failure for the provider to report when no recovery signal claimed it.</summary>
|
|
public string? DeferredError => _deferredToolArguments;
|
|
|
|
/// <summary>The most recent rejected rate limit, used as the failure detail when Claude exits without a terminal event.</summary>
|
|
public string? RateLimitFailure => _rateLimitFailure;
|
|
|
|
/// <summary>Rate-limit notices (warnings and rejections) in arrival order, formatted for a user notification.</summary>
|
|
public IReadOnlyList<string> Notices => _notices;
|
|
|
|
public int? ServedContextWindow { get; private set; }
|
|
public int? ServedMaxOutputTokens { get; private set; }
|
|
|
|
/// <summary>Maps one stdout record to the events to publish. Throws <see cref="ClaudeCodeException"/> on protocol failures.</summary>
|
|
public IReadOnlyList<AssistantMessageEvent> Accept(JsonElement record)
|
|
{
|
|
if (record.ValueKind != JsonValueKind.Object) throw Stream("Invalid Claude event record");
|
|
var type = Str(record, "type");
|
|
var subtype = Str(record, "subtype");
|
|
if (type == "system" && subtype == "init") return ValidateInit(record);
|
|
if (!SawInit)
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.ProtocolInit, $"Claude emitted a {RecordKind(type, subtype)} record before initialization");
|
|
if (SawResult) throw Stream("Claude emitted a record after its result");
|
|
|
|
switch (type)
|
|
{
|
|
case "stream_event":
|
|
// Once a handoff is latched the provider is terminating Claude, but records already in the pipe still
|
|
// arrive. Claude Code answers a tool denial or an output limit with another message of its own, so these
|
|
// events belong to a turn the core never asked for: neither published nor rejected as protocol drift.
|
|
if (HandoffLatched) return [];
|
|
return AcceptStreamEvent(Obj(record, "event") ?? throw Stream("Claude stream_event did not contain an event"));
|
|
case "result":
|
|
return AcceptResult(record);
|
|
case "rate_limit_event":
|
|
AcceptRateLimit(record);
|
|
return [];
|
|
case "assistant":
|
|
RejectInterruptedStream(record, type, subtype);
|
|
AcceptAssistant(record);
|
|
return [];
|
|
case "user":
|
|
case "system":
|
|
// Completed user echoes and non-init system status records are redundant with the partial stream events.
|
|
RejectInterruptedStream(record, type, subtype);
|
|
return [];
|
|
default:
|
|
throw Stream($"Unsupported Claude record type: {type ?? "untyped"}");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// The terminal <c>Done</c> event, produced once the provider has validated the process exit and private state.
|
|
/// Reason: ToolUse or Length after a latched handoff, else Stop/Length from the successful result.
|
|
/// </summary>
|
|
public AssistantMessageEvent Complete()
|
|
{
|
|
if (_finished) throw new InvalidOperationException("The response was already completed.");
|
|
ThrowDeferredFailure();
|
|
if (_openIndex is not null) throw Stream("Claude response completed with unclosed content blocks");
|
|
switch (PendingStop)
|
|
{
|
|
case StopReason.ToolUse:
|
|
if (!Builder.Current.Content.OfType<ToolCallContent>().Any()) throw Stream("Claude reported tool use without a tool call");
|
|
break;
|
|
case StopReason.Length:
|
|
if (!IsLengthStop(_stopReason)) throw Stream("Claude did not report an output limit stop");
|
|
break;
|
|
default:
|
|
if (!_successfulResult) throw Stream("Claude result was not ready for completion");
|
|
break;
|
|
}
|
|
_finished = true;
|
|
return Builder.Done(PendingStop ?? StopReason.Stop);
|
|
}
|
|
|
|
/// <summary>Whether an error message is the API's context overflow, in any of Claude Code's wordings.</summary>
|
|
public static bool IsOverflow(string? message) => message is not null && Overflow().IsMatch(message);
|
|
|
|
/// <summary>Prefixes an overflow message with <c>context_length_exceeded:</c> so the core compacts and retries.</summary>
|
|
public static string NormalizeOverflowMessage(string message) =>
|
|
message.StartsWith("context_length_exceeded:", StringComparison.Ordinal) || !IsOverflow(message)
|
|
? message
|
|
: $"context_length_exceeded: {message}";
|
|
|
|
// ---- init ----
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> ValidateInit(JsonElement record)
|
|
{
|
|
if (SawInit) throw Init("Claude emitted duplicate initialization");
|
|
|
|
if (!record.TryGetProperty("tools", out var toolsElement) || toolsElement.ValueKind != JsonValueKind.Array)
|
|
throw Init("Claude initialization omitted tools");
|
|
var observed = new List<string>();
|
|
foreach (var tool in toolsElement.EnumerateArray())
|
|
{
|
|
if (tool.ValueKind != JsonValueKind.String) throw Init("Claude initialization contained an invalid tool name");
|
|
observed.Add(tool.GetString()!);
|
|
}
|
|
var observedSet = observed.ToHashSet(StringComparer.Ordinal);
|
|
var expected = _catalog.TransportNames.ToHashSet(StringComparer.Ordinal);
|
|
if (observedSet.Count != observed.Count || !observedSet.SetEquals(expected))
|
|
throw Init($"Claude Code initialized with an unexpected tool set (expected: {FormatNames(expected)}; observed: {FormatNames(observedSet)})");
|
|
|
|
if (Str(record, "permissionMode") != "dontAsk") throw Init("Claude Code did not enter dontAsk permission mode");
|
|
|
|
if (!record.TryGetProperty("slash_commands", out var slashCommands) || slashCommands.ValueKind != JsonValueKind.Array)
|
|
throw Init("Claude initialization omitted customization inventories");
|
|
var loaded = new List<string>();
|
|
foreach (var field in new[] { "plugins", "skills", "slash_commands" })
|
|
{
|
|
if (record.TryGetProperty(field, out var inventory) && inventory.ValueKind == JsonValueKind.Array && inventory.GetArrayLength() > 0)
|
|
loaded.Add($"{field}: {InventoryNames(inventory)}");
|
|
}
|
|
if (loaded.Count > 0) throw Init($"Claude Code loaded unexpected customizations ({string.Join("; ", loaded)})");
|
|
|
|
var apiKeySource = Str(record, "apiKeySource");
|
|
if (apiKeySource != "none")
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.Auth, $"Claude Code is not using subscription auth (apiKeySource: {apiKeySource ?? "missing"})");
|
|
|
|
if (record.TryGetProperty("mcp_server_errors", out var mcpErrors) && mcpErrors.ValueKind != JsonValueKind.Null)
|
|
{
|
|
if (mcpErrors.ValueKind != JsonValueKind.Array) throw Init("Claude initialization contained invalid MCP server errors");
|
|
if (mcpErrors.GetArrayLength() > 0)
|
|
{
|
|
var details = mcpErrors.EnumerateArray().Select(e => $"{Str(e, "type") ?? "error"}: {Truncate(Str(e, "message") ?? "unknown", 512)}");
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.McpStartup, $"Claude Code reported MCP initialization errors: {string.Join("; ", details)}");
|
|
}
|
|
}
|
|
|
|
if (!record.TryGetProperty("mcp_servers", out var servers) || servers.ValueKind != JsonValueKind.Array)
|
|
throw Init("Claude initialization omitted MCP inventory");
|
|
if (_catalog.IsEmpty)
|
|
{
|
|
if (servers.GetArrayLength() > 0)
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.McpStartup, $"Claude Code loaded an unexpected MCP server ({InventoryNames(servers)})");
|
|
}
|
|
else
|
|
{
|
|
var single = servers.GetArrayLength() == 1 ? servers[0] : (JsonElement?)null;
|
|
if (single is null || Str(single.Value, "name") != _catalog.ServerName || Str(single.Value, "status") != "connected")
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.McpStartup,
|
|
$"The proposal MCP server did not initialize correctly (expected '{_catalog.ServerName}' connected; observed: {InventoryNames(servers)}{ServerStatuses(servers)})");
|
|
}
|
|
|
|
var model = Str(record, "model");
|
|
if (string.IsNullOrEmpty(model)) throw Init("Claude initialization omitted the resolved model");
|
|
ResponseModel = model;
|
|
Builder.SetResponseModel(model);
|
|
SawInit = true;
|
|
return [Builder.Start()];
|
|
}
|
|
|
|
// ---- stream events ----
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> AcceptStreamEvent(JsonElement ev)
|
|
{
|
|
var type = Str(ev, "type");
|
|
if (type == "message_start")
|
|
{
|
|
if (_messageStarted) throw Stream("Claude emitted duplicate message_start");
|
|
var message = Obj(ev, "message") ?? throw Stream("message_start.message must be an object");
|
|
if (Str(message, "model") is { Length: > 0 } model)
|
|
{
|
|
ResponseModel = model;
|
|
Builder.SetResponseModel(model);
|
|
}
|
|
_streamMessageId = Str(message, "id");
|
|
if (_streamMessageId is not null) Builder.SetResponseId(_streamMessageId);
|
|
ApplyUsage(message);
|
|
_messageStarted = true;
|
|
return [];
|
|
}
|
|
if (_messageStopped) throw Stream($"{type ?? "untyped"} arrived after message_stop");
|
|
if (type == "ping") return [];
|
|
if (!_messageStarted) throw Stream($"{type ?? "untyped"} arrived before message_start");
|
|
|
|
switch (type)
|
|
{
|
|
case "content_block_start":
|
|
return StartBlock(ev);
|
|
case "content_block_delta":
|
|
return DeltaBlock(ev);
|
|
case "content_block_stop":
|
|
return EndBlock(ev);
|
|
case "message_delta":
|
|
return AcceptMessageDelta(ev);
|
|
case "message_stop":
|
|
if (_openIndex is not null) throw Stream("Claude stopped with unclosed content blocks");
|
|
_messageStopped = true;
|
|
Latch();
|
|
return [];
|
|
default:
|
|
throw Stream($"Unsupported Claude stream event: {type ?? "untyped"}");
|
|
}
|
|
}
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> AcceptMessageDelta(JsonElement ev)
|
|
{
|
|
var delta = Obj(ev, "delta") ?? throw Stream("message_delta.delta must be an object");
|
|
if (delta.TryGetProperty("stop_reason", out var stop) && stop.ValueKind != JsonValueKind.Null)
|
|
{
|
|
// Claude exposes this as a string rather than a closed enum. Only tool_use and the length stops carry special
|
|
// meaning here, so future values are preserved as ordinary stops instead of rejecting a valid response.
|
|
if (stop.ValueKind != JsonValueKind.String || string.IsNullOrEmpty(stop.GetString()))
|
|
throw Stream($"Invalid Claude stop reason: {stop}");
|
|
_stopReason = stop.GetString();
|
|
}
|
|
ApplyUsage(ev);
|
|
Latch();
|
|
return [];
|
|
}
|
|
|
|
private void Latch()
|
|
{
|
|
if (_stopReason == "tool_use")
|
|
{
|
|
HandoffLatched = true;
|
|
PendingStop = StopReason.ToolUse;
|
|
}
|
|
else if (IsLengthStop(_stopReason))
|
|
{
|
|
// The API accepts input plus max_tokens beyond the window and stops at the window with
|
|
// model_context_window_exceeded; Claude Code answers both stops with a synthetic continuation turn.
|
|
HandoffLatched = true;
|
|
PendingStop = StopReason.Length;
|
|
}
|
|
}
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> StartBlock(JsonElement ev)
|
|
{
|
|
var index = Index(ev);
|
|
if (_openIndex is not null)
|
|
throw Stream(_openIndex == index ? $"Duplicate content block index {index}" : $"Content block {index} started while block {_openIndex} is open");
|
|
var source = Obj(ev, "content_block") ?? throw Stream("content_block_start.content_block must be an object");
|
|
var events = new List<AssistantMessageEvent>();
|
|
_partialJson.Clear();
|
|
_initialInput = null;
|
|
_openToolName = null;
|
|
switch (Str(source, "type"))
|
|
{
|
|
case "text":
|
|
events.Add(Builder.StartText());
|
|
if (Str(source, "text") is { Length: > 0 } text) events.Add(Builder.AppendText(text));
|
|
_openKind = BlockKind.Text;
|
|
break;
|
|
case "thinking":
|
|
events.Add(Builder.StartThinking());
|
|
if (Str(source, "thinking") is { Length: > 0 } thinking) events.Add(Builder.AppendThinking(thinking));
|
|
if (Str(source, "signature") is { Length: > 0 } signature) Builder.AppendThinkingSignature(signature);
|
|
_openKind = BlockKind.Thinking;
|
|
break;
|
|
case "redacted_thinking":
|
|
{
|
|
var data = Str(source, "data");
|
|
if (string.IsNullOrEmpty(data)) throw Stream("Claude emitted redacted thinking without opaque data");
|
|
events.Add(Builder.StartThinking(redacted: true));
|
|
Builder.AppendThinkingSignature(data);
|
|
_openKind = BlockKind.Thinking;
|
|
break;
|
|
}
|
|
case "tool_use":
|
|
{
|
|
var qualified = Str(source, "name") ?? "";
|
|
if (!_catalog.TransportToTool.TryGetValue(qualified, out var toolName))
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.ToolUnknown, $"Claude proposed an unknown tool: {qualified}");
|
|
var id = Str(source, "id");
|
|
if (string.IsNullOrEmpty(id)) throw Stream("Claude emitted a tool without an ID");
|
|
if (source.TryGetProperty("input", out var input) && input.ValueKind == JsonValueKind.Object && input.EnumerateObject().Any())
|
|
_initialInput = input.Clone();
|
|
events.Add(Builder.StartToolCall(id, toolName));
|
|
_openToolName = toolName;
|
|
_openKind = BlockKind.ToolCall;
|
|
break;
|
|
}
|
|
default:
|
|
throw Stream($"Unsupported Claude content block: {Str(source, "type") ?? "untyped"}");
|
|
}
|
|
_openIndex = index;
|
|
return events;
|
|
}
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> DeltaBlock(JsonElement ev)
|
|
{
|
|
var index = Index(ev);
|
|
if (_openIndex != index) throw Stream($"Delta for unknown content block {index}");
|
|
var delta = Obj(ev, "delta") ?? throw Stream("content_block_delta.delta must be an object");
|
|
var deltaType = Str(delta, "type");
|
|
switch (deltaType)
|
|
{
|
|
case "text_delta" when _openKind == BlockKind.Text && Str(delta, "text") is { } text:
|
|
return [Builder.AppendText(text)];
|
|
case "thinking_delta" when _openKind == BlockKind.Thinking && Str(delta, "thinking") is { } thinking:
|
|
return [Builder.AppendThinking(thinking)];
|
|
case "signature_delta" when _openKind == BlockKind.Thinking && Str(delta, "signature") is { } signature:
|
|
Builder.AppendThinkingSignature(signature);
|
|
return [];
|
|
case "input_json_delta" when _openKind == BlockKind.ToolCall && Str(delta, "partial_json") is { } partial:
|
|
_partialJson.Append(partial);
|
|
return [Builder.AppendToolCallDelta(partial)];
|
|
default:
|
|
throw Stream($"Delta {deltaType ?? "untyped"} did not match its content block");
|
|
}
|
|
}
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> EndBlock(JsonElement ev)
|
|
{
|
|
var index = Index(ev);
|
|
if (_openIndex != index) throw Stream($"Stop for unknown content block {index}");
|
|
var kind = _openKind;
|
|
_openIndex = null;
|
|
_openKind = BlockKind.None;
|
|
switch (kind)
|
|
{
|
|
case BlockKind.Text:
|
|
return [Builder.EndText()];
|
|
case BlockKind.Thinking:
|
|
return [Builder.EndThinking()];
|
|
case BlockKind.ToolCall:
|
|
{
|
|
var raw = _partialJson.ToString();
|
|
_partialJson.Clear();
|
|
if (raw.Length > 0)
|
|
{
|
|
JsonElement parsed;
|
|
try
|
|
{
|
|
using var document = JsonDocument.Parse(raw);
|
|
parsed = document.RootElement.Clone();
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
parsed = default;
|
|
}
|
|
if (parsed.ValueKind != JsonValueKind.Object)
|
|
{
|
|
// A connection drop inside streamed tool arguments truncates the JSON. Claude Code then closes the
|
|
// block, stops the message and announces its recovery, so hold this failure: if a recovery signal
|
|
// follows, the retryable interruption is the accurate explanation. The block is never published.
|
|
_deferredToolArguments = $"Claude emitted invalid arguments for tool {_openToolName}";
|
|
Builder.EndToolCall();
|
|
return [];
|
|
}
|
|
return [Builder.EndToolCall(parsed)];
|
|
}
|
|
return [_initialInput is { } initial ? Builder.EndToolCall(initial) : Builder.EndToolCall()];
|
|
}
|
|
default:
|
|
throw Stream($"Missing output block for source index {index}");
|
|
}
|
|
}
|
|
|
|
// ---- result ----
|
|
|
|
private IReadOnlyList<AssistantMessageEvent> AcceptResult(JsonElement record)
|
|
{
|
|
if (!record.TryGetProperty("is_error", out var isErrorElement) || isErrorElement.ValueKind is not (JsonValueKind.True or JsonValueKind.False))
|
|
throw Stream("Claude result omitted a boolean is_error field");
|
|
var isError = isErrorElement.GetBoolean();
|
|
string? resultText = null;
|
|
if (record.TryGetProperty("result", out var resultElement) && resultElement.ValueKind != JsonValueKind.Null)
|
|
{
|
|
if (resultElement.ValueKind != JsonValueKind.String) throw Stream("Claude result contained a non-string result field");
|
|
resultText = resultElement.GetString();
|
|
}
|
|
SawResult = true;
|
|
ApplyUsage(record);
|
|
ApplyModelUsage(record);
|
|
|
|
if (isError)
|
|
{
|
|
if (IsExpectedHandoffTermination(record)) return [];
|
|
ThrowDeferredFailure();
|
|
var status = record.TryGetProperty("api_error_status", out var statusElement) && statusElement.ValueKind == JsonValueKind.Number
|
|
? $" ({statusElement.GetRawText()})"
|
|
: "";
|
|
var detail = FirstNonEmpty(resultText, _assistantDiagnostic, ErrorsText(record), Str(record, "terminal_reason"), _rateLimitFailure) ?? "unknown error";
|
|
var message = $"Claude Code request failed{status}: {detail}";
|
|
if (IsOverflow(message))
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.ContextOverflow, $"context_length_exceeded: {message}");
|
|
throw Stream(message);
|
|
}
|
|
|
|
// After a latched handoff a success result belongs to Claude Code's own continuation turn; the proposal stands.
|
|
if (HandoffLatched) return [];
|
|
ThrowDeferredFailure();
|
|
// Only a successful result must account for every block it opened; an error result is reported as itself.
|
|
if (_openIndex is not null) throw Stream("Claude result arrived with unclosed content blocks");
|
|
if (_stopReason is null && Str(record, "stop_reason") is { Length: > 0 } resultStop) _stopReason = resultStop;
|
|
if (_stopReason == "tool_use") throw Stream("Claude returned success after a tool-use stop");
|
|
|
|
var events = new List<AssistantMessageEvent>();
|
|
if (Builder.Current.Content.Count == 0 && !string.IsNullOrEmpty(resultText))
|
|
{
|
|
events.Add(Builder.StartText());
|
|
events.Add(Builder.AppendText(resultText));
|
|
events.Add(Builder.EndText());
|
|
}
|
|
_successfulResult = true;
|
|
PendingStop = IsLengthStop(_stopReason) ? StopReason.Length : StopReason.Stop;
|
|
return events;
|
|
}
|
|
|
|
/// <summary>Whether an error result is only the provider's own handoff termination (SIGTERM after a tool-use or length stop).</summary>
|
|
private bool IsExpectedHandoffTermination(JsonElement record)
|
|
{
|
|
if (!HandoffLatched) return false;
|
|
if (Str(record, "subtype") != "error_during_execution") return false;
|
|
if (record.TryGetProperty("api_error_status", out var status) && status.ValueKind != JsonValueKind.Null) return false;
|
|
if (record.TryGetProperty("result", out var result) && result.ValueKind != JsonValueKind.Null) return false;
|
|
return true;
|
|
}
|
|
|
|
private void ThrowDeferredFailure()
|
|
{
|
|
var message = _deferredToolArguments;
|
|
if (message is null) return;
|
|
_deferredToolArguments = null;
|
|
throw Stream(message);
|
|
}
|
|
|
|
// ---- diagnostics ----
|
|
|
|
/// <summary>
|
|
/// Claude Code recovers from an API failure that arrives after this response began streaming by replaying the
|
|
/// request, continuing from the partial it kept, or re-requesting without streaming. The events already published
|
|
/// are append-only, so none of those recoveries can be published here: the provider reports one retryable failure
|
|
/// and the core re-runs the turn from the unchanged context in a fresh process (a prompt-cache hit).
|
|
/// </summary>
|
|
private void RejectInterruptedStream(JsonElement record, string? type, string? subtype)
|
|
{
|
|
// Before the response starts a retry is an ordinary pre-stream retry. After a tool-use or length stop the
|
|
// provider is already terminating Claude; records about Claude's own next request must not invalidate the proposal.
|
|
if (!_messageStarted || _stopReason == "tool_use" || IsLengthStop(_stopReason)) return;
|
|
string? cause = null;
|
|
if (type == "system" && subtype == "api_retry")
|
|
{
|
|
var category = Str(record, "error") is { Length: > 0 } error ? error : "unknown";
|
|
var status = record.TryGetProperty("error_status", out var statusElement) && statusElement.ValueKind == JsonValueKind.Number
|
|
? $", HTTP {statusElement.GetRawText()}"
|
|
: "";
|
|
// A recorded rate-limit rejection already carries the reset time; it is more useful than the category alone.
|
|
if (category == "rate_limit" && _rateLimitFailure is not null)
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.StreamInterrupted, _rateLimitFailure, retryable: true);
|
|
if (!RetryableInterruptions.Contains(category))
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.StreamInterrupted, $"Claude Code API request failed mid-response ({category}{status})");
|
|
cause = $"Claude Code began retrying: {category}{status}";
|
|
}
|
|
else if (type == "assistant")
|
|
{
|
|
if (Str(record, "error") is not null || Bool(record, "is_api_error_message") == true)
|
|
cause = "Claude Code reported a mid-response API error";
|
|
else if (!_messageStopped && Obj(record, "message") is { } message && Str(message, "id") is { } id && id != _streamMessageId)
|
|
// A different message id before this stream stops is Claude Code's non-streaming replacement.
|
|
cause = "Claude Code replaced the stream with a non-streaming request";
|
|
}
|
|
else if (type == "user" && Bool(record, "isSynthetic") == true)
|
|
{
|
|
// Claude Code's synthetic user turns ("Resume directly ...") each start another internal model turn.
|
|
cause = "Claude Code asked the model to continue a cut-off response";
|
|
}
|
|
// The fixed "stream ended before message_stop" phrasing is what the core's retry classifier matches.
|
|
if (cause is not null)
|
|
throw new ClaudeCodeException(ClaudeCodeException.Codes.StreamInterrupted, $"{InterruptedPhrase} ({cause})", retryable: true);
|
|
}
|
|
|
|
private void AcceptAssistant(JsonElement record)
|
|
{
|
|
if (_assistantDiagnostic is not null) return;
|
|
var error = Str(record, "error")?.Trim() ?? "";
|
|
var text = "";
|
|
if (Obj(record, "message") is { } message && message.TryGetProperty("content", out var content))
|
|
{
|
|
if (content.ValueKind == JsonValueKind.String) text = content.GetString()!.Trim();
|
|
else if (content.ValueKind == JsonValueKind.Array)
|
|
text = string.Join(" ", content.EnumerateArray()
|
|
.Where(block => block.ValueKind == JsonValueKind.Object && Str(block, "type") == "text")
|
|
.Select(block => Str(block, "text")?.Trim())
|
|
.Where(t => !string.IsNullOrEmpty(t)));
|
|
}
|
|
var diagnostic = string.Join(": ", new[] { error, text }.Where(part => part.Length > 0));
|
|
if (diagnostic.Length > 0) _assistantDiagnostic = diagnostic;
|
|
}
|
|
|
|
private void AcceptRateLimit(JsonElement record)
|
|
{
|
|
// Rate-limit events are advisory; malformed details must not fail an otherwise valid request.
|
|
if (!record.TryGetProperty("rate_limit_info", out var info) || info.ValueKind != JsonValueKind.Object) return;
|
|
var notice = ParseRateLimitNotice(info);
|
|
if (notice is null) return;
|
|
_notices.Add(FormatNotice(notice));
|
|
if (notice.Status == "rejected") _rateLimitFailure = RejectionMessage(notice);
|
|
}
|
|
|
|
private sealed record RateLimitNotice(
|
|
string Status,
|
|
string RateLimitType,
|
|
double? Utilization,
|
|
DateTimeOffset? ResetsAt,
|
|
string? OverageStatus,
|
|
DateTimeOffset? OverageResetsAt,
|
|
string? OverageDisabledReason,
|
|
bool IsUsingOverage);
|
|
|
|
private static RateLimitNotice? ParseRateLimitNotice(JsonElement info)
|
|
{
|
|
var primaryStatus = AlertStatus(Str(info, "status"));
|
|
var overageStatus = AlertStatus(Str(info, "overageStatus"));
|
|
var usingOverage = Bool(info, "isUsingOverage") == true;
|
|
// org_level_disabled is the normal repeated state without usage credits, so overage constrains a request only
|
|
// together with a plan rejection or actual overage use.
|
|
var overageBlocking = overageStatus == "rejected" && (primaryStatus == "rejected" || usingOverage);
|
|
if (primaryStatus is null && !overageBlocking) return null;
|
|
var status = primaryStatus == "rejected" || overageBlocking ? "rejected" : "allowed_warning";
|
|
var hasPrimary = primaryStatus is not null;
|
|
var overageRelevant = overageStatus is not null && (primaryStatus == "rejected" || usingOverage);
|
|
var rateLimitType = hasPrimary
|
|
? Str(info, "rateLimitType") is { } t && !string.IsNullOrWhiteSpace(t) ? t.Trim() : "unknown"
|
|
: "overage";
|
|
var resetsAt = Timestamp(info, hasPrimary ? "resetsAt" : "overageResetsAt");
|
|
var overageReset = Timestamp(info, "overageResetsAt");
|
|
var utilization = info.TryGetProperty("utilization", out var u) && u.ValueKind == JsonValueKind.Number && u.TryGetDouble(out var value) && value is >= 0 and <= 1
|
|
? value
|
|
: (double?)null;
|
|
var reason = Str(info, "overageDisabledReason") is { } r && !string.IsNullOrWhiteSpace(r) ? r.Trim() : null;
|
|
return new RateLimitNotice(
|
|
status,
|
|
rateLimitType,
|
|
hasPrimary ? utilization : null,
|
|
resetsAt,
|
|
overageRelevant ? overageStatus : null,
|
|
overageRelevant && hasPrimary ? overageReset : null,
|
|
overageRelevant ? reason : null,
|
|
usingOverage);
|
|
}
|
|
|
|
private static string RejectionMessage(RateLimitNotice notice)
|
|
{
|
|
var reason = notice.OverageDisabledReason is null ? "" : $"; {notice.OverageDisabledReason}";
|
|
var reset = notice.ResetsAt is null ? "" : $"; resets at {notice.ResetsAt.Value.UtcDateTime.ToString("O", CultureInfo.InvariantCulture)}";
|
|
return $"Claude rate limit rejected ({notice.RateLimitType}){reason}{reset}";
|
|
}
|
|
|
|
private static string FormatNotice(RateLimitNotice notice)
|
|
{
|
|
var reset = notice.ResetsAt is null ? "" : $"; resets at {notice.ResetsAt.Value.ToLocalTime().ToString("g", CultureInfo.CurrentCulture)}";
|
|
var overageReset = notice.OverageResetsAt is null ? "" : $"; overage resets at {notice.OverageResetsAt.Value.ToLocalTime().ToString("g", CultureInfo.CurrentCulture)}";
|
|
var overage = notice.OverageStatus is null ? "" : $"; overage {notice.OverageStatus}{(notice.OverageDisabledReason is null ? "" : $" ({notice.OverageDisabledReason})")}";
|
|
var usingOverage = notice.IsUsingOverage ? "; using overage" : "";
|
|
if (notice.Status == "rejected")
|
|
return $"[claude-code] Claude rate limited ({notice.RateLimitType}){reset}{overageReset}{overage}{usingOverage}";
|
|
var usage = notice.Utilization is null ? "usage is approaching the limit" : $"{Math.Floor(notice.Utilization.Value * 100)}% used";
|
|
return $"[claude-code] Claude rate limit warning: {usage} ({notice.RateLimitType}){reset}{overageReset}{overage}{usingOverage}";
|
|
}
|
|
|
|
private static string? AlertStatus(string? value) => value is "allowed_warning" or "rejected" ? value : null;
|
|
|
|
private static DateTimeOffset? Timestamp(JsonElement info, string name)
|
|
{
|
|
if (!info.TryGetProperty(name, out var element) || element.ValueKind != JsonValueKind.Number || !element.TryGetDouble(out var value) || value <= 0 || double.IsInfinity(value))
|
|
return null;
|
|
// The current wire value is Unix seconds; a plausible epoch-millisecond value is not converted twice.
|
|
var milliseconds = value >= EpochMillisecondsThreshold ? value : value * 1000;
|
|
try
|
|
{
|
|
return DateTimeOffset.FromUnixTimeMilliseconds((long)milliseconds);
|
|
}
|
|
catch (ArgumentOutOfRangeException)
|
|
{
|
|
return null;
|
|
}
|
|
}
|
|
|
|
// ---- usage ----
|
|
|
|
private void ApplyUsage(JsonElement container)
|
|
{
|
|
if (Obj(container, "usage") is not { } usage) return;
|
|
if (Int(usage, "input_tokens") is { } input) _input = input;
|
|
if (Int(usage, "output_tokens") is { } output) _output = output;
|
|
if (Int(usage, "cache_read_input_tokens") is { } cacheRead) _cacheRead = cacheRead;
|
|
if (Int(usage, "cache_creation_input_tokens") is { } cacheWrite) _cacheWrite = cacheWrite;
|
|
if (Obj(usage, "output_tokens_details") is { } details && Int(details, "thinking_tokens") is { } reasoning) _reasoning = reasoning;
|
|
Builder.SetUsage(new Usage(_input, _output, _cacheRead, _cacheWrite, _reasoning));
|
|
}
|
|
|
|
private void ApplyModelUsage(JsonElement record)
|
|
{
|
|
if (Obj(record, "modelUsage") is not { } modelUsage) return;
|
|
foreach (var entry in modelUsage.EnumerateObject())
|
|
{
|
|
if (entry.Value.ValueKind != JsonValueKind.Object) continue;
|
|
ServedContextWindow ??= Int(entry.Value, "contextWindow");
|
|
ServedMaxOutputTokens ??= Int(entry.Value, "maxOutputTokens");
|
|
}
|
|
}
|
|
|
|
// ---- helpers ----
|
|
|
|
private static bool IsLengthStop(string? reason) => reason is "max_tokens" or "model_context_window_exceeded";
|
|
|
|
private static ClaudeCodeException Stream(string message) => new(ClaudeCodeException.Codes.ProtocolStream, message);
|
|
|
|
private static ClaudeCodeException Init(string message) => new(ClaudeCodeException.Codes.ProtocolInit, message);
|
|
|
|
private static int Index(JsonElement ev)
|
|
{
|
|
if (!ev.TryGetProperty("index", out var index) || index.ValueKind != JsonValueKind.Number || !index.TryGetInt32(out var value) || value < 0)
|
|
throw Stream($"Invalid Claude content index: {(ev.TryGetProperty("index", out var raw) ? raw.ToString() : "missing")}");
|
|
return value;
|
|
}
|
|
|
|
private static string? Str(JsonElement obj, string name) =>
|
|
obj.ValueKind == JsonValueKind.Object && obj.TryGetProperty(name, out var value) && value.ValueKind == JsonValueKind.String ? value.GetString() : null;
|
|
|
|
private static JsonElement? Obj(JsonElement obj, string name) =>
|
|
obj.ValueKind == JsonValueKind.Object && obj.TryGetProperty(name, out var value) && value.ValueKind == JsonValueKind.Object ? value : null;
|
|
|
|
private static bool? Bool(JsonElement obj, string name) =>
|
|
obj.ValueKind == JsonValueKind.Object && obj.TryGetProperty(name, out var value) && value.ValueKind is JsonValueKind.True or JsonValueKind.False ? value.GetBoolean() : null;
|
|
|
|
private static int? Int(JsonElement obj, string name)
|
|
{
|
|
if (obj.ValueKind != JsonValueKind.Object || !obj.TryGetProperty(name, out var value) || value.ValueKind != JsonValueKind.Number) return null;
|
|
if (value.TryGetInt32(out var i)) return i;
|
|
return value.TryGetDouble(out var d) && double.IsFinite(d) ? (int)Math.Clamp(d, int.MinValue, int.MaxValue) : null;
|
|
}
|
|
|
|
private static string? FirstNonEmpty(params string?[] values) => values.FirstOrDefault(v => !string.IsNullOrWhiteSpace(v))?.Trim();
|
|
|
|
private static string? ErrorsText(JsonElement record)
|
|
{
|
|
if (!record.TryGetProperty("errors", out var errors) || errors.ValueKind != JsonValueKind.Array) return null;
|
|
var text = string.Join("; ", errors.EnumerateArray().Where(e => e.ValueKind == JsonValueKind.String).Select(e => e.GetString()!.Trim()).Where(s => s.Length > 0));
|
|
return text.Length == 0 ? null : text;
|
|
}
|
|
|
|
private static string RecordKind(string? type, string? subtype)
|
|
{
|
|
static string Part(string? value) => value is null ? "" : Truncate(Sanitize().Replace(value, ""), 48);
|
|
var t = Part(type);
|
|
var s = Part(subtype);
|
|
if (t.Length == 0) t = "untyped";
|
|
return s.Length == 0 ? t : $"{t}/{s}";
|
|
}
|
|
|
|
private static string InventoryNames(JsonElement entries)
|
|
{
|
|
var names = entries.EnumerateArray().Take(5).Select(entry =>
|
|
{
|
|
var name = entry.ValueKind == JsonValueKind.String ? entry.GetString() : Str(entry, "name");
|
|
if (string.IsNullOrEmpty(name)) return "unnamed";
|
|
return name.Contains('/') || name.Contains('\\') ? "<path>" : Truncate(name, 64);
|
|
});
|
|
var count = entries.GetArrayLength();
|
|
return $"{string.Join(", ", names)}{(count > 5 ? $", and {count - 5} more" : "")}";
|
|
}
|
|
|
|
private static string ServerStatuses(JsonElement servers)
|
|
{
|
|
var statuses = servers.EnumerateArray().Select(s => Str(s, "status")).Where(s => s is not null).Distinct().ToList();
|
|
return statuses.Count == 0 ? "" : $" [{string.Join(", ", statuses)}]";
|
|
}
|
|
|
|
private static string FormatNames(IEnumerable<string> names)
|
|
{
|
|
var joined = string.Join(", ", names.OrderBy(n => n, StringComparer.Ordinal));
|
|
return joined.Length == 0 ? "none" : joined;
|
|
}
|
|
|
|
private static string Truncate(string value, int max) => value.Length <= max ? value : value[..max];
|
|
|
|
[GeneratedRegex(@"prompt is too long|prompt too long|context window exceeded|context length exceeded|context_length_exceeded|input length", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant)]
|
|
private static partial Regex Overflow();
|
|
|
|
[GeneratedRegex(@"[^\w.-]", RegexOptions.CultureInvariant)]
|
|
private static partial Regex Sanitize();
|
|
}
|