generateStreamProto method

Stream<LLMStreamEvent> generateStreamProto(LLMGenerateRequest request)

Stream text generation using the lifecycle-owned generated-proto LLM ABI, with TRUE incremental token delivery (FLUTTER-IOS-006 resolved).

Threading model: the blocking rac_llm_generate_stream_proto call runs inside a short-lived worker isolate (Isolate.run). On iOS and Android, the Flutter plugin exports a native-port helper that copies proto bytes inside the C callback and posts owned Uint8List messages to the main isolate. That path is safe for MLX/Swift async and for native backends that invoke stream callbacks from worker threads.

Cancellation: onCancel → rac_llm_cancel_proto sets the lifecycle cancel flag checked per token; the engine aborts, the worker's blocking call returns, and the trailing rc sentinel tears down the port. (With the old main-isolate placement, cancel could not even run until the generation finished — the isolate was blocked inside the FFI call.)

No isLoaded gate (see generateProto): generation resolves via the commons model lifecycle, not this bridge's _handle.

Implementation

Stream<LLMStreamEvent> generateStreamProto(LLMGenerateRequest request) {
  if (RacNative.bindings.rac_llm_generate_stream_proto == null) {
    return Stream<LLMStreamEvent>.error(
      UnsupportedError('rac_llm_generate_stream_proto is unavailable'),
    );
  }
  if (RacNative.bindings.ra_flutter_llm_generate_stream_proto_native_port ==
      null) {
    return Stream<LLMStreamEvent>.error(
      UnsupportedError(
        'ra_flutter_llm_generate_stream_proto_native_port is unavailable',
      ),
    );
  }

  final controller = StreamController<LLMStreamEvent>(sync: false);
  final receivePort = ReceivePort();
  var sawTerminalEvent = false;
  var tornDown = false;

  void teardown() {
    if (tornDown) return;
    tornDown = true;
    receivePort.close();
  }

  receivePort.listen((Object? message) {
    if (message is Uint8List) {
      // One serialized LLMStreamEvent, already copied in the worker's
      // synchronous callback, delivered over the port in emission order.
      if (controller.isClosed) return;
      try {
        final event = LLMStreamEvent.fromBuffer(message);
        sawTerminalEvent = sawTerminalEvent || event.isFinal;
        controller.add(event);
        if (event.isFinal) {
          unawaited(controller.close());
        }
      } catch (e, st) {
        controller.addError(e, st);
        unawaited(controller.close());
      }
    } else if (message is int) {
      // rc sentinel — always the LAST message (same port as the tokens, so
      // FIFO ordering is guaranteed; the Isolate.run future has no such
      // ordering relative to port messages). Early-return rcs (parse /
      // no-model errors) produce no terminal event, so surface them.
      if (message != RAC_SUCCESS &&
          !sawTerminalEvent &&
          !controller.isClosed) {
        controller.addError(
          StateError(
            'rac_llm_generate_stream_proto failed: '
            '${RacResultCode.getMessage(message)}',
          ),
        );
      }
      if (!controller.isClosed) {
        unawaited(controller.close());
      }
      teardown();
    }
  });

  final requestBytes = request.writeToBuffer();
  final worker = _runLlmStreamNativePortWorker(
    requestBytes,
    receivePort.sendPort.nativePort,
  );

  unawaited(
    worker.catchError((Object e, StackTrace st) {
      // Worker isolate crashed (RemoteError) before the rc sentinel.
      if (!controller.isClosed) {
        controller.addError(e, st);
        unawaited(controller.close());
      }
      teardown();
      return RAC_SUCCESS;
    }),
  );

  // Cancel sets the per-token lifecycle cancel flag; the worker's blocking
  // call returns shortly after, emits a terminal "cancelled" event
  // (dropped — the controller is closing) and the rc sentinel closes the
  // port.
  controller.onCancel = cancelProto;

  return controller.stream;
}