chain-client-sdk-dart/lib/src/hub_client.dart

// HubClient — typed Dart wrapper around the F∆I gRPC surface.
//
// Holds a single grpc.ClientChannel keyed by [HubEndpoint]. Each
// 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';
import 'package:grpc/grpc.dart' as grpc;
import 'package:grpc/grpc_connection_interface.dart';

import 'channel_factory_io.dart'
    if (dart.library.html) 'channel_factory_web.dart' as channel_factory;
import 'generated/chain/v1/common.pb.dart' as pb_common;
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 {
  /// Hostname or IP. Defaults to localhost for the canonical
  /// `fai serve` setup.
  final String host;

  /// gRPC port. Defaults to 50051.
  final int port;

  /// When true, use TLS. Phase 1+ will add credential support.
  final bool secure;

  const HubEndpoint({
    this.host = '127.0.0.1',
    this.port = 50051,
    this.secure = false,
  });

  @override
  String toString() => '${secure ? "https" : "http"}://$host:$port';
}

/// Outcome of a [HubClient.probe] health check. Distinguishes
/// "the hub is down" from "the hub is up but rejected our
/// credentials" — the two need different operator guidance.
enum HubProbeResult {
  /// Hub responded with SERVING.
  serving,

  /// Hub responded, but not with SERVING (starting up or
  /// shutting down).
  notServing,

  /// Hub is reachable but rejected the call as UNAUTHENTICATED
  /// or PERMISSION_DENIED — the endpoint is fine, the token
  /// is missing, wrong, or lacks scope.
  authRejected,

  /// Connection-level failure (refused, timeout, TLS, DNS).
  unreachable,
}

/// Re-exports of generated protobuf types so callers don't have
/// to import the `generated/` directory directly.
typedef CapabilityEntry = pb.CapabilityEntry;
typedef CapabilityList = pb.CapabilityList;
typedef LoggedEvent = pb.LoggedEvent;
typedef PendingApprovalEntry = pb.PendingApprovalEntry;
typedef DeclaredService = pb.DeclaredService;
typedef VerifyEventChainResponse = pb.VerifyEventChainResponse;
typedef CheckUpdateResponse = pb.CheckUpdateResponse;
typedef ChannelStatusResponse = pb.ChannelStatusResponse;
typedef ChannelEntry = pb.ChannelEntry;
typedef StoreSource = pb.StoreSource;
typedef ListStoresResponse = pb.ListStoresResponse;
typedef AddStoreResponse = pb.AddStoreResponse;
typedef RemoveStoreResponse = pb.RemoveStoreResponse;
typedef SystemAiStatusResponse = pb.SystemAiStatusResponse;
typedef AskAiResponse = pb.AskAiResponse;
typedef ListSystemAiModelsResponse = pb.ListSystemAiModelsResponse;
typedef SystemAiModel = pb.SystemAiModel;
typedef PullSystemAiModelResponse = pb.PullSystemAiModelResponse;
typedef HardwareInfoResponse = pb.HardwareInfoResponse;
typedef ListSystemAiCuratedModelsResponse =
    pb.ListSystemAiCuratedModelsResponse;
typedef CuratedModel = pb.CuratedModel;
typedef ClearEventLogResponse = pb.ClearEventLogResponse;
typedef DaemonPathsResponse = pb.DaemonPathsResponse;
typedef AuthStatusResponse = pb.AuthStatusResponse;
typedef PlanSetupResponse = pb.PlanSetupResponse;
typedef AuthTokenInfo = pb.AuthTokenInfo;
typedef JwtValidatorInfo = pb.JwtValidatorInfo;
typedef UninstallModuleResponse = pb.UninstallModuleResponse;
typedef FetchModuleDocsResponse = pb.FetchModuleDocsResponse;
typedef GetInstalledModuleDocsResponse = pb.GetInstalledModuleDocsResponse;
typedef InvokePluginThemeResponse = pb.InvokePluginThemeResponse;
typedef InvokePluginTranslateResponse = pb.InvokePluginTranslateResponse;
typedef ListMcpClientsResponse = pb.ListMcpClientsResponse;
typedef McpClientStatus = pb.McpClientStatus;
typedef McpClientConfigEntry = pb.McpClientConfigEntry;
typedef ListN8nEndpointsResponse = pb.ListN8nEndpointsResponse;
typedef N8nEndpointStatus = pb.N8nEndpointStatus;
typedef N8nEndpointConfigEntry = pb.N8nEndpointConfigEntry;
typedef ModuleInfoResponse = pb.ModuleInfoResponse;
typedef FlowSummary = pb.FlowSummary;
typedef ProjectInfo = pb.ProjectInfo;
typedef StoreEntry = pb.StoreEntry;
typedef StoreSearchResponse = pb.StoreSearchResponse;
typedef InstallModuleResponse = pb.InstallModuleResponse;
typedef InstallProgressUpdate = pb.InstallProgressUpdate;
typedef PullProgressUpdate = pb.PullProgressUpdate;
typedef Payload = pb_common.Payload;
typedef SubmitResponse = pb.SubmitResponse;
// Live-stream (T2) + detached-invocation (T3) wire types, re-exported
// so consumers of [HubClient.submitStreaming] don't import the
// generated package directly.
typedef SubmitStreamEvent = pb.SubmitStreamEvent;
typedef InvocationStatus = pb.InvocationStatus;
typedef InvocationEntry = pb.InvocationEntry;
typedef InvocationList = pb.InvocationList;

