fix(inspector): render captured event streams

packages/experimental/inspector/src/host/inspection/network.ts

@@ -108,7 +108,7 @@ export function installFetchObserver(
       status: response.status,
       statusText: response.statusText,
       headers: headerEntries(response.headers),
-      mimeType: response.headers.get('content-type')?.split(';', 1)[0]?.trim() ?? '',
+      mimeType: response.headers.get('content-type')?.split(';', 1)[0]?.trim().toLowerCase() ?? '',
     })
 
     try {

packages/experimental/inspector/src/shared/network/event-source.ts

@@ -0,0 +1,77 @@
+/** Incremental UTF-8 parser for Server-Sent Events carried by captured responses. */
+
+import type { InspectorEventSourceMessage } from './observation.ts'
+
+/** Parse response bytes into consumer-neutral Server-Sent Event messages. */
+export class InspectorEventSourceParser {
+  private readonly decoder = new TextDecoder()
+  private line = ''
+  private eventName = ''
+  private eventId = ''
+  private data = ''
+  private afterCarriageReturn = false
+
+  /**
+   * Consume one response-body chunk.
+   * @param bytes - Next bytes in response order.
+   * @returns Complete events terminated by an empty line in this chunk.
+   */
+  push(bytes: Uint8Array): readonly InspectorEventSourceMessage[] {
+    return this.consume(this.decoder.decode(bytes, { stream: true }))
+  }
+
+  private consume(chunk: string): InspectorEventSourceMessage[] {
+    const messages: InspectorEventSourceMessage[] = []
+    let start = 0
+    for (let index = 0; index < chunk.length; index++) {
+      if (this.afterCarriageReturn && chunk[index] === '\n') {
+        this.afterCarriageReturn = false
+        start = index + 1
+        continue
+      }
+      this.afterCarriageReturn = false
+      if (chunk[index] !== '\r' && chunk[index] !== '\n') continue
+      this.line += chunk.slice(start, index)
+      const message = this.parseLine()
+      if (message !== undefined) messages.push(message)
+      this.line = ''
+      start = index + 1
+      this.afterCarriageReturn = chunk[index] === '\r'
+    }
+    this.line += chunk.slice(start)
+    return messages
+  }
+
+  private parseLine(): InspectorEventSourceMessage | undefined {
+    if (this.line.length === 0) {
+      const data = this.data
+      this.data = ''
+      const eventName = this.eventName
+      this.eventName = ''
+      if (data.length === 0) return undefined
+      return {
+        eventName: eventName || 'message',
+        eventId: this.eventId,
+        data: data.slice(0, -1),
+      }
+    }
+    if (this.line.startsWith(':')) return undefined
+    const colon = this.line.indexOf(':')
+    const field = colon === -1 ? this.line : this.line.slice(0, colon)
+    let value = colon === -1 ? '' : this.line.slice(colon + 1)
+    if (value.startsWith(' ')) value = value.slice(1)
+    switch (field) {
+      case 'event':
+        this.eventName = value
+        return undefined
+      case 'data':
+        this.data += `${value}\n`
+        return undefined
+      case 'id':
+        if (!value.includes('\0')) this.eventId = value
+        return undefined
+      default:
+        return undefined
+    }
+  }
+}

packages/experimental/inspector/src/shared/network/observation.ts

@@ -46,6 +46,13 @@ export interface FetchEndPayload extends FetchIdentity {
   readonly responseCaptureError?: string
 }
 
+/** One parsed Server-Sent Event independent of its CDP projection. */
+export interface InspectorEventSourceMessage {
+  readonly eventName: string
+  readonly eventId: string
+  readonly data: string
+}
+
 /** Fetch rejected before returning a Response. */
 export interface FetchErrorPayload extends FetchIdentity {
   readonly message: string

packages/experimental/inspector/src/worker/cdp/domains/network/session.ts

@@ -9,10 +9,15 @@ export interface NetworkSink {
   sendEvent(method: string, params: Readonly<Record<string, unknown>>): void
 }
 
+type RequestStartedEvent = Extract<NetworkStoreEvent, { readonly type: 'request-started' }>
+type NetworkResourceType = 'EventSource' | 'Fetch'
+
 /** Projects retained and live network observations into connection-local CDP state. */
 export class NetworkDomain {
   private readonly enabled = new Set<NetworkSink>()
   private readonly streamedRequests = new Map<NetworkSink, Set<string>>()
+  private readonly pendingStarts = new Map<NetworkSink, Map<string, RequestStartedEvent>>()
+  private readonly requestTypes = new Map<NetworkSink, Map<string, NetworkResourceType>>()
   private readonly unsubscribe: () => void
 
   constructor(private readonly store: NetworkStore) {
@@ -25,8 +30,10 @@ export class NetworkDomain {
    */
   enable(session: NetworkSink): void {
     if (this.enabled.has(session)) return
-    for (const event of this.store.replay()) this.send(session, event)
     this.enabled.add(session)
+    this.pendingStarts.set(session, new Map())
+    this.requestTypes.set(session, new Map())
+    for (const event of this.store.replay()) this.send(session, event)
   }
 
   /**
@@ -36,6 +43,8 @@ export class NetworkDomain {
   disable(session: NetworkSink): void {
     this.enabled.delete(session)
     this.streamedRequests.delete(session)
+    this.pendingStarts.delete(session)
+    this.requestTypes.delete(session)
   }
 
   /**
@@ -51,6 +60,8 @@ export class NetworkDomain {
     this.unsubscribe()
     this.enabled.clear()
     this.streamedRequests.clear()
+    this.pendingStarts.clear()
+    this.requestTypes.clear()
   }
 
   /**
@@ -112,6 +123,8 @@ export class NetworkDomain {
         requests.delete(event.requestKey)
         if (requests.size === 0) this.streamedRequests.delete(session)
       }
+      for (const requests of this.pendingStarts.values()) requests.delete(event.requestKey)
+      for (const requests of this.requestTypes.values()) requests.delete(event.requestKey)
       return
     }
     for (const session of this.enabled) this.send(session, event)
@@ -121,29 +134,17 @@ export class NetworkDomain {
     const timestamp = (event.timestampMs - performance.timeOrigin) / 1_000
     switch (event.type) {
       case 'request-started':
-        session.sendEvent('Network.requestWillBeSent', {
-          requestId: event.requestId,
-          loaderId: 'dsh-inspector-loader',
-          documentURL: 'dsh://host',
-          request: {
-            url: event.url,
-            method: event.method,
-            headers: cdpHeaders(event.headers),
-            hasPostData: event.hasBody,
-          },
-          timestamp,
-          wallTime: event.wallTimeMs / 1_000,
-          initiator: { type: 'other' },
-          type: 'Fetch',
-        })
+        this.pendingStarts.get(session)?.set(event.requestKey, event)
         return
-      case 'response-received':
+      case 'response-received': {
+        const resourceType = event.mimeType === 'text/event-stream' ? 'EventSource' : 'Fetch'
+        this.sendRequestStart(session, event.requestKey, resourceType)
         session.sendEvent('Network.responseReceived', {
           requestId: event.requestId,
           loaderId: 'dsh-inspector-loader',
           frameId: 'dsh-inspector-host-frame',
           timestamp,
-          type: 'Fetch',
+          type: resourceType,
           response: {
             url: event.url,
             status: event.status,
@@ -152,11 +153,21 @@ export class NetworkDomain {
             mimeType: event.mimeType,
             connectionReused: false,
             connectionId: 0,
-            encodedDataLength: 0,
+            encodedDataLength: resourceType === 'EventSource' ? -1 : 0,
             securityState: 'neutral',
           },
         })
         return
+      }
+      case 'event-source-message':
+        session.sendEvent('Network.eventSourceMessageReceived', {
+          requestId: event.requestId,
+          timestamp,
+          eventName: event.eventName,
+          eventId: event.eventId,
+          data: event.data,
+        })
+        return
       case 'response-data':
         session.sendEvent('Network.dataReceived', {
           requestId: event.requestId,
@@ -167,34 +178,62 @@ export class NetworkDomain {
         })
         return
       case 'request-finished':
+        this.sendRequestStart(session, event.requestKey, 'Fetch')
         session.sendEvent('Network.loadingFinished', {
           requestId: event.requestId,
           timestamp,
           encodedDataLength: event.encodedDataLength,
           dshInspectorTruncated: event.truncated,
         })
-        this.stopStreaming(event.requestKey)
+        this.stopRequest(session, event.requestKey)
         return
-      case 'request-failed':
+      case 'request-failed': {
+        this.sendRequestStart(session, event.requestKey, 'Fetch')
+        const resourceType = this.requestTypes.get(session)?.get(event.requestKey) ?? 'Fetch'
         session.sendEvent('Network.loadingFailed', {
           requestId: event.requestId,
           timestamp,
-          type: 'Fetch',
+          type: resourceType,
           errorText: event.errorText,
           canceled: event.canceled,
         })
-        this.stopStreaming(event.requestKey)
+        this.stopRequest(session, event.requestKey)
         return
+      }
       default:
         return assertNever(event)
     }
   }
 
-  private stopStreaming(requestKey: string): void {
-    for (const [session, requests] of this.streamedRequests) {
-      requests.delete(requestKey)
-      if (requests.size === 0) this.streamedRequests.delete(session)
-    }
+  private sendRequestStart(session: NetworkSink, requestKey: string, resourceType: NetworkResourceType): void {
+    const pending = this.pendingStarts.get(session)
+    const event = pending?.get(requestKey)
+    if (event === undefined) return
+    pending?.delete(requestKey)
+    this.requestTypes.get(session)?.set(requestKey, resourceType)
+    session.sendEvent('Network.requestWillBeSent', {
+      requestId: event.requestId,
+      loaderId: 'dsh-inspector-loader',
+      documentURL: 'dsh://host',
+      request: {
+        url: event.url,
+        method: event.method,
+        headers: cdpHeaders(event.headers),
+        hasPostData: event.hasBody,
+      },
+      timestamp: (event.timestampMs - performance.timeOrigin) / 1_000,
+      wallTime: event.wallTimeMs / 1_000,
+      initiator: { type: 'other' },
+      type: resourceType,
+    })
+  }
+
+  private stopRequest(session: NetworkSink, requestKey: string): void {
+    const streamed = this.streamedRequests.get(session)
+    streamed?.delete(requestKey)
+    if (streamed?.size === 0) this.streamedRequests.delete(session)
+    this.pendingStarts.get(session)?.delete(requestKey)
+    this.requestTypes.get(session)?.delete(requestKey)
   }
 }
 

packages/experimental/inspector/src/worker/inspection/network-store.ts

@@ -3,6 +3,7 @@
 import { Buffer } from 'node:buffer'
 import { FETCH_TOPICS } from '../../shared/bridge/messages/network.ts'
 import type { InspectorHeader } from '../../shared/network/observation.ts'
+import { InspectorEventSourceParser } from '../../shared/network/event-source.ts'
 import { isPlainObject } from '../../shared/json.ts'
 import type { InspectorSourceDescriptor } from '../../shared/bridge/messages/observation.ts'
 import type { IngestedInspectorRecord, InspectorRecordConsumer } from '../bridge/hub.ts'
@@ -50,6 +51,12 @@ export type NetworkStoreEvent =
     readonly data: string
     readonly byteLength: number
   }
+  | NetworkEventBase & {
+    readonly type: 'event-source-message'
+    readonly eventName: string
+    readonly eventId: string
+    readonly data: string
+  }
   | NetworkEventBase & {
     readonly type: 'request-finished'
     readonly encodedDataLength: number
@@ -62,6 +69,9 @@ export type NetworkStoreEvent =
   }
   | { readonly type: 'request-evicted'; readonly requestKey: string }
 
+type JournalNetworkEvent = Exclude<NetworkStoreEvent, {
+  readonly type: 'response-data' | 'event-source-message' | 'request-evicted'
+}>
 type ReplayableNetworkEvent = Exclude<NetworkStoreEvent, { readonly type: 'response-data' | 'request-evicted' }>
 
 interface CapturedRequest {
@@ -78,13 +88,15 @@ interface CapturedRequest {
   responseCaptureError?: string
   responseSeen: boolean
   completed: boolean
+  eventSourceParser: InspectorEventSourceParser | undefined
+  nextEventSourceId: number
 }
 
 /** Validated Network observation store independent of CDP connection state. */
 export class NetworkStore implements InspectorRecordConsumer {
   readonly topics = new Set<string>(FETCH_TOPICS)
   private readonly requests = new Map<string, CapturedRequest>()
-  private readonly journal: ReplayableNetworkEvent[] = []
+  private readonly journal: JournalNetworkEvent[] = []
   private readonly completed: string[] = []
   private readonly listeners = new Set<(event: NetworkStoreEvent) => void>()
   private journalBytes = 0
@@ -129,7 +141,26 @@ export class NetworkStore implements InspectorRecordConsumer {
    * @returns Events in observation order.
    */
   replay(): readonly ReplayableNetworkEvent[] {
-    return this.journal
+    const replay: ReplayableNetworkEvent[] = []
+    for (const event of this.journal) {
+      replay.push(event)
+      if (event.type !== 'response-received' || event.mimeType !== 'text/event-stream') continue
+      const request = this.requests.get(event.requestKey)
+      if (request === undefined) continue
+      const messages = new InspectorEventSourceParser().push(Buffer.concat(request.responseBody))
+      let eventId = 0
+      for (const message of messages) {
+        replay.push({
+          type: 'event-source-message',
+          requestKey: request.key,
+          requestId: request.requestId,
+          timestampMs: event.timestampMs,
+          ...message,
+          eventId: String(++eventId),
+        })
+      }
+    }
+    return replay
   }
 
   /**
@@ -191,6 +222,8 @@ export class NetworkStore implements InspectorRecordConsumer {
         responseBodyTruncated: false,
         responseSeen: false,
         completed: false,
+        eventSourceParser: undefined,
+        nextEventSourceId: 0,
       }
       this.requests.set(key, request)
       this.publish({
@@ -221,6 +254,10 @@ export class NetworkStore implements InspectorRecordConsumer {
       }
       case 'fetch/response':
         request.responseSeen = true
+        const mimeType = stringField(payload, 'mimeType').toLowerCase()
+        request.eventSourceParser = mimeType === 'text/event-stream'
+          ? new InspectorEventSourceParser()
+          : undefined
         this.publish({
           type: 'response-received',
           requestKey: key,
@@ -230,12 +267,23 @@ export class NetworkStore implements InspectorRecordConsumer {
           status: numberField(payload, 'status'),
           statusText: stringField(payload, 'statusText'),
           headers: headerField(payload, 'headers'),
-          mimeType: stringField(payload, 'mimeType'),
+          mimeType,
         })
         return
       case 'fetch/response-body-chunk': {
         const data = stringField(payload, 'data')
-        const byteLength = this.appendBody(request, 'response', data)
+        const bytes = this.appendBody(request, 'response', data)
+        const byteLength = bytes.byteLength
+        for (const message of request.eventSourceParser?.push(bytes) ?? []) {
+          this.emit({
+            type: 'event-source-message',
+            requestKey: key,
+            requestId: request.requestId,
+            timestampMs,
+            ...message,
+            eventId: String(++request.nextEventSourceId),
+          })
+        }
         this.emit({ type: 'response-data', requestKey: key, requestId: request.requestId, timestampMs, data, byteLength })
         return
       }
@@ -268,7 +316,7 @@ export class NetworkStore implements InspectorRecordConsumer {
     }
   }
 
-  private appendBody(request: CapturedRequest, side: 'request' | 'response', encoded: string): number {
+  private appendBody(request: CapturedRequest, side: 'request' | 'response', encoded: string): Buffer {
     const bytes = decodeBase64(encoded)
     this.evictCompletedFor(bytes.byteLength, request.key)
     const retained = bytes.subarray(0, Math.max(0, this.options.maxJournalBytes - this.journalBytes))
@@ -283,10 +331,10 @@ export class NetworkStore implements InspectorRecordConsumer {
     }
     this.journalBytes += retained.byteLength
     this.enforceRetention()
-    return bytes.byteLength
+    return bytes
   }
 
-  private complete(request: CapturedRequest, event: ReplayableNetworkEvent): void {
+  private complete(request: CapturedRequest, event: JournalNetworkEvent): void {
     if (request.completed) return
     request.completed = true
     this.publish(event)
@@ -294,7 +342,7 @@ export class NetworkStore implements InspectorRecordConsumer {
     this.enforceRetention()
   }
 
-  private publish(event: ReplayableNetworkEvent): void {
+  private publish(event: JournalNetworkEvent): void {
     this.journal.push(event)
     this.emit(event)
   }

packages/experimental/inspector/tests/event-source.host.spec.ts

@@ -0,0 +1,31 @@
+/** Consumer-neutral Server-Sent Event parsing behavior. */
+
+import { describe, expect, it } from 'vitest'
+import { InspectorEventSourceParser } from '../src/shared/network/event-source.ts'
+
+const encoder = new TextEncoder()
+
+describe('InspectorEventSourceParser', () => {
+  it('preserves parser state across chunks, CRLF boundaries, and UTF-8 boundaries', () => {
+    const parser = new InspectorEventSourceParser()
+    expect(parser.push(encoder.encode(': ignored\rid:first\revent: update\rdata: one\r'))).toEqual([])
+
+    const unicode = encoder.encode('\ndata: two 你\r\n\r\n')
+    const split = unicode.indexOf(0xe4) + 1
+    expect(parser.push(unicode.subarray(0, split))).toEqual([])
+    expect(parser.push(unicode.subarray(split))).toEqual([{
+      eventName: 'update',
+      eventId: 'first',
+      data: 'one\ntwo 你',
+    }])
+  })
+
+  it('retains valid ids, ignores comments and unknown fields, and emits empty data', () => {
+    const parser = new InspectorEventSourceParser()
+    expect(parser.push(encoder.encode('retry: 1000\nunknown\n\n'))).toEqual([])
+    expect(parser.push(encoder.encode('id: stable\ndata: value\n\nid: bad\0id\ndata:\n\n'))).toEqual([
+      { eventName: 'message', eventId: 'stable', data: 'value' },
+      { eventName: 'message', eventId: 'stable', data: '' },
+    ])
+  })
+})

packages/experimental/inspector/tests/integration.host.spec.ts

@@ -466,6 +466,20 @@ describe('experimental Inspector real Worker', () => {
           && (event.params?.response as Record<string, unknown> | undefined)?.mimeType === 'text/event-stream')
         requestId = received?.params?.requestId as string | undefined
         expect(requestId).toBeTypeOf('string')
+        expect(received?.params).toMatchObject({
+          type: 'EventSource',
+          response: { encodedDataLength: -1 },
+        })
+        expect(cdp!.events.find(event =>
+          event.method === 'Network.requestWillBeSent'
+          && event.params?.requestId === requestId)?.params?.type).toBe('EventSource')
+        expect(cdp!.events.find(event =>
+          event.method === 'Network.eventSourceMessageReceived'
+          && event.params?.requestId === requestId)?.params).toMatchObject({
+          eventName: 'message',
+          eventId: '1',
+          data: 'first',
+        })
         expect(cdp!.events.some(event =>
           event.method === 'Network.dataReceived'
           && event.params?.requestId === requestId)).toBe(true)
@@ -488,6 +502,13 @@ describe('experimental Inspector real Worker', () => {
             && typeof event.params?.data === 'string')
           .map(event => Buffer.from(String(event.params!.data), 'base64'))
         expect(Buffer.concat(streamed).toString('utf8')).toBe(laterChunk)
+        expect(cdp!.events.slice(laterEventOffset).find(event =>
+          event.method === 'Network.eventSourceMessageReceived'
+          && event.params?.requestId === requestId)?.params).toMatchObject({
+          eventName: 'update',
+          eventId: '2',
+          data: 'second\nline',
+        })
       })
 
       const body = await cdp.call('Network.getResponseBody', { requestId })

packages/experimental/inspector/tests/network.host.spec.ts

@@ -89,6 +89,50 @@ describe('Inspector Network domain', () => {
     expect(secondSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')?.[1]).toMatchObject({ data: later })
   })
 
+  it('projects and replays parsed Server-Sent Events through the CDP EventSource path', () => {
+    const liveSend = vi.fn()
+    const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 1_024 })
+    const network = new NetworkDomain(store)
+    network.enable({ sendEvent: liveSend })
+    store.append(source, eventStreamRecords('events'))
+
+    expect(liveSend).toHaveBeenNthCalledWith(1, 'Network.requestWillBeSent', expect.objectContaining({
+      type: 'EventSource',
+    }))
+    expect(liveSend).toHaveBeenCalledWith('Network.responseReceived', expect.objectContaining({
+      type: 'EventSource',
+    }))
+    expect(liveSend.mock.calls
+      .filter(call => call[0] === 'Network.eventSourceMessageReceived')
+      .map(call => call[1] as unknown))
+      .toEqual([
+        expect.objectContaining({ eventName: 'message', eventId: '1', data: 'first' }),
+        expect.objectContaining({ eventName: 'update', eventId: '2', data: 'second\nline' }),
+      ])
+    expect(liveSend.mock.calls.map(call => String(call[0]))).toEqual([
+      'Network.requestWillBeSent',
+      'Network.responseReceived',
+      'Network.eventSourceMessageReceived',
+      'Network.dataReceived',
+      'Network.eventSourceMessageReceived',
+      'Network.dataReceived',
+      'Network.loadingFinished',
+    ])
+
+    const replay = vi.fn()
+    network.enable({ sendEvent: replay })
+    expect(replay).toHaveBeenNthCalledWith(1, 'Network.requestWillBeSent', expect.objectContaining({
+      type: 'EventSource',
+    }))
+    expect(replay.mock.calls
+      .filter(call => call[0] === 'Network.eventSourceMessageReceived')
+      .map(call => call[1] as unknown))
+      .toEqual([
+        expect.objectContaining({ eventName: 'message', eventId: '1', data: 'first' }),
+        expect.objectContaining({ eventName: 'update', eventId: '2', data: 'second\nline' }),
+      ])
+  })
+
   it('bounds active request metadata and does not retain per-chunk events for replay', () => {
     const firstSend = vi.fn()
     const store = new NetworkStore({ maxRetainedRequests: 1, maxJournalBytes: 1_024 })
@@ -148,6 +192,50 @@ function requestRecords(localId: string, body: string): IngestedInspectorRecord[
   ]
 }
 
+function eventStreamRecords(localId: string): IngestedInspectorRecord[] {
+  const first = 'id: 1\ndata: first\n\n'
+  const second = 'id: 2\nevent: update\ndata: second\ndata: line\n\n'
+  return [
+    {
+      sequence: 1,
+      monotonicMs: 1,
+      topic: 'fetch/start',
+      payload: { requestId: localId, url: 'https://example.test/events', method: 'GET', headers: [], hasBody: false, wallTimeMs: 1 },
+    },
+    {
+      sequence: 2,
+      monotonicMs: 2,
+      topic: 'fetch/response',
+      payload: {
+        requestId: localId,
+        url: 'https://example.test/events',
+        status: 200,
+        statusText: 'OK',
+        headers: [['content-type', 'text/event-stream; charset=utf-8']],
+        mimeType: 'TEXT/EVENT-STREAM',
+      },
+    },
+    {
+      sequence: 3,
+      monotonicMs: 3,
+      topic: 'fetch/response-body-chunk',
+      payload: { requestId: localId, data: Buffer.from(first).toString('base64') },
+    },
+    {
+      sequence: 4,
+      monotonicMs: 4,
+      topic: 'fetch/response-body-chunk',
+      payload: { requestId: localId, data: Buffer.from(second).toString('base64') },
+    },
+    {
+      sequence: 5,
+      monotonicMs: 5,
+      topic: 'fetch/end',
+      payload: { requestId: localId, capturedBytes: first.length + second.length, responseBodyTruncated: false },
+    },
+  ]
+}
+
 function requestId(localId: string): string {
   return `${source.sourceId}:${source.generation}:${localId}`
 }

packages/experimental/inspector/tsconfig.client.json

@@ -79,6 +79,7 @@
     "src/shared/identity.ts",
     "src/shared/index.ts",
     "src/shared/json.ts",
+    "src/shared/network/event-source.ts",
     "src/shared/network/observation.ts",
     "src/shared/service.ts",
     "src/shared/validation.ts"

packages/experimental/inspector/tsconfig.host.json

@@ -81,6 +81,7 @@
     "src/shared/identity.ts",
     "src/shared/index.ts",
     "src/shared/json.ts",
+    "src/shared/network/event-source.ts",
     "src/shared/network/observation.ts",
     "src/shared/service.ts",
     "src/shared/validation.ts",