feat(inspector): project Host fetches through CDP Network

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

@@ -0,0 +1,4 @@
+/** Client network observation is not enabled in the current source producer. */
+
+/** Observation topics published by the Client network adapter. */
+export const NETWORK_TOPICS: readonly string[] = []

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

@@ -0,0 +1,244 @@
+/** Full `globalThis.fetch` capture that publishes without delaying response delivery. */
+
+import type { InspectorJsonValue } from '../../shared/json.ts'
+import type { InspectorPublisher } from '../../shared/bridge/publisher.ts'
+import { FETCH_TOPICS } from '../../shared/bridge/messages/network.ts'
+
+/** Observation topics published by the Host network adapter. */
+export const NETWORK_TOPICS: readonly string[] = FETCH_TOPICS
+
+/** Byte limits for request and response clone capture. */
+export interface FetchCaptureOptions {
+  readonly maxRequestBodyBytes: number
+  readonly maxResponseBodyBytes: number
+  readonly maxChunkBytes: number
+}
+
+interface CaptureOutcome {
+  readonly capturedBytes: number
+  readonly truncated: boolean
+  readonly captureError?: string
+}
+
+/** Active global fetch wrapper. */
+export interface FetchObserver {
+  /** Restore the prior fetch implementation, cancel clone readers, and await their settlement. */
+  stop(): Promise<void>
+}
+
+/**
+ * Install full fetch capture for every later call through `globalThis.fetch`.
+ * @param publisher - Host source that receives fetch lifecycle records.
+ * @param options - Per-body capture limits.
+ * @returns The owner that stops capture and awaits pending body readers.
+ */
+export function installFetchObserver(
+  publisher: InspectorPublisher,
+  options: FetchCaptureOptions,
+): FetchObserver {
+  const descriptor = Object.getOwnPropertyDescriptor(globalThis, 'fetch')
+  const original = globalThis.fetch
+  if (typeof original !== 'function') throw new Error('inspector: globalThis.fetch is unavailable')
+  if (descriptor !== undefined && !('value' in descriptor)) {
+    throw new Error('inspector: globalThis.fetch is an accessor and cannot be observed safely')
+  }
+
+  const controller = new AbortController()
+  const pending = new Set<Promise<void>>()
+  let nextRequestId = 0
+
+  const track = (promise: Promise<void>): void => {
+    pending.add(promise)
+    void promise.then(
+      () => { pending.delete(promise) },
+      () => { pending.delete(promise) },
+    )
+  }
+
+  const observedFetch: typeof fetch = async (input, init) => {
+    const request = new Request(input, init)
+    const requestId = `fetch-${++nextRequestId}`
+    publisher.publish('fetch/start', {
+      requestId,
+      url: request.url,
+      method: request.method,
+      headers: headerEntries(request.headers),
+      hasBody: request.body !== null,
+      wallTimeMs: Date.now(),
+    })
+
+    let requestClone: Request | undefined
+    try {
+      requestClone = request.clone()
+    } catch (error) {
+      publisher.publish('fetch/request-body-end', {
+        requestId,
+        capturedBytes: 0,
+        truncated: false,
+        captureError: renderError(error),
+      })
+    }
+    if (requestClone !== undefined) {
+      track(captureBody(
+        requestClone.body,
+        options.maxRequestBodyBytes,
+        options.maxChunkBytes,
+        controller.signal,
+        (data) => { publisher.publish('fetch/request-body-chunk', { requestId, data }) },
+      ).then((outcome) => {
+        publisher.publish('fetch/request-body-end', compactOutcome(requestId, outcome))
+      }))
+    }
+
+    let response: Response
+    try {
+      response = await Reflect.apply(original, globalThis, [request])
+    } catch (error) {
+      publisher.publish('fetch/error', {
+        requestId,
+        message: renderError(error),
+        canceled: request.signal.aborted || isAbortError(error),
+      })
+      throw error
+    }
+
+    publisher.publish('fetch/response', {
+      requestId,
+      url: response.url || request.url,
+      status: response.status,
+      statusText: response.statusText,
+      headers: headerEntries(response.headers),
+      mimeType: response.headers.get('content-type')?.split(';', 1)[0]?.trim() ?? '',
+    })
+
+    try {
+      const responseClone = response.clone()
+      track(captureBody(
+        responseClone.body,
+        options.maxResponseBodyBytes,
+        options.maxChunkBytes,
+        AbortSignal.any([controller.signal, request.signal]),
+        (data) => { publisher.publish('fetch/response-body-chunk', { requestId, data }) },
+      ).then((outcome) => {
+        if (request.signal.aborted) {
+          publisher.publish('fetch/error', {
+            requestId,
+            message: request.signal.reason === undefined
+              ? 'AbortError: request aborted during response body capture'
+              : renderError(request.signal.reason),
+            canceled: true,
+          })
+          return
+        }
+        publisher.publish('fetch/end', {
+          requestId,
+          capturedBytes: outcome.capturedBytes,
+          responseBodyTruncated: outcome.truncated,
+          ...(outcome.captureError === undefined ? {} : { responseCaptureError: outcome.captureError }),
+        })
+      }))
+    } catch (error) {
+      publisher.publish('fetch/end', {
+        requestId,
+        capturedBytes: 0,
+        responseBodyTruncated: false,
+        responseCaptureError: renderError(error),
+      })
+    }
+    return response
+  }
+
+  Object.defineProperty(observedFetch, 'name', { value: original.name, configurable: true })
+  Object.defineProperty(observedFetch, 'length', { value: original.length, configurable: true })
+  Object.defineProperty(globalThis, 'fetch', descriptor === undefined
+    ? { value: observedFetch, writable: true, configurable: true }
+    : { ...descriptor, value: observedFetch })
+
+  let stopped: Promise<void> | undefined
+  return {
+    stop(): Promise<void> {
+      if (stopped !== undefined) return stopped
+      stopped = (async () => {
+        const current = Object.getOwnPropertyDescriptor(globalThis, 'fetch')
+        if (current !== undefined && 'value' in current && current.value === observedFetch) {
+          if (descriptor === undefined) Reflect.deleteProperty(globalThis, 'fetch')
+          else Object.defineProperty(globalThis, 'fetch', descriptor)
+        }
+        controller.abort()
+        await Promise.allSettled([...pending])
+      })()
+      return stopped
+    },
+  }
+}
+
+async function captureBody(
+  body: ReadableStream<Uint8Array> | null,
+  limit: number,
+  chunkLimit: number,
+  signal: AbortSignal,
+  emit: (base64: string) => void,
+): Promise<CaptureOutcome> {
+  if (body === null) return { capturedBytes: 0, truncated: false }
+  const reader = body.getReader()
+  const abort = (): void => { void reader.cancel(signal.reason).catch(() => undefined) }
+  signal.addEventListener('abort', abort, { once: true })
+  let capturedBytes = 0
+  let truncated = false
+  try {
+    while (!signal.aborted) {
+      const item = await reader.read()
+      if (item.done) break
+      let offset = 0
+      while (offset < item.value.byteLength) {
+        const remaining = limit - capturedBytes
+        if (remaining <= 0) {
+          truncated = true
+          void reader.cancel('inspector body capture limit reached').catch(() => undefined)
+          return { capturedBytes, truncated }
+        }
+        const size = Math.min(chunkLimit, remaining, item.value.byteLength - offset)
+        const chunk = item.value.subarray(offset, offset + size)
+        emit(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength).toString('base64'))
+        capturedBytes += size
+        offset += size
+      }
+    }
+    if (signal.aborted) {
+      void reader.cancel(signal.reason).catch(() => undefined)
+      return { capturedBytes, truncated, captureError: 'inspector stopped during body capture' }
+    }
+    return { capturedBytes, truncated }
+  } catch (error) {
+    return { capturedBytes, truncated, captureError: renderError(error) }
+  } finally {
+    signal.removeEventListener('abort', abort)
+    reader.releaseLock()
+  }
+}
+
+function compactOutcome(requestId: string, outcome: CaptureOutcome): InspectorJsonValue {
+  return {
+    requestId,
+    capturedBytes: outcome.capturedBytes,
+    truncated: outcome.truncated,
+    ...(outcome.captureError === undefined ? {} : { captureError: outcome.captureError }),
+  }
+}
+
+function headerEntries(headers: Headers): [string, string][] {
+  return [...headers.entries()]
+}
+
+function isAbortError(error: unknown): boolean {
+  return error instanceof DOMException && error.name === 'AbortError'
+}
+
+function renderError(error: unknown): string {
+  if (error instanceof Error) return `${error.name}: ${error.message}`
+  try {
+    return String(error)
+  } catch {
+    return 'unrenderable fetch error'
+  }
+}

