diff --git a/lib/src/hub_client.dart b/lib/src/hub_client.dart index 52781ce..3760d3c 100644 --- a/lib/src/hub_client.dart +++ b/lib/src/hub_client.dart @@ -4,6 +4,7 @@ // public method maps onto exactly one RPC on Hub or HubAdmin, so // callers don't have to know which service hosts which call. +import 'dart:convert'; import 'dart:typed_data'; import 'package:fixnum/fixnum.dart'; @@ -17,6 +18,8 @@ import 'generated/chain/v1/hub.pb.dart' as pb; import 'generated/chain/v1/hub.pbgrpc.dart' as grpc_hub; import 'package:protobuf/well_known_types/google/protobuf/empty.pb.dart' show Empty; +import 'package:protobuf/well_known_types/google/protobuf/struct.pb.dart' + show Struct, Value, ListValue, NullValue; /// Connection coordinates for a hub endpoint. class HubEndpoint { @@ -752,12 +755,87 @@ class HubClient { return _admin.runSavedFlow(req); } + /// Submit a flow YAML to the hub for a one-shot run, with typed + /// inputs, and read its outputs from the response. + /// + /// Unlike [runSavedFlow] (which runs a flow already saved in the + /// hub by name), this sends the flow definition inline and exposes + /// the full [Submit] RPC. Two things only this path can do: + /// * pass **JSON-typed inputs** ([jsonInputs]) — for modules that + /// read `inputs.require_json(...)` (e.g. `econ.skm_score`, + /// `law.benefit_score`). Each value must be a JSON object (a + /// Dart `Map`); it is converted to a `google.protobuf.Struct`. + /// * read the flow's **outputs** off the response. The audit log + /// does not persist outputs — they come back only here. + /// + /// [textInputs] → text Payloads, [fileInputs] → bytes Payloads + /// (MIME from [fileMimeTypes], default `application/octet-stream`). + /// All maps default to empty. + Future submit({ + required String flowYaml, + Map textInputs = const {}, + Map fileInputs = const {}, + Map jsonInputs = const {}, + Map fileMimeTypes = const {}, + }) async { + final req = pb.SubmitRequest()..flowYaml = utf8.encode(flowYaml); + for (final entry in textInputs.entries) { + req.inputs[entry.key] = pb_common.Payload()..text = entry.value; + } + for (final entry in fileInputs.entries) { + final mime = fileMimeTypes[entry.key] ?? 'application/octet-stream'; + req.inputs[entry.key] = pb_common.Payload() + ..bytes = (pb_common.Bytes() + ..mimeType = mime + ..data = entry.value); + } + for (final entry in jsonInputs.entries) { + final value = entry.value; + if (value is! Map) { + throw ArgumentError( + 'jsonInputs["${entry.key}"] must be a JSON object (Map); ' + 'got ${value.runtimeType}', + ); + } + req.inputs[entry.key] = pb_common.Payload()..json = _structFromMap(value); + } + return _hub.submit(req); + } + /// Closes the gRPC channel. Idempotent. Future close() async { await _channel.shutdown(); } } +/// Convert a Dart JSON object into a `google.protobuf.Struct`. +/// Mirrors the proto3-JSON mapping the hub expects for a +/// `Payload.json` input. Keys are stringified; values recurse via +/// [_structValue]. +Struct _structFromMap(Map map) { + final s = Struct(); + map.forEach((k, v) => s.fields[k.toString()] = _structValue(v)); + return s; +} + +/// Convert a single Dart JSON value into a `google.protobuf.Value`. +Value _structValue(Object? v) { + if (v == null) return Value()..nullValue = NullValue.NULL_VALUE; + if (v is bool) return Value()..boolValue = v; + if (v is num) return Value()..numberValue = v.toDouble(); + if (v is String) return Value()..stringValue = v; + if (v is Map) return Value()..structValue = _structFromMap(v); + if (v is Iterable) { + final list = ListValue(); + for (final e in v) { + list.values.add(_structValue(e)); + } + return Value()..listValue = list; + } + // Unknown type — stringify so one stray value can't fail the run. + return Value()..stringValue = v.toString(); +} + /// One declared input in a saved flow's `inputs:` section. /// Returned by [HubClient.getFlowDefinition] so GUI clients /// can build a typed run-flow form. The [type] field carries