fix(mothership): fix superagent credentials (#4185)

* Fix

* Fix ajv csp issue

* Lint

Author: Sg312

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

@@ -58,12 +58,16 @@ import {
   WorkspaceFile,
   WorkspaceFileOperation,
 } from '@/lib/copilot/generated/tool-catalog-v1'
-import { parsePersistedStreamEventEnvelopeJson } from '@/lib/copilot/request/session/contract'
+import {
+  type ParseStreamEventEnvelopeFailure,
+  parsePersistedStreamEventEnvelope,
+  parsePersistedStreamEventEnvelopeJson,
+} from '@/lib/copilot/request/session/contract'
 import {
   type FilePreviewSession,
   isFilePreviewSession,
 } from '@/lib/copilot/request/session/file-preview-session-contract'
-import { isStreamBatchEvent, type StreamBatchEvent } from '@/lib/copilot/request/session/types'
+import type { StreamBatchEvent } from '@/lib/copilot/request/session/types'
 import {
   extractResourcesFromToolResult,
   isResourceToolName,
@@ -509,27 +513,75 @@ function isRecord(value: unknown): value is Record<string, unknown> {
   return Boolean(value) && typeof value === 'object' && !Array.isArray(value)
 }
 
+const STREAM_SCHEMA_ENFORCEMENT_PREFIX = 'Client stream schema enforcement failed.'
+
+class StreamSchemaValidationError extends Error {
+  constructor(message: string) {
+    super(message)
+    this.name = 'StreamSchemaValidationError'
+  }
+}
+
+function createStreamSchemaValidationError(
+  failure: ParseStreamEventEnvelopeFailure,
+  context?: string
+): StreamSchemaValidationError {
+  const details = failure.errors?.filter(Boolean).join('; ')
+  return new StreamSchemaValidationError(
+    [STREAM_SCHEMA_ENFORCEMENT_PREFIX, context, failure.message, details].filter(Boolean).join(' ')
+  )
+}
+
+function createBatchSchemaValidationError(message: string): StreamSchemaValidationError {
+  return new StreamSchemaValidationError([STREAM_SCHEMA_ENFORCEMENT_PREFIX, message].join(' '))
+}
+
+function isStreamSchemaValidationError(error: unknown): error is StreamSchemaValidationError {
+  return error instanceof StreamSchemaValidationError
+}
+
 function parseStreamBatchResponse(value: unknown): StreamBatchResponse {
   if (!isRecord(value)) {
     throw new Error('Invalid stream batch response')
   }
 
   const rawEvents = Array.isArray(value.events) ? value.events : []
   const events: StreamBatchEvent[] = []
-  for (const entry of rawEvents) {
-    if (!isStreamBatchEvent(entry)) {
-      throw new Error('Invalid stream batch event')
+  for (const [index, entry] of rawEvents.entries()) {
+    if (!isRecord(entry)) {
+      throw createBatchSchemaValidationError(`Reconnect batch event ${index + 1} is not an object.`)
+    }
+    if (
+      typeof entry.eventId !== 'number' ||
+      !Number.isFinite(entry.eventId) ||
+      typeof entry.streamId !== 'string'
+    ) {
+      throw createBatchSchemaValidationError(
+        `Reconnect batch event ${index + 1} is missing required metadata.`
+      )
     }
-    events.push(entry)
+
+    const parsedEvent = parsePersistedStreamEventEnvelope(entry.event)
+    if (!parsedEvent.ok) {
+      throw createStreamSchemaValidationError(parsedEvent, `Reconnect batch event ${index + 1}.`)
+    }
+
+    events.push({
+      eventId: entry.eventId,
+      streamId: entry.streamId,
+      event: parsedEvent.event,
+    })
   }
 
   const rawPreviewSessions = Array.isArray(value.previewSessions)
     ? value.previewSessions
     : undefined
   const previewSessions =
-    rawPreviewSessions?.map((session) => {
+    rawPreviewSessions?.map((session, index) => {
       if (!isFilePreviewSession(session)) {
-        throw new Error('Invalid stream preview session')
+        throw createBatchSchemaValidationError(
+          `Reconnect preview session ${index + 1} failed validation.`
+        )
       }
       return session
     }) ?? undefined
@@ -1579,12 +1631,14 @@ export function useChat(
 
           const parsedResult = parsePersistedStreamEventEnvelopeJson(raw)
           if (!parsedResult.ok) {
-            logger.warn('Failed to parse chat SSE event', {
+            const error = createStreamSchemaValidationError(parsedResult, 'Live SSE event.')
+            logger.error('Rejected chat SSE event due to client-side schema enforcement', {
               reason: parsedResult.reason,
               message: parsedResult.message,
               errors: parsedResult.errors,
+              error: error.message,
             })
-            continue
+            throw error
           }
           const parsed = parsedResult.event
 
@@ -2533,6 +2587,17 @@ export function useChat(
             }
             return true
           }
+          if (isStreamSchemaValidationError(err)) {
+            logger.error('Reconnect halted by client-side stream schema enforcement', {
+              streamId,
+              attempt: attempt + 1,
+              error: err.message,
+            })
+            if (streamGenRef.current === gen) {
+              setError(err.message)
+            }
+            return false
+          }
           logger.warn('Reconnect attempt failed', {
             streamId,
             attempt: attempt + 1,
@@ -2892,6 +2957,13 @@ export function useChat(
         }
       } catch (err) {
         if (err instanceof Error && err.name === 'AbortError') return consumedByTranscript
+        if (isStreamSchemaValidationError(err)) {
+          setError(err.message)
+          if (streamGenRef.current === gen) {
+            finalize({ error: true })
+          }
+          return consumedByTranscript
+        }
 
         const activeStreamId = streamIdRef.current
         if (activeStreamId && streamGenRef.current === gen) {

apps/sim/lib/copilot/request/session/contract.test.ts

@@ -43,6 +43,92 @@ describe('stream session contract parser', () => {
     })
   })
 
+  it('accepts contract session chat events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'session' as const,
+      payload: { kind: 'chat' as const, chatId: 'chat-1' },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
+  it('accepts contract complete events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'complete' as const,
+      payload: { status: 'complete' as const },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
+  it('accepts contract error events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'error' as const,
+      payload: { message: 'something went wrong' },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
+  it('accepts contract tool call events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'tool' as const,
+      payload: {
+        toolCallId: 'tc-1',
+        toolName: 'read',
+        phase: 'call' as const,
+        executor: 'sim' as const,
+        mode: 'sync' as const,
+      },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
+  it('accepts contract span events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'span' as const,
+      payload: { kind: 'subagent' as const, event: 'start' as const, agent: 'file' },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
+  it('accepts contract resource events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'resource' as const,
+      payload: {
+        op: 'upsert' as const,
+        resource: { id: 'r-1', type: 'file', title: 'test.md' },
+      },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
+  it('accepts contract run events', () => {
+    const event = {
+      ...BASE_ENVELOPE,
+      type: 'run' as const,
+      payload: { kind: 'compaction_start' as const },
+    }
+
+    expect(isContractStreamEventEnvelope(event)).toBe(true)
+    expect(parsePersistedStreamEventEnvelope(event).ok).toBe(true)
+  })
+
   it('accepts synthetic file preview events', () => {
     const event = {
       ...BASE_ENVELOPE,
@@ -82,7 +168,32 @@ describe('stream session contract parser', () => {
       throw new Error('expected invalid result')
     }
     expect(parsed.reason).toBe('invalid_stream_event')
-    expect(parsed.errors?.length).toBeGreaterThan(0)
+  })
+
+  it('rejects unknown event types', () => {
+    const parsed = parsePersistedStreamEventEnvelope({
+      ...BASE_ENVELOPE,
+      type: 'unknown_type',
+      payload: {},
+    })
+
+    expect(parsed.ok).toBe(false)
+    if (parsed.ok) {
+      throw new Error('expected invalid result')
+    }
+    expect(parsed.reason).toBe('invalid_stream_event')
+    expect(parsed.errors).toContain('unknown type="unknown_type"')
+  })
+
+  it('rejects non-object values', () => {
+    const parsed = parsePersistedStreamEventEnvelope('not an object')
+
+    expect(parsed.ok).toBe(false)
+    if (parsed.ok) {
+      throw new Error('expected invalid result')
+    }
+    expect(parsed.reason).toBe('invalid_stream_event')
+    expect(parsed.errors).toContain('value is not an object')
   })
 
   it('reports invalid JSON separately from schema failures', () => {