packages/experimental/inspector/src/shared/bridge/messages/network.ts

@@ -0,0 +1,12 @@
+/** Observation topic names carried by the internal bridge for captured fetches. */
+
+/** Complete set of fetch observation topics. */
+export const FETCH_TOPICS = [
+  'fetch/start',
+  'fetch/request-body-chunk',
+  'fetch/request-body-end',
+  'fetch/response',
+  'fetch/response-body-chunk',
+  'fetch/end',
+  'fetch/error',
+] as const

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

@@ -0,0 +1,53 @@
+/** Full-capture fetch observations sent to the Inspector Worker. */
+
+/** One header entry; arrays retain duplicate header names. */
+export type InspectorHeader = readonly [name: string, value: string]
+
+/** Common request identity. */
+export interface FetchIdentity {
+  readonly requestId: string
+}
+
+/** A high-level global fetch call began. */
+export interface FetchStartPayload extends FetchIdentity {
+  readonly url: string
+  readonly method: string
+  readonly headers: InspectorHeader[]
+  readonly hasBody: boolean
+  readonly wallTimeMs: number
+}
+
+/** One captured request-body chunk. */
+export interface FetchBodyChunkPayload extends FetchIdentity {
+  readonly data: string
+}
+
+/** Terminal state of one captured request body. */
+export interface FetchRequestBodyEndPayload extends FetchIdentity {
+  readonly capturedBytes: number
+  readonly truncated: boolean
+  readonly captureError?: string
+}
+
+/** Fetch resolved with response headers. */
+export interface FetchResponsePayload extends FetchIdentity {
+  readonly url: string
+  readonly status: number
+  readonly statusText: string
+  readonly headers: InspectorHeader[]
+  readonly mimeType: string
+}
+
+/** One captured response-body chunk. */
+/** Fetch capture reached a terminal response-body state. */
+export interface FetchEndPayload extends FetchIdentity {
+  readonly capturedBytes: number
+  readonly responseBodyTruncated: boolean
+  readonly responseCaptureError?: string
+}
+
+/** Fetch rejected before returning a Response. */
+export interface FetchErrorPayload extends FetchIdentity {
+  readonly message: string
+  readonly canceled: boolean
+}

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

