halkyon/packages/halkyon_base_server/test/integration/assistant/assistant_test.dart

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

import 'package:clock/clock.dart';
import 'package:halkyon_base_server/base.dart';
import 'package:halkyon_llm/halkyon_llm.dart';
import 'package:halkyon_llm/testing.dart';
import 'package:serverpod/serverpod.dart';
import 'package:test/test.dart';

import '../support/fixtures.dart';
import '../support/password_support.dart';
import '../test_tools/serverpod_test_tools.dart';

// The assistants an operator grants (Halkyon Learn's ADR 0021, step 2): what
// a person is offered and agrees to (D40), a question forwarded to the
// provider and its answer streamed back (D35), refused without a grant, an
// agreement, room within the limits or a question the provider can take;
// what each request counts (D38, D42), kept 3 months (D41) and deleted with
// the person; and the operator's grants and figures, refused to anyone else.
// The provider is a FakeLlmClient: halkyon_llm's own tests hold the
// protocols.

/// 22:30 UTC on 25 Oct 2026: the day's limit resets at midnight, the
/// month's on 1 Nov.
final _now = DateTime.utc(2026, 10, 25, 22, 30);

final _settings = {
  'ASSISTANTS': 'claude local',
  'ASSISTANT_ID_KEY': 'k' * assistantIdKeyMinLength,
  'ASSISTANT_CURRENCY': 'EUR',
  'ASSISTANT_CLAUDE_NAME': 'Claude Sonnet',
  'ASSISTANT_CLAUDE_PROTOCOL': 'anthropic',
  'ASSISTANT_CLAUDE_MODEL': 'claude-sonnet-5-5',
  'ASSISTANT_CLAUDE_KEY': 'sk-ant-test',
  'ASSISTANT_CLAUDE_IMAGES': 'on',
  'ASSISTANT_CLAUDE_PRICE_INPUT': '3',
  'ASSISTANT_CLAUDE_PRICE_OUTPUT': '15',
  'ASSISTANT_LOCAL_NAME': 'Qwen at home',
  'ASSISTANT_LOCAL_PROTOCOL': 'openai',
  'ASSISTANT_LOCAL_BASE_URL': 'http://ollama:11434/v1',
  'ASSISTANT_LOCAL_MODEL': 'qwen3:8b',
};

Assistants _assistants([Map<String, String> changes = const {}]) =>
    assistantsFrom({..._settings, ...changes}.get);

AssistantMessage _user(List<AssistantPart> parts) =>
    AssistantMessage(role: AssistantRole.user, parts: parts);

AssistantMessage _model(List<AssistantPart> parts) =>
    AssistantMessage(role: AssistantRole.assistant, parts: parts);

AssistantPart _text(String text) =>
    AssistantPart(kind: AssistantPartKind.text, text: text);

AssistantQuestion _question({
  String assistant = 'claude',
  List<AssistantMessage>? messages,
  List<AssistantTool> tools = const [],
  int maxOutputTokens = 2048,
}) => AssistantQuestion(
  assistant: assistant,
  system: 'Du hilfst beim Japanischlernen.',
  messages:
      messages ??
      [
        _user([_text('Was heißt „Hund“?')]),
      ],
  tools: tools,
  maxOutputTokens: maxOutputTokens,
);

Matcher _refused(AssistantRefusal refusal) => throwsA(
  isA<AssistantRefusedException>().having(
    (e) => e.refusal,
    'refusal',
    refusal,
  ),
);

Matcher _refusedForOperator(OperatorRefusal reason) => throwsA(
  isA<OperatorRefusedException>().having((e) => e.reason, 'reason', reason),
);

