fix(logs): export call fields to OpenTelemetry and tell concurrent polls apart

Scalar fields passed with a log line now become OpenTelemetry attributes, matching their top-level keys in the JSON line, so ids moved out of message text stay filterable in both sinks. Nested values stay in log.args. The request context still owns requestId.

The polled-webhook processor logs through a logger bound to the webhook id, so lines from concurrent polls in one tick stay distinguishable. Test expectations follow the new metadata shape.

Author: waleedlatif1

.agents/skills/add-trigger/SKILL.md

@@ -411,7 +411,7 @@ export const {service}PollingHandler: PollingProviderHandler = {
       await markWebhookSuccess(webhookId, logger)
       return 'success'
     } catch (error) {
-      logger.error(`Error processing {service} webhook ${webhookId}:`, error, { requestId })
+      logger.error('Error processing {service} webhook:', error)
       await markWebhookFailed(webhookId, logger)
       return 'failure'
     }

apps/sim/background/knowledge-connector-sync.test.ts

@@ -76,7 +76,8 @@ describe('knowledge connector sync worker', () => {
       processingDispatch: { failed: 1 },
     })
     expect(mockWarn).toHaveBeenCalledWith(
-      expect.stringContaining('1 source failures, 1 dispatch failures')
+      expect.stringContaining('1 source failures, 1 dispatch failures'),
+      { requestId: 'request-1' }
     )
   })
 

apps/sim/lib/mothership/chat/project-file-context.integration.ts

