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, LlmFailure?)> _reply( Fixture fixture, { LlmRequest? request, Sent? sent, int chunk = 7, }) async { final client = LlmClient( httpClient: replying(fixture, sent: sent, chunk: chunk), ); final events = []; try { await client.send(_endpoint, request ?? _question).forEach(events.add); } on LlmFailure catch (failure) { return (events, failure); } return (events, null); } String _text(List events) => events.whereType().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(); 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()); }, ); 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(), 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()]); }); 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'), ), ), ); 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(), throwsA( isA().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>.fromFuture(Completer>().future), 200, ), ), idleTimeout: const Duration(milliseconds: 20), ); expect( client.send(_endpoint, _question).drain(), throwsA( isA().having( (f) => f.kind, 'kind', LlmFailureKind.network, ), ), ); }); test('a reply that never begins fails as network', () { final client = HttpLlmClient( httpClient: MockClient.streaming( (_, _) => Completer().future, ), headersTimeout: const Duration(milliseconds: 20), ); expect( client.send(_endpoint, _question).drain(), throwsA( isA().having( (f) => f.kind, 'kind', LlmFailureKind.network, ), ), ); }); test('cancelling the answer aborts the request', () async { final sent = Sent(); final body = StreamController>(); final client = LlmClient( httpClient: MockClient.streaming((request, bodyStream) async { sent.request = request; await bodyStream.drain(); if (request case http.Abortable(:final abortTrigger?)) { abortTrigger.whenComplete(() => sent.aborted = true).ignore(); } return http.StreamedResponse(body.stream, 200); }), ); final first = Completer(); 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(); }); }); }