void main() {
  withServerpod(
    'Assistants',
    // The figures are counted on a session of their own, outside any test's
    // transaction: these tests delete what they make instead.
    rollbackDatabase: RollbackDatabase.disabled,
    (sessionBuilder, endpoints) {
      late Session session;
      late Serverpod pod;
      late FakeLlmClient provider;

      /// The instance's operator.
      late UuidValue olga;
      late UuidValue olgasLaptop;

      /// Identified; no operator.
      late UuidValue ben;
      late UuidValue bensPhone;

      setUp(() async {
        useBase(sessionBuilder, app: 'learn');
        session = sessionBuilder.build();
        pod = session.serverpod;
        await _deleteAll(session);
        provider = FakeLlmClient();
        useAssistants(pod, _assistants());
        useLlmClient(pod, provider);
        olga = await newPersonWithAddress(
          session,
          'olga@example.org',
          handle: 'Olga',
        );
        olgasLaptop = await signInNewDevice(
          session,
          olga,
          scopes: {identified, instanceOperator},
        );
        ben = await newPersonWithAddress(
          session,
          'ben@example.org',
          handle: 'ben',
        );
        bensPhone = await signInNewDevice(session, ben, scopes: {identified});
      });
      tearDown(() async {
        await assistantUsageCounted();
        useAssistants(pod, null);
        useLlmClient(pod, null);
        await _deleteAll(sessionBuilder.build());
      });

      TestSessionBuilder notSignedIn() => sessionBuilder.copyWith(
        authentication: AuthenticationOverride.unauthenticated(),
      );
      TestSessionBuilder asOlga() => sessionBuilder.copyWith(
        authentication: AuthenticationOverride.authenticationInfo(
          olga.uuid,
          {identified, instanceOperator},
          authId: olgasLaptop.uuid,
        ),
      );
      TestSessionBuilder asBen() => sessionBuilder.copyWith(
        authentication: AuthenticationOverride.authenticationInfo(
          ben.uuid,
          {identified},
          authId: bensPhone.uuid,
        ),
      );

      Future<T> atNow<T>(Future<T> Function() body) =>
          withClock(Clock.fixed(_now), body);

      /// Grants Ben [assistant] with [daily] and [monthly] limits, as Olga.
      Future<void> grantBen({
        String assistant = 'claude',
        int? daily,
        int? monthly,
      }) => endpoints.assistantOperator.grant(
        asOlga(),
        account: ben,
        assistant: assistant,
        dailyLimit: daily,
        monthlyLimit: monthly,
      );

      /// Grants Ben [assistant], and he agrees to it.
      Future<void> grantAndAgree({
        String assistant = 'claude',
        int? daily,
        int? monthly,
      }) async {
        await grantBen(assistant: assistant, daily: daily, monthly: monthly);
        await endpoints.assistant.agree(asBen(), assistant: assistant);
      }

      /// A grant of [assistant] to Ben from before, which the instance may
      /// set no more (D44).
      Future<void> grantedBefore(String assistant) =>
          AssistantGrant.db.insertRow(
            session,
            AssistantGrant(
              authUserId: ben,
              assistant: assistant,
              grantedAt: _now,
              limitUnit: AssistantLimitUnit.tokens,
            ),
          );

      /// What Ben's question is answered with, whole.
      Future<List<AssistantAnswer>> askAsBen(AssistantQuestion question) =>
          endpoints.assistant.ask(asBen(), question: question).toList();

      /// Ben's figures of [assistant], day by day.
      Future<List<AssistantFigures>> figuresOfBen([
        String assistant = 'claude',
      ]) => AssistantFigures.db.find(
        session,
        where: (t) => t.authUserId.equals(ben) & t.assistant.equals(assistant),
        orderBy: (t) => t.day,
      );

      /// Ben's use of [assistant] on [day], as counted before.
      Future<void> usedBefore(
        DateTime day, {
        String assistant = 'claude',
        int inputTokens = 0,
        int outputTokens = 0,
        int costMicros = 0,
      }) => AssistantFigures.db.insertRow(
        session,
        AssistantFigures(
          authUserId: ben,
          assistant: assistant,
          day: day,
          requests: 1,
          inputTokens: inputTokens,
          outputTokens: outputTokens,
          costMicros: costMicros,
        ),
      );

      group('asking (D35, point 7)', () {
        test('streams the answer as the model writes it, then the whole turn '
            "with what was counted, forwarded with the operator's endpoint and "
            "the person's opaque ID (D30)", () async {
          await grantAndAgree();
          provider.replyText(
            '„Hund“ heißt 犬.',
            usage: const LlmUsage(inputTokens: 1000, outputTokens: 200),
          );

          final answers = await askAsBen(_question());

          expect(
            [for (final answer in answers) ?answer.delta].join(),
            '„Hund“ heißt 犬.',
          );
          final last = answers.last;
          expect(last.delta, isNull);
          expect(last.stop, AssistantStop.end);
          expect(last.message!.role, AssistantRole.assistant);
          expect(last.message!.parts.single.text, '„Hund“ heißt 犬.');
          expect((last.inputTokens, last.outputTokens), (1000, 200));

          final (endpoint, request) = provider.sent.single;
          expect(endpoint.protocol, LlmProtocol.anthropic);
          expect(endpoint.model, 'claude-sonnet-5-5');
          expect(endpoint.apiKey, 'sk-ant-test');
          expect(request.system, 'Du hilfst beim Japanischlernen.');
          expect(request.messages.single.text, 'Was heißt „Hund“?');
          expect(request.maxOutputTokens, 2048);
          expect(request.personId, _assistants().personIdOf(ben));
          expect(request.personId, isNot(contains(ben.uuid)));
        });

        test('counts each request into the UTC day, at the prices of the '
            'time (D38, D42)', () async {
          await grantAndAgree();
          provider
            ..replyText(
              'a',
              usage: const LlmUsage(inputTokens: 1000, outputTokens: 200),
            )
            ..replyText(
              'b',
              usage: const LlmUsage(inputTokens: 500, outputTokens: 100),
            );

          await atNow(() => askAsBen(_question()));
          await atNow(() => askAsBen(_question()));

          final day = (await figuresOfBen()).single;
          expect(day.day, DateTime.utc(2026, 10, 25));
          expect(day.requests, 2);
          expect((day.inputTokens, day.outputTokens), (1500, 300));
          // 1,500 input tokens at 3 and 300 output at 15 per million.
          expect(day.costMicros, 1500 * 3 + 300 * 15);
        });

        test("sends the conversation as the person's app holds it: images to "
            'an assistant that reads them, tool calls with what the provider '
            'wants back, their results, blocks only the provider reads; and '
            'hands the tool calls back alike', () async {
          await grantAndAgree();
          provider.reply([
            LlmFinished(
              message: LlmMessage.assistant(const [
                LlmProviderBlock(
                  protocol: LlmProtocol.anthropic,
                  block: {'type': 'thinking', 'signature': 'EqQB'},
                ),
                LlmToolCall(
                  id: 'toolu_2',
                  name: 'find_cards',
                  input: {'query': 'Katze'},
                  extra: {'x': 1},
                ),
              ]),
              stop: LlmStop.toolCalls,
              usage: const LlmUsage(inputTokens: 10, outputTokens: 5),
            ),
          ]);
          final picture = Uint8List.fromList([1, 2, 3]);

          final answers = await askAsBen(
            _question(
              messages: [
                _user([
                  _text('Was steht da?'),
                  AssistantPart(
                    kind: AssistantPartKind.image,
                    mediaType: 'image/png',
                    bytes: ByteData.sublistView(picture),
                  ),
                ]),
                _model([
                  AssistantPart(
                    kind: AssistantPartKind.providerBlock,
                    protocol: 'anthropic',
                    json: '{"type":"thinking","signature":"EqQA"}',
                  ),
                  AssistantPart(
                    kind: AssistantPartKind.text,
                    text: 'Ich sehe nach.',
                    extra: '{"phase":"commentary"}',
                  ),
                  AssistantPart(
                    kind: AssistantPartKind.toolCall,
                    callId: 'toolu_1',
                    name: 'find_cards',
                    json: '{"query":"Hund"}',
                  ),
                ]),
                _user([
                  AssistantPart(
                    kind: AssistantPartKind.toolResult,
                    callId: 'toolu_1',
                    text: 'Keine Karte',
                    isError: true,
                  ),
                ]),
              ],
              tools: [
                AssistantTool(
                  name: 'find_cards',
                  description: 'Finds cards.',
                  inputSchema: '{"type":"object"}',
                ),
              ],
            ),
          );

          final request = provider.sent.single.$2;
          final [first, second, third] = request.messages;
          expect(first.role, LlmRole.user);
          final [text, image] = first.parts;
          expect((text as LlmText).text, 'Was steht da?');
          expect((image as LlmImage).mediaType, 'image/png');
          expect(image.bytes, picture);
          expect(second.role, LlmRole.assistant);
          final [block, said, call] = second.parts;
          expect((block as LlmProviderBlock).protocol, LlmProtocol.anthropic);
          expect(block.block, {'type': 'thinking', 'signature': 'EqQA'});
          expect((said as LlmText).extra, {'phase': 'commentary'});
          expect((call as LlmToolCall).id, 'toolu_1');
          expect(call.input, {'query': 'Hund'});
          final result = third.parts.single as LlmToolResult;
          expect((result.callId, result.content), ('toolu_1', 'Keine Karte'));
          expect(result.isError, isTrue);
          expect(request.tools.single.name, 'find_cards');
          expect(request.tools.single.inputSchema, {'type': 'object'});

          final last = answers.single;
          expect(last.stop, AssistantStop.toolCalls);
          final [kept, called] = last.message!.parts;
          expect(kept.kind, AssistantPartKind.providerBlock);
          expect(kept.protocol, 'anthropic');
          expect(jsonDecode(kept.json!), {
            'type': 'thinking',
            'signature': 'EqQB',
          });
          expect(called.kind, AssistantPartKind.toolCall);
          expect((called.callId, called.name), ('toolu_2', 'find_cards'));
          expect(jsonDecode(called.json!), {'query': 'Katze'});
          expect(jsonDecode(called.extra!), {'x': 1});
        });

        test("holds the ceiling on output to the instance's", () async {
          await grantAndAgree();
          provider
            ..replyText('a')
            ..replyText('b');

          await askAsBen(_question(maxOutputTokens: 100000));
          await askAsBen(_question(maxOutputTokens: 0));

          expect(
            [for (final (_, request) in provider.sent) request.maxOutputTokens],
            [assistantOutputCeiling, 1],
          );
        });

        test('an answer cancelled is counted to its end all the same, so '
            "cancelling can't slip past a limit", () async {
          await grantAndAgree();
          final held = _HeldClient();
          useLlmClient(pod, held);

          final answers = endpoints.assistant.ask(
            asBen(),
            question: _question(),
          );
          final first = Completer<AssistantAnswer>();
          final subscription = answers.listen((answer) {
            if (!first.isCompleted) {
              first.complete(answer);
            }
          });
          held.add(const LlmTextDelta('„Hund“'));
          await first.future;
          await subscription.cancel();
          held
            ..add(const LlmTextDelta(' heißt 犬.'))
            ..finish(const LlmUsage(inputTokens: 800, outputTokens: 400));
          await assistantUsageCounted();

          expect(held.cancelled, isFalse);
          final day = (await figuresOfBen()).single;
          expect(
            (day.requests, day.inputTokens, day.outputTokens),
            (1, 800, 400),
          );
        });

        group(
          "a provider's failure is told by kind, what it counted counted",
          () {
            final failures = {
              LlmFailureKind.key: AssistantRefusal.providerKey,
              LlmFailureKind.rateLimited: AssistantRefusal.providerBusy,
              LlmFailureKind.unavailable: AssistantRefusal.providerBusy,
              LlmFailureKind.request: AssistantRefusal.providerRefused,
              LlmFailureKind.provider: AssistantRefusal.providerFailed,
              LlmFailureKind.network: AssistantRefusal.providerFailed,
              LlmFailureKind.malformed: AssistantRefusal.providerFailed,
            };
            for (final MapEntry(key: kind, value: refusal)
                in failures.entries) {
              test('${kind.name}: ${refusal.name}', () async {
                await grantAndAgree();
                provider.fail(
                  kind,
                  usage: const LlmUsage(inputTokens: 300, outputTokens: 0),
                );

                await expectLater(askAsBen(_question()), _refused(refusal));

                final day = (await figuresOfBen()).single;
                expect((day.requests, day.inputTokens), (1, 300));
              });
            }

            test("with the provider's wait, where it says", () async {
              await grantAndAgree();
              provider.reply(
                const [],
                failure: const LlmFailure(
                  kind: LlmFailureKind.rateLimited,
                  detail: 'slow down',
                  retryAfter: Duration(seconds: 30),
                ),
              );

              await expectLater(
                askAsBen(_question()),
                throwsA(
                  isA<AssistantRefusedException>().having(
                    (e) => e.retryAfterSeconds,
                    'retryAfterSeconds',
                    30,
                  ),
                ),
              );
            });
          },
        );
      });

      group('asking is refused, and nothing goes to the provider', () {
        tearDown(() => expect(provider.sent, isEmpty));

        test('to someone not signed in', () async {
          await expectLater(
            endpoints.assistant
                .ask(notSignedIn(), question: _question())
                .toList(),
            throwsA(isA<ServerpodUnauthenticatedException>()),
          );
        });

        test('without a grant, for an assistant not set, and for one set no '
            'more (D44)', () async {
          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.notGranted),
          );
          await expectLater(
            askAsBen(_question(assistant: 'gpt')),
            _refused(AssistantRefusal.notGranted),
          );
          await grantAndAgree();
          useAssistants(pod, _assistants({'ASSISTANTS': 'local'}));
          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.notGranted),
          );
        });

        test('until the person agreed, and again once the assistant moved to '
            'another provider (D40)', () async {
          await grantBen();
          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.notAgreed),
          );
          await endpoints.assistant.agree(asBen(), assistant: 'claude');
          useAssistants(
            pod,
            _assistants({
              'ASSISTANT_CLAUDE_BASE_URL': 'https://claude.proxy.example.org',
            }),
          );
          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.notAgreed),
          );
        });

        test(
          "past a day's limit, saying it resets at UTC's midnight (D42)",
          () async {
            await grantAndAgree(daily: 10000);
            await usedBefore(DateTime.utc(2026, 10, 25), costMicros: 10000);

            await expectLater(
              atNow(() => askAsBen(_question())),
              throwsA(
                isA<AssistantRefusedException>()
                    .having(
                      (e) => e.refusal,
                      'refusal',
                      AssistantRefusal.limitReached,
                    )
                    .having(
                      (e) => e.resetsAt,
                      'resetsAt',
                      DateTime.utc(2026, 10, 26),
                    ),
              ),
            );
          },
        );

        test("past a month's limit, saying it resets on the next month's "
            "first, which a day's limit doesn't change", () async {
          await grantAndAgree(daily: 10000, monthly: 25000);
          await usedBefore(DateTime.utc(2026, 10, 3), costMicros: 15000);
          await usedBefore(DateTime.utc(2026, 10, 25), costMicros: 10000);

          await expectLater(
            atNow(() => askAsBen(_question())),
            throwsA(
              isA<AssistantRefusedException>()
                  .having(
                    (e) => e.refusal,
                    'refusal',
                    AssistantRefusal.limitReached,
                  )
                  .having(
                    (e) => e.resetsAt,
                    'resetsAt',
                    DateTime.utc(2026, 11),
                  ),
            ),
          );
        });

        test('in tokens, where the assistant has no prices', () async {
          await grantAndAgree(assistant: 'local', daily: 1000);
          await usedBefore(
            DateTime.utc(2026, 10, 25),
            assistant: 'local',
            inputTokens: 900,
            outputTokens: 100,
          );

          await expectLater(
            atNow(() => askAsBen(_question(assistant: 'local'))),
            _refused(AssistantRefusal.limitReached),
          );
        });

        test('once its prices came or went, until the operator sets the '
            'limits again', () async {
          await grantAndAgree(daily: 10000);
          useAssistants(
            pod,
            _assistants({
              'ASSISTANT_CLAUDE_PRICE_INPUT': '',
              'ASSISTANT_CLAUDE_PRICE_OUTPUT': '',
            }),
          );

          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.limitOutdated),
          );
        });

        test('with an image, to an assistant that reads none (D12)', () async {
          await grantAndAgree(assistant: 'local');

          await expectLater(
            askAsBen(
              _question(
                assistant: 'local',
                messages: [
                  _user([
                    AssistantPart(
                      kind: AssistantPartKind.image,
                      mediaType: 'image/png',
                      bytes: ByteData(3),
                    ),
                  ]),
                ],
              ),
            ),
            _refused(AssistantRefusal.imagesNotRead),
          );
        });

        group('for a question that cannot be sent', () {
          final malformed = <String, AssistantQuestion>{
            'no turn': _question(messages: []),
            "the last turn the model's": _question(
              messages: [
                _user([_text('?')]),
                _model([_text('!')]),
              ],
            ),
            'a turn without parts': _question(messages: [_user([])]),
            "a tool call in the person's turn": _question(
              messages: [
                _user([
                  AssistantPart(
                    kind: AssistantPartKind.toolCall,
                    callId: 'c',
                    name: 'n',
                    json: '{}',
                  ),
                ]),
              ],
            ),
            "an image in the model's turn": _question(
              messages: [
                _user([_text('?')]),
                _model([
                  AssistantPart(
                    kind: AssistantPartKind.image,
                    mediaType: 'image/png',
                    bytes: ByteData(3),
                  ),
                ]),
                _user([_text('?')]),
              ],
            ),
            'a text without its text': _question(
              messages: [
                _user([AssistantPart(kind: AssistantPartKind.text)]),
              ],
            ),
            "a tool call's input that is no JSON object": _question(
              messages: [
                _user([_text('?')]),
                _model([
                  AssistantPart(
                    kind: AssistantPartKind.toolCall,
                    callId: 'c',
                    name: 'n',
                    json: '[1]',
                  ),
                ]),
                _user([
                  AssistantPart(
                    kind: AssistantPartKind.toolResult,
                    callId: 'c',
                    text: '',
                  ),
                ]),
              ],
            ),
            'a block of a protocol unknown': _question(
              messages: [
                _user([_text('?')]),
                _model([
                  AssistantPart(
                    kind: AssistantPartKind.providerBlock,
                    protocol: 'lumo',
                    json: '{}',
                  ),
                ]),
                _user([_text('?')]),
              ],
            ),
            'an image of a type the providers refuse': _question(
              messages: [
                _user([
                  AssistantPart(
                    kind: AssistantPartKind.image,
                    mediaType: 'image/svg+xml',
                    bytes: ByteData(3),
                  ),
                ]),
              ],
            ),
            'a tool named as no provider takes': _question(
              tools: [
                AssistantTool(
                  name: 'find cards',
                  description: '',
                  inputSchema: '{"type":"object"}',
                ),
              ],
            ),
            "a tool's schema that is no JSON object": _question(
              tools: [
                AssistantTool(
                  name: 'find_cards',
                  description: '',
                  inputSchema: 'object',
                ),
              ],
            ),
          };
          for (final MapEntry(key: name, value: question)
              in malformed.entries) {
            test(name, () async {
              await grantAndAgree();
              await expectLater(
                askAsBen(question),
                _refused(AssistantRefusal.malformed),
              );
            });
          }
        });

        test('for a question too large: its text, or an image', () async {
          await grantAndAgree();
          await expectLater(
            askAsBen(
              _question(
                messages: [
                  _user([_text('犬' * (assistantQuestionMaxCharacters + 1))]),
                ],
              ),
            ),
            _refused(AssistantRefusal.tooLarge),
          );
          await expectLater(
            askAsBen(
              _question(
                messages: [
                  _user([
                    AssistantPart(
                      kind: AssistantPartKind.image,
                      mediaType: 'image/jpeg',
                      bytes: ByteData(assistantImageMaxBytes + 1),
                    ),
                  ]),
                ],
              ),
            ),
            _refused(AssistantRefusal.tooLarge),
          );
        });

        test('on a member, which offers no assistant yet (D43)', () async {
          await grantAndAgree();
          useBase(
            sessionBuilder,
            app: 'todo',
            accounts: MemberOfRegister(Uri.parse('http://learn.test:8082')),
          );

          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.notGranted),
          );
          expect(
            (await endpoints.assistant.offered(asBen())).assistants,
            isEmpty,
          );
        });
      });

      test('a person has at most $maxQuestionsAtOnce questions on their way '
          'at once, so parallel ones never pass a limit together', () async {
        await grantAndAgree();
        final held = [
          for (var i = 0; i < maxQuestionsAtOnce; i++) _HeldClient(),
        ];
        final client = _Clients(held);
        useLlmClient(pod, client);

        final asked = [
          for (var i = 0; i < maxQuestionsAtOnce; i++) askAsBen(_question()),
        ];
        await client.allSent;
        await expectLater(
          askAsBen(_question()),
          _refused(AssistantRefusal.tooManyAtOnce),
        );
        for (final one in held) {
          one.finish(const LlmUsage(inputTokens: 1, outputTokens: 1));
        }
        await Future.wait(asked);
        expect(client.sent, maxQuestionsAtOnce);
      });

      group('what a person is offered (point 6, D27)', () {
        test("the assistants granted that are set, in the operator's order, "
            'with what goes where, the limits and what is left', () async {
          await grantBen(assistant: 'local', daily: 5000);
          await grantAndAgree(daily: 10000, monthly: 50000);
          await grantedBefore('gone');
          await usedBefore(DateTime.utc(2026, 10, 3), costMicros: 7000);
          await usedBefore(DateTime.utc(2026, 10, 25), costMicros: 2500);

          final offered = await atNow(
            () => endpoints.assistant.offered(asBen()),
          );

          expect(offered.currency, 'EUR');
          expect(offered.ownKeys, isFalse);
          final [claude, local] = offered.assistants;
          expect(claude.id, 'claude');
          expect(claude.name, 'Claude Sonnet');
          expect(claude.protocol, 'anthropic');
          expect(claude.host, 'api.anthropic.com');
          expect(claude.model, 'claude-sonnet-5-5');
          expect(claude.images, isTrue);
          expect(claude.agreedAt, isNotNull);
          expect(claude.limitUnit, AssistantLimitUnit.costMicros);
          expect((claude.dailyLimit, claude.monthlyLimit), (10000, 50000));
          expect((claude.usedToday, claude.usedThisMonth), (2500, 9500));
          expect(claude.outdated, isFalse);
          expect(local.id, 'local');
          expect(local.host, 'ollama');
          expect(local.agreedAt, isNull);
          expect(local.limitUnit, AssistantLimitUnit.tokens);
          expect((local.usedToday, local.usedThisMonth), (0, 0));
        });

        test('an agreement to another provider counts as none; the grant '
            'whose prices came or went says so', () async {
          await grantAndAgree(daily: 10000);
          useAssistants(
            pod,
            _assistants({
              'ASSISTANT_CLAUDE_BASE_URL': 'https://claude.proxy.example.org',
              'ASSISTANT_CLAUDE_PRICE_INPUT': '',
              'ASSISTANT_CLAUDE_PRICE_OUTPUT': '',
            }),
          );

          final claude = (await endpoints.assistant.offered(asBen()))
              .assistants
              .single;
          expect(claude.host, 'claude.proxy.example.org');
          expect(claude.agreedAt, isNull);
          expect(claude.outdated, isTrue);
        });

        test('a new model of the same provider keeps the agreement', () async {
          await grantAndAgree();
          useAssistants(
            pod,
            _assistants({'ASSISTANT_CLAUDE_MODEL': 'claude-opus-5-5'}),
          );
          provider.replyText('犬');

          final offered = await endpoints.assistant.offered(asBen());
          expect(offered.assistants.single.agreedAt, isNotNull);
          await askAsBen(_question());
        });

        test('whether personal keys are allowed (D5, D6)', () async {
          useAssistants(pod, _assistants({'ASSISTANT_OWN_KEYS': 'on'}));
          expect((await endpoints.assistant.offered(asBen())).ownKeys, isTrue);
          expect(
            (await endpoints.instance.basics(notSignedIn())).assistantOwnKeys,
            isTrue,
          );
        });

        test('agreeing is refused without a grant, or for an assistant not '
            'set', () async {
          await expectLater(
            endpoints.assistant.agree(asBen(), assistant: 'claude'),
            _refused(AssistantRefusal.notGranted),
          );
          await grantedBefore('gone');
          await expectLater(
            endpoints.assistant.agree(asBen(), assistant: 'gone'),
            _refused(AssistantRefusal.notGranted),
          );
        });

        test(
          "a person's own figures, by month, without anyone else's",
          () async {
            await grantAndAgree();
            await usedBefore(
              DateTime.utc(2026, 9, 12),
              inputTokens: 10,
              outputTokens: 2,
              costMicros: 60,
            );
            await usedBefore(
              DateTime.utc(2026, 10, 3),
              inputTokens: 100,
              outputTokens: 20,
              costMicros: 600,
            );
            await usedBefore(
              DateTime.utc(2026, 10, 25),
              inputTokens: 50,
              outputTokens: 10,
              costMicros: 300,
            );
            await AssistantFigures.db.insertRow(
              session,
              AssistantFigures(
                authUserId: olga,
                assistant: 'claude',
                day: DateTime.utc(2026, 10, 25),
                requests: 9,
                inputTokens: 9,
                outputTokens: 9,
                costMicros: 9,
              ),
            );

            final figures = await endpoints.assistant.figures(asBen());

            expect(
              [
                for (final month in figures)
                  (
                    month.month,
                    month.requests,
                    month.inputTokens,
                    month.outputTokens,
                    month.costMicros,
                  ),
              ],
              [
                (DateTime.utc(2026, 10), 2, 150, 30, 900),
                (DateTime.utc(2026, 9), 1, 10, 2, 60),
              ],
            );
            expect(figures.map((month) => month.account), everyElement(ben));
            expect(figures.map((month) => month.handle), everyElement(isNull));
          },
        );

        test('are refused to someone not signed in', () async {
          await expectLater(
            endpoints.assistant.offered(notSignedIn()),
            throwsA(isA<ServerpodUnauthenticatedException>()),
          );
          await expectLater(
            endpoints.assistant.agree(notSignedIn(), assistant: 'claude'),
            throwsA(isA<ServerpodUnauthenticatedException>()),
          );
          await expectLater(
            endpoints.assistant.figures(notSignedIn()),
            throwsA(isA<ServerpodUnauthenticatedException>()),
          );
        });
      });

      group("the operator's calls (points 6 and 8)", () {
        test(
          'are refused to anyone but an operator, signed in or not',
          () async {
            await grantBen();
            final calls =
                <String, Future<Object?> Function(TestSessionBuilder)>{
                  'setUp': (as) => endpoints.assistantOperator.setUp(as),
                  'grants': (as) => endpoints.assistantOperator.grants(as),
                  'grant': (as) => endpoints.assistantOperator.grant(
                    as,
                    account: ben,
                    assistant: 'local',
                    dailyLimit: null,
                    monthlyLimit: null,
                  ),
                  'revoke': (as) => endpoints.assistantOperator.revoke(
                    as,
                    account: ben,
                    assistant: 'claude',
                  ),
                  'figures': (as) => endpoints.assistantOperator.figures(
                    as,
                    month: DateTime.utc(2026, 10),
                  ),
                };

            for (final MapEntry(key: name, value: call) in calls.entries) {
              await expectLater(
                call(asBen()),
                throwsA(isA<ServerpodInsufficientAccessException>()),
                reason: name,
              );
              await expectLater(
                call(notSignedIn()),
                throwsA(isA<ServerpodUnauthenticatedException>()),
                reason: name,
              );
            }

            final grants = await AssistantGrant.db.find(session);
            expect([for (final grant in grants) grant.assistant], ['claude']);
          },
        );

        test('see the assistants set, without their keys', () async {
          final setUp = await endpoints.assistantOperator.setUp(asOlga());

          expect(setUp.currency, 'EUR');
          expect(setUp.ownKeys, isFalse);
          final [claude, local] = setUp.assistants;
          expect(
            (
              claude.id,
              claude.name,
              claude.protocol,
              claude.host,
              claude.model,
            ),
            (
              'claude',
              'Claude Sonnet',
              'anthropic',
              'api.anthropic.com',
              'claude-sonnet-5-5',
            ),
          );
          expect((claude.priceInput, claude.priceOutput), (3.0, 15.0));
          expect(claude.images, isTrue);
          expect((local.protocol, local.host), ('openai', 'ollama'));
          expect(local.priceInput, isNull);
          expect(jsonEncode(setUp.toJson()), isNot(contains('sk-ant-test')));
        });

        test("grant in the unit of the assistant's pricing; granted again, "
            'the limits change and the agreement stays', () async {
          await grantAndAgree(daily: 10000);
          await grantBen(assistant: 'local', monthly: 200000);
          await grantBen(daily: 20000, monthly: 90000);

          final grants = await endpoints.assistantOperator.grants(asOlga());

          final [claude, local] = grants;
          expect((claude.account, claude.handle), (ben, 'ben'));
          expect(claude.assistant, 'claude');
          expect(claude.limitUnit, AssistantLimitUnit.costMicros);
          expect((claude.dailyLimit, claude.monthlyLimit), (20000, 90000));
          expect(claude.agreedAt, isNotNull);
          expect(claude.assistantSet, isTrue);
          expect(claude.outdated, isFalse);
          expect(local.limitUnit, AssistantLimitUnit.tokens);
          expect((local.dailyLimit, local.monthlyLimit), (null, 200000));
          expect(local.agreedAt, isNull);
        });

        test('a grant of an assistant set no more is shown so, and an '
            'outdated one too (D44)', () async {
          await grantBen(daily: 1);
          await grantBen(assistant: 'local');
          useAssistants(
            pod,
            _assistants({
              'ASSISTANTS': 'claude',
              'ASSISTANT_CLAUDE_PRICE_INPUT': '',
              'ASSISTANT_CLAUDE_PRICE_OUTPUT': '',
            }),
          );

          final grants = await endpoints.assistantOperator.grants(asOlga());

          final [claude, local] = grants;
          expect((claude.assistantSet, claude.outdated), (true, true));
          expect(local.assistantSet, isFalse);
        });

        test('granting is refused for an assistant not set, a limit below 0, '
            'and an account unknown', () async {
          await expectLater(
            grantBen(assistant: 'gpt'),
            _refusedForOperator(OperatorRefusal.assistantNotSet),
          );
          await expectLater(
            grantBen(daily: -1),
            _refusedForOperator(OperatorRefusal.invalidLimit),
          );
          await expectLater(
            endpoints.assistantOperator.grant(
              asOlga(),
              account: UuidValue.fromString(
                '0193a3d4-5c6e-7f80-9a1b-2c3d4e5f6a7b',
              ),
              assistant: 'claude',
              dailyLimit: null,
              monthlyLimit: null,
            ),
            _refusedForOperator(OperatorRefusal.notFound),
          );
          expect(await AssistantGrant.db.count(session), 0);
        });

        test('revoke a grant, which then refuses; a grant there is not is '
            'refused', () async {
          await grantAndAgree();
          await endpoints.assistantOperator.revoke(
            asOlga(),
            account: ben,
            assistant: 'claude',
          );

          await expectLater(
            askAsBen(_question()),
            _refused(AssistantRefusal.notGranted),
          );
          await expectLater(
            endpoints.assistantOperator.revoke(
              asOlga(),
              account: ben,
              assistant: 'claude',
            ),
            _refusedForOperator(OperatorRefusal.notFound),
          );
        });

        test("see everyone's figures of a month, per person and assistant, "
            'never what was asked', () async {
          await usedBefore(DateTime.utc(2026, 10, 3), costMicros: 600);
          await usedBefore(DateTime.utc(2026, 10, 25), costMicros: 300);
          await usedBefore(DateTime.utc(2026, 9, 30), costMicros: 1);
          await usedBefore(
            DateTime.utc(2026, 10, 25),
            assistant: 'local',
            inputTokens: 70,
          );
          await AssistantFigures.db.insertRow(
            session,
            AssistantFigures(
              authUserId: olga,
              assistant: 'claude',
              day: DateTime.utc(2026, 10),
              requests: 4,
              inputTokens: 40,
              outputTokens: 4,
              costMicros: 180,
            ),
          );

          final figures = await endpoints.assistantOperator.figures(
            asOlga(),
            month: DateTime.utc(2026, 10, 17, 9),
          );

          expect(
            {
              for (final row in figures)
                (row.handle, row.assistant): (
                  row.month,
                  row.requests,
                  row.inputTokens,
                  row.costMicros,
                ),
            },
            {
              ('ben', 'claude'): (DateTime.utc(2026, 10), 2, 0, 900),
              ('ben', 'local'): (DateTime.utc(2026, 10), 1, 70, 0),
              ('Olga', 'claude'): (DateTime.utc(2026, 10), 4, 40, 180),
            },
          );
        });
      });

      group('the figures (D41)', () {
        test('are kept for the UTC month and the two before it, the older '
            'days forgotten daily', () async {
          await usedBefore(DateTime.utc(2026, 7, 31));
          await usedBefore(DateTime.utc(2026, 8));
          await usedBefore(DateTime.utc(2026, 10, 25));

          final forgotten = await forgetOldAssistantFigures(session, now: _now);

          expect(forgotten, 1);
          expect(
            [for (final day in await figuresOfBen()) day.day],
            [DateTime.utc(2026, 8), DateTime.utc(2026, 10, 25)],
          );
        });

        test('go with the person, as do their grants', () async {
          await grantAndAgree();
          await usedBefore(DateTime.utc(2026, 10, 25));

          await deletePerson(session, ben);

          expect(await AssistantGrant.db.count(session), 0);
          expect(await AssistantFigures.db.count(session), 0);
        });
      });
    },
  );
}

