Migrate TrueNAS to supported JSON-RPC transport

2026-09-10 02:25:56 +00:00 · 2026-07-24 00:07:19 +01:00
parent 0c17922055
commit b81ba7dd06
19 changed files with 2145 additions and 371 deletions
@@ -615,8 +615,13 @@ claim, broad same-shape TrueNAS iteration is not default work above this
 declared floor.

 1. Architecture boundary:
-   TrueNAS is API-first. The unified agent may augment a TrueNAS system later,
-   but it is not required for bootstrap or baseline support. TrueNAS must
+   TrueNAS is API-first. SCALE 25.04 and later uses the supported versioned
+   JSON-RPC 2.0 WebSocket API at `/api/current`; REST is a connection-local,
+   version-gated compatibility boundary only for SCALE releases before 25.04
+   and TrueNAS CORE/FreeNAS. Authentication, permission, TLS, protocol, and
+   method errors on the current endpoint fail closed and never downgrade to
+   REST. The unified agent may augment a TrueNAS system later, but it is not
+   required for bootstrap or baseline support. TrueNAS must
   project into the canonical `agent`, `vm`, `app-container`, `storage`,
   `physical-disk`, and recovery contracts. TrueNAS app rows may carry a
   TrueNAS-native app facet sourced from `app.query` / `active_workloads`, and
@@ -627,7 +632,10 @@ declared floor.
   Supported now through the shared platform-connections flow and
   `/api/truenas/connections`. Operators can add, test, edit, retest, and
   delete stored TrueNAS connections without re-entering masked secrets on
-   ordinary edits. `truenas_disabled` is only an explicit server opt-out, not
+   ordinary edits. Current API-key setup includes the key owner's username for
+   the supported `auth.login_ex` flow; username-less stored keys remain an
+   upgrade bridge only while the appliance still exposes its deprecated login
+   method. `truenas_disabled` is only an explicit server opt-out, not
   the baseline state. Out of scope for this floor: a separate TrueNAS-first
   wizard, agent-required bootstrap, or disabled-by-default launch posture.
 3. Infrastructure visibility:
@@ -5359,7 +5359,13 @@ workspace can render real API-backed status and handoff context without
 inventing a settings-local shadow fetch path. `frontend-modern/src/api/truenas.ts`
 owns the browser normalizer for that observed summary and must preserve those
 native inventory facets instead of collapsing them into a generic app or
-storage count. Zero-value legacy
+storage count. The same redacted list payload carries an optional secret-free
+`transport` object with `mode`, `endpoint`, `tls`, `connected`,
+`authMechanism`, `applianceVersion`, `legacyReason`, `reconnects`,
+`lastError`, and `lastConnectedAt`. Settings must identify current
+`jsonrpc-websocket`, version-gated `legacy-rest`, and not-yet-negotiated states
+without exposing a credential or rebuilding transport state in the browser.
+Zero-value legacy
 `pollIntervalSeconds` config must normalize back to the canonical 60-second
 default at this same boundary instead of leaking ambiguous `0` values to the
 frontend.
@@ -222,6 +222,7 @@ node-local Agent evidence.
 45. `pkg/agents/kubernetes/report.go`
 46. `internal/monitoring/temperature.go`
 47. `internal/truenas/client.go`
+47a. `internal/truenas/transport.go`
 48. `internal/truenas/disk_health.go`
 49. `internal/truenas/provider.go`
 50. `internal/models/ceph_cluster_identity.go`
@@ -434,8 +435,9 @@ node-local Agent evidence.
    hard-coded cluster name.
 16. Add or change TrueNAS supplemental inventory only through the native
    TrueNAS provider path and unified-resource projection. TrueNAS apps are
-    API-owned application records: `app.query` is the preferred live inventory
-    source, with legacy REST allowed only as a compatibility fallback. The
+    API-owned application records: `app.query` is the live inventory source on
+    the negotiated current transport, with legacy REST allowed only for a
+    connection proven to run a release that predates the versioned API. The
    provider may preserve Docker-compatible runtime metadata for shared
    container tooling, but it must also publish the native app identity, state,
    version, update availability, workload containers, ports, images, volumes,
@@ -463,10 +465,31 @@ node-local Agent evidence.
    `internal/truenas/contract_test.go`. The canonical-ID migration semantics
    for rows minted under the retired hostname-keyed derivation live in the
    unified-resources contract (record-declared succession, item 27).
-    TrueNAS storage and alert inventory follow native query methods first:
+    Every live TrueNAS client owns one explicit, immutable transport decision.
+    SCALE 25.04 and later must use JSON-RPC 2.0 over a TLS WebSocket at
+    `/api/current`; authentication, authorization, TLS, protocol, or method
+    failures on that endpoint are authoritative and must never wake the
+    deprecated REST bridge. Only an unsupported-endpoint WebSocket handshake
+    may trigger a REST `/system/info` version probe, and REST may then be
+    selected only for recognized SCALE releases before 25.04 or TrueNAS
+    CORE/FreeNAS. Unknown or current versions fail closed. The decision and
+    persistent socket belong to one configured client, so reconnects or
+    legacy negotiation for one appliance cannot alter another appliance.
+    Current API-key authentication uses `auth.login_ex` with the key owner's
+    username and `API_KEY_PLAIN`; password authentication uses
+    `PASSWORD_PLAIN`. Username-less stored API keys may use the deprecated
+    login method only as an upgrade bridge, with explicit remediation when a
+    release removes that method. Read calls may reconnect with bounded backoff
+    and replay once after a transport failure. Mutating app calls must never
+    replay after dispatch because their outcome is ambiguous. Event reads must
+    retain the `core.subscribe` ID, call `core.unsubscribe` before reusing the
+    session, and discard a socket after a terminal stream read deadline.
+    Connection summaries expose only secret-free transport mode, endpoint,
+    TLS, authentication mechanism, appliance version, reconnect count, and
+    last-error timing diagnostics.
+    TrueNAS storage and alert inventory follow the negotiated transport:
    pools use `pool.query`, datasets use `pool.dataset.query`, disks use
-    `disk.query`, and alerts use `alert.list`, with legacy REST allowed only
-    as compatibility fallback. Inventory readers must only consume fields the
+    `disk.query`, and alerts use `alert.list`. Inventory readers must only consume fields the
    API actually serves on every supported TrueNAS line (CORE 13 REST-only
    included): `pool.dataset.query` carries no `mounted` field, so a listed
    dataset counts as mounted unless `locked` or an explicit value says
@@ -474,8 +497,9 @@ node-local Agent evidence.
    `extra.pools` join cannot cross the REST bridge, so per-disk pool
    membership and ZFS member state derive from the vdev topology that
    `pool.query` attaches unconditionally; and `disk.temperatures` is a
-    parameterized method the REST bridge only serves as POST, tried after
-    native JSON-RPC reporting. Missing disk telemetry is reported as
+    parameterized method the legacy REST bridge only serves as POST, while
+    current releases use native JSON-RPC reporting without per-method REST
+    fallback. Missing disk telemetry is reported as
    unknown, never as a failure signal. Regression coverage:
    `internal/truenas/client_api_shapes_test.go`. Unhealthy pool state
    from `pool.query` must emit a provider-native `zfs_pool_state` incident on
@@ -501,9 +525,11 @@ node-local Agent evidence.
    `sharing.nfs.query` publishes native `TrueNASData.Share` on canonical
    `network-share` resources parented to the owning dataset or pool when the
    API/path supplies that evidence. TrueNAS protection inventory follows the
-    same native-query rule: ZFS snapshots prefer `zfs.resource.snapshot.query`
-    with older `pool.snapshot.query`/REST compatibility fallback, and
-    replication tasks prefer `replication.query`.
+    same native-query rule: current connections prefer
+    `zfs.resource.snapshot.query`, with `pool.snapshot.query` allowed only as a
+    same-transport method-name compatibility path; version-gated legacy
+    connections use REST. Replication tasks use `replication.query` on current
+    connections.
    TrueNAS system services are also native appliance inventory: `service.query`
    is the preferred source for service name, boot enablement, runtime state,
    and process IDs. Pulse must publish that data through `TrueNASData.Services`
@@ -1682,9 +1708,9 @@ machine ID, capacity, and poll-health path instead of failing connection tests
 or background refreshes during JSON decoding.
 That same monitoring boundary now also owns live TrueNAS disk temperatures.
 `internal/truenas/client.go` and `internal/truenas/provider.go` must ingest
-`disk.temperatures` from the TrueNAS API, fall back to modern
-`reporting.get_data` `disktemp` when the dedicated endpoint is unavailable, and
-project those readings into the canonical physical-disk model and risk path
+legacy `disk.temperatures` from the REST API or `reporting.get_data` `disktemp`
+from the current JSON-RPC transport, and project those readings into the
+canonical physical-disk model and risk path
 instead of leaving temperature telemetry agent-only or adding a TrueNAS-local
 presentation shim.
 That same monitoring boundary also owns SMART-backed TrueNAS disk risk
@@ -1701,10 +1727,11 @@ null, empty, missing, unknown, or unavailable SMART telemetry to canonical
 native failure states such as `FAULTED`, `FAILED`, `OFFLINE`, `REMOVED`, and
 `UNAVAIL` must continue to produce canonical disk-health risk.
 That same boundary owns boot-pool and replication-target storage posture.
-`internal/truenas/client.go` must collect `boot.get_state` with the legacy REST
-bridge as compatibility fallback, merge it only within the current configured
-connection, and use its vdev leaves to enrich boot-disk pool membership and
-native ZFS state. `internal/monitoring/truenas_poller.go` must correlate
+`internal/truenas/client.go` must collect `boot.get_state` through the
+connection's negotiated transport, use REST only on a version-gated legacy
+connection, merge it only within the current configured connection, and use
+its vdev leaves to enrich boot-disk pool membership and native ZFS state.
+`internal/monitoring/truenas_poller.go` must correlate
 `replication.query` intent across providers within the same organization using
 local/PULL ownership or a unique configured/observed target-host match for
 remote PUSH tasks. The resulting `SET`/`REQUIRE` receive-side read-only posture
@@ -5255,6 +5255,7 @@
              "internal/truenas/disk_health.go",
              "internal/truenas/fixtures.go",
              "internal/truenas/provider.go",
+              "internal/truenas/transport.go",
              "internal/truenas/types.go"
            ],
            "allow_same_subsystem_tests": false,
@@ -5264,7 +5265,8 @@
              "internal/monitoring/monitor_polling_test.go",
              "internal/truenas/client_test.go",
              "internal/truenas/contract_test.go",
