createSession method

ToolCallingSessionHandle createSession(ToolCallingSessionCreateRequest request)

Create a tool-calling session that runs entirely in a dedicated worker isolate, so the inline llama.cpp generation loop (create + every step) never blocks the calling isolate. Events are copied SYNCHRONOUSLY inside a worker-owned NativeCallable.isolateLocal (commons' dispatcher hands out a stack-local buffer valid only for the call — see tool_calling_session.cpp dispatch_pending) and forwarded to this isolate over a port, where they are decoded onto the returned broadcast stream. Tool results are sent back to the worker via ToolCallingSessionHandle.stepWithResult; cancellation uses the thread-safe rac_tool_calling_session_cancel_proto directly.

Implementation

ToolCallingSessionHandle createSession(
  ToolCallingSessionCreateRequest request,
) {
  // Ownership transfers to ToolCallingSessionHandle, whose close() method
  // closes the controller after the worker port is quiesced.
  // ignore: close_sinks
  final controller = StreamController<ToolCallingSessionEvent>.broadcast();
  final fromWorker = ReceivePort();
  final idCompleter = Completer<int>();
  final handle = ToolCallingSessionHandle._(
    sessionId: idCompleter.future,
    events: controller,
    fromWorker: fromWorker,
  );

  fromWorker.listen((Object? message) {
    if (message is SendPort) {
      handle._toWorker = message;
    } else if (message is Uint8List) {
      if (controller.isClosed) return;
      try {
        controller.add(ToolCallingSessionEvent.fromBuffer(message));
      } catch (e, st) {
        controller.addError(e, st);
      }
    } else if (message is int) {
      _logger.debug('Tool calling session created: handle=$message');
      if (!idCompleter.isCompleted) idCompleter.complete(message);
    } else if (message is List && message.isNotEmpty && message[0] == 'err') {
      final err = StateError(
        message.length > 1 ? '${message[1]}' : 'tool-calling worker error',
      );
      if (!controller.isClosed) controller.addError(err);
      if (!idCompleter.isCompleted) idCompleter.completeError(err);
    }
  });

  final bytes = request.writeToBuffer();
  unawaited(() async {
    try {
      final iso = await Isolate.spawn(_toolSessionWorkerEntry, <dynamic>[
        fromWorker.sendPort,
        bytes,
      ]);
      handle._isolate = iso;
      if (handle._closed) iso.kill(priority: Isolate.immediate);
    } catch (e, st) {
      if (!controller.isClosed) controller.addError(e, st);
      if (!idCompleter.isCompleted) idCompleter.completeError(e, st);
    }
  }());

  return handle;
}