Add support for duration strings, any type, field mask and fix a bug around wrapper known type (#5)

* Add support for duration strings, fix a bug around wrapper known type

* Add oneof test for round trip

* Add support for any and fieldmask

* Add field mask with array ui

* Add drop down support for Any fields

* Add support for any type

* Add messageType labels

* Refactor

* Restore test

Author: pranavpsawant

backend/src/grpc/connect/connectInvoker.ts

@@ -11,8 +11,11 @@ import {
   type DescMethodUnary,
   type JsonValue as BufJsonValue,
   type MessageInitShape,
+  type Registry,
 } from '@bufbuild/protobuf'
+import type { FileDescriptorSet } from '@bufbuild/protobuf/wkt'
 import { Code, ConnectError, createClient, type CallOptions } from '@connectrpc/connect'
+import { buildDynamicRegistry } from './dynamicRegistry.js'
 import type { JsonValue } from '@grpc-studio/shared'
 import configManager from '../../config/configManager.js'
 import type { FormattedGrpcError, StreamCallbacks, StreamHandle } from '../../types/index.js'
@@ -46,40 +49,51 @@ export type OutboundHeaders = Record<string, string>
 export class ConnectInvoker {
   constructor(private readonly transportProvider: ConnectTransportProvider = connectTransportProvider) {}
 
-  async invokeUnary(method: DescMethodUnary, request: JsonValue | undefined, headers: OutboundHeaders): Promise<JsonValue> {
+  async invokeUnary(
+    method: DescMethodUnary,
+    request: JsonValue | undefined,
+    headers: OutboundHeaders,
+    fileDescriptorSet: FileDescriptorSet
+  ): Promise<JsonValue> {
+    const registry = buildDynamicRegistry(fileDescriptorSet)
     const response = await this.getClientMethod<UnaryClientMethod>(method)(
-      toProtoRequest(method.input, request),
+      toProtoRequest(method.input, request, registry),
       this.buildCallOptions(configManager.getClientConfig().rpc.unaryDeadlineMs, headers)
     )
 
-    return toJsonResponse(method.output, response)
+    return toJsonResponse(method.output, response, registry)
   }
 
   async startServerStream(
     method: DescMethodServerStreaming,
     request: JsonValue | undefined,
     callbacks: StreamCallbacks,
-    headers: OutboundHeaders
+    headers: OutboundHeaders,
+    fileDescriptorSet: FileDescriptorSet
   ): Promise<StreamHandle> {
+    const registry = buildDynamicRegistry(fileDescriptorSet)
     return this.startResponseStream(
       callbacks,
       async (signal) => this.getClientMethod<ServerStreamingClientMethod>(method)(
-        toProtoRequest(method.input, request),
+        toProtoRequest(method.input, request, registry),
         this.buildCallOptions(configManager.getClientConfig().rpc.streamDeadlineMs, headers, signal)
       ),
-      method.output
+      method.output,
+      registry
     )
   }
 
   async startClientStream(
     method: DescMethodClientStreaming,
     requests: RequestStream,
     callbacks: StreamCallbacks,
-    headers: OutboundHeaders
+    headers: OutboundHeaders,
+    fileDescriptorSet: FileDescriptorSet
   ): Promise<StreamHandle> {
     const abortController = new AbortController()
+    const registry = buildDynamicRegistry(fileDescriptorSet)
 
-    this.pumpClientStream(method, requests, callbacks, headers, abortController.signal)
+    this.pumpClientStream(method, requests, callbacks, headers, abortController.signal, registry)
       .catch((error) => {
         // Catch unhandled errors from pump to prevent unhandled promise rejections
         if (!abortController.signal.aborted) {
@@ -98,26 +112,30 @@ export class ConnectInvoker {
     method: DescMethodBiDiStreaming,
     requests: RequestStream,
     callbacks: StreamCallbacks,
-    headers: OutboundHeaders
+    headers: OutboundHeaders,
+    fileDescriptorSet: FileDescriptorSet
   ): Promise<StreamHandle> {
+    const registry = buildDynamicRegistry(fileDescriptorSet)
     return this.startResponseStream(
       callbacks,
       async (signal) => this.getClientMethod<BidiStreamingClientMethod>(method)(
-        toConnectRequestStream(method, requests),
+        toConnectRequestStream(method, requests, registry),
         this.buildCallOptions(configManager.getClientConfig().rpc.streamDeadlineMs, headers, signal)
       ),
-      method.output
+      method.output,
+      registry
     )
   }
 
   private startResponseStream(
     callbacks: StreamCallbacks,
     createStream: (signal: AbortSignal) => Promise<AsyncIterable<unknown>>,
-    responseDesc: DescMessage
+    responseDesc: DescMessage,
+    registry: Registry
   ): StreamHandle {
     const abortController = new AbortController()
 
-    this.pumpResponseStream(createStream, responseDesc, callbacks, abortController.signal)
+    this.pumpResponseStream(createStream, responseDesc, callbacks, abortController.signal, registry)
       .catch((error) => {
         // Catch unhandled errors from pump to prevent unhandled promise rejections
         if (!abortController.signal.aborted) {
@@ -136,13 +154,14 @@ export class ConnectInvoker {
     createStream: (signal: AbortSignal) => Promise<AsyncIterable<unknown>>,
     responseDesc: DescMessage,
     callbacks: StreamCallbacks,
-    signal: AbortSignal
+    signal: AbortSignal,
+    registry: Registry
   ): Promise<void> {
     try {
       const stream = await createStream(signal)
 
       for await (const message of stream) {
-        callbacks.onData(toJsonResponse(responseDesc, message))
+        callbacks.onData(toJsonResponse(responseDesc, message, registry))
       }
 
       callbacks.onEnd()
@@ -158,15 +177,16 @@ export class ConnectInvoker {
     requests: RequestStream,
     callbacks: StreamCallbacks,
     headers: OutboundHeaders,
-    signal: AbortSignal
+    signal: AbortSignal,
+    registry: Registry
   ): Promise<void> {
     try {
       const response = await this.getClientMethod<ClientStreamingClientMethod>(method)(
-        toConnectRequestStream(method, requests),
+        toConnectRequestStream(method, requests, registry),
         this.buildCallOptions(configManager.getClientConfig().rpc.streamDeadlineMs, headers, signal)
       )
 
-      callbacks.onData(toJsonResponse(method.output, response))
+      callbacks.onData(toJsonResponse(method.output, response, registry))
       callbacks.onEnd()
     } catch (error) {
       if (!signal.aborted) {
@@ -206,19 +226,20 @@ export function formatConnectError(error: unknown): FormattedGrpcError {
   }
 }
 
-function toProtoRequest(desc: DescMessage, data: JsonValue | undefined): MessageInitShape<DescMessage> {
-  return fromJson(desc, (data ?? {}) as BufJsonValue, { ignoreUnknownFields: true }) as MessageInitShape<DescMessage>
+function toProtoRequest(desc: DescMessage, data: JsonValue | undefined, registry: Registry): MessageInitShape<DescMessage> {
+  return fromJson(desc, (data ?? {}) as BufJsonValue, { ignoreUnknownFields: true, registry }) as MessageInitShape<DescMessage>
 }
 
-function toJsonResponse(desc: DescMessage, message: unknown): JsonValue {
-  return toJson(desc, message as never, { alwaysEmitImplicit: true, useProtoFieldName: true }) as JsonValue
+function toJsonResponse(desc: DescMessage, message: unknown, registry: Registry): JsonValue {
+  return toJson(desc, message as never, { alwaysEmitImplicit: true, useProtoFieldName: true, registry }) as JsonValue
 }
 
 async function* toConnectRequestStream(
   method: DescMethod,
-  requests: RequestStream
+  requests: RequestStream,
+  registry: Registry
 ): AsyncIterable<MessageInitShape<DescMessage>> {
   for await (const request of requests) {
-    yield toProtoRequest(method.input, request)
+    yield toProtoRequest(method.input, request, registry)
   }
 }

backend/src/grpc/connect/dynamicRegistry.ts

@@ -0,0 +1,55 @@
+// Copyright (c) 2026 Electronic Arts Inc. All rights reserved.
+
+/**
+ * Dynamic registry builder that combines well-known types with reflected schemas.
+ * This enables google.protobuf.Any fields to contain custom message types discovered
+ * via reflection, not just WKTs.
+ */
+
+import { createRegistry, createFileRegistry, type DescMessage, type Registry } from '@bufbuild/protobuf'
+import * as wkt from '@bufbuild/protobuf/wkt'
+import type { FileDescriptorSet } from '@bufbuild/protobuf/wkt'
+import logger from '../../utils/logger.js'
+
+const registryLogger = logger.child({ module: 'dynamic-registry' })
+
+// Static registry containing all well-known types
+const wktRegistry: Registry = createRegistry(
+  ...(Object.values(wkt).filter(
+    (v) => typeof v === 'object' && v !== null && 'typeName' in v && 'fields' in v
+  ) as DescMessage[])
+)
+
+/**
+ * Builds a Registry containing both WKTs and all message types from the given FileDescriptorSet.
+ * This registry can be passed to fromJson/toJson so google.protobuf.Any fields can contain
+ * custom types discovered through reflection.
+ */
+export function buildDynamicRegistry(fileDescriptorSet: FileDescriptorSet): Registry {
+  const messageDescriptors: DescMessage[] = []
+
+  // Parse the raw FileDescriptorSet into a FileRegistry to get DescFile/DescMessage objects
+  const fileRegistry = createFileRegistry(fileDescriptorSet)
+
+  // Extract all message descriptors from the parsed FileRegistry
+  for (const file of fileRegistry.files) {
+    for (const message of file.messages) {
+      messageDescriptors.push(message)
+    }
+  }
+
+  registryLogger.debug('Building dynamic registry', {
+    wktCount: Array.from(wktRegistry).length,
+    reflectedCount: messageDescriptors.length,
+  })
+
+  // Create a combined registry with WKTs + reflected types
+  return createRegistry(...Array.from(wktRegistry), ...messageDescriptors)
+}
+
+/**
+ * Returns the base WKT-only registry for cases where no FileRegistry is available.
+ */
+export function getWktRegistry(): Registry {
+  return wktRegistry
+}

backend/src/repositories/reflectionSchemaRepository.ts

@@ -5,8 +5,8 @@
  * through gRPC reflection.
  */
 
-import { createFileRegistry, type FileRegistry } from '@bufbuild/protobuf'
-import type { FileDescriptorSet } from '@bufbuild/protobuf/wkt'
+import { createFileRegistry, type FileRegistry, create } from '@bufbuild/protobuf'
+import { FileDescriptorSetSchema, type FileDescriptorSet } from '@bufbuild/protobuf/wkt'
 import * as reflectionClient from '../grpc/reflection/reflectionClient.js'
 import * as descriptorSetFromReflection from '../grpc/reflection/descriptorSetFromReflection.js'
 import * as reflectionMetadataCache from '../cache/reflectionMetadataCache.js'
@@ -123,6 +123,32 @@ export class ReflectionSchemaRepository {
     }
   }
 
+  async getAllFileDescriptorSet(): Promise<FileDescriptorSet> {
+    // Build a FileDescriptorSet containing ALL files from ALL services.
+    // This is used for building a comprehensive registry for Any field decoding,
+    // since Any can contain messages from any reflected service.
+    const services = await this.listServices()
+    const allDescriptorSets: FileDescriptorSet[] = []
+
+    for (const serviceName of services) {
+      try {
+        const descriptorSet = await this.getFileDescriptorSet(serviceName)
+        allDescriptorSets.push(descriptorSet)
+      } catch (error) {
+        repositoryLogger.warn('Failed to get descriptor set for service, skipping', {
+          serviceName,
+          error: error instanceof Error ? error.message : String(error),
+        })
+      }
+    }
+
+    // Merge all files from all descriptor sets
+    const allFiles = allDescriptorSets.flatMap(ds => ds.file)
+
+    // Create a proper FileDescriptorSet instance using @bufbuild/protobuf's create helper
+    return create(FileDescriptorSetSchema, { file: allFiles })
+  }
+
   clearCache(): void {
     serviceNamesCache.clear()
     fileDescriptorSetCache.clear()

backend/src/services/grpcMethodInvokerService.ts

@@ -22,7 +22,7 @@ const VALID_NAME_RE = /^[a-zA-Z_][\w.]*$/
 
 export class GrpcMethodInvokerService {
   constructor(
-    private readonly schemaRepository: Pick<ReflectionSchemaRepository, 'getFileRegistry'> = reflectionSchemaRepository,
+    private readonly schemaRepository: Pick<ReflectionSchemaRepository, 'getFileRegistry' | 'getAllFileDescriptorSet'> = reflectionSchemaRepository,
     private readonly transportProvider: ConnectTransportProvider = connectTransportProvider,
     private readonly connectInvoker: ConnectInvoker = new ConnectInvoker(transportProvider)
   ) {}
@@ -34,12 +34,14 @@ export class GrpcMethodInvokerService {
   async invokeUnary(serviceName: string, methodName: string, request: JsonValue | undefined): Promise<UnaryResult> {
     return instrumentUnaryCall(serviceName, methodName, async () => {
       try {
+        // Use global descriptor set so Any fields can contain types from any reflected service
+        const fileDescriptorSet = await this.schemaRepository.getAllFileDescriptorSet()
         const method = await this.resolveMethod(serviceName, methodName)
         assertUnaryMethod(method)
         const headers = await headerManager.getOutboundHeaders(userContextMiddleware.getCurrentUserContext())
         return {
           success: true,
-          data: await this.connectInvoker.invokeUnary(method, request, headers),
+          data: await this.connectInvoker.invokeUnary(method, request, headers, fileDescriptorSet),
           completedAtMs: Date.now(),
         }
       } catch (error) {
@@ -58,10 +60,11 @@ export class GrpcMethodInvokerService {
   ): Promise<StreamHandle> {
     return instrumentStreamCall(serviceName, methodName, 'server_streaming', callbacks, async (wrappedCallbacks) => {
       try {
+        const fileDescriptorSet = await this.schemaRepository.getAllFileDescriptorSet()
         const method = await this.resolveMethod(serviceName, methodName)
         assertServerStreamingMethod(method)
         const headers = await headerManager.getOutboundHeaders(userContextMiddleware.getCurrentUserContext())
-        return await this.connectInvoker.startServerStream(method, request, wrappedCallbacks, headers)
+        return await this.connectInvoker.startServerStream(method, request, wrappedCallbacks, headers, fileDescriptorSet)
       } catch (error) {
         wrappedCallbacks.onError(formatConnectError(error))
         return noopStreamHandle()
@@ -77,10 +80,11 @@ export class GrpcMethodInvokerService {
   ): Promise<StreamHandle> {
     return instrumentStreamCall(serviceName, methodName, 'client_streaming', callbacks, async (wrappedCallbacks) => {
       try {
+        const fileDescriptorSet = await this.schemaRepository.getAllFileDescriptorSet()
         const method = await this.resolveMethod(serviceName, methodName)
         assertClientStreamingMethod(method)
         const headers = await headerManager.getOutboundHeaders(userContextMiddleware.getCurrentUserContext())
-        return await this.connectInvoker.startClientStream(method, requests, wrappedCallbacks, headers)
+        return await this.connectInvoker.startClientStream(method, requests, wrappedCallbacks, headers, fileDescriptorSet)
       } catch (error) {
         wrappedCallbacks.onError(formatConnectError(error))
         return noopStreamHandle()
@@ -96,10 +100,11 @@ export class GrpcMethodInvokerService {
   ): Promise<StreamHandle> {
     return instrumentStreamCall(serviceName, methodName, 'bidi_streaming', callbacks, async (wrappedCallbacks) => {
       try {
+        const fileDescriptorSet = await this.schemaRepository.getAllFileDescriptorSet()
         const method = await this.resolveMethod(serviceName, methodName)
         assertBidiStreamingMethod(method)
         const headers = await headerManager.getOutboundHeaders(userContextMiddleware.getCurrentUserContext())
-        return await this.connectInvoker.startBidiStream(method, requests, wrappedCallbacks, headers)
+        return await this.connectInvoker.startBidiStream(method, requests, wrappedCallbacks, headers, fileDescriptorSet)
       } catch (error) {
         wrappedCallbacks.onError(formatConnectError(error))
         return noopStreamHandle()