-              "internal/truenas/provider_test.go"
+              "internal/truenas/provider_test.go",
+              "internal/truenas/transport_test.go"
            ]
          },
          {
@@ -19,6 +19,7 @@ describe('TrueNASAPI', () => {
        host: ' truenas.local ',
        port: 443,
        apiKey: ' ******** ',
+        username: ' pulse-readonly ',
        useHttps: true,
        insecureSkipVerify: false,
        enabled: true,
@@ -40,6 +41,15 @@ describe('TrueNASAPI', () => {
          disks: 8,
          recoveryArtifacts: 18,
        },
+        transport: {
+          mode: 'jsonrpc-websocket',
+          endpoint: ' wss://truenas.local/api/current ',
+          tls: true,
+          connected: true,
+          authMechanism: ' api-key-plain ',
+          applianceVersion: ' TrueNAS-SCALE-25.10.4 ',
+          reconnects: 1,
+        },
      },
    ]);

@@ -50,7 +60,7 @@ describe('TrueNASAPI', () => {
        host: 'truenas.local',
        port: 443,
        apiKey: '********',
-        username: undefined,
+        username: 'pulse-readonly',
        password: undefined,
        useHttps: true,
        insecureSkipVerify: false,
@@ -77,6 +87,18 @@ describe('TrueNASAPI', () => {
          disks: 8,
          recoveryArtifacts: 18,
        },
+        transport: {
+          mode: 'jsonrpc-websocket',
+          endpoint: 'wss://truenas.local/api/current',
+          tls: true,
+          connected: true,
+          authMechanism: 'api-key-plain',
+          applianceVersion: 'TrueNAS-SCALE-25.10.4',
+          legacyReason: undefined,
+          reconnects: 1,
+          lastError: undefined,
+          lastConnectedAt: undefined,
+        },
        monitorDatasets: false,
        monitorPools: false,
        monitorReplication: false,
@@ -21,6 +21,7 @@ type RawTrueNASConnection = Partial<TrueNASConnection>;
 type RawTrueNASConnectionPollError = Partial<TrueNASConnectionPollError>;
 type RawTrueNASConnectionPoll = Partial<TrueNASConnectionPollStatus>;
 type RawTrueNASConnectionObservedSummary = Partial<TrueNASConnectionObservedSummary>;
+type RawTrueNASConnectionTransportStatus = Partial<TrueNASConnectionTransportStatus>;

 export interface TrueNASConnectionPollError {
  at?: string;
@@ -50,6 +51,19 @@ export interface TrueNASConnectionObservedSummary {
  recoveryArtifacts: number;
 }

+export interface TrueNASConnectionTransportStatus {
+  mode: 'negotiating' | 'jsonrpc-websocket' | 'legacy-rest';
+  endpoint?: string;
+  tls: boolean;
+  connected: boolean;
+  authMechanism?: string;
+  applianceVersion?: string;
+  legacyReason?: string;
+  reconnects?: number;
+  lastError?: string;
+  lastConnectedAt?: string;
+}
+
 export interface TrueNASConnection {
  id: string;
  name: string;
@@ -68,6 +82,7 @@ export interface TrueNASConnection {
  monitorReplication: boolean;
  poll?: TrueNASConnectionPollStatus;
  observed?: TrueNASConnectionObservedSummary;
+  transport?: TrueNASConnectionTransportStatus;
 }

 export interface TrueNASConnectionInput {
@@ -134,6 +149,27 @@ const normalizeTrueNASConnectionObservedSummary = (
  };
 };

+const normalizeTrueNASConnectionTransportStatus = (
+  transport: RawTrueNASConnectionTransportStatus | undefined,
+): TrueNASConnectionTransportStatus | undefined => {
+  if (!transport || typeof transport !== 'object') return undefined;
+  const rawMode = optionalTrimmedString(transport.mode);
+  const mode =
+    rawMode === 'jsonrpc-websocket' || rawMode === 'legacy-rest' ? rawMode : 'negotiating';
+  return {
+    mode,
+    endpoint: optionalTrimmedString(transport.endpoint),
+    tls: strictBoolean(transport.tls),
+    connected: strictBoolean(transport.connected),
+    authMechanism: optionalTrimmedString(transport.authMechanism),
+    applianceVersion: optionalTrimmedString(transport.applianceVersion),
+    legacyReason: optionalTrimmedString(transport.legacyReason),
+    reconnects: finiteNumberOrUndefined(transport.reconnects),
+    lastError: optionalTrimmedString(transport.lastError),
+    lastConnectedAt: optionalTrimmedString(transport.lastConnectedAt),
+  };
+};
+
 const normalizeTrueNASConnection = (connection: RawTrueNASConnection): TrueNASConnection => ({
  id: trimmedString(connection.id),
  name: optionalTrimmedString(connection.name) ?? '',
@@ -152,6 +188,7 @@ const normalizeTrueNASConnection = (connection: RawTrueNASConnection): TrueNASCo
  monitorReplication: strictBoolean(connection.monitorReplication),
  poll: normalizeTrueNASConnectionPoll(connection.poll),
  observed: normalizeTrueNASConnectionObservedSummary(connection.observed),
+  transport: normalizeTrueNASConnectionTransportStatus(connection.transport),
 });

 const serializeTrueNASConnectionInput = (input: TrueNASConnectionInput) => ({
@@ -28,6 +28,22 @@ export interface TrueNASCredentialSlotProps {
  deleteError?: string | null;
 }

+const describeTransport = (transport: NonNullable<TrueNASConnection['transport']>): string => {
+  let description: string;
+  if (transport.mode === 'jsonrpc-websocket') {
+    description = `Transport: JSON-RPC WebSocket${transport.tls ? ' over TLS' : ''}; authentication: ${transport.authMechanism || 'session'}.`;
+  } else if (transport.mode === 'legacy-rest') {
+    description =
+      `Transport: legacy REST (${transport.applianceVersion || 'legacy TrueNAS'}). ${transport.legacyReason || ''}`.trim();
+  } else {
+    description = 'Transport negotiation has not completed yet.';
+  }
+  if (transport.lastError) {
+    description += ` Last error: ${transport.lastError}`;
+  }
+  return description;
+};
+
 export const TrueNASCredentialSlot: Component<TrueNASCredentialSlotProps> = (props) => {
  let primed = false;

@@ -132,21 +148,36 @@ export const TrueNASCredentialSlot: Component<TrueNASCredentialSlotProps> = (pro
        </div>

        <Show when={props.state.form().authMode === 'apiKey'}>
-          <label class={formField}>
-            <span class={formLabel}>API key</span>
-            <input
-              class={formControl}
-              type="password"
-              value={props.state.form().apiKey}
-              onInput={(event) => props.state.updateForm({ apiKey: event.currentTarget.value })}
-              placeholder={
-                props.state.form().hasStoredApiKey ? 'Saved API key retained unless replaced' : ''
-              }
-            />
-            <Show when={props.state.form().hasStoredApiKey}>
-              <span class={formHelpText}>Leave this blank to keep the saved API key.</span>
-            </Show>
-          </label>
+          <div class="grid gap-4 sm:grid-cols-2">
+            <label class={formField}>
+              <span class={formLabel}>API key owner username</span>
+              <input
+                class={formControl}
+                value={props.state.form().username}
+                onInput={(event) => props.state.updateForm({ username: event.currentTarget.value })}
+                placeholder="pulse-monitor"
+              />
+              <span class={formHelpText}>
+                Use the TrueNAS user that owns this key. Required by the supported login flow on
+                current and future releases.
+              </span>
+            </label>
+            <label class={formField}>
+              <span class={formLabel}>API key</span>
+              <input
+                class={formControl}
+                type="password"
+                value={props.state.form().apiKey}
+                onInput={(event) => props.state.updateForm({ apiKey: event.currentTarget.value })}
+                placeholder={
+                  props.state.form().hasStoredApiKey ? 'Saved API key retained unless replaced' : ''
+                }
+              />
+              <Show when={props.state.form().hasStoredApiKey}>
+                <span class={formHelpText}>Leave this blank to keep the saved API key.</span>
+              </Show>
+            </label>
+          </div>
        </Show>

        <Show when={props.state.form().authMode === 'userpass'}>
@@ -236,6 +267,17 @@ export const TrueNASCredentialSlot: Component<TrueNASCredentialSlotProps> = (pro
          </div>
        </div>

+        <Show when={props.editingConnection?.transport}>
+          {(transport) => (
+            <CalloutCard
+              tone={transport().mode === 'legacy-rest' ? 'warning' : 'info'}
+              scale="compact"
+              padding="sm"
+              description={describeTransport(transport())}
+            />
+          )}
+        </Show>
+
        <div class="space-y-3 rounded-md border border-border bg-surface p-4">
          <div>
            <div class="text-sm font-semibold text-base-content">Collection scope</div>
@@ -80,6 +80,7 @@ describe('useTrueNASSettingsPanelState', () => {
        name: 'tower',
        host: 'truenas.local',
        apiKey: '********',
+        username: 'pulse-readonly',
        pollIntervalSeconds: 90,
        useHttps: true,
        insecureSkipVerify: false,
@@ -91,6 +92,7 @@ describe('useTrueNASSettingsPanelState', () => {
      name: 'tower',
      host: 'truenas.local',
      apiKey: '********',
+      username: 'pulse-readonly',
      pollIntervalSeconds: 90,
      useHttps: true,
      insecureSkipVerify: false,
@@ -105,6 +107,7 @@ describe('useTrueNASSettingsPanelState', () => {
        name: 'tower',
        host: 'truenas.local',
        apiKey: '********',
+        username: 'pulse-readonly',
        pollIntervalSeconds: 90,
        useHttps: true,
        insecureSkipVerify: false,
@@ -128,6 +131,7 @@ describe('useTrueNASSettingsPanelState', () => {
        host: 'truenas.local',
        apiKey: '********',
        pollIntervalSeconds: 90,
+        username: 'pulse-readonly',
      }),
    );
    expect(TrueNASAPI.updateConnection).toHaveBeenCalledWith(
@@ -136,7 +140,7 @@ describe('useTrueNASSettingsPanelState', () => {
        host: 'truenas.local',
        apiKey: '********',
        pollIntervalSeconds: 90,
-        username: '',
+        username: 'pulse-readonly',
        password: '',
      }),
    );
@@ -73,7 +73,7 @@ const buildFormStateFromConnection = (
    pollIntervalSeconds: String(connection.pollIntervalSeconds ?? 60),
    authMode,
    apiKey: '',
-    username: authMode === 'userpass' ? connection.username || '' : '',
+    username: connection.username || '',
    password: '',
    useHttps: connection.useHttps,
    insecureSkipVerify: connection.insecureSkipVerify,
@@ -163,7 +163,7 @@ const buildConnectionInput = (form: TrueNASConnectionFormState): TrueNASConnecti

  if (form.authMode === 'apiKey') {
    input.apiKey = form.apiKey.trim() || (form.hasStoredApiKey ? REDACTED_SECRET : '');
-    input.username = '';
+    input.username = form.username.trim();
    input.password = '';
  } else {
    input.apiKey = '';
@@ -6,6 +6,7 @@ import (
 	"github.com/rcourtman/pulse-go-rewrite/internal/config"
 	"github.com/rcourtman/pulse-go-rewrite/internal/mock"
 	"github.com/rcourtman/pulse-go-rewrite/internal/monitoring"
+	"github.com/rcourtman/pulse-go-rewrite/internal/truenas"
 )

 func mockTrueNASConnectionResponses() []trueNASConnectionResponse {
@@ -43,6 +44,15 @@ func mockTrueNASConnectionResponses() []trueNASConnectionResponse {
 			LastSuccessAt:   mockPlatformTimePointer(fixture.CollectedAt),
 		},
 		Observed: observed,
+		Transport: &truenas.TransportStatus{
+			Mode:             truenas.TransportJSONRPC,
+			Endpoint:         "wss://truenas-main/api/current",
+			TLS:              true,
+			Connected:        true,
+			AuthMechanism:    "api-key-plain",
+			ApplianceVersion: "TrueNAS-SCALE-25.10.4",
+			LastConnectedAt:  mockPlatformTimePointer(fixture.CollectedAt),
+		},
 	}}
 }

@@ -105,8 +115,9 @@ func mockTrueNASLedgerInputs() ([]config.TrueNASInstance, map[string]monitoring.
 	for _, response := range responses {
 		instances = append(instances, response.TrueNASInstance)
 		summaries[response.TrueNASInstance.ID] = monitoring.TrueNASConnectionSummary{
-			Poll:     response.Poll,
-			Observed: response.Observed,
+			Poll:      response.Poll,
+			Observed:  response.Observed,
+			Transport: response.Transport,
 		}
 	}
 	return instances, summaries
@@ -28,8 +28,9 @@ type TrueNASHandlers struct {

 type trueNASConnectionResponse struct {
 	config.TrueNASInstance
-	Poll     *monitoring.TrueNASConnectionPollStatus      `json:"poll,omitempty"`
-	Observed *monitoring.TrueNASConnectionObservedSummary `json:"observed,omitempty"`
+	Poll      *monitoring.TrueNASConnectionPollStatus      `json:"poll,omitempty"`
+	Observed  *monitoring.TrueNASConnectionObservedSummary `json:"observed,omitempty"`
+	Transport *truenas.TransportStatus                     `json:"transport,omitempty"`
 }

 type trueNASClient interface {
@@ -126,6 +127,7 @@ func (h *TrueNASHandlers) HandleList(w http.ResponseWriter, r *http.Request) {
 		if summary, ok := summaries[strings.TrimSpace(item.ID)]; ok {
 			response.Poll = summary.Poll
 			response.Observed = summary.Observed
+			response.Transport = summary.Transport
 		}
 		redacted = append(redacted, response)
 	}
@@ -439,6 +439,12 @@ func TestTrueNASHandlers_HandleList_ReturnsMockConnectionsInMockMode(t *testing.
 	if listed[0].Observed == nil || listed[0].Observed.Systems != 1 || listed[0].Observed.StoragePools == 0 {
 		t.Fatalf("expected populated mock observed summary, got %+v", listed[0].Observed)
 	}
+	if listed[0].Transport == nil ||
+		listed[0].Transport.Mode != truenas.TransportJSONRPC ||
+		!listed[0].Transport.Connected ||
+		!listed[0].Transport.TLS {
+		t.Fatalf("expected current mock transport diagnostics, got %+v", listed[0].Transport)
+	}
 }

 func TestTrueNASHandlers_HandleList_IncludesPollAndObservedSummary(t *testing.T) {
@@ -520,6 +526,18 @@ func TestTrueNASHandlers_HandleList_IncludesPollAndObservedSummary(t *testing.T)
 	if listed[0].Observed.StoragePools != 1 || listed[0].Observed.Datasets != 1 || listed[0].Observed.Disks != 1 {
 		t.Fatalf("unexpected observed counts: %+v", listed[0].Observed)
 	}
+	if listed[0].Transport == nil ||
+		listed[0].Transport.Mode != truenas.TransportLegacyREST ||
+		listed[0].Transport.ApplianceVersion != "TrueNAS-SCALE-24.10.2" {
+		t.Fatalf("unexpected transport diagnostics: %+v", listed[0].Transport)
+	}
+	encodedTransport, err := json.Marshal(listed[0].Transport)
+	if err != nil {
+		t.Fatalf("encode transport diagnostics: %v", err)
+	}
+	if strings.Contains(string(encodedTransport), connection.APIKey) {
+		t.Fatalf("transport diagnostics exposed API key: %s", encodedTransport)
+	}
 }

 func TestTrueNASHandlers_HandleDelete_RemovesAndHandlesUnknownID(t *testing.T) {
@@ -62,8 +62,9 @@ type TrueNASConnectionObservedSummary struct {
 // TrueNASConnectionSummary merges poll health with the most recent discovered
 // platform contribution for one configured TrueNAS connection.
 type TrueNASConnectionSummary struct {
-	Poll     *TrueNASConnectionPollStatus      `json:"poll,omitempty"`
-	Observed *TrueNASConnectionObservedSummary `json:"observed,omitempty"`
+	Poll      *TrueNASConnectionPollStatus      `json:"poll,omitempty"`
+	Observed  *TrueNASConnectionObservedSummary `json:"observed,omitempty"`
+	Transport *truenas.TransportStatus          `json:"transport,omitempty"`
 }

 type trueNASConnectionRuntimeStatus struct {
@@ -390,6 +391,10 @@ func (p *TrueNASPoller) ConnectionSummaries(orgID string, instances []config.Tru
 		if p != nil {
 			p.mu.Lock()
 			status := p.cloneRuntimeStatusLocked(orgID, connID)
+			var provider *truenas.Provider
+			if byConnection := p.providersByOrg[orgID]; byConnection != nil {
+				provider = byConnection[connID]
+			}
 			p.mu.Unlock()
 			if status != nil {
 				summary.Poll.LastAttemptAt = cloneTimePointer(status.LastAttemptAt)
@@ -398,6 +403,10 @@ func (p *TrueNASPoller) ConnectionSummaries(orgID string, instances []config.Tru
 				summary.Poll.LastError = cloneTrueNASConnectionPollError(status.LastError)
 				summary.Observed = cloneTrueNASObservedSummary(status.Observed)
 			}
+			if provider != nil {
+				transport := provider.TransportStatus()
+				summary.Transport = &transport
+			}
 		}

 		summaries[connID] = summary
@@ -1484,6 +1493,29 @@ func classifyTrueNASError(err error, connectionID string) *internalerrors.Monito
 			errType = internalerrors.ErrorTypeAPI
 		}
 	} else {
+		var authErr *truenas.RPCAuthError
+		if errors.As(err, &authErr) {
+			errType = internalerrors.ErrorTypeAuth
+			retryable = false
+		}
+		var handshakeErr *truenas.RPCHandshakeError
+		if errors.As(err, &handshakeErr) &&
+			(handshakeErr.StatusCode == http.StatusUnauthorized || handshakeErr.StatusCode == http.StatusForbidden) {
+			errType = internalerrors.ErrorTypeAuth
+			retryable = false
+		}
+		var rpcErr *truenas.RPCError
+		if errors.As(err, &rpcErr) {
+			message := strings.ToLower(rpcErr.Message + " " + rpcErr.Reason)
+			if strings.HasPrefix(rpcErr.Method, "auth.") ||
+				strings.EqualFold(rpcErr.Errname, "EACCES") ||
+				strings.EqualFold(rpcErr.Errname, "EPERM") ||
+				strings.Contains(message, "not authorized") ||
+				strings.Contains(message, "permission") {
+				errType = internalerrors.ErrorTypeAuth
+				retryable = false
+			}
+		}
 		// Transport-level errors: timeout takes precedence over generic connection failures.
 		var urlErr *url.Error
 		if (errors.As(err, &urlErr) && urlErr.Timeout()) || errors.Is(err, context.DeadlineExceeded) {
@@ -310,6 +310,11 @@ func TestTrueNASPollerConnectionSummariesExposeObservedCounts(t *testing.T) {
 	if summary.Observed.Systems != 1 || summary.Observed.StoragePools != 1 || summary.Observed.Datasets != 1 || summary.Observed.Disks != 1 {
 		t.Fatalf("unexpected observed counts: %+v", summary.Observed)
 	}
+	if summary.Transport == nil ||
+		summary.Transport.Mode != truenas.TransportLegacyREST ||
+		!summary.Transport.Connected {
+		t.Fatalf("unexpected connection-local transport summary: %+v", summary.Transport)
+	}
 }

 func TestTrueNASObservedSummaryIncludesNativeRuntimeAndSharingFacets(t *testing.T) {
@@ -1678,6 +1683,33 @@ func TestClassifyTrueNASError(t *testing.T) {
 			expectedType:  "auth",
 			expectedRetry: false,
 		},
+		{
+			name:          "websocket handshake 403 classifies as auth",
+			err:           &truenas.RPCHandshakeError{StatusCode: 403, Err: fmt.Errorf("forbidden")},
+			expectedType:  "auth",
+			expectedRetry: false,
+		},
+		{
+			name: "JSON-RPC login response classifies as auth",
+			err: &truenas.RPCAuthError{
+				Mechanism:    "api-key-plain",
+				ResponseType: "AUTH_ERR",
+			},
+			expectedType:  "auth",
+			expectedRetry: false,
+		},
+		{
+			name: "JSON-RPC permission error classifies as auth",
+			err: &truenas.RPCError{
+				Code:    -32001,
+				Method:  "pool.query",
+				Message: "Method call error",
+				Reason:  "Not authorized",
+				Errname: "EACCES",
+			},
+			expectedType:  "auth",
+			expectedRetry: false,
+		},
 		{
 			name:          "context.DeadlineExceeded classifies as timeout",
 			err:           context.DeadlineExceeded,
@@ -152,8 +152,11 @@ func TestClientGetters(t *testing.T) {

 func TestClientAuthHeaderAPIKey(t *testing.T) {
 	server := newMockServer(t, map[string]apiResponse{
-		"/api/v2.0/system/info": {body: `{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`},
+		"/api/v2.0/system/info": {body: `{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`},
 	}, func(t *testing.T, request *http.Request) {
+		if request.URL.Path == "/api/current" {
+			return
+		}
 		if got := request.Header.Get("Authorization"); got != "Bearer test-key" {
 			t.Fatalf("expected bearer auth header, got %q", got)
 		}
@@ -168,8 +171,11 @@ func TestClientAuthHeaderAPIKey(t *testing.T) {

 func TestClientAuthHeaderBasic(t *testing.T) {
 	server := newMockServer(t, map[string]apiResponse{
-		"/api/v2.0/system/info": {body: `{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`},
+		"/api/v2.0/system/info": {body: `{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`},
 	}, func(t *testing.T, request *http.Request) {
+		if request.URL.Path == "/api/current" {
+			return
+		}
 		username, password, ok := request.BasicAuth()
 		if !ok {
 			t.Fatalf("expected basic auth")
@@ -187,36 +193,22 @@ func TestClientAuthHeaderBasic(t *testing.T) {
 }

 func TestGetSystemInfoAcceptsStructuredBuildTime(t *testing.T) {
-	server := newMockServer(t, map[string]apiResponse{
-		"/api/v2.0/system/info": {
-			body: `{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":1}`,
-		},
-	}, nil)
-	t.Cleanup(server.Close)
-
-	client := mustClientForServer(t, server.URL, ClientConfig{APIKey: "test-key"})
-	system, err := client.GetSystemInfo(context.Background())
-	if err != nil {
-		t.Fatalf("GetSystemInfo() error = %v", err)
+	var response systemInfoResponse
+	if err := json.Unmarshal([]byte(`{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":1}`), &response); err != nil {
+		t.Fatalf("Unmarshal() error = %v", err)
 	}
+	system := systemInfoFromResponse(response)
 	if system.Build != "2026-05-14T18:24:01+02:00" {
 		t.Fatalf("Build = %q, want structured buildtime date", system.Build)
 	}
 }

 func TestGetSystemInfoAcceptsFractionalUptimeSeconds(t *testing.T) {
-	server := newMockServer(t, map[string]apiResponse{
-		"/api/v2.0/system/info": {
-			body: `{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":360144.629139547}`,
-		},
-	}, nil)
-	t.Cleanup(server.Close)
-
-	client := mustClientForServer(t, server.URL, ClientConfig{APIKey: "test-key"})
-	system, err := client.GetSystemInfo(context.Background())
-	if err != nil {
-		t.Fatalf("GetSystemInfo() error = %v", err)
+	var response systemInfoResponse
+	if err := json.Unmarshal([]byte(`{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":360144.629139547}`), &response); err != nil {
+		t.Fatalf("Unmarshal() error = %v", err)
 	}
+	system := systemInfoFromResponse(response)
 	if system.UptimeSeconds != 360144 {
 		t.Fatalf("UptimeSeconds = %d, want truncated fractional uptime", system.UptimeSeconds)
 	}
@@ -224,7 +216,7 @@ func TestGetSystemInfoAcceptsFractionalUptimeSeconds(t *testing.T) {

 func TestTestConnectionSuccessAndFailure(t *testing.T) {
 	successServer := newMockServer(t, map[string]apiResponse{
-		"/api/v2.0/system/info": {body: `{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`},
+		"/api/v2.0/system/info": {body: `{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`},
 	}, nil)
 	t.Cleanup(successServer.Close)

@@ -667,6 +659,7 @@ func TestGetAppsEnrichesStatsFromRPC(t *testing.T) {
 				},
 			},
 		})
+		expectRPCUnsubscribe(t, conn, "sub-1")
 	})
 	t.Cleanup(server.Close)

@@ -813,24 +806,22 @@ func TestGetReplicationTasksUsesNativeReplicationQueryShape(t *testing.T) {
 }

 func TestStartAndStopAppUseRPCMethods(t *testing.T) {
-	var rpcCalls int
+	var rpcSessions int
 	server := newMockServerWithRPC(t, defaultAPIResponses(), nil, func(t *testing.T, conn *websocket.Conn) {
-		rpcCalls++
+		rpcSessions++
 		authReq := readRPCRequest(t, conn)
 		if authReq.Method != "auth.login_with_api_key" {
 			t.Fatalf("expected api-key auth method, got %q", authReq.Method)
 		}
 		writeRPCResult(t, conn, authReq.ID, true)

-		actionReq := readRPCRequest(t, conn)
-		expectedMethod := "app.start"
-		if rpcCalls == 2 {
-			expectedMethod = "app.stop"
+		for _, expectedMethod := range []string{"app.start", "app.stop"} {
+			actionReq := readRPCRequest(t, conn)
+			if actionReq.Method != expectedMethod {
+				t.Fatalf("expected %s, got %q", expectedMethod, actionReq.Method)
+			}
+			writeRPCResult(t, conn, actionReq.ID, true)
 		}
-		if actionReq.Method != expectedMethod {
-			t.Fatalf("expected %s, got %q", expectedMethod, actionReq.Method)
-		}
-		writeRPCResult(t, conn, actionReq.ID, true)
 	})
 	t.Cleanup(server.Close)

@@ -841,8 +832,8 @@ func TestStartAndStopAppUseRPCMethods(t *testing.T) {
 	if err := client.StopApp(context.Background(), "nextcloud"); err != nil {
 		t.Fatalf("StopApp() error = %v", err)
 	}
-	if rpcCalls != 2 {
-		t.Fatalf("expected two RPC app-action sessions, got %d", rpcCalls)
+	if rpcSessions != 1 {
+		t.Fatalf("expected one persistent RPC app-action session, got %d", rpcSessions)
 	}
 }

@@ -884,6 +875,7 @@ func TestGetAppLogsUsesRPCSubscription(t *testing.T) {
 				"timestamp": "2026-03-29T18:01:00Z",
 			},
 		})
+		expectRPCUnsubscribe(t, conn, "sub-logs")
 		time.Sleep(defaultAppLogIdleWait + 100*time.Millisecond)
 	})
 	t.Cleanup(server.Close)
@@ -967,6 +959,7 @@ func TestGetSystemTelemetryFromRPC(t *testing.T) {
 				},
 			},
 		})
+		expectRPCUnsubscribe(t, conn, "sub-1")
 	})
 	t.Cleanup(server.Close)

@@ -1130,6 +1123,7 @@ func TestGetSystemTelemetryIgnoresUnavailableTemperatureRPC(t *testing.T) {
 				"cpu": map[string]any{"usage": 41},
 			},
 		})
+		expectRPCUnsubscribe(t, conn, "sub-1")
 	})
 	t.Cleanup(server.Close)

@@ -1199,6 +1193,7 @@ func TestGetDisksToleratesUnavailableTemperatureEndpoint(t *testing.T) {
 }

 func TestGetDiskTemperaturesFallsBackToReportingRPC(t *testing.T) {
+	connectionCount := 0
 	server := newMockServerWithRPC(t, map[string]apiResponse{
 		"/api/v2.0/disk": {
 			body: `[{"identifier":"{disk-1}","name":"sda","serial":"SER-A","size":1000000,"model":"Seagate","type":"HDD","pool":"tank","bus":"SATA","rotationrate":7200,"status":"ONLINE"}]`,
@@ -1214,7 +1209,15 @@ func TestGetDiskTemperaturesFallsBackToReportingRPC(t *testing.T) {
 		}
 		writeRPCResult(t, conn, authReq.ID, true)

+		connectionCount++
 		temperatureReq := readRPCRequest(t, conn)
+		if connectionCount == 1 {
+			if temperatureReq.Method != "disk.query" {
+				t.Fatalf("expected disk.query, got %q", temperatureReq.Method)
+			}
+			writeRPCResult(t, conn, temperatureReq.ID, defaultRoutePayloadMaps(t, "/api/v2.0/disk")[:1])
+			return
+		}
 		if temperatureReq.Method != "reporting.get_data" {
 			t.Fatalf("expected reporting.get_data, got %q", temperatureReq.Method)
 		}
@@ -1613,7 +1616,7 @@ func TestClientTLSFingerprintPinning(t *testing.T) {
 			return
 		}
 		writer.Header().Set("Content-Type", "application/json")
-		_, _ = writer.Write([]byte(`{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`))
+		_, _ = writer.Write([]byte(`{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`))
 	})

 	tlsServer := httptest.NewTLSServer(handler)
@@ -1741,6 +1744,11 @@ func newMockServer(t *testing.T, responses map[string]apiResponse, assertRequest

 		response, ok := responses[request.URL.Path]
 		if !ok {
+			if request.URL.Path == "/api/v2.0/system/info" {
+				writer.Header().Set("Content-Type", "application/json")
+				_, _ = writer.Write([]byte(`{"hostname":"legacy-test","version":"TrueNAS-SCALE-24.10.2","buildtime":"24.10.2","uptime_seconds":1}`))
+				return
+			}
 			http.NotFound(writer, request)
 			return
 		}
@@ -1857,6 +1865,20 @@ func writeRPCNotification(t *testing.T, conn *websocket.Conn, method string, par
 	}
 }

+func expectRPCUnsubscribe(t *testing.T, conn *websocket.Conn, subscriptionID string) {
+	t.Helper()
+
+	request := readRPCRequest(t, conn)
+	if request.Method != "core.unsubscribe" {
+		t.Fatalf("expected core.unsubscribe, got %q", request.Method)
+	}
+	params, ok := request.Params.([]any)
+	if !ok || len(params) != 1 || params[0] != subscriptionID {
+		t.Fatalf("core.unsubscribe params = %#v, want [%q]", request.Params, subscriptionID)
+	}
+	writeRPCResult(t, conn, request.ID, nil)
+}
+
 func mustClientForServer(t *testing.T, serverURL string, config ClientConfig) *Client {
 	t.Helper()

@@ -1877,6 +1899,7 @@ func mustClientForServer(t *testing.T, serverURL string, config ClientConfig) *C
 	if err != nil {
 		t.Fatalf("NewClient() error = %v", err)
 	}
+	client.allowInsecureRPC = true
 	return client
 }

@@ -72,6 +72,10 @@ type systemMetricHistoryFetcher interface {
 	SystemMetricHistory(ctx context.Context, duration time.Duration) (*SystemMetricHistory, error)
 }

+type transportStatusFetcher interface {
+	TransportStatus() TransportStatus
+}
+
 // APIFetcher loads snapshots from the live TrueNAS API client.
 type APIFetcher struct {
 	Client *Client
@@ -128,6 +132,15 @@ func (f *APIFetcher) SystemMetricHistory(ctx context.Context, duration time.Dura
 	return f.Client.GetSystemMetricHistory(ctx, duration)
 }

+// TransportStatus returns the underlying client's non-secret connection-local
+// transport diagnostics.
+func (f *APIFetcher) TransportStatus() TransportStatus {
+	if f == nil || f.Client == nil {
+		return TransportStatus{Mode: TransportUnknown}
+	}
+	return f.Client.TransportStatus()
+}
+
 // FixtureFetcher loads snapshots from static fixture data.
 type FixtureFetcher struct {
 	Snapshot FixtureSnapshot
@@ -210,6 +223,18 @@ func (p *Provider) Refresh(ctx context.Context) error {
 	return nil
 }

+// TransportStatus returns non-secret diagnostics when the provider is backed
+// by the live API client. Fixture providers remain in the negotiating state.
+func (p *Provider) TransportStatus() TransportStatus {
+	if p == nil {
+		return TransportStatus{Mode: TransportUnknown}
+	}
+	if source, ok := p.fetcher.(transportStatusFetcher); ok {
+		return source.TransportStatus()
+	}
+	return TransportStatus{Mode: TransportUnknown}
+}
+
 // ControlApp executes a native start/stop action against a TrueNAS app and
 // refreshes the cached snapshot so downstream readers observe canonical state.
 func (p *Provider) ControlApp(ctx context.Context, appID, action string) (*FixtureSnapshot, error) {
@@ -0,0 +1,556 @@
+package truenas
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+)
+
+var errRPCStreamSessionConsumed = errors.New("truenas rpc stream session cannot be reused")
+
+type discardRPCSessionError struct {
+	err error
+}
+
+func (e *discardRPCSessionError) Error() string {
+	return fmt.Sprintf("discard truenas rpc session: %v", e.err)
+}
+
+func (e *discardRPCSessionError) Unwrap() error { return e.err }
+
+func discardRPCSessionForStreamError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var transportErr *RPCTransportError
+	if errors.As(err, &transportErr) {
+		return err
+	}
+	return &discardRPCSessionError{err: err}
+}
+
+// TransportMode is the connection-local API transport selected for an
+// appliance. A client never changes modes after successful negotiation.
+type TransportMode string
+
+const (
+	TransportUnknown    TransportMode = "negotiating"
+	TransportJSONRPC    TransportMode = "jsonrpc-websocket"
+	TransportLegacyREST TransportMode = "legacy-rest"
+)
+
+// TransportStatus is a non-secret diagnostic projection of transport and
+// authentication state for one appliance.
+type TransportStatus struct {
+	Mode             TransportMode `json:"mode"`
+	Endpoint         string        `json:"endpoint,omitempty"`
+	TLS              bool          `json:"tls"`
+	Connected        bool          `json:"connected"`
+	AuthMechanism    string        `json:"authMechanism,omitempty"`
+	ApplianceVersion string        `json:"applianceVersion,omitempty"`
+	LegacyReason     string        `json:"legacyReason,omitempty"`
+	Reconnects       int           `json:"reconnects,omitempty"`
+	LastError        string        `json:"lastError,omitempty"`
+	LastConnectedAt  *time.Time    `json:"lastConnectedAt,omitempty"`
+}
+
+// RPCError is a structured JSON-RPC method error. Data is intentionally
+// reduced to non-secret middleware diagnostics.
+type RPCError struct {
+	Code    int
+	Method  string
+	Message string
+	Reason  string
+	Errname string
+}
+
+func (e *RPCError) Error() string {
+	if e == nil {
+		return "truenas rpc error"
+	}
+	detail := strings.TrimSpace(e.Reason)
+	if detail == "" {
+		detail = strings.TrimSpace(e.Message)
+	}
+	if e.Errname != "" {
+		return fmt.Sprintf("truenas rpc %s failed: code=%d errname=%s message=%q", e.Method, e.Code, e.Errname, detail)
+	}
+	return fmt.Sprintf("truenas rpc %s failed: code=%d message=%q", e.Method, e.Code, detail)
+}
+
+// RPCTransportError distinguishes a failed websocket exchange from an
+// authoritative middleware method error.
+type RPCTransportError struct {
+	Method string
+	Phase  string
+	Err    error
+}
+
+func (e *RPCTransportError) Error() string {
+	return fmt.Sprintf("%s truenas rpc %s %s: %v", e.Phase, e.Method, transportPhaseNoun(e.Phase), e.Err)
+}
+
+func (e *RPCTransportError) Unwrap() error { return e.Err }
+
+func transportPhaseNoun(phase string) string {
+	if phase == "write" {
+		return "request"
+	}
+	return "response"
+}
+
+type RPCHandshakeError struct {
+	StatusCode int
+	Err        error
+}
+
+func (e *RPCHandshakeError) Error() string {
+	if e.StatusCode > 0 {
+		return fmt.Sprintf("dial truenas rpc websocket failed: status=%d: %v", e.StatusCode, e.Err)
+	}
+	return fmt.Sprintf("dial truenas rpc websocket failed: %v", e.Err)
+}
+
+func (e *RPCHandshakeError) Unwrap() error { return e.Err }
+
+// RPCAuthError is an authoritative authentication refusal returned after the
+// websocket handshake. It is distinct from a retryable transport failure and
+// intentionally carries no credential material.
+type RPCAuthError struct {
+	Mechanism    string
+	ResponseType string
+}
+
+func (e *RPCAuthError) Error() string {
+	if e == nil {
+		return "truenas rpc authentication failed"
+	}
+	if strings.TrimSpace(e.ResponseType) == "" {
+		return fmt.Sprintf("truenas rpc %s authentication failed", e.Mechanism)
+	}
+	return fmt.Sprintf("truenas rpc %s authentication failed: response_type=%q", e.Mechanism, e.ResponseType)
+}
+
+// TransportStatus returns a secret-free snapshot. It never waits for an
+// in-flight API call.
+func (c *Client) TransportStatus() TransportStatus {
+	if c == nil {
+		return TransportStatus{Mode: TransportUnknown}
+	}
+	c.statusMu.RLock()
+	defer c.statusMu.RUnlock()
+	status := c.status
+	if status.LastConnectedAt != nil {
+		value := *status.LastConnectedAt
+		status.LastConnectedAt = &value
+	}
+	return status
+}
+
+func (c *Client) updateTransportStatus(update func(*TransportStatus)) {
+	if c == nil {
+		return
+	}
+	c.statusMu.Lock()
+	update(&c.status)
+	c.statusMu.Unlock()
+}
+
+func (c *Client) ensureTransport(ctx context.Context) (TransportMode, error) {
+	if c == nil {
+		return TransportUnknown, fmt.Errorf("truenas client is nil")
+	}
+	c.rpcMu.Lock()
+	defer c.rpcMu.Unlock()
+	return c.ensureTransportLocked(ctx)
+}
+
+func (c *Client) useLegacyREST(ctx context.Context) (bool, error) {
+	mode, err := c.ensureTransport(ctx)
+	return mode == TransportLegacyREST, err
+}
+
+func (c *Client) ensureTransportLocked(ctx context.Context) (TransportMode, error) {
+	if c.closed {
+		return TransportUnknown, fmt.Errorf("truenas client is closed")
+	}
+	if c.mode == TransportLegacyREST {
+		return c.mode, nil
+	}
+	if c.mode == TransportJSONRPC && c.rpc != nil {
+		return c.mode, nil
+	}
+
+	reconnecting := c.mode == TransportJSONRPC
+	rpc, authMechanism, err := c.openAuthenticatedRPC(ctx)
+	if err == nil {
+		now := time.Now().UTC()
+		c.rpc = rpc
+		c.mode = TransportJSONRPC
+		c.updateTransportStatus(func(status *TransportStatus) {
+			status.Mode = TransportJSONRPC
+			status.Endpoint = c.rpcURL
+			status.Connected = true
+			status.AuthMechanism = authMechanism
+			status.LegacyReason = ""
+			status.LastError = ""
+			status.LastConnectedAt = &now
+			if reconnecting {
+				status.Reconnects++
+			}
+		})
+		return c.mode, nil
+	}
+
+	if reconnecting {
+		lockedErr := fmt.Errorf("truenas JSON-RPC reconnect failed; transport is immutable for this client and REST downgrade is disabled: %w", err)
+		c.recordTransportError(lockedErr)
+		return TransportJSONRPC, lockedErr
+	}
+
+	if !isUnsupportedWebSocketEndpoint(err) {
+		c.recordTransportError(err)
+		return TransportUnknown, err
+	}
+
+	var response systemInfoResponse
+	if restErr := c.getJSON(ctx, http.MethodGet, "/system/info", &response); restErr != nil {
+		joined := fmt.Errorf("truenas JSON-RPC endpoint is unavailable and legacy version probe failed: websocket=%w rest=%v", err, restErr)
+		c.recordTransportError(joined)
+		return TransportUnknown, joined
+	}
+	if !supportsLegacyREST(response.Version) {
+		version := strings.TrimSpace(response.Version)
+		if version == "" {
+			version = "unknown"
+		}
+		rejected := fmt.Errorf("truenas %s requires JSON-RPC WebSocket at /api/current; refusing REST downgrade after websocket endpoint failure", version)
+		c.recordTransportError(rejected)
+		return TransportUnknown, rejected
+	}
+
+	c.mode = TransportLegacyREST
+	version := strings.TrimSpace(response.Version)
+	c.updateTransportStatus(func(status *TransportStatus) {
+		status.Mode = TransportLegacyREST
+		status.Endpoint = c.baseURL
+		status.Connected = true
+		status.AuthMechanism = legacyRESTAuthMechanism(c.config)
+		status.ApplianceVersion = version
+		status.LegacyReason = "JSON-RPC /api/current is unavailable on this recognized legacy TrueNAS release"
+		status.LastError = ""
+	})
+	return c.mode, nil
+}
+
+func (c *Client) openAuthenticatedRPC(ctx context.Context) (*trueNASRPCClient, string, error) {
+	conn, err := c.dialRPC(ctx)
+	if err != nil {
+		return nil, "", err
+	}
+	if !c.config.UseHTTPS && !c.allowInsecureRPC {
+		_ = conn.Close()
+		return nil, "", fmt.Errorf("truenas JSON-RPC credentials require TLS; configure an https endpoint (certificate verification can be pinned or explicitly disabled)")
+	}
+	rpc := &trueNASRPCClient{conn: conn, nextID: 1}
+	authMechanism, err := rpc.authenticate(ctx, c.config)
+	if err != nil {
+		_ = conn.Close()
+		return nil, "", err
+	}
+	return rpc, authMechanism, nil
+}
+
+func (c *Client) callRPC(ctx context.Context, method string, params any, result any) error {
+	return c.callRPCWithRetry(ctx, method, params, result, true)
+}
+
+func (c *Client) withRPC(ctx context.Context, operation func(*trueNASRPCClient) error) error {
+	if c == nil {
+		return fmt.Errorf("truenas client is nil")
+	}
+	c.rpcMu.Lock()
+	defer c.rpcMu.Unlock()
+
+	mode, err := c.ensureTransportLocked(ctx)
+	if err != nil {
+		return err
+	}
+	if mode != TransportJSONRPC || c.rpc == nil {
+		return fmt.Errorf("truenas JSON-RPC operation is unavailable over negotiated transport %s", mode)
+	}
+	if err := operation(c.rpc); err != nil {
+		if errors.Is(err, errRPCStreamSessionConsumed) {
+			c.closeRPCLocked()
+			c.reconnect = 0
+			return nil
+		}
+		var discardErr *discardRPCSessionError
+		if errors.As(err, &discardErr) {
+			c.closeRPCLocked()
+			c.recordTransportError(err)
+			return err
+		}
+		var transportErr *RPCTransportError
+		if !errors.As(err, &transportErr) {
+			c.recordRPCOperationError(err)
+			return err
+		}
+		c.closeRPCLocked()
+		c.recordTransportError(err)
+		if err := c.waitReconnectBackoff(ctx); err != nil {
+			return err
+		}
+		rpc, authMechanism, reconnectErr := c.openAuthenticatedRPC(ctx)
+		if reconnectErr != nil {
+			c.recordTransportError(reconnectErr)
+			return reconnectErr
+		}
+		c.rpc = rpc
+		c.reconnect++
+		now := time.Now().UTC()
+		c.updateTransportStatus(func(status *TransportStatus) {
+			status.Connected = true
+			status.AuthMechanism = authMechanism
+			status.Reconnects++
+			status.LastConnectedAt = &now
+			status.LastError = ""
+		})
+		if err := operation(c.rpc); err != nil {
+			var secondTransportErr *RPCTransportError
+			if errors.As(err, &secondTransportErr) {
+				c.closeRPCLocked()
+				c.recordTransportError(err)
+			} else {
+				c.recordRPCOperationError(err)
+			}
+			return err
+		}
+	}
+	c.reconnect = 0
+	return nil
+}
+
+func (c *Client) callRPCAction(ctx context.Context, method string, params any, result any) error {
+	return c.callRPCWithRetry(ctx, method, params, result, false)
+}
+
+func (c *Client) callRPCWithRetry(ctx context.Context, method string, params any, result any, retryRead bool) error {
+	if c == nil {
+		return fmt.Errorf("truenas client is nil")
+	}
+	c.rpcMu.Lock()
+	defer c.rpcMu.Unlock()
+
+	mode, err := c.ensureTransportLocked(ctx)
+	if err != nil {
+		return err
+	}
+	if mode != TransportJSONRPC || c.rpc == nil {
+		return fmt.Errorf("truenas rpc %s is unavailable over negotiated transport %s", method, mode)
+	}
+
+	err = c.rpc.call(ctx, method, params, result)
+	if err == nil {
+		c.reconnect = 0
+		c.updateTransportStatus(func(status *TransportStatus) {
+			status.Connected = true
+			status.LastError = ""
+		})
+		return nil
+	}
+	var transportErr *RPCTransportError
+	if !errors.As(err, &transportErr) {
+		c.recordRPCOperationError(err)
+		return err
+	}
+
+	c.closeRPCLocked()
+	c.recordTransportError(err)
+	if !retryRead {
+		return fmt.Errorf("truenas action %s transport failed after dispatch; outcome is unknown and Pulse will not replay it: %w", method, err)
+	}
+
+	if err := c.waitReconnectBackoff(ctx); err != nil {
+		return err
+	}
+	rpc, authMechanism, reconnectErr := c.openAuthenticatedRPC(ctx)
+	if reconnectErr != nil {
+		c.recordTransportError(reconnectErr)
+		return reconnectErr
+	}
+	c.rpc = rpc
+	c.mode = TransportJSONRPC
+	c.reconnect++
+	now := time.Now().UTC()
+	c.updateTransportStatus(func(status *TransportStatus) {
+		status.Connected = true
+		status.AuthMechanism = authMechanism
+		status.Reconnects++
+		status.LastConnectedAt = &now
+		status.LastError = ""
+	})
+	if err := c.rpc.call(ctx, method, params, result); err != nil {
+		var secondTransportErr *RPCTransportError
+		if errors.As(err, &secondTransportErr) {
+			c.closeRPCLocked()
+			c.recordTransportError(err)
+		} else {
+			c.recordRPCOperationError(err)
+		}
+		return err
+	}
+	c.reconnect = 0
+	c.updateTransportStatus(func(status *TransportStatus) {
+		status.Connected = true
+		status.LastError = ""
+	})
+	return nil
+}
+
+func (c *Client) waitReconnectBackoff(ctx context.Context) error {
+	attempt := c.reconnect
+	if attempt > 4 {
+		attempt = 4
+	}
+	delay := 100 * time.Millisecond * time.Duration(1<<attempt)
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
+func (c *Client) closeRPCLocked() {
+	if c.rpc != nil && c.rpc.conn != nil {
+		_ = c.rpc.conn.Close()
+	}
+	c.rpc = nil
+	c.updateTransportStatus(func(status *TransportStatus) {
+		status.Connected = false
+	})
+}
+
+func (c *Client) recordTransportError(err error) {
+	message := c.sanitizeTransportError(err)
+	c.updateTransportStatus(func(status *TransportStatus) {
+		status.Connected = false
+		status.LastError = message
+	})
+}
+
+func (c *Client) recordRPCOperationError(err error) {
+	message := c.sanitizeTransportError(err)
+	c.updateTransportStatus(func(status *TransportStatus) {
+		status.Connected = true
+		status.LastError = message
+	})
+}
+
+func (c *Client) sanitizeTransportError(err error) string {
+	if err == nil {
+		return ""
+	}
+	message := err.Error()
+	if c == nil {
+		return message
+	}
+	for _, secret := range []string{c.config.APIKey, c.config.Password} {
+		secret = strings.TrimSpace(secret)
+		if secret != "" {
+			message = strings.ReplaceAll(message, secret, "[redacted]")
+		}
+	}
+	return message
+}
+
+func isUnsupportedWebSocketEndpoint(err error) bool {
+	var handshake *RPCHandshakeError
+	if !errors.As(err, &handshake) {
+		return false
+	}
+	switch handshake.StatusCode {
+	case http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusGone, http.StatusNotImplemented:
+		return true
+	default:
+		return false
+	}
+}
+
+func supportsLegacyREST(version string) bool {
+	normalized := strings.ToUpper(strings.TrimSpace(version))
+	if normalized == "" {
+		return false
+	}
+	if strings.Contains(normalized, "SCALE") {
+		major, minor, ok := trueNASYearRelease(normalized)
+		return ok && (major < 25 || (major == 25 && minor < 4))
+	}
+	// TrueNAS CORE 13 does not expose the versioned JSON-RPC 2.0 endpoint.
+	return strings.Contains(normalized, "CORE") || strings.HasPrefix(normalized, "TRUENAS-13") || strings.HasPrefix(normalized, "FREENAS-")
+}
+
+func trueNASYearRelease(version string) (int, int, bool) {
+	for _, field := range strings.FieldsFunc(version, func(r rune) bool {
+		return r < '0' || r > '9'
+	}) {
+		if len(field) != 2 && len(field) != 4 {
+			continue
+		}
+		major, err := strconv.Atoi(field)
+		if err != nil || major < 20 || major > 99 {
+			continue
+		}
+		rest := version[strings.Index(version, field)+len(field):]
+		for _, minorField := range strings.FieldsFunc(rest, func(r rune) bool {
+			return r < '0' || r > '9'
+		}) {
+			minor, err := strconv.Atoi(minorField)
+			if err == nil {
+				return major, minor, true
+			}
+		}
+	}
+	return 0, 0, false
+}
+
+func legacyRESTAuthMechanism(config ClientConfig) string {
+	if strings.TrimSpace(config.APIKey) != "" {
+		return "rest-bearer-api-key"
+	}
+	return "rest-basic-password"
+}
+
+func rpcErrorFromWire(method string, wire *trueNASRPCError) *RPCError {
+	rpcErr := &RPCError{Method: method}
+	if wire == nil {
+		return rpcErr
+	}
+	rpcErr.Code = wire.Code
+	rpcErr.Message = strings.TrimSpace(wire.Message)
+	if len(wire.Data) > 0 && string(wire.Data) != "null" {
+		var data map[string]any
+		if json.Unmarshal(wire.Data, &data) == nil {
+			rpcErr.Reason = strings.TrimSpace(readStringAny(data, "reason", "message"))
+			rpcErr.Errname = strings.TrimSpace(readStringAny(data, "errname", "name"))
+		}
+	}
+	return rpcErr
+}
+
+func isMethodUnavailable(err error) bool {
+	var rpcErr *RPCError
+	if !errors.As(err, &rpcErr) {
+		return false
+	}
+	message := strings.ToLower(rpcErr.Message + " " + rpcErr.Reason)
+	return rpcErr.Code == -32601 || strings.Contains(message, "method not found") || strings.Contains(message, "not found")
+}
@@ -0,0 +1,798 @@
+package truenas
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/gorilla/websocket"
+)
+
+type protocolFixtureReply struct {
+	result        any
+	err           *trueNASRPCError
+	notifications []protocolFixtureNotification
+	close         bool
+}
+
+type protocolFixtureNotification struct {
+	method string
+	params any
+}
+
+type protocolFixture struct {
+	server       *httptest.Server
+	sessions     atomic.Int32
+	restRequests atomic.Int32
+}
+
+func newProtocolFixture(
+	t *testing.T,
+	handle func(session int, request trueNASRPCRequest) protocolFixtureReply,
+	handleREST http.HandlerFunc,
+) *protocolFixture {
+	t.Helper()
+
+	fixture := &protocolFixture{}
+	upgrader := websocket.Upgrader{}
+	fixture.server = httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if request.URL.Path != "/api/current" || !websocket.IsWebSocketUpgrade(request) {
+			fixture.restRequests.Add(1)
+			if handleREST == nil {
+				http.NotFound(writer, request)
+				return
+			}
+			handleREST(writer, request)
+			return
+		}
+
+		conn, err := upgrader.Upgrade(writer, request, nil)
+		if err != nil {
+			t.Errorf("upgrade protocol fixture websocket: %v", err)
+			return
+		}
+		defer func() { _ = conn.Close() }()
+		session := int(fixture.sessions.Add(1))
+
+		for {
+			var rpcRequest trueNASRPCRequest
+			if err := conn.ReadJSON(&rpcRequest); err != nil {
+				return
+			}
+			reply := handle(session, rpcRequest)
+			if reply.close {
+				return
+			}
+			response := trueNASRPCResponse{
+				JSONRPC: "2.0",
+				ID:      rpcRequest.ID,
+				Error:   reply.err,
+			}
+			if reply.err == nil {
+				raw, err := json.Marshal(reply.result)
+				if err != nil {
+					t.Errorf("marshal protocol fixture result: %v", err)
+					return
+				}
+				response.Result = raw
+			}
+			if err := conn.WriteJSON(response); err != nil {
+				return
+			}
+			for _, notification := range reply.notifications {
+				params, err := json.Marshal(notification.params)
+				if err != nil {
+					t.Errorf("marshal protocol fixture notification: %v", err)
+					return
+				}
+				if err := conn.WriteJSON(trueNASRPCResponse{
+					JSONRPC: "2.0",
+					Method:  notification.method,
+					Params:  params,
+				}); err != nil {
+					return
+				}
+			}
+		}
+	}))
+	t.Cleanup(fixture.server.Close)
+	return fixture
+}
+
+func protocolFixtureClient(t *testing.T, serverURL string, config ClientConfig) *Client {
+	t.Helper()
+	config.InsecureSkipVerify = true
+	client := mustClientForServer(t, serverURL, config)
+	t.Cleanup(client.Close)
+	return client
+}
+
+func TestJSONRPCTransportUsesLoginExForReadonlyAPIKey(t *testing.T) {
+	var sawLogin atomic.Bool
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		switch request.Method {
+		case "auth.login_ex":
+			params, ok := request.Params.([]any)
+			if !ok || len(params) != 1 {
+				t.Errorf("auth.login_ex params = %#v", request.Params)
+				return protocolFixtureReply{close: true}
+			}
+			login, ok := params[0].(map[string]any)
+			if !ok {
+				t.Errorf("auth.login_ex login payload = %#v", params[0])
+				return protocolFixtureReply{close: true}
+			}
+			if login["mechanism"] != "API_KEY_PLAIN" || login["username"] != "pulse-readonly" || login["api_key"] != "readonly-key" {
+				t.Errorf("auth.login_ex login payload = %#v", login)
+			}
+			sawLogin.Store(true)
+			return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}}
+		case "system.info":
+			return protocolFixtureReply{result: map[string]any{
+				"hostname":       "nas-readonly",
+				"version":        "TrueNAS-SCALE-25.10.4",
+				"system_serial":  "READONLY-1",
+				"uptime_seconds": 42,
+				"physical_cores": 8,
+				"physmem":        16 * 1024 * 1024 * 1024,
+			}}
+		default:
+			t.Errorf("unexpected rpc method %q", request.Method)
+			return protocolFixtureReply{close: true}
+		}
+	}, nil)
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{
+		APIKey:   "readonly-key",
+		Username: "pulse-readonly",
+	})
+	system, err := client.GetSystemInfo(context.Background())
+	if err != nil {
+		t.Fatalf("GetSystemInfo() error = %v", err)
+	}
+	if !sawLogin.Load() || system.Hostname != "nas-readonly" || system.Version != "TrueNAS-SCALE-25.10.4" {
+		t.Fatalf("unexpected readonly system result: %+v", system)
+	}
+	if fixture.restRequests.Load() != 0 {
+		t.Fatalf("modern JSON-RPC session made %d REST requests", fixture.restRequests.Load())
+	}
+	status := client.TransportStatus()
+	if status.Mode != TransportJSONRPC || !status.Connected || status.AuthMechanism != "api-key-plain" || !status.TLS {
+		t.Fatalf("unexpected transport status: %+v", status)
+	}
+}
+
+func TestJSONRPCLegacyAPIKeyLoginRemovalRequestsOwnerUsername(t *testing.T) {
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		if request.Method != "auth.login_with_api_key" {
+			t.Errorf("unexpected rpc method %q", request.Method)
+			return protocolFixtureReply{close: true}
+		}
+		return protocolFixtureReply{err: &trueNASRPCError{
+			Code:    -32601,
+			Message: "Method not found",
+		}}
+	}, nil)
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "upgrade-key"})
+	_, err := client.GetSystemInfo(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "requires the API key owner username") {
+		t.Fatalf("GetSystemInfo() error = %v, want upgrade guidance", err)
+	}
+	if fixture.restRequests.Load() != 0 {
+		t.Fatalf("removed login method made %d REST downgrade requests", fixture.restRequests.Load())
+	}
+}
+
+func TestJSONRPCPermissionFailureDoesNotDowngradeToREST(t *testing.T) {
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		switch request.Method {
+		case "auth.login_ex":
+			return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}}
+		case "system.info":
+			data, _ := json.Marshal(map[string]any{
+				"reason":  "Not authorized",
+				"errname": "EACCES",
+			})
+			return protocolFixtureReply{err: &trueNASRPCError{
+				Code:    -32001,
+				Message: "Method call error",
+				Data:    data,
+			}}
+		default:
+			return protocolFixtureReply{close: true}
+		}
+	}, func(writer http.ResponseWriter, _ *http.Request) {
+		t.Error("REST must not be called after an authoritative JSON-RPC permission error")
+		http.Error(writer, "unexpected", http.StatusInternalServerError)
+	})
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{
+		APIKey:   "readonly-key",
+		Username: "pulse-readonly",
+	})
+	_, err := client.GetSystemInfo(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "EACCES") {
+		t.Fatalf("GetSystemInfo() error = %v, want structured permission failure", err)
+	}
+	if fixture.restRequests.Load() != 0 {
+		t.Fatalf("permission failure made %d REST requests", fixture.restRequests.Load())
+	}
+	status := client.TransportStatus()
+	if status.Mode != TransportJSONRPC || !status.Connected {
+		t.Fatalf("permission failure changed transport authority: %+v", status)
+	}
+}
+
+func TestJSONRPCReadonlyActionDenialIsAuthoritativeAndNotRetried(t *testing.T) {
+	var actionCalls atomic.Int32
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		switch request.Method {
+		case "auth.login_ex":
+			return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}}
+		case "app.start":
+			actionCalls.Add(1)
+			data, _ := json.Marshal(map[string]any{
+				"reason":  "Not authorized",
+				"errname": "EACCES",
+			})
+			return protocolFixtureReply{err: &trueNASRPCError{
+				Code:    -32001,
+				Message: "Method call error",
+				Data:    data,
+			}}
+		default:
+			return protocolFixtureReply{close: true}
+		}
+	}, func(writer http.ResponseWriter, _ *http.Request) {
+		t.Error("REST must not be called after an authoritative action denial")
+		http.Error(writer, "unexpected", http.StatusInternalServerError)
+	})
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{
+		APIKey:   "readonly-key",
+		Username: "pulse-readonly",
+	})
+	err := client.StartApp(context.Background(), "nextcloud")
+	var rpcErr *RPCError
+	if !errors.As(err, &rpcErr) || rpcErr.Errname != "EACCES" || rpcErr.Method != "app.start" {
+		t.Fatalf("StartApp() error = %v, want structured app.start permission denial", err)
+	}
+	if actionCalls.Load() != 1 || fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 {
+		t.Fatalf("actionCalls=%d sessions=%d restRequests=%d, denial was retried or downgraded",
+			actionCalls.Load(), fixture.sessions.Load(), fixture.restRequests.Load())
+	}
+}
+
+func TestTransportNegotiationAllowsOnlyRecognizedLegacyREST(t *testing.T) {
+	t.Run("legacy scale", func(t *testing.T) {
+		server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			switch request.URL.Path {
+			case "/api/current":
+				http.NotFound(writer, request)
+			case "/api/v2.0/system/info":
+				_, _ = writer.Write([]byte(`{"hostname":"legacy","version":"TrueNAS-SCALE-24.10.2","uptime_seconds":1}`))
+			case "/api/v2.0/pool":
+				_, _ = writer.Write([]byte(`[{"id":1,"name":"tank","status":"ONLINE"}]`))
+			default:
+				http.NotFound(writer, request)
+			}
+		}))
+		t.Cleanup(server.Close)
+
+		client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "legacy-key"})
+		pools, err := client.GetPools(context.Background())
+		if err != nil {
+			t.Fatalf("GetPools() legacy error = %v", err)
+		}
+		if len(pools) != 1 || pools[0].Name != "tank" {
+			t.Fatalf("unexpected legacy pools: %+v", pools)
+		}
+		if status := client.TransportStatus(); status.Mode != TransportLegacyREST || status.ApplianceVersion != "TrueNAS-SCALE-24.10.2" {
+			t.Fatalf("unexpected legacy status: %+v", status)
+		}
+	})
+
+	t.Run("legacy core", func(t *testing.T) {
+		server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			switch request.URL.Path {
+			case "/api/current":
+				http.NotFound(writer, request)
+			case "/api/v2.0/system/info":
+				_, _ = writer.Write([]byte(`{"hostname":"core","version":"TrueNAS-13.0-U6.4","uptime_seconds":1}`))
+			case "/api/v2.0/pool":
+				_, _ = writer.Write([]byte(`[{"id":1,"name":"core-tank","status":"ONLINE"}]`))
+			default:
+				http.NotFound(writer, request)
+			}
+		}))
+		t.Cleanup(server.Close)
+
+		client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "legacy-core-key"})
+		pools, err := client.GetPools(context.Background())
+		if err != nil {
+			t.Fatalf("GetPools() CORE error = %v", err)
+		}
+		if len(pools) != 1 || pools[0].Name != "core-tank" {
+			t.Fatalf("unexpected CORE pools: %+v", pools)
+		}
+		if status := client.TransportStatus(); status.Mode != TransportLegacyREST ||
+			status.ApplianceVersion != "TrueNAS-13.0-U6.4" {
+			t.Fatalf("unexpected CORE status: %+v", status)
+		}
+	})
+
+	t.Run("current scale fail closed", func(t *testing.T) {
+		var poolRESTCalls atomic.Int32
+		server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+			switch request.URL.Path {
+			case "/api/current":
+				http.NotFound(writer, request)
+			case "/api/v2.0/system/info":
+				_, _ = writer.Write([]byte(`{"hostname":"current","version":"TrueNAS-SCALE-25.10.4","uptime_seconds":1}`))
+			case "/api/v2.0/pool":
+				poolRESTCalls.Add(1)
+				_, _ = writer.Write([]byte(`[]`))
+			default:
+				http.NotFound(writer, request)
+			}
+		}))
+		t.Cleanup(server.Close)
+
+		client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "current-key"})
+		_, err := client.GetPools(context.Background())
+		if err == nil || !strings.Contains(err.Error(), "refusing REST downgrade") {
+			t.Fatalf("GetPools() error = %v, want fail-closed downgrade error", err)
+		}
+		if poolRESTCalls.Load() != 0 {
+			t.Fatalf("current SCALE used pool REST %d times", poolRESTCalls.Load())
+		}
+	})
+}
+
+func TestJSONRPCCredentialsRequireTLSWithoutSendingAuthentication(t *testing.T) {
+	var rpcRequests atomic.Int32
+	var restRequests atomic.Int32
+	upgrader := websocket.Upgrader{}
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if request.URL.Path != "/api/current" || !websocket.IsWebSocketUpgrade(request) {
+			restRequests.Add(1)
+			http.NotFound(writer, request)
+			return
+		}
+		conn, err := upgrader.Upgrade(writer, request, nil)
+		if err != nil {
+			t.Errorf("upgrade plaintext websocket: %v", err)
+			return
+		}
+		defer func() { _ = conn.Close() }()
+		var rpcRequest trueNASRPCRequest
+		if conn.ReadJSON(&rpcRequest) == nil {
+			rpcRequests.Add(1)
+		}
+	}))
+	t.Cleanup(server.Close)
+
+	client := mustClientForServer(t, server.URL, ClientConfig{
+		APIKey:   "must-not-cross-plaintext",
+		Username: "pulse-readonly",
+	})
+	client.allowInsecureRPC = false
+	t.Cleanup(client.Close)
+	_, err := client.GetPools(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "credentials require TLS") {
+		t.Fatalf("GetPools() error = %v, want TLS requirement", err)
+	}
+	if rpcRequests.Load() != 0 || restRequests.Load() != 0 {
+		t.Fatalf("plaintext negotiation sent rpc=%d REST=%d requests after websocket upgrade",
+			rpcRequests.Load(), restRequests.Load())
+	}
+}
+
+func TestTransportNegotiationIsIsolatedAcrossAppliances(t *testing.T) {
+	modern := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		switch request.Method {
+		case "auth.login_ex":
+			return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}}
+		case "pool.query":
+			return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "modern", "status": "ONLINE"}}}
+		default:
+			return protocolFixtureReply{close: true}
+		}
+	}, nil)
+	legacyServer := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		switch request.URL.Path {
+		case "/api/current":
+			http.NotFound(writer, request)
+		case "/api/v2.0/system/info":
+			_, _ = writer.Write([]byte(`{"hostname":"legacy","version":"TrueNAS-SCALE-24.10.2"}`))
+		case "/api/v2.0/pool":
+			_, _ = writer.Write([]byte(`[{"id":2,"name":"legacy","status":"ONLINE"}]`))
+		default:
+			http.NotFound(writer, request)
+		}
+	}))
+	t.Cleanup(legacyServer.Close)
+
+	modernClient := protocolFixtureClient(t, modern.server.URL, ClientConfig{
+		APIKey:   "modern-key",
+		Username: "pulse-readonly",
+	})
+	legacyClient := protocolFixtureClient(t, legacyServer.URL, ClientConfig{APIKey: "legacy-key"})
+
+	modernPools, modernErr := modernClient.GetPools(context.Background())
+	legacyPools, legacyErr := legacyClient.GetPools(context.Background())
+	if modernErr != nil || legacyErr != nil {
+		t.Fatalf("multi-appliance polls failed: modern=%v legacy=%v", modernErr, legacyErr)
+	}
+	if len(modernPools) != 1 || modernPools[0].Name != "modern" ||
+		len(legacyPools) != 1 || legacyPools[0].Name != "legacy" {
+		t.Fatalf("unexpected isolated inventories: modern=%+v legacy=%+v", modernPools, legacyPools)
+	}
+	modernStatus := modernClient.TransportStatus()
+	legacyStatus := legacyClient.TransportStatus()
+	if modernStatus.Mode != TransportJSONRPC || modernStatus.Endpoint == legacyStatus.Endpoint ||
+		legacyStatus.Mode != TransportLegacyREST {
+		t.Fatalf("transport decisions leaked across clients: modern=%+v legacy=%+v", modernStatus, legacyStatus)
+	}
+}
+
+func TestNegotiatedJSONRPCTransportNeverRenegotiatesToREST(t *testing.T) {
+	var websocketAttempts atomic.Int32
+	var restRequests atomic.Int32
+	upgrader := websocket.Upgrader{}
+	server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		switch request.URL.Path {
+		case "/api/current":
+			if websocketAttempts.Add(1) > 1 {
+				http.NotFound(writer, request)
+				return
+			}
+			conn, err := upgrader.Upgrade(writer, request, nil)
+			if err != nil {
+				t.Errorf("upgrade initial websocket: %v", err)
+				return
+			}
+			defer func() { _ = conn.Close() }()
+			for {
+				var rpcRequest trueNASRPCRequest
+				if err := conn.ReadJSON(&rpcRequest); err != nil {
+					return
+				}
+				var result any
+				switch rpcRequest.Method {
+				case "auth.login_with_api_key":
+					result = true
+				case "pool.query":
+					result = []map[string]any{{"id": 1, "name": "modern", "status": "ONLINE"}}
+				default:
+					t.Errorf("unexpected rpc method %q", rpcRequest.Method)
+					return
+				}
+				raw, err := json.Marshal(result)
+				if err != nil {
+					t.Errorf("marshal rpc result: %v", err)
+					return
+				}
+				if err := conn.WriteJSON(trueNASRPCResponse{
+					JSONRPC: "2.0",
+					ID:      rpcRequest.ID,
+					Result:  raw,
+				}); err != nil {
+					return
+				}
+			}
+		case "/api/v2.0/system/info", "/api/v2.0/pool":
+			restRequests.Add(1)
+			_, _ = writer.Write([]byte(`{"version":"TrueNAS-SCALE-24.10.2"}`))
+		default:
+			http.NotFound(writer, request)
+		}
+	}))
+	t.Cleanup(server.Close)
+
+	client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "modern-key"})
+	if _, err := client.GetPools(context.Background()); err != nil {
+		t.Fatalf("initial GetPools() error = %v", err)
+	}
+
+	client.rpcMu.Lock()
+	client.closeRPCLocked()
+	client.rpcMu.Unlock()
+
+	_, err := client.GetPools(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "transport is immutable") {
+		t.Fatalf("GetPools() reconnect error = %v, want immutable transport failure", err)
+	}
+	if restRequests.Load() != 0 {
+		t.Fatalf("negotiated JSON-RPC transport made %d REST downgrade requests", restRequests.Load())
+	}
+	status := client.TransportStatus()
+	if status.Mode != TransportJSONRPC || status.Connected {
+		t.Fatalf("immutable reconnect changed transport authority: %+v", status)
+	}
+}
+
+func TestJSONRPCReadReconnectsButActionIsNeverReplayed(t *testing.T) {
+	t.Run("read reconnect", func(t *testing.T) {
+		var poolCalls atomic.Int32
+		fixture := newProtocolFixture(t, func(session int, request trueNASRPCRequest) protocolFixtureReply {
+			if request.Method == "auth.login_with_api_key" {
+				return protocolFixtureReply{result: true}
+			}
+			if request.Method != "pool.query" {
+				return protocolFixtureReply{close: true}
+			}
+			call := poolCalls.Add(1)
+			if session == 1 && call == 2 {
+				return protocolFixtureReply{close: true}
+			}
+			return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE"}}}
+		}, nil)
+
+		client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "legacy-auth-key"})
+		if _, err := client.GetPools(context.Background()); err != nil {
+			t.Fatalf("first GetPools() error = %v", err)
+		}
+		if _, err := client.GetPools(context.Background()); err != nil {
+			t.Fatalf("reconnected GetPools() error = %v", err)
+		}
+		if fixture.sessions.Load() != 2 || poolCalls.Load() != 3 {
+			t.Fatalf("sessions=%d poolCalls=%d, want one replay on a new session", fixture.sessions.Load(), poolCalls.Load())
+		}
+		if status := client.TransportStatus(); status.Reconnects != 1 || status.Mode != TransportJSONRPC {
+			t.Fatalf("unexpected reconnect status: %+v", status)
+		}
+	})
+
+	t.Run("action outcome unknown", func(t *testing.T) {
+		var actionCalls atomic.Int32
+		fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+			if request.Method == "auth.login_with_api_key" {
+				return protocolFixtureReply{result: true}
+			}
+			if request.Method == "app.start" {
+				actionCalls.Add(1)
+				return protocolFixtureReply{close: true}
+			}
+			return protocolFixtureReply{close: true}
+		}, nil)
+
+		client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "app-write-key"})
+		err := client.StartApp(context.Background(), "nextcloud")
+		if err == nil || !strings.Contains(err.Error(), "outcome is unknown") {
+			t.Fatalf("StartApp() error = %v, want non-replay outcome", err)
+		}
+		if actionCalls.Load() != 1 || fixture.sessions.Load() != 1 {
+			t.Fatalf("actionCalls=%d sessions=%d, action was replayed", actionCalls.Load(), fixture.sessions.Load())
+		}
+	})
+}
+
+func TestJSONRPCQueryReturnsCompleteUnpaginatedInventory(t *testing.T) {
+	poolFixture := make([]map[string]any, 0, 225)
+	for i := 0; i < 225; i++ {
+		poolFixture = append(poolFixture, map[string]any{"id": i + 1, "name": "pool-" + strings.Repeat("x", i%4), "status": "ONLINE"})
+	}
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		if request.Method == "auth.login_with_api_key" {
+			return protocolFixtureReply{result: true}
+		}
+		if request.Method != "pool.query" {
+			return protocolFixtureReply{close: true}
+		}
+		params, ok := request.Params.([]any)
+		if !ok || len(params) != 2 {
+			t.Errorf("pool.query params = %#v", request.Params)
+		} else if options, ok := params[1].(map[string]any); !ok || len(options) != 0 {
+			t.Errorf("pool.query options must request the authoritative complete set, got %#v", params[1])
+		}
+		return protocolFixtureReply{result: poolFixture}
+	}, nil)
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "readonly-key"})
+	pools, err := client.GetPools(context.Background())
+	if err != nil {
+		t.Fatalf("GetPools() error = %v", err)
+	}
+	if len(pools) != len(poolFixture) {
+		t.Fatalf("GetPools() returned %d pools, want %d", len(pools), len(poolFixture))
+	}
+}
+
+func TestJSONRPCFetchSnapshotKeepsAllModernInventoryOnOneTransport(t *testing.T) {
+	methodCalls := make(map[string]int)
+	var methodCallsMu sync.Mutex
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		methodCallsMu.Lock()
+		methodCalls[request.Method]++
+		methodCallsMu.Unlock()
+
+		switch request.Method {
+		case "auth.login_ex":
+			return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}}
+		case "system.info":
+			return protocolFixtureReply{result: map[string]any{
+				"hostname":       "inventory-nas",
+				"version":        "TrueNAS-SCALE-25.10.4",
+				"system_serial":  "INVENTORY-1",
+				"uptime_seconds": 60,
+				"physical_cores": 4,
+				"physmem":        8 * 1024 * 1024 * 1024,
+			}}
+		case "reporting.get_data":
+			return protocolFixtureReply{result: []any{}}
+		case "core.subscribe":
+			params, _ := request.Params.([]any)
+			event, _ := params[0].(string)
+			if strings.HasPrefix(event, "reporting.realtime:") {
+				return protocolFixtureReply{
+					result: "sub-realtime",
+					notifications: []protocolFixtureNotification{{
+						method: "collection_update",
+						params: map[string]any{
+							"collection": event,
+							"fields":     map[string]any{"cpu": map[string]any{"usage": 12}},
+						},
+					}},
+				}
+			}
+			if strings.HasPrefix(event, "app.stats:") {
+				return protocolFixtureReply{
+					result: "sub-app-stats",
+					notifications: []protocolFixtureNotification{{
+						method: "collection_update",
+						params: map[string]any{
+							"collection": event,
+							"fields":     []any{},
+						},
+					}},
+				}
+			}
+			t.Errorf("unexpected subscription event %q", event)
+			return protocolFixtureReply{close: true}
+		case "core.unsubscribe":
+			return protocolFixtureReply{result: nil}
+		case "pool.query", "pool.dataset.query", "disk.query", "alert.list",
+			"service.query", "app.query", "vm.query", "sharing.smb.query",
+			"sharing.nfs.query", "zfs.resource.snapshot.query", "replication.query":
+			return protocolFixtureReply{result: []any{}}
+		case "boot.get_state":
+			return protocolFixtureReply{result: map[string]any{
+				"id": "boot-pool", "name": "boot-pool", "status": "ONLINE",
+			}}
+		default:
+			t.Errorf("unexpected rpc method %q", request.Method)
+			return protocolFixtureReply{close: true}
+		}
+	}, nil)
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{
+		APIKey:   "readonly-key",
+		Username: "pulse-readonly",
+	})
+	snapshot, err := client.FetchSnapshot(context.Background())
+	if err != nil {
+		t.Fatalf("FetchSnapshot() error = %v", err)
+	}
+	if snapshot.System.Hostname != "inventory-nas" || snapshot.System.CPUPercent != 12 {
+		t.Fatalf("unexpected snapshot system: %+v", snapshot.System)
+	}
+
+	for _, method := range []string{
+		"system.info",
+		"pool.query",
+		"boot.get_state",
+		"pool.dataset.query",
+		"disk.query",
+		"alert.list",
+		"service.query",
+		"app.query",
+		"vm.query",
+		"sharing.smb.query",
+		"sharing.nfs.query",
+		"zfs.resource.snapshot.query",
+		"replication.query",
+	} {
+		methodCallsMu.Lock()
+		count := methodCalls[method]
+		methodCallsMu.Unlock()
+		if count != 1 {
+			t.Errorf("%s calls = %d, want exactly one", method, count)
+		}
+	}
+	methodCallsMu.Lock()
+	subscribeCalls := methodCalls["core.subscribe"]
+	unsubscribeCalls := methodCalls["core.unsubscribe"]
+	methodCallsMu.Unlock()
+	if subscribeCalls != 2 || unsubscribeCalls != 2 {
+		t.Fatalf("subscription lifecycle calls: subscribe=%d unsubscribe=%d, want 2/2", subscribeCalls, unsubscribeCalls)
+	}
+	if fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 {
+		t.Fatalf("snapshot sessions=%d REST=%d, want one modern session and no REST",
+			fixture.sessions.Load(), fixture.restRequests.Load())
+	}
+}
+
+func TestJSONRPCMalformedSubscriptionEventDiscardsSession(t *testing.T) {
+	fixture := newProtocolFixture(t, func(session int, request trueNASRPCRequest) protocolFixtureReply {
+		switch request.Method {
+		case "auth.login_with_api_key":
+			return protocolFixtureReply{result: true}
+		case "core.subscribe":
+			return protocolFixtureReply{
+				result: "sub-malformed",
+				notifications: []protocolFixtureNotification{{
+					method: "collection_update",
+					params: map[string]any{
+						"collection": "app.stats:{\"interval\":2}",
+						"fields":     "not-an-array",
+					},
+				}},
+			}
+		case "pool.query":
+			if session != 2 {
+				t.Errorf("pool.query reused malformed subscription session %d", session)
+			}
+			return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE"}}}
+		default:
+			t.Errorf("unexpected rpc method %q", request.Method)
+			return protocolFixtureReply{close: true}
+		}
+	}, nil)
+
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "readonly-key"})
+	if _, err := client.GetAppStats(context.Background()); err == nil ||
+		!strings.Contains(err.Error(), "decode truenas app.stats notification") {
+		t.Fatalf("GetAppStats() error = %v, want malformed event error", err)
+	}
+	pools, err := client.GetPools(context.Background())
+	if err != nil {
+		t.Fatalf("GetPools() after discarded stream error = %v", err)
+	}
+	if len(pools) != 1 || pools[0].Name != "tank" || fixture.sessions.Load() != 2 {
+		t.Fatalf("unexpected post-discard state: pools=%+v sessions=%d", pools, fixture.sessions.Load())
+	}
+	if status := client.TransportStatus(); status.Reconnects != 1 {
+		t.Fatalf("discarded stream reconnect was not observable: %+v", status)
+	}
+}
+
+func TestTransportDiagnosticsAreConnectionLocalAndSecretFree(t *testing.T) {
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		if request.Method == "auth.login_with_api_key" {
+			data, _ := json.Marshal(map[string]any{
+				"reason":  "invalid credential never-expose-this-key",
+				"errname": "EINVAL",
+			})
+			return protocolFixtureReply{err: &trueNASRPCError{
+				Code:    -32001,
+				Message: "Method call error",
+				Data:    data,
+			}}
+		}
+		return protocolFixtureReply{close: true}
+	}, nil)
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "never-expose-this-key"})
+
+	if _, err := client.GetPools(context.Background()); err == nil {
+		t.Fatal("expected authentication failure")
+	}
+	status := client.TransportStatus()
+	encoded, err := json.Marshal(status)
+	if err != nil {
+		t.Fatalf("Marshal(status) error = %v", err)
+	}
+	if strings.Contains(string(encoded), "never-expose-this-key") {
+		t.Fatalf("transport status exposed API key: %s", encoded)
+	}
+	if status.Mode != TransportUnknown {
+		t.Fatalf("failed authentication selected a transport: %+v", status)
+	}
+}