/// A provider whose one answer the test writes as it goes.
final class _HeldClient implements LlmClient {
  final _events = StreamController<LlmEvent>();
  final _text = StringBuffer();
  final sentOnce = Completer<void>();
  bool _finished = false;

  /// Whether the answer was cancelled before it ended.
  bool cancelled = false;

  void add(LlmTextDelta delta) {
    _text.write(delta.text);
    _events.add(delta);
  }

  void finish(LlmUsage usage) {
    _events.add(
      LlmFinished(
        message: LlmMessage.assistant([LlmText(_text.toString())]),
        stop: LlmStop.end,
        usage: usage,
      ),
    );
    _finished = true;
    unawaited(_events.close());
  }

  @override
  Stream<LlmEvent> send(LlmEndpoint endpoint, LlmRequest request) {
    // Called as the stream ends, too.
    _events.onCancel = () => cancelled = !_finished;
    sentOnce.complete();
    return _events.stream;
  }

  @override
  void close() {}
}

/// Providers answering one request each, in turn.
final class _Clients implements LlmClient {
  _Clients(this._clients);

  final List<_HeldClient> _clients;
  int sent = 0;

  Future<void> get allSent =>
      Future.wait([for (final client in _clients) client.sentOnce.future]);

  @override
  Stream<LlmEvent> send(LlmEndpoint endpoint, LlmRequest request) =>
      _clients[sent++].send(endpoint, request);

  @override
  void close() {}
}

Future<void> _deleteAll(Session session) async {
  await AssistantGrant.db.deleteWhere(session, where: (t) => t.id > 0);
  await AssistantFigures.db.deleteWhere(session, where: (t) => t.id > 0);
  await deleteAll(session);
}

extension on Map<String, String> {
  String? get(String name) => this[name];
}