@@ -694,7 +694,7 @@ describe('open tabs that no longer resolve', () => {
         })
         expect(contexts[1]?.content).toContain(browserTab.url)
         expect(logs.calls('info', 'Skipped resource attachment that no longer exists')).toEqual([
-          { resourceType: 'file' },
+          { requestId: expect.any(String), resourceType: 'file' },
         ])
         expect(logs.calls('error', 'Failed to resolve resource attachment')).toHaveLength(1)
         expect(
@@ -737,7 +737,7 @@ describe('open tabs that no longer resolve', () => {
         expect(contexts[0]?.content).toContain(browserTab.url)
         expect(
           logs.calls('warn', 'Skipped resource attachment the user can no longer access')
-        ).toEqual([{ resourceType: 'file' }])
+        ).toEqual([{ requestId: expect.any(String), resourceType: 'file' }])
         expect(logs.calls('info', 'Skipped resource attachment that no longer exists')).toEqual([])
         expect(logs.calls('error', 'Failed to resolve resource attachment')).toEqual([])
       } finally {

apps/sim/lib/webhooks/processor.ts

@@ -1011,11 +1011,13 @@ export async function processPolledWebhookEvent(
     return { success: false, error: 'Webhook has no provider', statusCode: 400 }
   }
   const provider = foundWebhook.provider
+  // Concurrent polls in one tick share a request; the webhook id tells their lines apart.
+  const log = logger.withMetadata({ webhookId: foundWebhook.id })
 
   const ticket = tryAdmit()
   if (!ticket) {
     const gateFailure = ADMISSION_ERROR_DESCRIPTOR.GATE_CAPACITY
-    logger.warn('Admission gate rejected polled webhook event', { requestId })
+    log.warn('Admission gate rejected polled webhook event', { requestId })
     return {
       success: false,
       error: 'Server at capacity',
@@ -1032,7 +1034,7 @@ export async function processPolledWebhookEvent(
     if (foundWebhook.blockId) {
       const blockExists = await blockExistsInDeployment(foundWorkflow.id, foundWebhook.blockId)
       if (!blockExists) {
-        logger.info(
+        log.info(
           `Trigger block ${foundWebhook.blockId} not found in deployment for workflow ${foundWorkflow.id}`,
           { requestId }
         )
@@ -1052,7 +1054,7 @@ export async function processPolledWebhookEvent(
         typeof errorBody.error === 'string'
           ? errorBody.error
           : 'Preprocessing failed'
-      logger.warn('Polled webhook preprocessing failed', {
+      log.warn('Polled webhook preprocessing failed', {
         requestId,
         statusCode,
         error: errorMessage,
@@ -1084,7 +1086,7 @@ export async function processPolledWebhookEvent(
     const billingAttribution = preprocessResult.billingAttribution
     const workspaceId = foundWorkflow.workspaceId ?? undefined
     if (!actorUserId || !billingAttribution || !workspaceId) {
-      logger.error(`Missing billing context for webhook ${foundWebhook.id}`, { requestId })
+      log.error(`Missing billing context for webhook ${foundWebhook.id}`, { requestId })
       return { success: false, error: 'Unable to resolve billing account', statusCode: 500 }
     }
 
@@ -1146,7 +1148,7 @@ export async function processPolledWebhookEvent(
         maxDurationSeconds,
       })
       reservationTransferred = true
-      logger.info(
+      log.info(
         `Queued polling webhook execution task ${jobId} for ${provider} webhook via job queue`,
         { requestId }
       )
@@ -1164,12 +1166,12 @@ export async function processPolledWebhookEvent(
           executeWebhookJob(payload, signal),
       })
       reservationTransferred = true
-      logger.info(`Queued ${provider} webhook execution ${jobId} via inline backend`, { requestId })
+      log.info(`Queued ${provider} webhook execution ${jobId} via inline backend`, { requestId })
     }
 
     return { success: true, executionId }
   } catch (error: unknown) {
-    logger.error('Failed to process polled webhook event:', error, { requestId })
+    log.error('Failed to process polled webhook event:', error, { requestId })
     return { success: false, error: 'Internal server error', statusCode: 500 }
   } finally {
     if (!reservationTransferred && reservedExecutionId) {

apps/sim/lib/workflows/streaming/streaming.test.ts

@@ -1444,7 +1444,7 @@ describe('createStreamingResponse agent-events-v1', () => {
       )
     )
     expect(
-      workflowStreamingLogger.error.mock.calls.some(([message]) => message === 'Stream error')
+      workflowStreamingLogger.error.mock.calls.some((call: unknown[]) => call[0] === 'Stream error')
     ).toBe(false)
   })
 

packages/logger/src/index.test.ts

@@ -137,6 +137,29 @@ describe('Logger', () => {
       expect(lines()[0]).toMatchObject({ requestId: 'task-run' })
     })
 
+    test('exports scalar fields passed with a line as OpenTelemetry attributes', () => {
+      const emit = vi.fn()
+      const getLoggerSpy = vi
+        .spyOn(logs, 'getLogger')
+        .mockReturnValue({ emit, enabled: () => true })
+      try {
+        const logger = jsonLogger()
+        logger.info('Outside a request', { requestId: 'task-run', documentId: 'doc-1' })
+        runWithRequestContext({ requestId: 'req-context' }, () => {
+          logger.info('Inside a request', { requestId: 'replay-key', attempt: 2, nested: { a: 1 } })
+        })
+        const attributes = emit.mock.calls.map(
+          ([record]) => (record as { attributes: Record<string, string> }).attributes
+        )
+        expect(attributes[0]).toMatchObject({ requestId: 'task-run', documentId: 'doc-1' })
+        expect(attributes[1]).toMatchObject({ requestId: 'req-context', attempt: '2' })
+        expect(attributes[1]).not.toHaveProperty('nested')
+        expect(attributes[1]['log.args']).toContain('"nested"')
+      } finally {
+        getLoggerSpy.mockRestore()
+      }
+    })
+
     test('never lets an absent optional field erase one the line carries', () => {
       const logger = jsonLogger().withMetadata({ workflowId: 'workflow-1' })
       runWithRequestContext({ requestId: 'req-context' }, () => {

packages/logger/src/index.ts

@@ -504,7 +504,7 @@ export class Logger {
     const effectiveMetadata = reqCtx
       ? { ...requestContextMetadata(reqCtx), ...this.metadata, ...owned }
       : this.metadata
-    emitOtelLogRecord(level, this.module, message, effectiveMetadata, args)
+    emitOtelLogRecord(level, this.module, message, effectiveMetadata, args, owned)
     const metadataEntries = Object.entries(filterUndefined(effectiveMetadata))
     const metadataStr =
       metadataEntries.length > 0
@@ -638,6 +638,12 @@ const OTEL_LOG_SEVERITY: Record<LogLevel, { number: SeverityNumber; text: string
 
 const OTEL_LOG_ARG_MAX_CHARS = 2000
 
+const isOtelScalar = (value: unknown): value is string | number | boolean | bigint =>
+  typeof value === 'string' ||
+  typeof value === 'number' ||
+  typeof value === 'boolean' ||
+  typeof value === 'bigint'
+
 /**
  * Fans every accepted log line out through the OTel Logs API. Until an
  * application installs a global LoggerProvider (apps/sim does in
@@ -652,14 +658,28 @@ function emitOtelLogRecord(
   module: string,
   message: string,
   metadata: Record<string, unknown>,
-  args: unknown[]
+  args: unknown[],
+  owned: Record<string, unknown>
 ): void {
   try {
     const severity = OTEL_LOG_SEVERITY[level]
     const attributes: Record<string, string> = { 'log.module': module }
     for (const [key, value] of Object.entries(filterUndefined(metadata))) {
       attributes[key] = String(value)
     }
+    // Scalar fields passed with the line become attributes, as they are top-level keys in the JSON
+    // line, so `{ documentId }` filters the same way in both; nested values stay in `log.args`.
+    for (const arg of args) {
+      if (!isRecordLike(arg) || arg instanceof Error) continue
+      try {
+        for (const [key, value] of Object.entries(arg)) {
+          if (isOtelScalar(value)) attributes[key] = String(value)
+        }
+      } catch {
+        // A throwing getter loses only its own argument's fields; `log.args` still records it.
+      }
+    }
+    for (const [key, value] of Object.entries(owned)) attributes[key] = String(value)
     const error = primaryError(args)
     if (error) {
       const logged = toLoggedError(error)