/// Typed client for the F∆I Hub gRPC surface. Construct once,
/// reuse for the process lifetime, call [close] on shutdown.
///
/// The channel type is picked at compile time:
///   * native targets → plain HTTP/2 gRPC via [ClientChannel].
///   * web targets    → HTTP/1.1 gRPC-Web via [GrpcWebClientChannel].
///                       Requires the hub to be running with the
///                       `tonic-web` layer enabled.
class HubClient {
  final HubEndpoint endpoint;
  final ClientChannelBase _channel;
  final grpc_hub.HubClient _hub;
  final grpc_hub.HubAdminClient _admin;

  HubClient._(this.endpoint, this._channel, grpc.CallOptions? defaultOptions)
      : _hub = grpc_hub.HubClient(_channel, options: defaultOptions),
        _admin = grpc_hub.HubAdminClient(_channel, options: defaultOptions);

  /// Construct a client.
  ///
  /// Supply [authToken] when the hub is configured with
  /// `auth.tokens:` in operator config — the value is sent as
  /// `authorization: Bearer <authToken>` metadata on every
  /// gRPC call. Without a token, calls are unauthenticated
  /// (the hub may reject them with `UNAUTHENTICATED` (16) when
  /// auth is required).
  factory HubClient({
    HubEndpoint endpoint = const HubEndpoint(),
    String? authToken,
  }) {
    final defaultOptions = (authToken == null || authToken.isEmpty)
        ? null
        : grpc.CallOptions(
            metadata: {'authorization': 'Bearer $authToken'},
          );
    return HubClient._(
      endpoint,
      channel_factory.createChannel(endpoint),
      defaultOptions,
    );
  }

  /// Liveness probe. Returns true when the hub responds with
  /// SERVING. False on any failure (includes connection refused).
  Future<bool> healthy() async {
    try {
      final r = await _hub.health(Empty());
      return r.state == pb.HealthStatus_State.SERVING;
    } catch (_) {
      return false;
    }
  }

  /// Like [healthy], but keeps enough of the failure to tell an
  /// auth rejection apart from a dead endpoint, so UIs can say
  /// "check your token" instead of a misleading "unreachable".
  Future<HubProbeResult> probe() async {
    try {
      final r = await _hub.health(Empty());
      return r.state == pb.HealthStatus_State.SERVING
          ? HubProbeResult.serving
          : HubProbeResult.notServing;
    } on grpc.GrpcError catch (e) {
      if (e.code == grpc.StatusCode.unauthenticated ||
          e.code == grpc.StatusCode.permissionDenied) {
        return HubProbeResult.authRejected;
      }
      return HubProbeResult.unreachable;
    } catch (_) {
      return HubProbeResult.unreachable;
    }
  }

  /// All capabilities provided by installed modules.
  Future<List<CapabilityEntry>> listCapabilities() async {
    final r = await _admin.listCapabilities(Empty());
    return r.capabilities;
  }

  /// Federation satellites currently connected to this hub
  /// (primary side). Empty when none are connected or this hub does
  /// not accept satellites.
  Future<List<pb.SatelliteEntry>> listSatellites() async {
    final r = await _admin.listSatellites(Empty());
    return r.satellites;
  }

  /// Issue a single-use federation bootstrap token enrolling
  /// `satelliteName`. The response also carries the primary's CA
  /// certificate so the operator can install token + CA on the
  /// satellite in one step (secure first connect).
  Future<pb.IssueBootstrapTokenResponse> issueBootstrapToken(
    String satelliteName, {
    String region = '',
    int ttlSeconds = 0,
  }) {
    final req = pb.IssueBootstrapTokenRequest()
      ..satelliteName = satelliteName
      ..region = region
      ..ttlSeconds = Int64(ttlSeconds);
    return _admin.issueBootstrapToken(req);
  }

  /// Detailed info for one installed module: name, version,
  /// provided capabilities, declared permissions, on-disk path.
  Future<ModuleInfoResponse> moduleInfo(String moduleName) {
    final req = pb.ModuleInfoRequest()..moduleName = moduleName;
    return _admin.moduleInfo(req);
  }

  /// Recent events from the audit log, newest-first. Pass an
  /// empty `types` list for all event types. A non-empty
  /// [project] scopes the view to that project's runs (the slug
  /// is normalized hub-side, so a display name works too).
  Future<List<LoggedEvent>> eventLog({
    int limit = 100,
    List<String> types = const [],
    String project = '',
  }) async {
    final req = pb.EventLogRequest()
      ..limit = limit
      ..eventTypes.addAll(types)
      ..project = project;
    final r = await _admin.eventLog(req);
    return r.events;
  }

