halkyon/packages/halkyon_llm/test/anthropic_test.dart

import 'dart:async';
import 'dart:convert';
import 'dart:typed_data';

import 'package:halkyon_llm/halkyon_llm.dart';
import 'package:http/http.dart' as http;
import 'package:http/testing.dart';
import 'package:test/test.dart';

import 'support/fixture.dart';

final _endpoint = LlmEndpoint(
  protocol: LlmProtocol.anthropic,
  model: 'claude-sonnet-5-5',
  apiKey: 'sk-ant-test',
);

final _question = LlmRequest(
  system: 'Du hilfst beim Japanischlernen.',
  messages: [
    LlmMessage.user([const LlmText('Was heißt „Hund“ auf Japanisch?')]),
  ],
  maxOutputTokens: 4096,
  personId: 'p_8Zq3',
);

final _documented = fixturesOf('documented', 'anthropic');

/// What [request] streams, against [fixture], each event, then the error
/// the stream ended with, if any.
Future<(List<LlmEvent>, LlmFailure?)> _reply(
  Fixture fixture, {
  LlmRequest? request,
  Sent? sent,
  int chunk = 7,
}) async {
  final client = LlmClient(
    httpClient: replying(fixture, sent: sent, chunk: chunk),
  );
  final events = <LlmEvent>[];
  try {
    await client.send(_endpoint, request ?? _question).forEach(events.add);
  } on LlmFailure catch (failure) {
    return (events, failure);
  }
  return (events, null);
}

String _text(List<LlmEvent> events) =>
    events.whereType<LlmTextDelta>().map((delta) => delta.text).join();