@@ -0,0 +1,211 @@
+/** CDP Network projection over the Worker-owned normalized network store. */
+
+import { Buffer } from 'node:buffer'
+import type { InspectorHeader } from '../../../../shared/network/observation.ts'
+import type { NetworkStore, NetworkStoreEvent } from '../../../inspection/network-store.ts'
+
+/** CDP session slice used by the Network domain. */
+export interface NetworkSink {
+  sendEvent(method: string, params: Readonly<Record<string, unknown>>): void
+}
+
+/** 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 unsubscribe: () => void
+
+  constructor(private readonly store: NetworkStore) {
+    this.unsubscribe = store.subscribe((event) => { this.receive(event) })
+  }
+
+  /**
+   * Enable Network for one DevTools connection and replay retained lifecycle events.
+   * @param session - Connection receiving replay and subsequent events.
+   */
+  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)
+  }
+
+  /**
+   * Stop Network events for one DevTools connection.
+   * @param session - Connection leaving the enabled set.
+   */
+  disable(session: NetworkSink): void {
+    this.enabled.delete(session)
+    this.streamedRequests.delete(session)
+  }
+
+  /**
+   * Forget a closed DevTools connection.
+   * @param session - Closed DevTools connection.
+   */
+  detach(session: NetworkSink): void {
+    this.disable(session)
+  }
+
+  /** Release the repository subscription and all connection-local state. */
+  close(): void {
+    this.unsubscribe()
+    this.enabled.clear()
+    this.streamedRequests.clear()
+  }
+
+  /**
+   * Handle one Worker-local Network method.
+   * @param method - CDP method name.
+   * @param params - Parsed request parameters.
+   * @param session - Calling DevTools connection.
+   * @returns The CDP result fields.
+   */
+  handle(method: string, params: Readonly<Record<string, unknown>>, session: NetworkSink): unknown {
+    switch (method) {
+      case 'Network.enable':
+        this.enable(session)
+        return {}
+      case 'Network.disable':
+        this.disable(session)
+        return {}
+      case 'Network.getResponseBody': {
+        const body = this.store.responseBody(params.requestId)
+        return {
+          body: Buffer.from(body.bytes).toString('base64'),
+          base64Encoded: true,
+          dshInspectorTruncated: body.truncated,
+          ...(body.captureError === undefined ? {} : { dshInspectorCaptureError: body.captureError }),
+        }
+      }
+      case 'Network.getRequestPostData': {
+        const body = this.store.requestBody(params.requestId)
+        return {
+          postData: Buffer.from(body.bytes).toString('utf8'),
+          dshInspectorTruncated: body.truncated,
+          ...(body.captureError === undefined ? {} : { dshInspectorCaptureError: body.captureError }),
+        }
+      }
+      case 'Network.streamResourceContent': {
+        const body = this.store.responseBody(params.requestId)
+        if (typeof params.requestId !== 'string') throw new Error('Network requestId must be a string')
+        if (!body.complete) {
+          let requests = this.streamedRequests.get(session)
+          if (requests === undefined) this.streamedRequests.set(session, requests = new Set())
+          requests.add(params.requestId)
+        }
+        return { bufferedData: Buffer.from(body.bytes).toString('base64') }
+      }
+      case 'Network.setCacheDisabled':
+      case 'Network.setBypassServiceWorker':
+      case 'Network.setExtraHTTPHeaders':
+      case 'Network.clearBrowserCache':
+      case 'Network.clearBrowserCookies':
+        return {}
+      default:
+        throw new Error(`unsupported Network method ${method}`)
+    }
+  }
+
+  private receive(event: NetworkStoreEvent): void {
+    if (event.type === 'request-evicted') {
+      for (const [session, requests] of this.streamedRequests) {
+        requests.delete(event.requestKey)
+        if (requests.size === 0) this.streamedRequests.delete(session)
+      }
+      return
+    }
+    for (const session of this.enabled) this.send(session, event)
+  }
+
+  private send(session: NetworkSink, event: Exclude<NetworkStoreEvent, { readonly type: 'request-evicted' }>): void {
+    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',
+        })
+        return
+      case 'response-received':
+        session.sendEvent('Network.responseReceived', {
+          requestId: event.requestId,
+          loaderId: 'dsh-inspector-loader',
+          frameId: 'dsh-inspector-host-frame',
+          timestamp,
+          type: 'Fetch',
+          response: {
+            url: event.url,
+            status: event.status,
+            statusText: event.statusText,
+            headers: cdpHeaders(event.headers),
+            mimeType: event.mimeType,
+            connectionReused: false,
+            connectionId: 0,
+            encodedDataLength: 0,
+            securityState: 'neutral',
+          },
+        })
+        return
+      case 'response-data':
+        session.sendEvent('Network.dataReceived', {
+          requestId: event.requestId,
+          timestamp,
+          dataLength: event.byteLength,
+          encodedDataLength: event.byteLength,
+          ...(this.streamedRequests.get(session)?.has(event.requestKey) === true ? { data: event.data } : {}),
+        })
+        return
+      case 'request-finished':
+        session.sendEvent('Network.loadingFinished', {
+          requestId: event.requestId,
+          timestamp,
+          encodedDataLength: event.encodedDataLength,
+          dshInspectorTruncated: event.truncated,
+        })
+        this.stopStreaming(event.requestKey)
+        return
+      case 'request-failed':
+        session.sendEvent('Network.loadingFailed', {
+          requestId: event.requestId,
+          timestamp,
+          type: 'Fetch',
+          errorText: event.errorText,
+          canceled: event.canceled,
+        })
+        this.stopStreaming(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)
+    }
+  }
+}
+
+function cdpHeaders(entries: readonly InspectorHeader[]): Record<string, string> {
+  const headers: Record<string, string> = Object.create(null) as Record<string, string>
+  for (const [name, value] of entries) {
+    headers[name] = headers[name] === undefined ? value : `${headers[name]}\n${value}`
+  }
+  return headers
+}
+
+function assertNever(value: never): never {
+  throw new Error(`Unexpected network event: ${JSON.stringify(value)}`)
+}

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