  /// The shared hub's project registry (`general` first): the
  /// lightweight labels grouping flows, runs, approvals and audit
  /// events. Sealed areas never appear here — they are their own
  /// hub instance a client connects to directly.
  Future<List<ProjectInfo>> listProjects() async {
    final r = await _admin.listProjects(Empty());
    return r.projects;
  }

  /// Read the operator's `default_scope:` — the ordered list of
  /// provider segments the bare-form capability resolver
  /// consults when a flow names a capability without an explicit
  /// `<provider>/` prefix. Also returns a sorted list of catalog-
  /// known publisher segments so editors can offer a typeahead
  /// shortlist (arbitrary entries are still allowed).
  Future<({List<String> scope, List<String> knownProviders})>
      getDefaultScope() async {
    final r = await _admin.getDefaultScope(Empty());
    return (
      scope: List<String>.from(r.scope),
      knownProviders: List<String>.from(r.knownProviders),
    );
  }

  /// Replace the operator's `default_scope:`. The hub validates
  /// the list (non-empty after trimming), persists to
  /// `~/.chain/config.yaml`, and hot-swaps the in-memory copy
  /// without a daemon restart. Returns the snapshot.
  Future<({List<String> scope, List<String> knownProviders})>
      setDefaultScope(List<String> scope) async {
    final req = pb.SetDefaultScopeRequest()..scope.addAll(scope);
    final r = await _admin.setDefaultScope(req);
    return (
      scope: List<String>.from(r.scope),
      knownProviders: List<String>.from(r.knownProviders),
    );
  }

  /// Re-read `auth.tokens:` from operator config + env vars
  /// and atomically swap the hub's live token store. Used by
  /// operators to rotate tokens without restarting the daemon.
  /// Returns the new token count after the swap. Throws when
  /// the new config is invalid; the existing store stays in
  /// place in that case so a botched rotation can't lock the
  /// operator out.
  Future<int> reloadAuth() async {
    final r = await _admin.reloadAuth(Empty());
    return r.tokenCount;
  }

  /// Read-only snapshot of the hub's authentication policy as
  /// persisted in the operator config: active validator, whether
  /// anonymous calls are accepted, and the static token entries
  /// (names, scope grants, env-var name + set-flag — never the
  /// secret value). Admin-scoped like [reloadAuth].
  Future<AuthStatusResponse> authStatus() {
    return _admin.authStatus(Empty());
  }

  /// Assemble the guided-setup plan for [scenario] / [intent] /
  /// [target] (kebab-case wire values) — a pure computation on the
  /// hub, no state change. Field-for-field what `chain init
  /// --plan-json` prints, so GUIs can render the setup preview over
  /// the existing connection instead of spawning the CLI.
  Future<PlanSetupResponse> planSetup({
    required String scenario,
    required String intent,
    required String target,
    bool requireApproval = false,
    bool dataMustStayLocal = false,
    bool allowUnsignedModules = false,
  }) {
    return _admin.planSetup(
      pb.PlanSetupRequest(
        scenario: scenario,
        intent: intent,
        target: target,
        requireApproval: requireApproval,
        dataMustStayLocal: dataMustStayLocal,
        allowUnsignedModules: allowUnsignedModules,
      ),
    );
  }

  /// Server-streaming audit-log subscription. The hub first
  /// replays up to [backfill] historical events (oldest-last so
  /// the receiver sees them in chronological order), then keeps
  /// the stream open and forwards every newly-appended event as
  /// soon as it lands.
  ///
  /// Cancel the returned [Stream]'s subscription to close the
  /// RPC. If the server reports `RESOURCE_EXHAUSTED` ("stream
  /// lagged"), reconnect with backfill to resync.
  Stream<LoggedEvent> streamEvents({
    int backfill = 50,
    List<String> types = const [],
    String project = '',
  }) {
    final req = pb.StreamEventsRequest()
      ..backfill = backfill
      ..eventTypes.addAll(types)
      ..project = project;
    return _admin.streamEvents(req);
  }

  /// Pending approvals filtered by status and/or project.
  /// Defaults to all.
  Future<List<PendingApprovalEntry>> listApprovals({
    List<String> statuses = const [],
    int limit = 100,
    String project = '',
  }) async {
    final req = pb.ListApprovalsRequest()
      ..limit = limit
      ..statuses.addAll(statuses)
      ..project = project;
    final r = await _admin.listApprovals(req);
    return r.entries;
  }

  /// Approve a pending approval. The flow that posted it resumes
  /// on its next polling tick.
  Future<void> approve({
    required String approvalId,
    required String reviewer,
  }) async {
    final req = pb.DecideApprovalRequest()
      ..approvalId = approvalId
      ..reviewer = reviewer
      ..decision = pb.DecideApprovalRequest_Decision.APPROVE;
    await _admin.decideApproval(req);
  }

