fix(oci-streaming): normalize absent pool settings and filters

Author: Bill Leoutsakos

apps/sim/blocks/blocks/oci_streaming.ts

@@ -882,7 +882,7 @@ export const OciStreamingBlock: BlockConfig<OciStreamingResponse> = {
           'ociRegion',
           'requestId',
         ]) {
-          result[key] = params[key] === '' ? undefined : params[key]
+          result[key] = params[key] === '' || params[key] == null ? undefined : params[key]
         }
         result.partitions = parseOptionalNumberInput(params.partitions, 'partitions')
         result.retentionInHours = parseOptionalNumberInput(

apps/sim/lib/internal/oci-streaming/execute-tool.test.ts

@@ -15,6 +15,9 @@ vi.mock('@/lib/internal/oci/client.server', () => ({ createOciClient: mocks.crea
 import { AuthType } from '@/lib/auth/hybrid'
 import { executeOciStreamingTool } from '@/lib/internal/oci-streaming/execute-tool'
 import type { InternalToolOperationCall } from '@/lib/internal/tool-operations/types'
+import { OciStreamingBlock } from '@/blocks/blocks/oci_streaming'
+import { ociStreamingListStreamPoolsTool } from '@/tools/oci_streaming/list_stream_pools'
+import type { OciStreamingListStreamPoolsParams } from '@/tools/oci_streaming/types'
 
 function call(overrides: Partial<InternalToolOperationCall> = {}): InternalToolOperationCall {
   return {
@@ -52,6 +55,38 @@ describe('OCI Streaming trusted credential boundary', () => {
     })
   })
 
+  it.each([null, ''])(
+    'omits blank workflow filters after merging the block patch: %s',
+    async (blank) => {
+      const inputs = {
+        operation: 'oci_streaming_list_stream_pools',
+        ociCredential: 'supplied-reference',
+        compartmentId: 'compartment-1',
+        name: blank,
+        page: blank,
+        lifecycleState: blank,
+        sortBy: blank,
+        sortOrder: blank,
+      }
+      const transformed = OciStreamingBlock.tools.config!.params!(inputs)
+      const operationInput = ociStreamingListStreamPoolsTool.operation.input({
+        ...inputs,
+        ...transformed,
+      } as OciStreamingListStreamPoolsParams)
+      const response = await executeOciStreamingTool(
+        call({ toolId: 'oci_streaming_list_stream_pools', input: operationInput })
+      )
+      expect(response.status).toBe(200)
+      expect(mocks.request).toHaveBeenCalledTimes(1)
+      expect(mocks.request.mock.calls[0][0].queryPairs).not.toEqual(
+        expect.arrayContaining([['name', 'null']])
+      )
+      expect(mocks.request.mock.calls[0][0].queryPairs).not.toEqual(
+        expect.arrayContaining([['page', '']])
+      )
+    }
+  )
+
   it('passes only the resolved credential ID and trusted workspace to the foundation', async () => {
     const response = await executeOciStreamingTool(call())
     expect(response.status).toBe(200)

apps/sim/lib/internal/oci-streaming/operations.test.ts

@@ -64,6 +64,81 @@ function harness() {
 afterEach(() => vi.useRealTimers())
 
 describe('OCI Streaming request and response semantics', () => {
+  it.each(['create_stream_pool', 'get_stream_pool', 'update_stream_pool'])(
+    'projects inactive pool settings for %s without replaying the request',
+    async (operation) => {
+      const h = harness()
+      const pool = {
+        id: 'pool',
+        name: 'events',
+        compartmentId: 'compartment',
+        lifecycleState: 'ACTIVE',
+        timeCreated: '2026-01-01T00:00:00Z',
+        kafkaSettings: { bootstrapServers: null },
+        customEncryptionKey: { kmsKeyId: null, keyState: 'NONE' },
+        privateEndpointSettings: { nsgIds: null, privateEndpointIp: null, subnetId: null },
+      }
+      h.request.mockResolvedValue(response(JSON.stringify(pool), { etag: 'v1' }))
+      const params =
+        operation === 'create_stream_pool'
+          ? { name: 'events', compartmentId: 'compartment' }
+          : operation === 'update_stream_pool'
+            ? { streamPoolId: 'pool', name: 'events' }
+            : { streamPoolId: 'pool' }
+      const result = await h.execute({ operation, ...params })
+      expect(result.output).toMatchObject({
+        streamPool: {
+          ...pool,
+          kafkaSettings: { bootstrapServers: undefined },
+          customEncryptionKey: { kmsKeyId: undefined, keyState: 'NONE' },
+          privateEndpointSettings: {
+            nsgIds: undefined,
+            privateEndpointIp: undefined,
+            subnetId: undefined,
+          },
+        },
+        etag: 'v1',
+      })
+      expect(h.request).toHaveBeenCalledTimes(1)
+    }
+  )
+
+  it('preserves populated pool settings and rejects malformed settings', async () => {
+    const h = harness()
+    const pool = {
+      id: 'pool',
+      name: 'events',
+      compartmentId: 'compartment',
+      lifecycleState: 'ACTIVE',
+      timeCreated: '2026-01-01T00:00:00Z',
+      kafkaSettings: { bootstrapServers: 'server:9092' },
+      customEncryptionKey: { kmsKeyId: 'key' },
+      privateEndpointSettings: {
+        nsgIds: ['nsg'],
+        privateEndpointIp: '10.0.0.1',
+        subnetId: 'subnet',
+      },
+    }
+    h.request.mockResolvedValue(response(JSON.stringify(pool)))
+    expect(
+      (await h.execute({ operation: 'get_stream_pool', streamPoolId: 'pool' })).output.streamPool
+    ).toEqual(pool)
+    h.request.mockResolvedValue(
+      response(JSON.stringify({ ...pool, privateEndpointSettings: { nsgIds: 'nsg' } }))
+    )
+    await expect(
+      h.execute({ operation: 'get_stream_pool', streamPoolId: 'pool' })
+    ).rejects.toThrow()
+    expect(
+      ociStreamingInputSchema.safeParse({
+        operation: 'create_stream_pool',
+        ociCredential: 'credential',
+        name: 'events',
+        compartmentId: 'compartment',
+        customEncryptionKeyDetails: { kmsKeyId: null },
+      }).success
+    ).toBe(false)
+  })
   it('accepts full resource names, bounds group paths, and preserves empty tag updates', () => {
     const create = {
       operation: 'create_stream',

apps/sim/lib/internal/oci-streaming/schema.ts

@@ -392,16 +392,33 @@ export const streamPoolSummarySchema = z.object({ ...resource, isPrivate: z.bool
 export const streamPoolSchema = streamPoolSummarySchema.extend({
   endpointFqdn: z.string().nullish(),
   lifecycleStateDetails: z.string().nullish(),
-  kafkaSettings: kafkaSettings.strip().extend({ bootstrapServers: z.string().optional() }),
+  kafkaSettings: kafkaSettings.strip().extend({
+    bootstrapServers: z
+      .string()
+      .nullish()
+      .transform((value) => value ?? undefined),
+  }),
   customEncryptionKey: z.object({
-    kmsKeyId: z.string().optional(),
+    kmsKeyId: z
+      .string()
+      .nullish()
+      .transform((value) => value ?? undefined),
     keyState: z.string().optional(),
   }),
   privateEndpointSettings: z
     .object({
-      nsgIds: z.array(z.string()).optional(),
-      privateEndpointIp: z.string().optional(),
-      subnetId: z.string().optional(),
+      nsgIds: z
+        .array(z.string())
+        .nullish()
+        .transform((value) => value ?? undefined),
+      privateEndpointIp: z
+        .string()
+        .nullish()
+        .transform((value) => value ?? undefined),
+      subnetId: z
+        .string()
+        .nullish()
+        .transform((value) => value ?? undefined),
     })
     .nullish(),
 })