@@ -0,0 +1,429 @@
+/** Worker-owned repository of normalized fetch observations and captured bodies. */
+
+import { Buffer } from 'node:buffer'
+import { FETCH_TOPICS } from '../../shared/bridge/messages/network.ts'
+import type { InspectorHeader } from '../../shared/network/observation.ts'
+import { isPlainObject } from '../../shared/json.ts'
+import type { InspectorSourceDescriptor } from '../../shared/bridge/messages/observation.ts'
+import type { IngestedInspectorRecord, InspectorRecordConsumer } from '../bridge/hub.ts'
+
+/** Bounded retention policy for observed network requests. */
+export interface NetworkStoreOptions {
+  readonly maxRetainedRequests: number
+  readonly maxJournalBytes: number
+}
+
+/** Captured body data returned without a CDP representation. */
+export interface CapturedNetworkBody {
+  readonly bytes: Uint8Array
+  readonly truncated: boolean
+  readonly captureError?: string
+  readonly complete: boolean
+}
+
+interface NetworkEventBase {
+  readonly requestKey: string
+  readonly requestId: string
+  readonly timestampMs: number
+}
+
+/** Transport-independent changes emitted by the network repository. */
+export type NetworkStoreEvent =
+  | NetworkEventBase & {
+    readonly type: 'request-started'
+    readonly wallTimeMs: number
+    readonly url: string
+    readonly method: string
+    readonly headers: readonly InspectorHeader[]
+    readonly hasBody: boolean
+  }
+  | NetworkEventBase & {
+    readonly type: 'response-received'
+    readonly url: string
+    readonly status: number
+    readonly statusText: string
+    readonly headers: readonly InspectorHeader[]
+    readonly mimeType: string
+  }
+  | NetworkEventBase & {
+    readonly type: 'response-data'
+    readonly data: string
+    readonly byteLength: number
+  }
+  | NetworkEventBase & {
+    readonly type: 'request-finished'
+    readonly encodedDataLength: number
+    readonly truncated: boolean
+  }
+  | NetworkEventBase & {
+    readonly type: 'request-failed'
+    readonly errorText: string
+    readonly canceled: boolean
+  }
+  | { readonly type: 'request-evicted'; readonly requestKey: string }
+
+type ReplayableNetworkEvent = Exclude<NetworkStoreEvent, { readonly type: 'response-data' | 'request-evicted' }>
+
+interface CapturedRequest {
+  readonly key: string
+  readonly requestId: string
+  readonly sourceId: string
+  readonly requestBody: Buffer[]
+  readonly responseBody: Buffer[]
+  requestBodyBytes: number
+  responseBodyBytes: number
+  requestBodyTruncated: boolean
+  responseBodyTruncated: boolean
+  requestCaptureError?: string
+  responseCaptureError?: string
+  responseSeen: boolean
+  completed: boolean
+}
+
+/** 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 completed: string[] = []
+  private readonly listeners = new Set<(event: NetworkStoreEvent) => void>()
+  private journalBytes = 0
+
+  constructor(private readonly options: NetworkStoreOptions) {}
+
+  replace(source: InspectorSourceDescriptor, records: readonly IngestedInspectorRecord[]): void {
+    this.close(source, 'source state replaced')
+    this.append(source, records)
+  }
+
+  append(source: InspectorSourceDescriptor, records: readonly IngestedInspectorRecord[]): void {
+    for (const record of records) {
+      if (!this.topics.has(record.topic)) continue
+      try {
+        this.ingest(source, record)
+      } catch {
+        // A malformed domain payload loses only that observation; later records remain independently useful.
+      }
+    }
+  }
+
+  close(source: InspectorSourceDescriptor, reason: string): void {
+    for (const request of this.requests.values()) {
+      if (request.sourceId !== source.sourceId || request.completed) continue
+      request.completed = true
+      this.publish({
+        type: 'request-failed',
+        requestKey: request.key,
+        requestId: request.requestId,
+        timestampMs: performance.timeOrigin + performance.now(),
+        errorText: reason,
+        canceled: true,
+      })
+      this.completed.push(request.key)
+    }
+    this.enforceRetention()
+  }
+
+  /**
+   * Read retained request lifecycle events.
+   * @returns Events in observation order.
+   */
+  replay(): readonly ReplayableNetworkEvent[] {
+    return this.journal
+  }
+
+  /**
+   * Subscribe to live request changes and eviction.
+   * @param listener - Consumer called synchronously after each accepted change.
+   * @returns A disposer removing the consumer.
+   */
+  subscribe(listener: (event: NetworkStoreEvent) => void): () => void {
+    this.listeners.add(listener)
+    return () => { this.listeners.delete(listener) }
+  }
+
+  /**
+   * Read one retained request body.
+   * @param requestId - Public request id assigned by this store.
+   * @returns Captured bytes and truncation metadata.
+   */
+  requestBody(requestId: unknown): CapturedNetworkBody {
+    const request = this.requestById(requestId)
+    return body(request.requestBody, request.requestBodyTruncated, request.requestCaptureError, request.completed)
+  }
+
+  /**
+   * Read one retained response body after response headers have arrived.
+   * @param requestId - Public request id assigned by this store.
+   * @returns Captured bytes and truncation metadata.
+   */
+  responseBody(requestId: unknown): CapturedNetworkBody {
+    const request = this.requestById(requestId)
+    if (!request.responseSeen) throw new Error('response headers have not arrived')
+    return body(request.responseBody, request.responseBodyTruncated, request.responseCaptureError, request.completed)
+  }
+
+  /** Release subscribers and all retained request data. */
+  dispose(): void {
+    this.listeners.clear()
+    this.requests.clear()
+    this.journal.length = 0
+    this.completed.length = 0
+    this.journalBytes = 0
+  }
+
+  private ingest(source: InspectorSourceDescriptor, record: IngestedInspectorRecord): void {
+    const payload = requirePayload(record.payload)
+    const localId = stringField(payload, 'requestId')
+    const key = `${source.sourceId}:${source.generation}:${localId}`
+    const timestampMs = source.timeOriginMs + record.monotonicMs
+    if (record.topic === 'fetch/start') {
+      if (this.requests.has(key)) throw new Error('fetch observation reused an active request id')
+      const request: CapturedRequest = {
+        key,
+        requestId: key,
+        sourceId: source.sourceId,
+        requestBody: [],
+        responseBody: [],
+        requestBodyBytes: 0,
+        responseBodyBytes: 0,
+        requestBodyTruncated: false,
+        responseBodyTruncated: false,
+        responseSeen: false,
+        completed: false,
+      }
+      this.requests.set(key, request)
+      this.publish({
+        type: 'request-started',
+        requestKey: key,
+        requestId: request.requestId,
+        timestampMs,
+        wallTimeMs: numberField(payload, 'wallTimeMs'),
+        url: stringField(payload, 'url'),
+        method: stringField(payload, 'method'),
+        headers: headerField(payload, 'headers'),
+        hasBody: booleanField(payload, 'hasBody'),
+      })
+      this.enforceRetention()
+      return
+    }
+    const request = this.requests.get(key)
+    if (request === undefined) return
+    switch (record.topic) {
+      case 'fetch/request-body-chunk':
+        this.appendBody(request, 'request', stringField(payload, 'data'))
+        return
+      case 'fetch/request-body-end': {
+        request.requestBodyTruncated ||= booleanField(payload, 'truncated')
+        const captureError = optionalStringField(payload, 'captureError')
+        if (captureError !== undefined) request.requestCaptureError = captureError
+        return
+      }
+      case 'fetch/response':
+        request.responseSeen = true
+        this.publish({
+          type: 'response-received',
+          requestKey: key,
+          requestId: request.requestId,
+          timestampMs,
+          url: stringField(payload, 'url'),
+          status: numberField(payload, 'status'),
+          statusText: stringField(payload, 'statusText'),
+          headers: headerField(payload, 'headers'),
+          mimeType: stringField(payload, 'mimeType'),
+        })
+        return
+      case 'fetch/response-body-chunk': {
+        const data = stringField(payload, 'data')
+        const byteLength = this.appendBody(request, 'response', data)
+        this.emit({ type: 'response-data', requestKey: key, requestId: request.requestId, timestampMs, data, byteLength })
+        return
+      }
+      case 'fetch/end': {
+        request.responseBodyTruncated ||= booleanField(payload, 'responseBodyTruncated')
+        const captureError = optionalStringField(payload, 'responseCaptureError')
+        if (captureError !== undefined) request.responseCaptureError = captureError
+        this.complete(request, {
+          type: 'request-finished',
+          requestKey: key,
+          requestId: request.requestId,
+          timestampMs,
+          encodedDataLength: request.responseBodyBytes,
+          truncated: request.responseBodyTruncated,
+        })
+        return
+      }
+      case 'fetch/error':
+        this.complete(request, {
+          type: 'request-failed',
+          requestKey: key,
+          requestId: request.requestId,
+          timestampMs,
+          errorText: stringField(payload, 'message'),
+          canceled: booleanField(payload, 'canceled'),
+        })
+        return
+      default:
+        return
+    }
+  }
+
+  private appendBody(request: CapturedRequest, side: 'request' | 'response', encoded: string): number {
+    const bytes = decodeBase64(encoded)
+    this.evictCompletedFor(bytes.byteLength, request.key)
+    const retained = bytes.subarray(0, Math.max(0, this.options.maxJournalBytes - this.journalBytes))
+    if (side === 'request') {
+      if (retained.byteLength > 0) request.requestBody.push(retained)
+      request.requestBodyBytes += retained.byteLength
+      request.requestBodyTruncated ||= retained.byteLength < bytes.byteLength
+    } else {
+      if (retained.byteLength > 0) request.responseBody.push(retained)
+      request.responseBodyBytes += retained.byteLength
+      request.responseBodyTruncated ||= retained.byteLength < bytes.byteLength
+    }
+    this.journalBytes += retained.byteLength
+    this.enforceRetention()
+    return bytes.byteLength
+  }
+
+  private complete(request: CapturedRequest, event: ReplayableNetworkEvent): void {
+    if (request.completed) return
+    request.completed = true
+    this.publish(event)
+    this.completed.push(request.key)
+    this.enforceRetention()
+  }
+
+  private publish(event: ReplayableNetworkEvent): void {
+    this.journal.push(event)
+    this.emit(event)
+  }
+
+  private emit(event: NetworkStoreEvent): void {
+    for (const listener of [...this.listeners]) {
+      try {
+        listener(event)
+      } catch {
+        // One presentation adapter cannot interrupt repository ingestion or sibling consumers.
+      }
+    }
+  }
+
+  private enforceRetention(): void {
+    while (this.requests.size > this.options.maxRetainedRequests || this.journalBytes > this.options.maxJournalBytes) {
+      const key = this.completed.shift() ?? this.oldestActiveRequestKey()
+      if (key === undefined) return
+      const request = this.requests.get(key)
+      if (request === undefined) continue
+      if (!request.completed) {
+        request.completed = true
+        this.publish({
+          type: 'request-failed',
+          requestKey: request.key,
+          requestId: request.requestId,
+          timestampMs: performance.timeOrigin + performance.now(),
+          errorText: 'Inspector retained-request limit exceeded',
+          canceled: true,
+        })
+      }
+      this.evict(request)
+    }
+  }
+
+  private oldestActiveRequestKey(): string | undefined {
+    for (const request of this.requests.values()) {
+      if (!request.completed) return request.key
+    }
+    return undefined
+  }
+
+  private evictCompletedFor(bytes: number, protectedKey: string): void {
+    while (this.journalBytes + bytes > this.options.maxJournalBytes) {
+      const index = this.completed.findIndex(key => key !== protectedKey)
+      if (index === -1) return
+      const [key] = this.completed.splice(index, 1)
+      if (key === undefined) return
+      const request = this.requests.get(key)
+      if (request !== undefined) this.evict(request)
+    }
+  }
+
+  private evict(request: CapturedRequest): void {
+    this.journalBytes -= request.requestBodyBytes + request.responseBodyBytes
+    this.requests.delete(request.key)
+    for (let index = this.journal.length - 1; index >= 0; index--) {
+      if (this.journal[index]?.requestKey === request.key) this.journal.splice(index, 1)
+    }
+    this.emit({ type: 'request-evicted', requestKey: request.key })
+  }
+
+  private requestById(value: unknown): CapturedRequest {
+    if (typeof value !== 'string') throw new Error('Network requestId must be a string')
+    const request = [...this.requests.values()].find(candidate => candidate.requestId === value)
+    if (request === undefined) throw new Error(`No resource with given identifier: ${value}`)
+    return request
+  }
+}
+
+function body(
+  chunks: readonly Buffer[],
+  truncated: boolean,
+  captureError: string | undefined,
+  complete: boolean,
+): CapturedNetworkBody {
+  return {
+    bytes: Buffer.concat(chunks),
+    truncated,
+    complete,
+    ...(captureError === undefined ? {} : { captureError }),
+  }
+}
+
+function decodeBase64(value: string): Buffer {
+  if (value.length === 0 || value.length % 4 !== 0 || !/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/u.test(value)) {
+    throw new Error('fetch payload body chunk must be canonical base64')
+  }
+  const bytes = Buffer.from(value, 'base64')
+  if (bytes.toString('base64') !== value) throw new Error('fetch payload body chunk must be canonical base64')
+  return bytes
+}
+
+function requirePayload(value: unknown): Readonly<Record<string, unknown>> {
+  if (!isPlainObject(value)) throw new Error('fetch payload must be an object')
+  return value
+}
+
+function stringField(value: Readonly<Record<string, unknown>>, name: string): string {
+  const field = value[name]
+  if (typeof field !== 'string') throw new Error(`fetch payload ${name} must be a string`)
+  return field
+}
+
+function optionalStringField(value: Readonly<Record<string, unknown>>, name: string): string | undefined {
+  const field = value[name]
+  if (field !== undefined && typeof field !== 'string') throw new Error(`fetch payload ${name} must be a string`)
+  return field
+}
+
+function numberField(value: Readonly<Record<string, unknown>>, name: string): number {
+  const field = value[name]
+  if (typeof field !== 'number' || !Number.isFinite(field)) throw new Error(`fetch payload ${name} must be finite`)
+  return field
+}
+
+function booleanField(value: Readonly<Record<string, unknown>>, name: string): boolean {
+  const field = value[name]
+  if (typeof field !== 'boolean') throw new Error(`fetch payload ${name} must be boolean`)
+  return field
+}
+
+function headerField(value: Readonly<Record<string, unknown>>, name: string): InspectorHeader[] {
+  const field = value[name]
+  if (!Array.isArray(field)) throw new Error(`fetch payload ${name} must be a header list`)
+  return field.map((entry) => {
+    if (!Array.isArray(entry) || entry.length !== 2 || typeof entry[0] !== 'string' || typeof entry[1] !== 'string') {
+      throw new Error(`fetch payload ${name} contains an invalid header`)
+    }
+    return [entry[0], entry[1]] as const
+  })
+}