  /// Reject a pending approval. The flow fails with the supplied
  /// reason on its next polling tick.
  Future<void> reject({
    required String approvalId,
    required String reviewer,
    required String reason,
  }) async {
    final req = pb.DecideApprovalRequest()
      ..approvalId = approvalId
      ..reviewer = reviewer
      ..reason = reason
      ..decision = pb.DecideApprovalRequest_Decision.REJECT;
    await _admin.decideApproval(req);
  }

  /// Walk the event-log hash chain on the hub. Returns total /
  /// verified counts and an `tamperedAt` id (empty when clean).
  Future<VerifyEventChainResponse> verifyEventChain() {
    return _admin.verifyEventChain(Empty());
  }

  /// Wipe every audit event and seed a fresh `chain.reset`
  /// marker carrying [reviewer] + [reason]. The hub refuses on
  /// `beta` / `production` channels; on `local` / `dev` it
  /// returns the count of events that were purged.
  Future<ClearEventLogResponse> clearEventLog({
    required String reviewer,
    required String reason,
  }) {
    return _admin.clearEventLog(pb.ClearEventLogRequest(
      reviewer: reviewer,
      reason: reason,
    ));
  }

  /// Host services declared in the operator config. Each entry
  /// is a name + endpoint + tags; reachability is up to the
  /// caller (typically a follow-up HTTP probe).
  Future<List<DeclaredService>> listServices() async {
    final r = await _admin.listServices(Empty());
    return r.services;
  }

  /// Declare a host service in the operator config (`services:`).
  /// Returns whether a daemon restart is required for it to take
  /// effect (true today — services are read at start). Throws
  /// ALREADY_EXISTS when the name is taken.
  Future<bool> declareService({
    required String name,
    required String endpoint,
    String healthPath = '',
    List<String> tags = const [],
  }) async {
    final r = await _admin.declareService(pb.DeclareServiceRequest(
      name: name,
      endpoint: endpoint,
      healthPath: healthPath,
      tags: tags,
    ));
    return r.restartRequired;
  }

  /// Compare local hub version vs. release-channel manifest.
  /// Network failures land in `manifestReachable=false` rather
  /// than as exceptions — Studio renders that case as a quiet
  /// "release host unreachable" pill instead of an error.
  Future<CheckUpdateResponse> checkUpdate() {
    return _admin.checkUpdate(Empty());
  }

  /// Active channel name + per-channel daemon status. Studio's
  /// settings page calls this to render which channel is current
  /// and whether each daemon is up.
  Future<ChannelStatusResponse> channelStatus() {
    return _admin.channelStatus(Empty());
  }

  /// Filesystem paths the daemon owns (log file, db, config,
  /// flows, modules, pid). Studio's Doctor page surfaces these
  /// as "Open in file manager" buttons so Windows operators can
  /// inspect everything without a shell.
  Future<DaemonPathsResponse> daemonPaths() {
    return _admin.daemonPaths(Empty());
  }

  /// Remove an installed module by name. Wipes the module
  /// directory + re-scans the registry; writes a
  /// `module.uninstalled` audit event. Returned response carries
  /// the version that was removed.
  Future<UninstallModuleResponse> uninstallModule(
    String name, {
    String version = '',
  }) {
    return _admin.uninstallModule(
      pb.UninstallModuleRequest(name: name, version: version),
    );
  }

  /// Fetch a module's README markdown via the hub. The hub
  /// authenticates against the registry with FAI_REGISTRY_TOKEN
  /// when set, so Studio doesn't have to handle credentials.
  /// Errors come back inside [FetchModuleDocsResponse.errorKind]
  /// (not as exceptions) so the UI can render fallbacks.
  ///
  /// Deprecated: prefer [getInstalledModuleDocs], which reads
  /// `MODULE.md`/`MODULE.<locale>.md` shipped inside the bundle.
  @Deprecated('Use getInstalledModuleDocs (reads from bundle, no network)')
  Future<FetchModuleDocsResponse> fetchModuleDocs(
    String name, {
    String locale = '',
  }) {
    return _admin.fetchModuleDocs(
      pb.FetchModuleDocsRequest(name: name, locale: locale),
    );
  }

  /// Read the installed module's `MODULE.md` (or `MODULE.<locale>.md`)
  /// from disk under `~/.chain/modules/<module>/`. No network. Works in
  /// air-gapped installs. The response distinguishes three states via
  /// [GetInstalledModuleDocsResponse.notInstalled] and an empty
  /// [GetInstalledModuleDocsResponse.text]:
  ///
  /// - text non-empty → docs found, render them
  /// - text empty + notInstalled=false → installed but bundle had
  ///   no inline docs
  /// - text empty + notInstalled=true → module is not installed
  Future<GetInstalledModuleDocsResponse> getInstalledModuleDocs(
    String name, {
    String locale = '',
  }) {
    return _admin.getInstalledModuleDocs(
      pb.GetInstalledModuleDocsRequest(name: name, locale: locale),
    );
  }

