SharpClawCode/src/SharpClaw.Code.Providers/Internal/AnthropicSdkStreamAdapter.cs at 3c040f219874f3338e29215dbfded06b7077d0b6 · clawdotnet/SharpClawCode · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
using System.Runtime.CompilerServices;
using Anthropic.Models.Messages;
using SharpClaw.Code.Infrastructure.Abstractions;
using SharpClaw.Code.Providers.Models;
using SharpClaw.Code.Protocol.Models;

namespace SharpClaw.Code.Providers.Internal;

/// <summary>
/// Converts Anthropic SDK <see cref="RawMessageStreamEvent"/> sequences into SharpClaw <see cref="ProviderEvent"/> values.
/// </summary>
internal static class AnthropicSdkStreamAdapter
{
    /// <summary>
    /// Adapts a raw Anthropic message stream into provider events.
    /// </summary>
    public static async IAsyncEnumerable<ProviderEvent> AdaptAsync(
        IAsyncEnumerable<RawMessageStreamEvent> messageStream,
        string requestId,
        ISystemClock clock,
        [EnumeratorCancellation] CancellationToken cancellationToken)
    {
        // Track the current tool_use block being accumulated across events.
        string? pendingToolUseId = null;
        string? pendingToolName = null;
        System.Text.StringBuilder? pendingToolInputBuilder = null;

        IAsyncEnumerator<RawMessageStreamEvent>? enumerator = null;
        try
        {
            enumerator = messageStream.GetAsyncEnumerator(cancellationToken);
            while (true)
            {
                bool moved;
                string? streamError = null;
                try
                {
                    moved = await enumerator.MoveNextAsync().ConfigureAwait(false);
                }
                catch (OperationCanceledException)
                {
                    throw;
                }
                catch (Exception ex)
                {
                    cancellationToken.ThrowIfCancellationRequested();
                    streamError = ProviderStreamFailureClassifier.Describe(ex);
                    moved = false;
                }

                if (streamError is not null)
                {
                    yield return ProviderStreamEventFactory.Failed(requestId, clock, streamError);
                    yield break;
                }

                if (!moved)
                {
                    break;
                }

                var ev = enumerator.Current;

                if (ev.TryPickContentBlockStart(out var blockStart))
                {
                    if (blockStart.ContentBlock.TryPickToolUse(out var toolUse))
                    {
                        pendingToolUseId = toolUse.ID;
                        pendingToolName = toolUse.Name;
                        pendingToolInputBuilder = new System.Text.StringBuilder();
                    }
                }
                else if (ev.TryPickContentBlockDelta(out var blockDelta))
                {
                    var (deltaText, partialJson) = ExtractDeltas(blockDelta.Delta);
                    if (!string.IsNullOrEmpty(deltaText))
                    {
                        yield return ProviderStreamEventFactory.Delta(requestId, clock, deltaText);
                    }
                    else if (partialJson is not null && pendingToolInputBuilder is not null)
                    {
                        pendingToolInputBuilder.Append(partialJson);
                    }
                }
                else if (ev.TryPickContentBlockStop(out _))
                {
                    if (pendingToolUseId is not null && pendingToolName is not null && pendingToolInputBuilder is not null)
                    {
                        var toolInputJson = pendingToolInputBuilder.ToString();
                        yield return ProviderStreamEventFactory.ToolUse(requestId, clock, pendingToolUseId, pendingToolName, toolInputJson);
                        pendingToolUseId = null;
                        pendingToolName = null;
                        pendingToolInputBuilder = null;
                    }
                }
                else if (ev.TryPickStop(out _))
                {
                    yield return ProviderStreamEventFactory.Completed(requestId, clock, null);
                    yield break;
                }
            }
        }
        finally
        {
            if (enumerator is not null)
            {
                await enumerator.DisposeAsync().ConfigureAwait(false);
            }
        }

        yield return ProviderStreamEventFactory.Completed(requestId, clock, null);
    }

    private static (string? Text, string? PartialJson) ExtractDeltas(RawContentBlockDelta delta)
    {
        string? text = null;
        string? partialJson = null;

        delta.Match<int>(
            textDelta => { text = textDelta.Text; return 0; },
            inputJsonDelta => { partialJson = inputJsonDelta.PartialJson; return 0; },
            _ => 0,
            _ => 0,
            _ => 0);

        return (text, partialJson);
    }
}