packages/experimental/inspector/tests/fetch-observer.host.spec.ts

@@ -0,0 +1,124 @@
+/** Host fetch observation behavior. */
+
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { installFetchObserver, type FetchObserver } from '../src/host/inspection/network.ts'
+import type { InspectorRecordInput } from '../src/shared/bridge/messages/observation.ts'
+import type { InspectorJsonValue } from '../src/shared/json.ts'
+
+describe('full fetch observer', () => {
+  const originalDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'fetch')
+  let observer: FetchObserver | undefined
+
+  afterEach(async () => {
+    await observer?.stop()
+    observer = undefined
+    if (originalDescriptor === undefined) Reflect.deleteProperty(globalThis, 'fetch')
+    else Object.defineProperty(globalThis, 'fetch', originalDescriptor)
+  })
+
+  it('captures complete URL, headers, request body, response headers, and response body', async () => {
+    const records: InspectorRecordInput[] = []
+    const native = vi.fn(async (request: Request) => {
+      expect(await request.clone().text()).toBe('secret request body')
+      return new Response('complete response body', {
+        status: 201,
+        statusText: 'Created',
+        headers: { authorization: 'response secret', 'content-type': 'text/plain' },
+      })
+    })
+    Object.defineProperty(globalThis, 'fetch', { value: native, writable: true, configurable: true })
+    observer = installFetchObserver({
+      publish(topic: string, payload: InspectorJsonValue, monotonicMs = performance.now()) {
+        records.push({ topic, payload, monotonicMs })
+      },
+    }, { maxRequestBodyBytes: 1_024, maxResponseBodyBytes: 1_024, maxChunkBytes: 4 })
+
+    const response = await fetch('https://example.test/path?token=visible', {
+      method: 'POST',
+      headers: { authorization: 'Bearer visible' },
+      body: 'secret request body',
+    })
+    expect(await response.text()).toBe('complete response body')
+    await vi.waitFor(() => { expect(records.some(record => record.topic === 'fetch/end')).toBe(true) })
+
+    const start = payload(records, 'fetch/start')
+    expect(start).toMatchObject({
+      url: 'https://example.test/path?token=visible',
+      method: 'POST',
+    })
+    expect(start.headers).toEqual(expect.arrayContaining([['authorization', 'Bearer visible']]))
+    expect(decodeChunks(records, 'fetch/request-body-chunk')).toBe('secret request body')
+    const responseRecord = payload(records, 'fetch/response')
+    expect(responseRecord.status).toBe(201)
+    expect(responseRecord.headers).toEqual(expect.arrayContaining([['authorization', 'response secret']]))
+    expect(decodeChunks(records, 'fetch/response-body-chunk')).toBe('complete response body')
+    expect(payload(records, 'fetch/request-body-end')).toMatchObject({ truncated: false })
+    expect(payload(records, 'fetch/end')).toMatchObject({ responseBodyTruncated: false })
+  })
+
+  it('marks bodies truncated without changing the caller response', async () => {
+    const records: InspectorRecordInput[] = []
+    Object.defineProperty(globalThis, 'fetch', {
+      value: vi.fn(() => Promise.resolve(new Response('response-long'))),
+      writable: true,
+      configurable: true,
+    })
+    observer = installFetchObserver({
+      publish(topic: string, payload: InspectorJsonValue, monotonicMs = performance.now()) {
+        records.push({ topic, payload, monotonicMs })
+      },
+    }, { maxRequestBodyBytes: 4, maxResponseBodyBytes: 4, maxChunkBytes: 2 })
+
+    const response = await fetch('https://example.test/', { method: 'POST', body: 'request-long' })
+    expect(await response.text()).toBe('response-long')
+    await vi.waitFor(() => { expect(records.some(record => record.topic === 'fetch/end')).toBe(true) })
+
+    expect(decodeChunks(records, 'fetch/request-body-chunk')).toBe('requ')
+    expect(payload(records, 'fetch/request-body-end')).toMatchObject({ capturedBytes: 4, truncated: true })
+    expect(decodeChunks(records, 'fetch/response-body-chunk')).toBe('resp')
+    expect(payload(records, 'fetch/end')).toMatchObject({ capturedBytes: 4, responseBodyTruncated: true })
+  })
+
+  it('reports cancellation after response headers as a canceled request', async () => {
+    const records: InspectorRecordInput[] = []
+    Object.defineProperty(globalThis, 'fetch', {
+      value: vi.fn(async (request: Request) => new Response(new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(Buffer.from('first'))
+          request.signal.addEventListener('abort', () => {
+            controller.error(new DOMException('aborted', 'AbortError'))
+          }, { once: true })
+        },
+      }))),
+      writable: true,
+      configurable: true,
+    })
+    observer = installFetchObserver({
+      publish(topic: string, payload: InspectorJsonValue, monotonicMs = performance.now()) {
+        records.push({ topic, payload, monotonicMs })
+      },
+    }, { maxRequestBodyBytes: 1_024, maxResponseBodyBytes: 1_024, maxChunkBytes: 4 })
+    const abort = new AbortController()
+
+    const response = await fetch('https://example.test/cancel-body', { signal: abort.signal })
+    abort.abort()
+    await expect(response.text()).rejects.toThrow()
+    await vi.waitFor(() => { expect(records.some(record => record.topic === 'fetch/error')).toBe(true) })
+
+    expect(payload(records, 'fetch/error')).toMatchObject({ canceled: true })
+    expect(records.some(record => record.topic === 'fetch/end')).toBe(false)
+  })
+})
+
+function payload(records: readonly InspectorRecordInput[], topic: string): Record<string, unknown> {
+  const record = records.find(candidate => candidate.topic === topic)
+  expect(record).toBeDefined()
+  return record!.payload as Record<string, unknown>
+}
+
+function decodeChunks(records: readonly InspectorRecordInput[], topic: string): string {
+  return Buffer.concat(records
+    .filter(record => record.topic === topic)
+    .map(record => Buffer.from(String((record.payload as Record<string, unknown>).data), 'base64')))
+    .toString('utf8')
+}

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