  /// Invoke an installed Studio plugin's `theme` hook and
  /// receive a Material 3 ColorScheme for the brightness
  /// ("light" or "dark") the caller asked for.
  ///
  /// [capability] is the full Studio-plugin capability name
  /// (e.g. "studio.theme.solarized"). [brightness] must be
  /// "light" or "dark"; other values surface as
  /// `INVALID_ARGUMENT` from the hub. A plugin that doesn't
  /// recognise the brightness returns Declined, which the
  /// hub maps to `FAILED_PRECONDITION` — Studio's
  /// friendly-error mapper renders it with a recovery hint.
  ///
  /// The first end-to-end RPC of the Studio-plugin
  /// subsystem; `translate` and `output-view` hooks get
  /// their own methods once the matching Studio surfaces
  /// ship.
  Future<InvokePluginThemeResponse> invokePluginTheme({
    required String capability,
    required String brightness,
  }) {
    return _admin.invokePluginTheme(
      pb.InvokePluginThemeRequest(
        capability: capability,
        brightness: brightness,
      ),
    );
  }

  /// Invoke an installed Studio plugin's `translate` hook and
  /// receive translated text. [fromLocale] may be empty
  /// (auto-detect); [toLocale] must be a BCP-47 code (the
  /// host always knows the target).
  ///
  /// Errors surface as gRPC exceptions:
  ///   * NOT_FOUND       — plugin not installed.
  ///   * FAILED_PRECONDITION — plugin returned Declined or
  ///                     Misconfigured (e.g. unreachable
  ///                     LLM endpoint).
  ///   * INTERNAL        — generic catch-all.
  Future<InvokePluginTranslateResponse> invokePluginTranslate({
    required String capability,
    required String text,
    required String toLocale,
    String fromLocale = '',
  }) {
    return _admin.invokePluginTranslate(
      pb.InvokePluginTranslateRequest(
        capability: capability,
        text: text,
        fromLocale: fromLocale,
        toLocale: toLocale,
      ),
    );
  }

  /// List every configured MCP server with the last-discovery
  /// snapshot. Drives Studio's MCP-clients editor.
  Future<ListMcpClientsResponse> listMcpClients() {
    return _admin.listMcpClients(Empty());
  }

  /// Re-run discovery against every configured MCP server and
  /// update the running daemon's store-index in place.
  Future<ListMcpClientsResponse> refreshMcpClients() {
    return _admin.refreshMcpClients(Empty());
  }

  /// Persist a new MCP server in `~/.chain/config.yaml` and
  /// trigger discovery in one round-trip.
  Future<ListMcpClientsResponse> addMcpClient({
    required String name,
    required String endpoint,
    String apiKeyEnv = '',
    String description = '',
  }) {
    return _admin.addMcpClient(pb.McpClientConfigEntry(
      name: name,
      endpoint: endpoint,
      apiKeyEnv: apiKeyEnv,
      description: description,
    ));
  }

  /// Remove an MCP server by name + drop its synthetic
  /// capabilities from the running store-index.
  Future<ListMcpClientsResponse> removeMcpClient(String name) {
    return _admin.removeMcpClient(pb.RemoveMcpClientRequest(name: name));
  }

  /// Sister methods for n8n-endpoint federation. Same shape
  /// as the MCP-client surface above; Studio's n8n editor
  /// reuses the MCP editor patterns wholesale.
  Future<ListN8nEndpointsResponse> listN8nEndpoints() =>
      _admin.listN8nEndpoints(Empty());
  Future<ListN8nEndpointsResponse> refreshN8nEndpoints() =>
      _admin.refreshN8nEndpoints(Empty());
  Future<ListN8nEndpointsResponse> addN8nEndpoint({
    required String name,
    required String baseUrl,
    String apiKeyEnv = '',
    String description = '',
  }) {
    return _admin.addN8nEndpoint(pb.N8nEndpointConfigEntry(
      name: name,
      baseUrl: baseUrl,
      apiKeyEnv: apiKeyEnv,
      description: description,
    ));
  }
  Future<ListN8nEndpointsResponse> removeN8nEndpoint(String name) {
    return _admin.removeN8nEndpoint(pb.RemoveN8nEndpointRequest(name: name));
  }

  /// Read-only system-AI status. `enabled=false` lets Studio
  /// render "configure in Settings" without making a request.
  Future<SystemAiStatusResponse> systemAiStatus() {
    return _admin.systemAiStatus(Empty());
  }

  /// One-shot prompt to the configured System AI. Errors come
  /// back inside [AskAiResponse.errorKind] (not as exceptions)
  /// so Studio can map each category to its inline-fix copy.
  ///
  /// Cache-aware: identical prompts hit the persistent
  /// per-(model, privacy_mode, prompt) cache; the response
  /// carries `cached: true` plus the original generation
  /// latency. Pass [forceFresh] to skip the cache (Studio's
  /// Regenerate button).
  Future<AskAiResponse> askAi(String prompt, {bool forceFresh = false}) {
    return _admin.askAi(
      pb.AskAiRequest(prompt: prompt, forceFresh: forceFresh),
    );
  }

