Keep timeout fixtures specific to the stalled RPC surface

The first snapshot control also stalled the unrelated app-stat stream. Model its valid empty reply and the proper stream field shapes in the two-connection poller fixture without weakening timeout, session, inventory or recovery assertions. Clarify that initial transport negotiation is bounded separately from each serialized RPC operation; runtime timeout policy is unchanged.

Contract-Neutral: Correct synthetic fixture responses and clarify the existing budget description; no runtime or wire behaviour changes.
Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -135,10 +135,11 @@ This proves ingestion behaviour, not reporter installation or release delivery.
 
 ### TrueNAS RPC operation budgets and poll isolation
 
-The configured client timeout (30 seconds by default) bounds each logical
-JSON-RPC operation, including its session-lock wait, handshake/authentication,
+The configured client timeout (30 seconds by default) bounds each serialized
+RPC operation, including its session-lock wait, any inline authentication,
 method exchange or subscription, and the permitted read retry/backoff. They
-share one budget; a shorter caller deadline remains effective. A cancelled
+share one budget; initial transport negotiation before a method is separately
+bounded by the same timeout. A shorter caller deadline remains effective. A cancelled
 waiter neither dispatches a request nor alters the current owner's socket or
 transport status. RPC/stream readers receive the bounded context, and a
 timed-out socket is discarded before a subsequent operation authenticates a

internal/monitoring/truenas_poller_test.go

@@ -1893,8 +1893,9 @@ func TestTrueNASPollerUnresponsiveRPCDoesNotFreezeOtherConnections(t *testing.T)
 			defer conn.Close()
 			for {
 				var request struct {
-					ID     int64  `json:"id"`
-					Method string `json:"method"`
+					ID     int64             `json:"id"`
+					Method string            `json:"method"`
+					Params []json.RawMessage `json:"params"`
 				}
 				if err := conn.ReadJSON(&request); err != nil {
 					return
@@ -1917,9 +1918,18 @@ func TestTrueNASPollerUnresponsiveRPCDoesNotFreezeOtherConnections(t *testing.T)
 					return
 				}
 				if request.Method == "core.subscribe" {
+					var event string
+					if len(request.Params) != 1 || json.Unmarshal(request.Params[0], &event) != nil {
+						t.Error("subscription did not supply one event")
+						return
+					}
+					var fields any = map[string]any{"cpu": map[string]any{"usage": 12}}
+					if strings.HasPrefix(event, "app.stats:") {
+						fields = []any{}
+					}
 					if err := conn.WriteJSON(map[string]any{
 						"jsonrpc": "2.0", "method": "collection_update",
-						"params": map[string]any{"collection": "reporting.realtime", "fields": map[string]any{"cpu": map[string]any{"usage": 12}}},
+						"params": map[string]any{"collection": event, "fields": fields},
 					}); err != nil {
 						return
 					}

internal/truenas/transport.go

@@ -45,7 +45,7 @@ func (m *rpcSessionMutex) Lock()   { _ = m.LockContext(context.Background()) }
 func (m *rpcSessionMutex) Unlock() { <-m.token }
 
 // rpcOperationContext applies the same configured timeout as HTTP requests to
-// a whole RPC operation: lock wait, negotiation/authentication, exchange or
+// a serialized RPC/negotiation operation: lock wait, inline authentication, exchange or
 // subscription, and its one permitted read retry share a single budget. A
 // shorter caller deadline is never extended. Keepalive has its own lifetime.
 func (c *Client) rpcOperationContext(ctx context.Context) (context.Context, context.CancelFunc) {

internal/truenas/transport_test.go

@@ -1386,6 +1386,15 @@ func TestJSONRPCSnapshotContinuesAfterTelemetryTimeout(t *testing.T) {
 		case "system.info":
 			return protocolFixtureReply{result: map[string]any{"hostname": "inventory-timeout", "version": "TrueNAS-SCALE-25.10.7", "system_serial": "FIXTURE-1", "physmem": 1024}}
 		case "core.subscribe":
+			params, _ := request.Params.([]any)
+			event, _ := params[0].(string)
+			if strings.HasPrefix(event, "app.stats:") {
+				// Only live system telemetry is silent in this fixture. App
+				// inventory's separate best-effort subscription still replies.
+				return protocolFixtureReply{result: "empty-app-stats", notifications: []protocolFixtureNotification{{
+					method: "collection_update", params: map[string]any{"collection": event, "fields": []any{}},
+				}}}
+			}
 			return protocolFixtureReply{result: "silent-realtime"}
 		case "pool.query":
 			return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE"}}}