diff --git a/packages/experimental/inspector/src/host/inspection/network.ts b/packages/experimental/inspector/src/host/inspection/network.ts index fcbdadb76d..aa5225a8c5 100644 --- a/packages/experimental/inspector/src/host/inspection/network.ts +++ b/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 { diff --git a/packages/experimental/inspector/src/shared/network/event-source.ts b/packages/experimental/inspector/src/shared/network/event-source.ts new file mode 100644 index 0000000000..29c40a5a1e --- /dev/null +++ b/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 + } + } +} diff --git a/packages/experimental/inspector/src/shared/network/observation.ts b/packages/experimental/inspector/src/shared/network/observation.ts index b3266dc7c7..c1cfadbb7d 100644 --- a/packages/experimental/inspector/src/shared/network/observation.ts +++ b/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 diff --git a/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts b/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts index 1988eca41a..c7ad184007 100644 --- a/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts +++ b/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts @@ -9,10 +9,15 @@ export interface NetworkSink { sendEvent(method: string, params: Readonly>): void } +type RequestStartedEvent = Extract +type NetworkResourceType = 'EventSource' | 'Fetch' + /** Projects retained and live network observations into connection-local CDP state. */ export class NetworkDomain { private readonly enabled = new Set() private readonly streamedRequests = new Map>() + private readonly pendingStarts = new Map>() + private readonly requestTypes = new Map>() 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) } } diff --git a/packages/experimental/inspector/src/worker/inspection/network-store.ts b/packages/experimental/inspector/src/worker/inspection/network-store.ts index 1599667503..cf1d80b90c 100644 --- a/packages/experimental/inspector/src/worker/inspection/network-store.ts +++ b/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 type ReplayableNetworkEvent = Exclude 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(FETCH_TOPICS) private readonly requests = new Map() - 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) } diff --git a/packages/experimental/inspector/tests/event-source.host.spec.ts b/packages/experimental/inspector/tests/event-source.host.spec.ts new file mode 100644 index 0000000000..cec6f87345 --- /dev/null +++ b/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: '' }, + ]) + }) +}) diff --git a/packages/experimental/inspector/tests/integration.host.spec.ts b/packages/experimental/inspector/tests/integration.host.spec.ts index 1e0ab8469a..79fcfe60ea 100644 --- a/packages/experimental/inspector/tests/integration.host.spec.ts +++ b/packages/experimental/inspector/tests/integration.host.spec.ts @@ -466,6 +466,20 @@ describe('experimental Inspector real Worker', () => { && (event.params?.response as Record | 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 }) diff --git a/packages/experimental/inspector/tests/network.host.spec.ts b/packages/experimental/inspector/tests/network.host.spec.ts index 05273e38e1..a658571a8d 100644 --- a/packages/experimental/inspector/tests/network.host.spec.ts +++ b/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}` } diff --git a/packages/experimental/inspector/tsconfig.client.json b/packages/experimental/inspector/tsconfig.client.json index 0c5a2b1f72..b24fdcb91f 100644 --- a/packages/experimental/inspector/tsconfig.client.json +++ b/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" diff --git a/packages/experimental/inspector/tsconfig.host.json b/packages/experimental/inspector/tsconfig.host.json index 45c311d863..9ba0233518 100644 --- a/packages/experimental/inspector/tsconfig.host.json +++ b/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",