  /// Drop every cached System-AI explanation. Used by Settings
  /// when the operator wants a clean slate without changing
  /// model/privacy mode (which would also flush).
  Future<int> clearSystemLlmCache() async {
    final r = await _admin.clearSystemLlmCache(Empty());
    return r.purged.toInt();
  }

  /// Drop a single cached entry by prompt. Studio's Regenerate
  /// button calls this so the next askAi falls through to the
  /// live provider.
  Future<void> forgetCachedExplanation(String prompt) async {
    await _admin.forgetCachedExplanation(pb.AskAiRequest(prompt: prompt));
  }

  /// Persist a new System-AI configuration. Validates +
  /// writes back to ~/.chain/config.yaml + hot-reloads the live
  /// hub state in one call. Returns the resulting status so
  /// the UI can refresh in one round-trip.
  Future<SystemAiStatusResponse> updateSystemAi({
    required String provider,
    required String endpoint,
    required String model,
    required String apiKeyEnv,
    required String privacyMode,
  }) {
    return _admin.updateSystemAi(pb.UpdateSystemAiRequest(
      provider: provider,
      endpoint: endpoint,
      model: model,
      apiKeyEnv: apiKeyEnv,
      privacyMode: privacyMode,
    ));
  }

  /// Probe a (possibly draft) System-AI configuration. Pass the
  /// form values from the editor so "Test connection" works
  /// without a save first. Empty fields fall back to the live
  /// in-memory config.
  Future<AskAiResponse> testSystemAi({
    String provider = '',
    String endpoint = '',
    String model = '',
    String apiKeyEnv = '',
    String privacyMode = '',
  }) {
    return _admin.testSystemAi(pb.TestSystemAiRequest(
      provider: provider,
      endpoint: endpoint,
      model: model,
      apiKeyEnv: apiKeyEnv,
      privacyMode: privacyMode,
    ));
  }

  /// List the models the configured provider exposes via
  /// `GET /v1/models`. Same draft-vs-live semantics as
  /// [testSystemAi]. Errors come back inside
  /// [ListSystemAiModelsResponse.errorKind].
  Future<ListSystemAiModelsResponse> listSystemAiModels({
    String provider = '',
    String endpoint = '',
    String apiKeyEnv = '',
  }) {
    return _admin.listSystemAiModels(pb.TestSystemAiRequest(
      provider: provider,
      endpoint: endpoint,
      apiKeyEnv: apiKeyEnv,
    ));
  }

  /// Pull (download + install) an Ollama model via /api/pull.
  /// Synchronous; takes minutes for large models. Errors come
  /// back inside [PullSystemAiModelResponse.errorKind].
  Future<PullSystemAiModelResponse> pullSystemAiModel({
    required String endpoint,
    required String model,
    String apiKeyEnv = '',
  }) {
    return _admin.pullSystemAiModel(pb.PullSystemAiModelRequest(
      endpoint: endpoint,
      model: model,
      apiKeyEnv: apiKeyEnv,
    ));
  }

  /// Detected host hardware. Cached server-side; cheap to call.
  /// Returned tier is one of `tiny`/`small`/`balanced`/`large`/
  /// `unknown`. See `docs/architecture/system-ai.md` →
  /// "Hardware tiers".
  Future<HardwareInfoResponse> hardwareInfo() {
    return _admin.hardwareInfo(Empty());
  }

  /// Bundled curated model database. Used by Studio's editor
  /// to colour-code models by hardware suitability. Refresh
  /// requires a hub redeploy — `lastReviewed` shows how stale
  /// the curation is.
  Future<ListSystemAiCuratedModelsResponse>
      listSystemAiCuratedModels() {
    return _admin.listSystemAiCuratedModels(Empty());
  }

  /// Search the hub's bundled store index. All filters are
  /// optional; an empty query returns the first [limit] entries
  /// the index ships with.
  Future<List<StoreEntry>> searchStore({
    String query = '',
    String category = '',
    String tag = '',
    String status = '',
    String provider = '',
    String namespace = '',
    int limit = 50,
  }) async {
    final r = await _admin.searchStore(pb.SearchStoreRequest(
      query: query,
      category: category,
      tag: tag,
      status: status,
      provider: provider,
      namespace: namespace,
      limit: limit,
    ));
    return r.entries;
  }

  /// List the configured module stores (operator config `stores:`) plus
  /// the bundled seed, each with how many modules it contributes to the
  /// live index. Backs the Studio store manager.
  Future<List<StoreSource>> listStores() async {
    final r = await _admin.listStores(pb.ListStoresRequest());
    return r.stores;
  }

  /// Full store listing including the operator policy snapshot
  /// (`requireSignatures`, `trustedPublisherCount`) the hub reports
  /// alongside the sources since 0.23 — store UIs use it to state
  /// honestly whether installs are signature-checked at all.
  /// [listStores] stays as the plain source list.
  Future<ListStoresResponse> listStoresFull() =>
      _admin.listStores(pb.ListStoresRequest());