@@ -0,0 +1,153 @@
+/** Worker-side Network projection behavior. */
+
+import { describe, expect, it, vi } from 'vitest'
+import { NetworkDomain, type NetworkSink } from '../src/worker/cdp/domains/network/session.ts'
+import { NetworkStore } from '../src/worker/inspection/network-store.ts'
+import { inspectorId } from '../src/shared/bridge/ids.ts'
+import type { InspectorSourceDescriptor } from '../src/shared/bridge/messages/observation.ts'
+import type { IngestedInspectorRecord } from '../src/worker/bridge/hub.ts'
+
+const source: InspectorSourceDescriptor = {
+  sourceId: inspectorId<'InspectorSourceId'>('host-network', 'sourceId'),
+  generation: inspectorId<'InspectorSourceGeneration'>('network-generation', 'generation'),
+  kind: 'host',
+  label: 'Host',
+  timeOriginMs: performance.timeOrigin,
+  capabilities: [],
+}
+
+describe('Inspector Network domain', () => {
+  it('bounds incomplete bodies and marks the retained prefix truncated', () => {
+    const sendEvent = vi.fn()
+    const sink: NetworkSink = { sendEvent }
+    const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 4 })
+    const network = new NetworkDomain(store)
+    network.enable(sink)
+    store.append(source, requestRecords('first', 'abcdef'))
+
+    const response = network.handle('Network.getResponseBody', { requestId: requestId('first') }, sink)
+    expect(response).toEqual({
+      body: Buffer.from('abcd').toString('base64'),
+      base64Encoded: true,
+      dshInspectorTruncated: true,
+    })
+    const dataEvent = sendEvent.mock.calls.find(call => call[0] === 'Network.dataReceived')
+    expect(dataEvent?.[1]).toMatchObject({ dataLength: 6, encodedDataLength: 6 })
+    expect(dataEvent?.[1]).not.toHaveProperty('data')
+  })
+
+  it('evicts completed requests before retaining a later body', () => {
+    const sink: NetworkSink = { sendEvent: vi.fn() }
+    const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 4 })
+    const network = new NetworkDomain(store)
+    store.append(source, requestRecords('first', 'aaaa'))
+    store.append(source, requestRecords('second', 'bbbb'))
+
+    expect(() => network.handle('Network.getResponseBody', { requestId: requestId('first') }, sink)).toThrow(
+      'No resource with given identifier',
+    )
+    expect(network.handle('Network.getResponseBody', { requestId: requestId('second') }, sink)).toEqual({
+      body: Buffer.from('bbbb').toString('base64'),
+      base64Encoded: true,
+      dshInspectorTruncated: false,
+    })
+  })
+
+  it('streams later response chunks only to CDP sessions that opted in', () => {
+    const firstSend = vi.fn()
+    const secondSend = vi.fn()
+    const first: NetworkSink = { sendEvent: firstSend }
+    const second: NetworkSink = { sendEvent: secondSend }
+    const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 1_024 })
+    const network = new NetworkDomain(store)
+    network.enable(first)
+    network.enable(second)
+    const records = requestRecords('stream', 'data: first\n\n')
+    store.append(source, records.slice(0, 2))
+
+    expect(network.handle('Network.streamResourceContent', { requestId: requestId('stream') }, first)).toEqual({
+      bufferedData: '',
+    })
+    store.append(source, records.slice(2, 3))
+
+    const firstData = firstSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')
+    const secondData = secondSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')
+    expect(firstData?.[1]).toMatchObject({ data: Buffer.from('data: first\n\n').toString('base64') })
+    expect(secondData?.[1]).not.toHaveProperty('data')
+    expect(network.handle('Network.streamResourceContent', { requestId: requestId('stream') }, second)).toEqual({
+      bufferedData: Buffer.from('data: first\n\n').toString('base64'),
+    })
+
+    const later = Buffer.from('data: second\n\n').toString('base64')
+    store.append(source, [{
+      sequence: 4,
+      monotonicMs: 4,
+      topic: 'fetch/response-body-chunk',
+      payload: { requestId: 'stream', data: later },
+    }])
+    expect(firstSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')?.[1]).toMatchObject({ data: later })
+    expect(secondSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')?.[1]).toMatchObject({ data: later })
+  })
+
+  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 })
+    const network = new NetworkDomain(store)
+    network.enable({ sendEvent: firstSend })
+    store.append(source, requestRecords('active-first', 'first').slice(0, 1))
+    store.append(source, requestRecords('active-second', 'second').slice(0, 1))
+
+    expect(firstSend).toHaveBeenCalledWith('Network.loadingFailed', expect.objectContaining({
+      requestId: requestId('active-first'),
+      canceled: true,
+    }))
+    expect(() => network.handle(
+      'Network.getRequestPostData',
+      { requestId: requestId('active-first') },
+      { sendEvent: vi.fn() },
+    )).toThrow('No resource with given identifier')
+    expect(() => { store.append(source, requestRecords('active-first', 'first').slice(1)) }).not.toThrow()
+
+    store.append(source, requestRecords('active-second', 'second').slice(1))
+    const replay = vi.fn()
+    network.enable({ sendEvent: replay })
+    expect(replay.mock.calls.some(call => call[0] === 'Network.dataReceived')).toBe(false)
+    expect(replay).toHaveBeenCalledTimes(3)
+    expect(replay).toHaveBeenNthCalledWith(1, 'Network.requestWillBeSent', expect.any(Object))
+    expect(replay).toHaveBeenNthCalledWith(2, 'Network.responseReceived', expect.any(Object))
+    expect(replay).toHaveBeenNthCalledWith(3, 'Network.loadingFinished', expect.any(Object))
+  })
+})
+
+function requestRecords(localId: string, body: string): IngestedInspectorRecord[] {
+  return [
+    {
+      sequence: 1,
+      monotonicMs: 1,
+      topic: 'fetch/start',
+      payload: { requestId: localId, url: 'https://example.test/', method: 'GET', headers: [], hasBody: false, wallTimeMs: 1 },
+    },
+    {
+      sequence: 2,
+      monotonicMs: 2,
+      topic: 'fetch/response',
+      payload: { requestId: localId, url: 'https://example.test/', status: 200, statusText: 'OK', headers: [], mimeType: 'text/plain' },
+    },
+    {
+      sequence: 3,
+      monotonicMs: 3,
+      topic: 'fetch/response-body-chunk',
+      payload: { requestId: localId, data: Buffer.from(body).toString('base64') },
+    },
+    {
+      sequence: 4,
+      monotonicMs: 4,
+      topic: 'fetch/end',
+      payload: { requestId: localId, capturedBytes: body.length, responseBodyTruncated: false },
+    },
+  ]
+}
+
+function requestId(localId: string): string {
+  return `${source.sourceId}:${source.generation}:${localId}`
+}