void main() {
  group('the request', () {
    test(
      "goes to the Messages API with the key and the API's version",
      () async {
        final sent = Sent();
        await _reply(_documented['text']!, sent: sent);
        expect(sent.request.method, 'POST');
        expect(
          sent.request.url,
          Uri.parse('https://api.anthropic.com/v1/messages'),
        );
        expect(sent.request.headers, containsPair('x-api-key', 'sk-ant-test'));
        expect(
          sent.request.headers,
          containsPair('anthropic-version', '2023-06-01'),
        );
        expect(
          sent.request.headers['content-type'],
          startsWith('application/json'),
        );
        expect(
          sent.request.headers.keys,
          isNot(contains('anthropic-dangerous-direct-browser-access')),
          reason: 'only in a browser',
        );
      },
    );

    test('takes a base URL of its own, with its path', () async {
      final sent = Sent();
      final client = LlmClient(
        httpClient: replying(_documented['text']!, sent: sent),
      );
      await client
          .send(
            LlmEndpoint(
              protocol: LlmProtocol.anthropic,
              model: 'm',
              apiKey: 'k',
              baseUrl: 'https://proxy.example.org/anthropic/',
            ),
            _question,
          )
          .drain<void>();
      expect(
        sent.request.url,
        Uri.parse('https://proxy.example.org/anthropic/v1/messages'),
      );
    });

    test('streams the conversation with the system prompt, the ceiling and the '
        "person's opaque ID", () async {
      final sent = Sent();
      await _reply(_documented['text']!, sent: sent);
      expect(sent.body, {
        'model': 'claude-sonnet-5-5',
        'max_tokens': 4096,
        'stream': true,
        'system': 'Du hilfst beim Japanischlernen.',
        'messages': [
          {
            'role': 'user',
            'content': [
              {'type': 'text', 'text': 'Was heißt „Hund“ auf Japanisch?'},
            ],
          },
        ],
        'metadata': {'user_id': 'p_8Zq3'},
      });
    });

    test('leaves out an empty system prompt, no tools and no person', () async {
      final sent = Sent();
      await _reply(
        _documented['text']!,
        sent: sent,
        request: LlmRequest(
          messages: [
            LlmMessage.user([const LlmText('Hallo')]),
          ],
          maxOutputTokens: 100,
        ),
      );
      expect(
        sent.body.keys,
        unorderedEquals(['model', 'max_tokens', 'stream', 'messages']),
      );
    });

    test('sends tools, images, tool calls, results first, and its own blocks '
        'back unchanged', () async {
      final sent = Sent();
      const thinking = LlmProviderBlock(
        protocol: LlmProtocol.anthropic,
        block: {'type': 'thinking', 'thinking': '', 'signature': 'EqQB'},
      );
      await _reply(
        _documented['text']!,
        sent: sent,
        request: LlmRequest(
          messages: [
            LlmMessage.user([
              const LlmText('Was steht da?'),
              LlmImage(
                mediaType: 'image/png',
                bytes: Uint8List.fromList([1, 2, 3]),
              ),
            ]),
            LlmMessage.assistant([
              thinking,
              const LlmText(''),
              const LlmProviderBlock(
                protocol: LlmProtocol.openai,
                block: {'type': 'other'},
              ),
              const LlmToolCall(
                id: 'toolu_1',
                name: 'find_cards',
                input: {'query': '犬'},
              ),
            ]),
            LlmMessage.user([
              const LlmText('Und?'),
              const LlmToolResult(
                callId: 'toolu_1',
                content: '[]',
                isError: true,
              ),
            ]),
          ],
          tools: const [
            LlmTool(
              name: 'find_cards',
              description: 'Finds cards.',
              inputSchema: {
                'type': 'object',
                'properties': {
                  'query': {'type': 'string'},
                },
              },
            ),
          ],
          maxOutputTokens: 100,
        ),
      );
      expect(sent.body['messages'], [
        {
          'role': 'user',
          'content': [
            {'type': 'text', 'text': 'Was steht da?'},
            {
              'type': 'image',
              'source': {
                'type': 'base64',
                'media_type': 'image/png',
                'data': 'AQID',
              },
            },
          ],
        },
        {
          'role': 'assistant',
          'content': [
            {'type': 'thinking', 'thinking': '', 'signature': 'EqQB'},
            {
              'type': 'tool_use',
              'id': 'toolu_1',
              'name': 'find_cards',
              'input': {'query': '犬'},
            },
          ],
        },
        {
          'role': 'user',
          'content': [
            {
              'type': 'tool_result',
              'tool_use_id': 'toolu_1',
              'content': '[]',
              'is_error': true,
            },
            {'type': 'text', 'text': 'Und?'},
          ],
        },
      ]);
      expect(sent.body['tools'], [
        {
          'name': 'find_cards',
          'description': 'Finds cards.',
          'input_schema': {
            'type': 'object',
            'properties': {
              'query': {'type': 'string'},
            },
          },
        },
      ]);
    });

    test('leaves out a turn with nothing to send', () async {
      final sent = Sent();
      await _reply(
        _documented['text']!,
        sent: sent,
        request: LlmRequest(
          messages: [
            LlmMessage.user([const LlmText('a')]),
            LlmMessage.assistant([const LlmText('')]),
            LlmMessage.user([const LlmText('b')]),
          ],
          maxOutputTokens: 100,
        ),
      );
      expect(sent.body['messages'], hasLength(2));
    });
  });

  group('documented replies', () {
    for (final chunk in [1, 7, 4096]) {
      test('text streams as written, in pieces of $chunk bytes', () async {
        final (events, failure) = await _reply(
          _documented['text']!,
          chunk: chunk,
        );
        expect(failure, isNull);
        expect(_text(events), '„Hund“ heißt auf Japanisch 犬 (いぬ, inu).');
        final finished = events.last as LlmFinished;
        expect(finished.stop, LlmStop.end);
        expect(
          finished.usage,
          const LlmUsage(inputTokens: 412, outputTokens: 38),
        );
        expect(finished.message.role, LlmRole.assistant);
        expect(
          finished.message.text,
          '„Hund“ heißt auf Japanisch 犬 (いぬ, inu).',
        );
      });
    }

    test(
      'a thinking block is kept, its signature whole, to be sent back',
      () async {
        final (events, _) = await _reply(_documented['text']!);
        final parts = (events.last as LlmFinished).message.parts;
        expect(parts, hasLength(2));
        final block = parts.first as LlmProviderBlock;
        expect(block.protocol, LlmProtocol.anthropic);
        expect(block.block, {
          'type': 'thinking',
          'thinking': '',
          'signature': 'EqQBCgIYAhIMxk3Lr9vQm2HbW7pTGgy1d2nXo8Rf4KsJ0eQiMNp6',
        });
        expect(parts.last, isA<LlmText>());
      },
    );

    test('a tool call comes whole, its input put together', () async {
      final (events, failure) = await _reply(_documented['tool']!);
      expect(failure, isNull);
      expect(_text(events), 'Ich sehe im Studienset nach.');
      final finished = events.last as LlmFinished;
      expect(finished.stop, LlmStop.toolCalls);
      expect(
        finished.usage,
        const LlmUsage(inputTokens: 472, outputTokens: 89),
      );
      final [call] = finished.message.toolCalls;
      expect(call.id, 'toolu_01T1x1fJ34qAmk2tNTrN7Up6');
      expect(call.name, 'find_cards');
      expect(call.input, {'query': 'Hund', 'limit': 5});
    });

    test(
      'an error streamed mid-answer fails it, with what was counted',
      () async {
        final (events, failure) = await _reply(
          _documented['overloaded_midstream']!,
        );
        expect(_text(events), '„Hund“');
        expect(events.whereType<LlmFinished>(), isEmpty);
        expect(failure!.kind, LlmFailureKind.unavailable);
        expect(failure.detail, 'Overloaded');
        expect(
          failure.usage,
          const LlmUsage(inputTokens: 412, outputTokens: 1),
        );
      },
    );

    final statuses = {
      'status_400': (LlmFailureKind.request, 400, null),
      'status_401': (LlmFailureKind.key, 401, null),
      'status_402': (LlmFailureKind.key, 402, null),
      'status_429': (
        LlmFailureKind.rateLimited,
        429,
        const Duration(seconds: 30),
      ),
      'status_529': (LlmFailureKind.unavailable, 529, null),
    };
    for (final MapEntry(key: name, value: (kind, status, retryAfter))
        in statuses.entries) {
      test('$name fails as ${kind.name}', () async {
        final (events, failure) = await _reply(_documented[name]!);
        expect(events, isEmpty);
        expect(failure!.kind, kind);
        expect(failure.status, status);
        expect(failure.retryAfter, retryAfter);
        expect(failure.usage, LlmUsage.none);
      });
    }

    test("keeps the provider's words for the logs", () async {
      final (_, failure) = await _reply(_documented['status_401']!);
      expect(failure!.detail, 'invalid x-api-key');
    });
  });

  group('replies off the documentation', () {
    Fixture stream(String events) => Fixture.parse(
      'inline',
      Uint8List.fromList(
        utf8.encode('HTTP 200\ncontent-type: text/event-stream\n\n$events'),
      ),
    );
    const start =
        'event: message_start\ndata: {"type":"message_start","message":'
        '{"usage":{"input_tokens":10,"output_tokens":1}}}\n\n';
    const text =
        'event: content_block_start\ndata: {"type":"content_block_start",'
        '"index":0,"content_block":{"type":"text","text":""}}\n\n'
        'event: content_block_delta\ndata: {"type":"content_block_delta",'
        '"index":0,"delta":{"type":"text_delta","text":"Hi"}}\n\n'
        'event: content_block_stop\ndata: {"type":"content_block_stop",'
        '"index":0}\n\n';
    const end =
        'event: message_delta\ndata: {"type":"message_delta","delta":'
        '{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}\n\n'
        'event: message_stop\ndata: {"type":"message_stop"}\n\n';

    test('events and deltas of kinds yet to come are passed over', () async {
      final (events, failure) = await _reply(
        stream(
          '${start}event: future_event\ndata: {"type":"future_event"}\n\n'
          'event: content_block_start\ndata: {"type":"content_block_start",'
          '"index":0,"content_block":{"type":"future_block"}}\n\n'
          'event: content_block_delta\ndata: {"type":"content_block_delta",'
          '"index":0,"delta":{"type":"future_delta"}}\n\n'
          'event: content_block_stop\ndata: {"type":"content_block_stop",'
          '"index":0}\n\n'
          '${text.replaceAll('"index":0', '"index":1')}$end',
        ),
      );
      expect(failure, isNull);
      expect((events.last as LlmFinished).message.parts, [isA<LlmText>()]);
    });

    test('a redacted thinking block is kept as it came', () async {
      final (events, _) = await _reply(
        stream(
          '${start}event: content_block_start\ndata: {"type":"content_block_start",'
          '"index":0,"content_block":{"type":"redacted_thinking","data":"EmwKAhgB"}}\n\n'
          'event: content_block_stop\ndata: {"type":"content_block_stop",'
          '"index":0}\n\n$end',
        ),
      );
      final block =
          (events.last as LlmFinished).message.parts.single as LlmProviderBlock;
      expect(block.block, {'type': 'redacted_thinking', 'data': 'EmwKAhgB'});
    });

    test('usage with input in message_delta takes the later count', () async {
      final (events, _) = await _reply(
        stream(
          '$start$text${end.replaceAll('"output_tokens":5', '"input_tokens":12,"cache_read_input_tokens":3,"output_tokens":5')}',
        ),
      );
      expect(
        (events.last as LlmFinished).usage,
        const LlmUsage(inputTokens: 15, outputTokens: 5),
      );
    });

    test('an empty answer is an empty text', () async {
      final (events, _) = await _reply(stream('$start$end'));
      expect((events.last as LlmFinished).message.text, '');
    });

    for (final (reason, stop) in [
      ('max_tokens', LlmStop.maxTokens),
      ('model_context_window_exceeded', LlmStop.maxTokens),
      ('stop_sequence', LlmStop.end),
      ('refusal', LlmStop.refused),
      ('pause_turn', LlmStop.other),
    ]) {
      test('stop reason $reason is ${stop.name}', () async {
        final (events, _) = await _reply(
          stream('$start$text${end.replaceAll('end_turn', reason)}'),
        );
        expect((events.last as LlmFinished).stop, stop);
      });
    }

    test('a reply that ends before message_stop fails as malformed, with what '
        'was counted', () async {
      final (events, failure) = await _reply(stream('$start$text'));
      expect(_text(events), 'Hi');
      expect(failure!.kind, LlmFailureKind.malformed);
      expect(failure.usage, const LlmUsage(inputTokens: 10, outputTokens: 1));
    });

    test("a tool call whose input isn't JSON fails as malformed", () async {
      final (_, failure) = await _reply(
        stream(
          '${start}event: content_block_start\ndata: {"type":"content_block_start",'
          '"index":0,"content_block":{"type":"tool_use","id":"t","name":"n","input":{}}}\n\n'
          'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,'
          '"delta":{"type":"input_json_delta","partial_json":"{\\"query\\": "}}\n\n'
          'event: content_block_stop\ndata: {"type":"content_block_stop",'
          '"index":0}\n\n$end',
        ),
      );
      expect(failure!.kind, LlmFailureKind.malformed);
    });

    test('a field of another type fails as malformed, without quoting the reply', () async {
      final (_, failure) = await _reply(
        stream(
          'event: content_block_start\ndata: {"type":"content_block_start",'
          '"index":"Geheimnis","content_block":{"type":"text","text":""}}\n\n',
        ),
      );
      expect(failure!.kind, LlmFailureKind.malformed);
      expect(failure.detail, isNot(contains('Geheimnis')));
    });

    test('a status without a JSON body fails by its status', () async {
      final (_, failure) = await _reply(
        Fixture.parse(
          'inline',
          Uint8List.fromList(
            utf8.encode('HTTP 502\ncontent-type: text/html\n\n<html>'),
          ),
        ),
      );
      expect(failure!.kind, LlmFailureKind.provider);
      expect(failure.status, 502);
    });

    test("what looks like a key in the provider's words is cut", () async {
      final (_, failure) = await _reply(
        Fixture.parse(
          'inline',
          Uint8List.fromList(
            utf8.encode(
              'HTTP 401\n\n{"type":"error","error":{"type":"authentication_error",'
              '"message":"Incorrect API key provided: sk-ant-api03-abcdEFGH1234"}}',
            ),
          ),
        ),
      );
      expect(failure!.detail, 'Incorrect API key provided: sk-…');
    });
  });

  group('the connection', () {
    test('a provider out of reach fails as network', () {
      final client = LlmClient(
        httpClient: MockClient(
          (_) => throw http.ClientException('Connection refused'),
        ),
      );
      expect(
        client.send(_endpoint, _question).drain<void>(),
        throwsA(
          isA<LlmFailure>().having(
            (f) => f.kind,
            'kind',
            LlmFailureKind.network,
          ),
        ),
      );
    });

    test('a reply that goes silent fails as network', () {
      final client = HttpLlmClient(
        httpClient: MockClient.streaming(
          (_, _) async => http.StreamedResponse(
            Stream<List<int>>.fromFuture(Completer<List<int>>().future),
            200,
          ),
        ),
        idleTimeout: const Duration(milliseconds: 20),
      );
      expect(
        client.send(_endpoint, _question).drain<void>(),
        throwsA(
          isA<LlmFailure>().having(
            (f) => f.kind,
            'kind',
            LlmFailureKind.network,
          ),
        ),
      );
    });

    test('a reply that never begins fails as network', () {
      final client = HttpLlmClient(
        httpClient: MockClient.streaming(
          (_, _) => Completer<http.StreamedResponse>().future,
        ),
        headersTimeout: const Duration(milliseconds: 20),
      );
      expect(
        client.send(_endpoint, _question).drain<void>(),
        throwsA(
          isA<LlmFailure>().having(
            (f) => f.kind,
            'kind',
            LlmFailureKind.network,
          ),
        ),
      );
    });

    test('cancelling the answer aborts the request', () async {
      final sent = Sent();
      final body = StreamController<List<int>>();
      final client = LlmClient(
        httpClient: MockClient.streaming((request, bodyStream) async {
          sent.request = request;
          await bodyStream.drain<void>();
          if (request case http.Abortable(:final abortTrigger?)) {
            abortTrigger.whenComplete(() => sent.aborted = true).ignore();
          }
          return http.StreamedResponse(body.stream, 200);
        }),
      );
      final first = Completer<void>();
      final subscription = client.send(_endpoint, _question).listen((_) {
        if (!first.isCompleted) {
          first.complete();
        }
      });
      body.add(
        utf8.encode(
          'event: content_block_start\ndata: {"type":"content_block_start",'
          '"index":0,"content_block":{"type":"text","text":"Hi"}}\n\n',
        ),
      );
      await first.future;
      await subscription.cancel();
      await pumpEventQueue();
      expect(sent.aborted, isTrue);
      await body.close();
    });
  });
}