  /// Register a new module store. The hub fetches + validates its index,
  /// persists it to config, and merges it live so its modules are
  /// immediately searchable + installable.
  Future<AddStoreResponse> addStore({
    required String name,
    required String url,
    String bearerEnv = '',
    String basicUser = '',
    String basicPasswordEnv = '',
    String pinnedPubkeyPem = '',
  }) {
    return _admin.addStore(pb.AddStoreRequest(
      name: name,
      url: url,
      bearerEnv: bearerEnv,
      basicUser: basicUser,
      basicPasswordEnv: basicPasswordEnv,
      pinnedPubkeyPem: pinnedPubkeyPem,
    ));
  }

  /// Drop a configured store by name. The bundled seed cannot be removed.
  Future<RemoveStoreResponse> removeStore(String name) {
    return _admin.removeStore(pb.RemoveStoreRequest(name: name));
  }

  /// Install a module from a `.chain` bundle. [source] is either a
  /// URL or a local filesystem path; [expectedSha256] is an
  /// optional hex digest the hub verifies before unpacking.
  Future<InstallModuleResponse> installModule({
    required String source,
    String expectedSha256 = '',
  }) {
    return _admin.installModule(pb.InstallModuleRequest(
      source: source,
      expectedSha256: expectedSha256,
    ));
  }

  /// Install a module while watching it happen.
  ///
  /// Emits an update per phase and, during the download, as bytes
  /// arrive. The stream completes when the install succeeds, and
  /// carries the hub's failure text as a stream error when it does
  /// not, so callers handle failures the same way as with
  /// [installModule]. Use [onFinished] to receive the installed
  /// module's name and version.
  ///
  /// A caller that only needs the outcome should keep using the
  /// unary [installModule]; this exists so a waiting operator can
  /// see that a slow download is progressing rather than hung.
  Stream<InstallProgressUpdate> installModuleStreaming({
    required String source,
    String expectedSha256 = '',
    String version = '',
    void Function(InstallModuleResponse result)? onFinished,
  }) async* {
    final stream = _admin.installModuleStream(pb.InstallModuleRequest(
      source: source,
      expectedSha256: expectedSha256,
      version: version,
    ));
    await for (final event in stream) {
      if (event.hasUpdate()) {
        yield event.update;
      } else if (event.hasFinished()) {
        onFinished?.call(event.finished);
        return;
      } else if (event.hasFailed()) {
        throw Exception(event.failed);
      }
    }
  }

  /// Pull a model while watching it happen.
  ///
  /// A model pull moves gigabytes; the unary [pullSystemAiModel]
  /// leaves the caller waiting minutes with nothing to show. Emits the
  /// backend's own status plus byte counts per layer, and calls
  /// [onFinished] with the terminal response, whose `errorKind` is
  /// non-empty when the pull failed.
  Stream<PullProgressUpdate> pullSystemAiModelStreaming({
    required String endpoint,
    required String model,
    String apiKeyEnv = '',
    void Function(PullSystemAiModelResponse result)? onFinished,
  }) async* {
    final stream = _admin.pullSystemAiModelStream(pb.PullSystemAiModelRequest(
      endpoint: endpoint,
      model: model,
      apiKeyEnv: apiKeyEnv,
    ));
    await for (final event in stream) {
      if (event.hasUpdate()) {
        yield event.update;
      } else if (event.hasFinished()) {
        onFinished?.call(event.finished);
        return;
      }
    }
  }

  /// All saved flows known to the hub. Each entry carries the
  /// flow name, on-disk path and byte size.
  Future<List<FlowSummary>> listFlows() async {
    final r = await _admin.listFlows(Empty());
    return r.flows;
  }

  /// Return the parsed input schema of a saved flow so a GUI
  /// client can render an input form before calling
  /// [runSavedFlow]. Each entry's `type` is the verbatim type
  /// tag from the flow YAML (`text`, `bytes`, `json`, `file`,
  /// or anything else the flow author put there).
  Future<List<FlowInputDef>> getFlowDefinition(String name) async {
    final req = pb.GetFlowDefinitionRequest()..name = name;
    final resp = await _admin.getFlowDefinition(req);
    return resp.inputs
        .map((spec) => FlowInputDef(name: spec.name, type: spec.type))
        .toList();
  }

  /// Run a saved flow by name with the supplied named inputs.
  /// Two input shapes are supported in the same call so a
  /// flow that mixes text and binary inputs (e.g. extract
  /// taking a `document: bytes`) flows through one RPC:
  ///   * [textInputs] — string values wrapped as text Payloads
  ///   * [fileInputs] — raw bytes wrapped as bytes Payloads.
  ///                     MIME type defaults to
  ///                     `application/octet-stream`; clients that
  ///                     know the real type pass it via
  ///                     [fileMimeTypes] (same key). Modules
  ///                     typically gate on MIME — e.g.
  ///                     `text.extract` rejects octet-stream — so
  ///                     accurate types matter.
  /// All maps default to empty so existing text-only callers
  /// are backward-compatible. Keys must not collide between
  /// [textInputs] and [fileInputs]; on collision the bytes-shaped
  /// entry wins (the less-common case is more likely to be the
  /// operator's intent).
  Future<SubmitResponse> runSavedFlow({
    required String name,
    Map<String, String> textInputs = const {},
    Map<String, Uint8List> fileInputs = const {},
    Map<String, String> fileMimeTypes = const {},
    String project = '',
  }) async {
    // Project override for this run; empty keeps the saved flow's
    // own `project:`, then `general` (resolved hub-side).
    final req = pb.RunSavedFlowRequest()
      ..name = name
      ..project = project;
    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';
      final bytes = pb_common.Bytes()
        ..mimeType = mime
        ..data = entry.value;
      req.inputs[entry.key] = pb_common.Payload()..bytes = bytes;
    }
    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<SubmitResponse> submit({
    required String flowYaml,
    Map<String, String> textInputs = const {},
    Map<String, Uint8List> fileInputs = const {},
    Map<String, Object?> jsonInputs = const {},
    Map<String, String> fileMimeTypes = const {},
    bool detach = false,
  }) async {
    return _hub.submit(_buildSubmitRequest(
      flowYaml: flowYaml,
      textInputs: textInputs,
      fileInputs: fileInputs,
      jsonInputs: jsonInputs,
      fileMimeTypes: fileMimeTypes,
      detach: detach,
    ));
  }

  /// Submit a flow and follow its execution live (server streaming).
  ///
  /// Yields [pb.SubmitStreamEvent]s in order: `stepStarted` /
  /// `stepFinished` markers, `moduleEvent`s (a module's ephemeral
  /// `host.emit-event`s), and finally either `finalResult` (== the
  /// unary [submit] response) or `error`. Works over native gRPC and,
  /// on web targets, gRPC-Web. Same inputs as [submit].
  ///
  /// ```dart
  /// await for (final ev in hub.submitStreaming(flowYaml: yaml)) {
  ///   if (ev.hasModuleEvent()) print('${ev.moduleEvent.name}');
  ///   if (ev.hasFinalResult()) print(ev.finalResult.outputs);
  /// }
  /// ```
  Stream<SubmitStreamEvent> submitStreaming({
    required String flowYaml,
    Map<String, String> textInputs = const {},
    Map<String, Uint8List> fileInputs = const {},
    Map<String, Object?> jsonInputs = const {},
    Map<String, String> fileMimeTypes = const {},
  }) {
    return _hub.submitStream(_buildSubmitRequest(
      flowYaml: flowYaml,
      textInputs: textInputs,
      fileInputs: fileInputs,
      jsonInputs: jsonInputs,
      fileMimeTypes: fileMimeTypes,
    ));
  }

  /// Status of a detached invocation (T3). Throws a [GrpcError] with
  /// code `notFound` for an unknown id.
  Future<InvocationStatus> getInvocationStatus(String invocationId) {
    return _hub.getInvocationStatus(pb_common.InvocationId()..value = invocationId);
  }

  /// Retained result of a succeeded detached invocation (T3). Throws
  /// `failedPrecondition` if it is still running / failed, `notFound`
  /// if unknown or already evicted.
  Future<SubmitResponse> getInvocationResult(String invocationId) {
    return _hub.getInvocationResult(pb_common.InvocationId()..value = invocationId);
  }

  /// Cancel a running detached invocation (T3). Returns true if it was
  /// running and got signalled, false if already finished or unknown.
  Future<bool> cancelInvocation(String invocationId) async {
    final r =
        await _hub.cancelInvocation(pb_common.InvocationId()..value = invocationId);
    return r.cancelled;
  }

  /// Every tracked detached invocation (newest-first) for a monitor
  /// surface — Studio's detached-runs list. Empty when detached
  /// invocations are disabled or none have run this process.
  Future<List<InvocationEntry>> listInvocations() async {
    final r = await _hub.listInvocations(Empty());
    return r.invocations;
  }

  /// The full detached-runs monitor payload: tracked invocations
  /// plus [InvocationList.detachedEnabled], so a UI can distinguish
  /// "feature off" from "on, but no runs yet".
  Future<InvocationList> listInvocationsFull() =>
      _hub.listInvocations(Empty());

  pb.SubmitRequest _buildSubmitRequest({
    required String flowYaml,
    required Map<String, String> textInputs,
    required Map<String, Uint8List> fileInputs,
    required Map<String, Object?> jsonInputs,
    required Map<String, String> fileMimeTypes,
    bool detach = false,
  }) {
    final req = pb.SubmitRequest()
      ..flowYaml = utf8.encode(flowYaml)
      ..detach = detach;
    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 req;
  }

  /// Closes the gRPC channel. Idempotent.
  Future<void> 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<dynamic, dynamic> 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
/// the flow YAML's verbatim type tag — opaque to the SDK so
/// the type system can grow without an SDK release.
class FlowInputDef {
  /// Input key (the name the flow YAML uses).
  final String name;
  /// Type tag — typically `text`, `bytes`, `json`, or `file`.
  final String type;

  const FlowInputDef({required this.name, required this.type});
}