diff --git a/.changeset/native-network-interceptor-adapters.md b/.changeset/native-network-interceptor-adapters.md
new file mode 100644
index 00000000..60152ee2
--- /dev/null
+++ b/.changeset/native-network-interceptor-adapters.md
@@ -0,0 +1,6 @@
+---
+'@wdio/selenium-devtools': minor
+'@wdio/nightwatch-devtools': minor
+---
+
+Capture a native app's network in the Network tab through the appium-interceptor plugin, as the WebdriverIO service does. When the session's capabilities set `appium:intercept: true` and Appium runs with `--use-plugins=appium-interceptor`, each request the plugin's proxy recorded is listed with method, URL, status, headers and bodies, in live and trace mode, and lands in the per-test trace of the test that made it. A new session, including one that replaces the previous one mid-run, gets its own listener. Without the plugin the session runs as before and a warning is logged.
diff --git a/.changeset/native-network-interceptor.md b/.changeset/native-network-interceptor.md
new file mode 100644
index 00000000..603d285c
--- /dev/null
+++ b/.changeset/native-network-interceptor.md
@@ -0,0 +1,5 @@
+---
+'@wdio/devtools-service': minor
+---
+
+Capture a native app's network in the Network tab through the appium-interceptor plugin. A native Appium session has no document and no BiDi network events, so its requests were never visible. When the session sets `appium:intercept: true` and Appium runs with `--use-plugins=appium-interceptor`, the service now lists each request the plugin's proxy recorded, with method, URL, status, headers and bodies, in live and trace mode. Without the plugin the session runs as before and a warning is logged.
diff --git a/CLAUDE.md b/CLAUDE.md
index e06da11a..533471c9 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -274,7 +274,7 @@ Documented divergences from the conventions above. They exist today as debt to b
 - **Network has three sources, and the page's Resource Timing is the floor.** BiDi (all four adapters) and Nightwatch's Chrome performance log are driver-side and carry headers and status. The page collector carries two: its patched `fetch`/`XMLHttpRequest` (the only source with bodies) and the document's Resource Timing buffer (`script/src/collectors/resourceTiming.ts`): every load with timing phases and size, a status on Chromium only, never headers. Resource Timing is the only network source a browser without BiDi has, which is Safari and every Appium session (Appium's BiDi reference lists `session.*` alone). The collector reads the buffer at construction, which is what covers loads made before a `<script>`-injected collector existed, then observes; a second evaluation of the bundle in the same document keeps the first collector (next entry), so the buffer is read once per document. Its two page sources are deduplicated inside the collector: the patch *claims* each call (absolute URL without fragment, which is the form the buffer names, plus start time), a `fetch`/`xmlhttprequest` entry within 1 s of a claim is taken so the row with headers and body is the one kept, and because the entry can arrive before the call has answered, a taken entry is *held* until the patch records the call (the entry is then its second listing and dropped) or drops it (no content-type, a rejected fetch, an XHR `error`/`abort`/`timeout`: the held entry becomes the request's only record). Recording a failed call as a failed row instead was tried and reverted: the service and Nightwatch keep the page's patched rows beside BiDi's, so on Chrome every blocked beacon was listed twice (Nightwatch example 35 → 67 failure rows); without BiDi a failed call therefore reads as unknown rather than failed, which is the entry's own account of it. Against a driver-side source the gate is `processTracePayload`'s `skipResourceTimingRequests`, which drops only the Resource Timing rows and leaves the patched rows their bodies, and **the gate is evidence, not a subscription**: measured on Appium 3.7's XCUITest driver, selenium-webdriver's BiDi inspectors attach cleanly and then deliver nothing, so gating on `bidiActive` (or on `browser.isBidi`) withheld the page's rows and the trace had zero network rows. Each adapter therefore latches on the first BiDi network event it receives (service `#bidiNetworkObserved`, set in `handleNetworkRequestStarted`; Selenium and Nightwatch `bidiNetworkObserved`, set in their `buildBidiSinks`; Python `stats["observed"]`, read by `bidi.network_serving`) and gates on that. Nightwatch's performance log is deliberately **not** a gate either: a capability can name a log the driver never serves (XCUITest), and the log is read after the drain in any case, so `perfLogs.ts` `upgradeResourceTimingRows` pairs each log entry with the page's row for the same load (URL, nearest `timestamp`, the log's being dated from CDP's `wallTime` because the monotonic clock only diffs within a session) and lays the log's method, headers, status and size over that row, keeping the row's id, clock and measured phases, and re-sends it under the id, which the dashboard replaces in place. The reverse order is covered too: a load that completes between the drain and the log read reaches the session as the log's row first, so the session keeps it in `loadsListedByLog` until the next drain, where `withoutLoadsTheLogListed` spends it on the page's one row for it (same URL, nearest `timestamp` within 2 s). A ledger spent per load, not a scan of the rows the session holds: the scan read the collector's own patched fetch row, or a row the log had already upgraded, as the listing of a *second* request to that URL inside the window and dropped that request outright (a fetch answered with a content-type-less 204 after one answered with JSON lost its only row). Spent at the next drain that reaches the driver, a collector-less document included, and forgotten at a session replacement (`armReplacedSession`), because the page observed the load before the log could list it: a row absent from that drain is never coming, and a load kept longer would take the row of the next request to the same URL. Residual, Chrome only (Safari and Appium serve no log, so the ledger stays empty there): the log also lists the collector's own patched fetch/XHR calls, whose row is not a Resource Timing row, so `upgradeResourceTimingRows` appends the log's row beside the patched one (main ingested both too) and the ledger can spend it on the row of a different request to that URL inside the window, as it can on the next document's load of a shared asset; the request then survives as its own log row, without the page's phases, so the cost is a row's shape rather than a request. Python withholds the whole page batch once BiDi is serving, as selenium-devtools does (whose console gate, `bidiConsoleObserved`, is evidence-based for the same reason); both lose the patched rows' bodies there rather than list each fetch twice beside its BiDi row, which is what the service's keep does. Residuals: a drain that lands before the first BiDi event of a page is delivered can pass that page's rows once, and Nightwatch clears both flags at a session rotation, so loads in the window before BiDi re-attaches can be listed by both.
   - **A second evaluation of the collector bundle in one document keeps the first collector.** Every adapter probes for the collector before injecting, but two callers can pass the probe before either injection has run (Nightwatch's navigation proxy and its drain's recovery, Selenium's navigation hook and its recovery), and the second evaluation used to replace `window.wdioTraceCollector` and drop everything the first had buffered. Measured on Android Chrome through Appium, where there is no BiDi and the `<script>` path is the only one: Nightwatch injected twice 11 ms apart and the trace had **0 network rows** for a run that loaded two pages, because the replacement also could not re-read the Resource Timing buffer without listing the first collector's loads again. `collector.ts` now exports `alreadyInstalled` and keeps the instance it finds, and `index.ts` skips its bootstrap on that flag, so the second script is a no-op rather than a second observer feeding the same collector; the adapters' probes only save the round trips. The `chrome-native:` new-tab document a mobile Chrome session opens on is ignored with the rest of the browser's own schemes, since a session starts there before its first navigation; that list is `shared/browser-schemes.ts`, read by core's injection gate and the collector's request filter alike, so the two cannot drift again.
   - **selenium-webdriver's `NetworkInspector` hands every network event to every subscriber's callback** (`_on` adds one unfiltered socket listener per subscription), so `core/bidi.ts` `handleBidiRequestSent` also receives response events and `handleBidiResponseCompleted` request events. Unguarded, the request handler pushed a second, never-completed row per request and the response handler finalized rows on request events: measured, every request in a Selenium Chrome trace listed twice (a status-0 row beside the completed one) and 57 of 135 rows in a Nightwatch one. Each handler now ignores the other's event by shape (`response` present or absent). Two more shapes the inspector does not model: a redirect hop re-sends the request under the same id, so rows are kept by id for the session (bounded at 5000, oldest out) and a request for a known id reopens its row under the first hop's clock rather than pushing one; and a failed fetch, which core closes as a failure (status 0, the browser's `errorText`, as the WDIO service's `handleNetworkFetchError` already did) through the inspector's `fetchError` hook, reading `network.fetchError` off the inspector's own connection only for an inspector without one; that handler ignores the request and response events the same dispatch hands it, by `errorText`. Without that, every blocked analytics beacon sat in the Network tab as pending for the life of the session; a browser that refuses the subscription keeps the inspector and only that gap.
-  - A Resource Timing row states `source: 'resource-timing'`, the list colours it `unknown` rather than `pending` when it has no status, the detail card names what it could not see, and the HAR writer adds `_source` and `_resourceType`: HAR's required `status` is written as 0 for it, which read back as-is is a transport failure, and an empty `content.mimeType` cannot restore the type. The method is `''` for a navigation, fetch or beacon (Resource Timing names none) and `GET` only for a parser-initiated load. A native app session has no document and therefore no network capture; only a device- or proxy-level capture outside the adapters sees that traffic.
+  - A Resource Timing row states `source: 'resource-timing'`, the list colours it `unknown` rather than `pending` when it has no status, the detail card names what it could not see, and the HAR writer adds `_source` and `_resourceType`: HAR's required `status` is written as 0 for it, which read back as-is is a transport failure, and an empty `content.mimeType` cannot restore the type. The method is `''` for a navigation, fetch or beacon (Resource Timing names none) and `GET` only for a parser-initiated load. A native app session has no document and therefore no page-side network capture. All four adapters read its traffic from the appium-interceptor plugin's proxy when the session sets `appium:intercept: true`: `core/appium-interceptor.ts` `NativeNetworkCapture` owns the per-session listener and its drains, and each JS adapter supplies only an execute path that renders no rows (the service and Nightwatch raw WebDriver HTTP bound to the session, Selenium the unpatched `executeScript`) plus its drain points (after each command, before a per-test slice is written, at session end). Every read returns the listener's whole history, bodies included, so the per-command drains are throttled to `NATIVE_NETWORK_DRAIN_INTERVAL_MS` while the test-end and session-end drains never are. Python mirrors it in `selenium_devtools/interceptor_network.py`, since it ships no Node, and drains when each test reports its outcome. `appium:intercept` is honoured flat or grouped under `appium:options`. The plugin records no timing, so each row is stamped when first read, and requests made before the listener starts (an app's launch traffic) are not listed.
 - Retry-aware trace policies share one mechanism: adapters feed a per-attempt **outcome ledger** (`core/attempt-tracker.ts` `TestAttemptTracker.recordStart(uid, specFile)` + `recordOutcome(uid, state, attempt)`) keyed by the **retry-stable** uid, and the finalizer reads the scoped views (`all`/`forSpec`/`forTest`) so `trace-retention.ts` evaluates **group-by-test** — `retain-on-failure` keys on each test's *final* attempt (no over-retaining a fail-then-pass), `retain-on-first-failure` on *attempt 0*. `recordStart` on a second attempt stamps the prior attempt `failed` (a retry only follows a failure), which corrects runners that swallow the intermediate failure — e.g. Mocha via a `--require` plugin never surfaces the retried attempt's failure, so the ledger would otherwise see `[passed, passed]`. An empty scoped view falls back to `testMetadata` (never fail-open-retains).
   - **WDIO + Selenium: verified end-to-end** (manual fail-then-pass runs: retained under `retain-on-first-failure`, dropped under `retain-on-failure`).
   - **Nightwatch: `retain-on-failure` works; the other retry-aware policies degrade.** Its `--retries` re-runs the testcase *internally* without re-firing the plugin's per-test hooks, and the per-testcase results carry no attempt/retry field (retries live only in undocumented, version-varying Nightwatch internals — `suiteRetries.testRetriesCount` / `reporter.testResults.retryTest`), so the ledger sees only the final attempt for the `describe/it` and exports-object interfaces. Cucumber scenarios expose per-scenario hooks, so the feed captures their attempts. Not cleanly fixable without depending on those internals.
@@ -343,7 +343,7 @@ Documented divergences from the conventions above. They exist today as debt to b
 - **Every per-action snapshot probe is timeout-guarded** (`core/action-snapshot.ts` `probe`, `SNAPSHOT_DRIVER_PROBE_TIMEOUT_MS`), not just the in-page scripts. Nightwatch's `browser.getCurrentUrl()`/`getTitle()` are QUEUED commands: called from inside the plugin's own command hook they enqueue behind the command still running and never resolve, and one unguarded probe in the capture's `Promise.all` stranded the whole snapshot — 10 of 14 captures never settled, so those actions reached the trace with no DOM, no a11y tree and no element rects. Nightwatch now runs all four probes (url/title/screenshot/script) over the raw WebDriver HTTP transport in `nightwatch-devtools/src/helpers/webdriverHttp.ts`, bypassing the queue entirely — the pattern `takeScreenshotViaHttp` already used. Relatedly, `runWith` treats a `null` script result as its fallback: a driver that answers `null` instead of rejecting (no-such-session, transport-swallowed script error) otherwise handed the serializers a non-array and lost the entire snapshot, screenshot and all.
 - The per-test `screenshot` and `video` options live on the **WDIO `ServiceOptions` only** — not `BaseDevToolsOptions` — because only the service implements them (an option belongs on an adapter until a second adapter consumes it, mirroring the core-helper rule; putting them on the shared base made them appear available in Selenium/Nightwatch and broke those adapters' `Required<>` option types). The policy *types* (`TraceScreenshotPolicy`/`TraceVideoPolicy`) and the capture/slice/encode logic (`core/screenshot-artifact.ts`, `core/video-slice.ts`) are framework-agnostic, so Selenium/Nightwatch adoption was wiring-only — now done (Selenium adds the options on its own `DevToolsOptions` with full inline attach; Nightwatch adds them produce-only — see the Allure-attach entry below). All are gated to `traceGranularity:'test'` (per-test inline Allure); coarser granularities keep artifacts in the manifest. Video records the screencast continuously and slices per-test by wall-time — the session frame buffer is bounded by `maxBufferFrames` (default 2000; decimates keeping first/last), and on non-Chrome the polling recorder issues many `takeScreenshot`s that flood `@wdio/allure-reporter` (pair with `disableWebdriverStepsReporting`).
 - The `filmstrip` option (dense screencast into the trace) is on **`BaseDevToolsOptions`** — the counterexample to the screenshot/video entry above — because all three adapters implement it (the "second consumer → base" rule realized). Core owns the work (`core/screencast-trace.ts` `thinScreencastFrames`/`buildDenseScreencast`; slice windowing in `spec-trace-helpers.ts`); adapters only default the option, un-gate the recorder in trace mode when it's set, and feed `recorder.frames` into the finalize context. Each adapter captures frames while the recorder is still alive (service `onReload` → `#filmstripFrames`; Selenium `onDriverEnd` drain before nulling; Nightwatch `#finalizeCurrentScreencast` snapshot before delegating), and each finalize context spreads `[...accumulated, ...(live recorder frames)]` so a **mid-run** per-spec/per-test slice flush (which fires before the recorder is drained) isn't blank. When dense frames are present they **supersede** the sparse per-action filmstrip (the per-action DOM `elements`/`snapshot` are carried independently by the `frame-snapshot` events, so no DOM data is lost); a run without dense frames keeps the sparse filmstrip, byte-stable with before. Thinning is applied at export; the live session frame buffer is bounded by `maxBufferFrames` (default 2000; see the screenshot/video entry above). Per-test filmstrip slicing follows the same per-test-hook availability as `traceGranularity:'test'` (works for WDIO mocha/cucumber, Selenium mocha, Nightwatch exports-object/cucumber; Nightwatch BDD `describe/it` degrades to session scope per the entry below), and non-Chrome polling carries the same reporter-noise caveat.
-- Per-test artifact **Allure attachment** is cross-adapter via a pluggable sink in `core/allure-artifacts.ts` (`AllureAttachSink`; `captureAndAttachScreenshot`/`captureAndAttachVideo`/`attachTraceArtifact`/`lastRenderedScreenshot` — moved out of the WDIO service). The service supplies a `@wdio/allure-reporter` sink; **Selenium** supplies an `allure-js-commons` `attachment()` sink (runtime-agnostic — attaches under any allure runner adapter — gated on `globalThis.allureTestRuntime`, dynamic-imported as an optional peer dep). An **undefined sink = produce-only** (write file + manifest, skip attach). Caveats: (a) **Selenium inline-attach needs awaited runner hooks** — its mocha `afterEach`/jest `afterEach`/cucumber `After` are now async + awaited so the async produce+attach lands while the allure adapter still holds the current test open (fire-and-forget attaches to the next test or drops); Gherkin `AfterStep`→`onTestEnd` stays fire-and-forget. (b) **Nightwatch inline-attach is unsupported** — `nightwatch-allure` is post-hoc with no live attach API, and `allure-js-commons` no-ops in a Nightwatch run (nothing calls `setGlobalTestRuntime`); Nightwatch is produce-only. Revisit if an `allure-js-commons`-based Nightwatch runtime adapter appears (then it's a sink swap). (c) **video is standalone** in both adapters (recorder starts for `video != off` OR filmstrip in trace mode; the orphan session-webm encode is trace-gated to stop-only).
+- Per-test artifact **Allure attachment** is cross-adapter via a pluggable sink in `core/allure-artifacts.ts` (`AllureAttachSink`; `captureAndAttachScreenshot`/`captureAndAttachVideo`/`attachTraceArtifact`/`lastRenderedScreenshot` — moved out of the WDIO service). The service supplies a `@wdio/allure-reporter` sink; **Selenium** supplies an `allure-js-commons` `attachment()` sink (runtime-agnostic — attaches under any allure runner adapter — gated on `globalThis.allureTestRuntime`, dynamic-imported as an optional peer dep). An **undefined sink = produce-only** (write file + manifest, skip attach). Caveats: (a) **Selenium inline-attach needs awaited runner hooks** — its mocha `afterEach`/jest `afterEach`/cucumber `After` are now async + awaited so the async produce+attach lands while the allure adapter still holds the current test open (fire-and-forget attaches to the next test or drops); Gherkin `AfterStep` is awaited too, so a step's trace slice, native app requests included, is written before the next step records its boundary. (b) **Nightwatch inline-attach is unsupported** — `nightwatch-allure` is post-hoc with no live attach API, and `allure-js-commons` no-ops in a Nightwatch run (nothing calls `setGlobalTestRuntime`); Nightwatch is produce-only. Revisit if an `allure-js-commons`-based Nightwatch runtime adapter appears (then it's a sink swap). (c) **video is standalone** in both adapters (recorder starts for `video != off` OR filmstrip in trace mode; the orphan session-webm encode is trace-gated to stop-only).
 - **node:assert error display** (`core/assert-patcher.ts` `describeAssertFailure`): node auto-generates a per-character COLORED diff as the AssertionError message; once any consumer strips the ANSI (allure-mocha, the app console filter, plain terminals) the actual/expected interleave into mush (`'ExampleThis DIs Nomt…'`). The patcher pulls node's clean `.actual`/`.expected` into a `CollapsedAssertResult` AND **rewrites the auto-generated message** (and its echo in the stack) as a value-bearing `Expected: … / Received: …` block. Because `toError` returns the thrown error itself, this rewrite reaches every consumer — the trace's Errors tab, the runner console, and **allure-mocha's error box** (which would otherwise show the ANSI-stripped mush). It only touches node's auto-generated messages; a user-supplied message and errors without `actual`/`expected` (e.g. `assert.ok`) pass through unchanged. This is a deliberate mutation of the thrown assert error — node's auto-message is a display artifact, and the rewrite is strictly more readable and matches `@wdio/allure-reporter`'s `Expected/Received` shape.
 - Nightwatch native asserts are buffered at call time and emitted in one batch at test-end (Nightwatch exposes no per-assertion execution/outcome hook reachable from the plugin — `client.queue.tree` / `client.reporter` aren't on `browser`). Streaming mid-run would flash every assert green before its outcome is known, so rows are held and reconciled against `results.testcases[*].assertions` at `afterEach` (the flat `results.assertions` only reflects the last testcase). The exporter re-sorts commands by timestamp (`buildActionEvents`) so the batched rows land at their real timeline positions rather than clustering at flush time. `actual` is parsed from the failure message (`but got: …`, failures only) into a collapsed `{passed, expected, actual}` result; `expected` is the assertion arg — the label still mirrors the call args, not the derived values.
 - Nightwatch's BDD `describe/it` interface fires the plugin's global `beforeEach`/`afterEach` **once per module** (with an empty `currentTest.name`), not per `it` — Nightwatch runs the individual `it`s internally with no per-testcase hook or event reachable from the plugin (the lib emits only transport-level events). So `traceGranularity:'test'` records a single slice boundary (the first test) and collapses to one session-scoped slice; retry-aware/per-test retention (`tracePolicy`) degrades to session scope for this interface. Per-test slicing works for adapters/styles that expose per-test hooks (WDIO mocha/cucumber, Selenium mocha; likely Nightwatch's exports-object style, unverified). Session-granularity trace is unaffected. **No per-test support has been built for this interface** — real per-`it` slicing needs a hook Nightwatch doesn't surface; deferred to a full-picture pass, not attempted yet.
diff --git a/examples/MOBILE.md b/examples/MOBILE.md
index ccbcb0d1..54f4ef91 100644
--- a/examples/MOBILE.md
+++ b/examples/MOBILE.md
@@ -449,7 +449,7 @@ a failure as "the device disagreed" rather than "the example is broken".
 
 ## Known gaps
 
-Both are tracked, and both are visible in these examples rather than hidden:
+All are tracked, and all are visible in these examples rather than hidden:
 
 - **Selenium and Nightwatch publish no viewport** (#373). The reader then falls
   back to `1280 × 720`, which is landscape — and the player picks the stacked
@@ -457,3 +457,8 @@ Both are tracked, and both are visible in these examples rather than hidden:
   does not get the device column until that is fixed. WDIO and Python do.
 - **A native session gets no accessibility tree from Selenium, Nightwatch or
   Python** (#372). Only the WDIO service derives one from the app's page source.
+- **A native session captures no network by default** (#444). All four
+  adapters can read it from the appium-interceptor plugin when the session
+  sets `appium:intercept: true` (see each adapter's README). The Clock app
+  these examples drive makes no requests, so there is nothing to see here
+  either way.
diff --git a/packages/core/src/appium-interceptor.ts b/packages/core/src/appium-interceptor.ts
new file mode 100644
index 00000000..407903de
--- /dev/null
+++ b/packages/core/src/appium-interceptor.ts
@@ -0,0 +1,287 @@
+// Native app network, read from the appium-interceptor plugin's proxy: a native app has no document and Appium serves no BiDi network events.
+
+import {
+  getRequestType,
+  isNativeAppSession,
+  type NetworkRequest
+} from '@wdio/devtools-shared'
+import { errorMessage } from './error.js'
+
+const INTERCEPT_CAPABILITY = 'appium:intercept'
+const APPIUM_OPTIONS = 'appium:options'
+
+/** Per-command drains are throttled to this: each read returns the listener's whole history, bodies included. */
+export const NATIVE_NETWORK_DRAIN_INTERVAL_MS = 2000
+
+/** Runs an Appium execute method (`interceptor: …`) outside the client's own command path, so it never renders as an action row. */
+export type InterceptorExecute = (
+  script: string,
+  args: unknown[]
+) => Promise<unknown>
+
+export type NativeNetworkLog = (level: 'info' | 'warn', message: string) => void
+
+/** One entry of the plugin's `interceptor: getInterceptedData`. */
+export interface InterceptedRequest {
+  url: string
+  method: string
+  statusCode: number
+  requestHeaders?: Record<string, string | string[]>
+  requestBody?: unknown
+  responseHeaders?: Record<string, string | string[]>
+  responseBody?: unknown
+}
+
+export function flattenInterceptedHeaders(
+  headers: InterceptedRequest['requestHeaders']
+): Record<string, string> {
+  const flat: Record<string, string> = {}
+  for (const [name, value] of Object.entries(headers ?? {})) {
+    // The plugin files the response status line as a header ("http/1.1 201 created").
+    if (/\s/.test(name)) {
+      continue
+    }
+    flat[name.toLowerCase()] = Array.isArray(value) ? value.join(', ') : value
+  }
+  return flat
+}
+
+export function interceptedBodyText(body: unknown): string | undefined {
+  if (body === undefined || body === null || body === '') {
+    return undefined
+  }
+  return typeof body === 'string' ? body : JSON.stringify(body)
+}
+
+/** The plugin records no timing, so a row is stamped when it was first read. */
+export function interceptedToNetworkRequest(
+  entry: InterceptedRequest,
+  id: string,
+  observedAt: { timestamp: number; startTime: number }
+): NetworkRequest {
+  const responseHeaders = flattenInterceptedHeaders(entry.responseHeaders)
+  const responseBody = interceptedBodyText(entry.responseBody)
+  return {
+    id,
+    url: entry.url,
+    method: entry.method,
+    status: entry.statusCode,
+    type: getRequestType(entry.url, responseHeaders['content-type']),
+    timestamp: observedAt.timestamp,
+    startTime: observedAt.startTime,
+    endTime: observedAt.startTime,
+    requestHeaders: flattenInterceptedHeaders(entry.requestHeaders),
+    responseHeaders,
+    response: {
+      fromCache: false,
+      headers: responseHeaders,
+      mimeType: responseHeaders['content-type']?.split(';')[0]?.trim() ?? '',
+      status: entry.statusCode
+    },
+    requestBody: interceptedBodyText(entry.requestBody),
+    responseBody,
+    size:
+      responseBody === undefined ? undefined : Buffer.byteLength(responseBody)
+  }
+}
+
+export class InterceptorNetworkPoller {
+  #execute: InterceptorExecute
+  #listenerId?: string
+  #seen = 0
+  // Drains are chained: two overlapping reads would both list the entries past the same `#seen`.
+  #tail: Promise<NetworkRequest[]> = Promise.resolve([])
+
+  constructor(execute: InterceptorExecute) {
+    this.#execute = execute
+  }
+
+  async start(): Promise<boolean> {
+    const id = await this.#execute('interceptor: startListening', [{}])
+    this.#listenerId = typeof id === 'string' ? id : undefined
+    return this.#listenerId !== undefined
+  }
+
+  drain(): Promise<NetworkRequest[]> {
+    // A failed read is reported to its own caller; the next drain still runs.
+    this.#tail = this.#tail.catch(() => []).then(() => this.#readFresh())
+    return this.#tail
+  }
+
+  async #readFresh(): Promise<NetworkRequest[]> {
+    const listenerId = this.#listenerId
+    if (!listenerId) {
+      return []
+    }
+    const data = await this.#execute('interceptor: getInterceptedData', [
+      { id: listenerId }
+    ])
+    // A transport that answers failures with null would otherwise read as "nothing new".
+    if (!Array.isArray(data)) {
+      throw new Error('the plugin returned no intercepted data')
+    }
+    if (data.length <= this.#seen) {
+      return []
+    }
+    const observedAt = { timestamp: Date.now(), startTime: performance.now() }
+    const fresh = (data as InterceptedRequest[])
+      .slice(this.#seen)
+      .map((entry, offset) =>
+        interceptedToNetworkRequest(
+          entry,
+          `interceptor-${listenerId}-${this.#seen + offset}`,
+          observedAt
+        )
+      )
+    this.#seen = data.length
+    return fresh
+  }
+}
+
+/** Whether one capability set asks for the proxy, directly or grouped under `appium:options`. */
+function asksForIntercept(caps: unknown): boolean {
+  if (!caps || typeof caps !== 'object') {
+    return false
+  }
+  const map = caps as { get?: (key: string) => unknown } & Record<
+    string,
+    unknown
+  >
+  const read = (key: string) =>
+    typeof map.get === 'function' ? map.get(key) : map[key]
+  if (read(INTERCEPT_CAPABILITY) === true) {
+    return true
+  }
+  const grouped = read(APPIUM_OPTIONS)
+  return (
+    !!grouped &&
+    typeof grouped === 'object' &&
+    (grouped as Record<string, unknown>).intercept === true
+  )
+}
+
+/** Whether requested capabilities ask appium-interceptor for its proxy: flat, W3C-wrapped, or a selenium `Capabilities` map. */
+export function requestsInterceptedNetwork(capabilities: unknown): boolean {
+  if (asksForIntercept(capabilities)) {
+    return true
+  }
+  if (!capabilities || typeof capabilities !== 'object') {
+    return false
+  }
+  const { alwaysMatch, firstMatch } = capabilities as {
+    alwaysMatch?: unknown
+    firstMatch?: unknown
+  }
+  return [alwaysMatch, ...(Array.isArray(firstMatch) ? firstMatch : [])].some(
+    asksForIntercept
+  )
+}
+
+/** Where a session's intercepted requests come from, resolved by each adapter from its own client. */
+export interface InterceptedNetworkTarget {
+  requested: boolean
+  capabilities: unknown
+  /** Undefined when the adapter cannot reach the driver outside its own command path. */
+  execute: InterceptorExecute | undefined
+  /** Why `execute` is undefined, in the adapter's own terms. */
+  unavailableReason?: string
+}
+
+export interface NetworkRequestSink {
+  recordNetworkRequests(requests: NetworkRequest[]): void
+}
+
+/** Per-session lifecycle every adapter shares: start a listener per session, drain it into the capturer. */
+export class NativeNetworkCapture {
+  #sink: NetworkRequestSink
+  #onLog: NativeNetworkLog
+  #now: () => number
+  #lastDrainAt = Number.NEGATIVE_INFINITY
+  #poller?: InterceptorNetworkPoller
+  // A start superseded by a later one (session replaced mid-handshake) must not install its poller.
+  #generation = 0
+
+  constructor(
+    sink: NetworkRequestSink,
+    onLog: NativeNetworkLog,
+    now: () => number = Date.now
+  ) {
+    this.#sink = sink
+    this.#onLog = onLog
+    this.#now = now
+  }
+
+  /** Called per session: a listener dies with its session. */
+  async start(target: InterceptedNetworkTarget | undefined): Promise<void> {
+    const generation = ++this.#generation
+    this.#poller = undefined
+    if (!target?.requested || !isNativeAppSession(target.capabilities)) {
+      return
+    }
+    if (!target.execute) {
+      this.#onLog(
+        'warn',
+        `${INTERCEPT_CAPABILITY} is set but ${target.unavailableReason ?? 'the driver cannot be reached outside the command path'}; native network is not captured`
+      )
+      return
+    }
+    const poller = new InterceptorNetworkPoller(target.execute)
+    const started = await poller.start().catch((err: unknown) => {
+      this.#onLog(
+        'warn',
+        `appium-interceptor listener failed to start: ${errorMessage(err)}`
+      )
+      return false
+    })
+    if (generation !== this.#generation) {
+      return
+    }
+    if (!started) {
+      this.#onLog(
+        'warn',
+        `${INTERCEPT_CAPABILITY} is set but the interceptor listener did not start; is Appium running with --use-plugins=appium-interceptor?`
+      )
+      return
+    }
+    this.#onLog('info', 'Native network capture active via appium-interceptor')
+    this.#poller = poller
+    this.#lastDrainAt = Number.NEGATIVE_INFINITY
+  }
+
+  /** Whether a listener is running, so a caller can skip a drain that cannot find anything. */
+  get active(): boolean {
+    return this.#poller !== undefined
+  }
+
+  /** At session end, so a drain after the driver quit does not reach a dead session. */
+  stop(): void {
+    this.#generation++
+    this.#poller = undefined
+  }
+
+  /** Lists what the proxy recorded since the last call. A test or session boundary must not pass `throttle`. */
+  async forward(options: { throttle?: boolean } = {}): Promise<void> {
+    const poller = this.#poller
+    if (!poller) {
+      return
+    }
+    const now = this.#now()
+    if (
+      options.throttle &&
+      now - this.#lastDrainAt < NATIVE_NETWORK_DRAIN_INTERVAL_MS
+    ) {
+      return
+    }
+    this.#lastDrainAt = now
+    const requests = await poller.drain().catch((err: unknown) => {
+      this.#onLog(
+        'warn',
+        `Reading appium-interceptor data failed: ${errorMessage(err)}`
+      )
+      return []
+    })
+    if (requests.length > 0) {
+      this.#sink.recordNetworkRequests(requests)
+    }
+  }
+}
diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts
index 7b340587..540a05cc 100644
--- a/packages/core/src/index.ts
+++ b/packages/core/src/index.ts
@@ -21,6 +21,7 @@ export * from './screenshot-artifact.js'
 export * from './video-slice.js'
 export * from './with-timeout.js'
 export * from './webdriver-http.js'
+export * from './appium-interceptor.js'
 export * from './viewport.js'
 export * from './assert-patcher.js'
 export * from './element-snapshot.js'
diff --git a/packages/core/src/session-capturer.ts b/packages/core/src/session-capturer.ts
index b2c1971a..e997f3c4 100644
--- a/packages/core/src/session-capturer.ts
+++ b/packages/core/src/session-capturer.ts
@@ -495,14 +495,9 @@ export abstract class SessionCapturerBase {
       this.sendUpstream('consoleLogs', tagged)
     }
 
-    const reqs = this.#acceptedNetworkRequests(networkRequests, opts)
-    if (reqs.length > 0) {
-      // A re-sent id replaces its row, as the dashboard does.
-      for (const req of reqs) {
-        this.#upsertNetworkRequest(req)
-      }
-      this.sendUpstream('networkRequests', reqs)
-    }
+    this.recordNetworkRequests(
+      this.#acceptedNetworkRequests(networkRequests, opts)
+    )
 
     if (Array.isArray(mutations) && mutations.length > 0) {
       const batch = mutations as TraceMutation[]
@@ -533,6 +528,17 @@ export abstract class SessionCapturerBase {
     }))
   }
 
+  /** Lists network rows and sends them in one message; a re-sent id replaces its row, as the dashboard does. */
+  recordNetworkRequests(reqs: NetworkRequest[]): void {
+    if (reqs.length === 0) {
+      return
+    }
+    for (const req of reqs) {
+      this.#upsertNetworkRequest(req)
+    }
+    this.sendUpstream('networkRequests', reqs)
+  }
+
   /** Row index by request id; checked against the row before use, since adapters also write rows by index. */
   #networkRequestIndex = new Map<string, number>()
 
diff --git a/packages/core/tests/appium-interceptor.test.ts b/packages/core/tests/appium-interceptor.test.ts
new file mode 100644
index 00000000..d9c29506
--- /dev/null
+++ b/packages/core/tests/appium-interceptor.test.ts
@@ -0,0 +1,401 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import {
+  InterceptorNetworkPoller,
+  NATIVE_NETWORK_DRAIN_INTERVAL_MS,
+  NativeNetworkCapture,
+  requestsInterceptedNetwork,
+  flattenInterceptedHeaders,
+  interceptedBodyText,
+  interceptedToNetworkRequest,
+  type InterceptedRequest
+} from '../src/appium-interceptor.js'
+
+const execute = vi.fn<(script: string, args: unknown[]) => Promise<unknown>>()
+
+const entry = (over: Partial<InterceptedRequest> = {}): InterceptedRequest => ({
+  url: 'https://api.example.com/items',
+  method: 'GET',
+  statusCode: 200,
+  responseHeaders: { 'content-type': 'application/json; charset=utf-8' },
+  responseBody: { ok: true },
+  ...over
+})
+
+describe('flattenInterceptedHeaders', () => {
+  it('joins repeated headers and drops the status line the plugin files as a header', () => {
+    expect(
+      flattenInterceptedHeaders({
+        'http/1.1 201 created': '',
+        'set-cookie': ['a=1', 'b=2'],
+        'content-type': 'text/plain'
+      })
+    ).toEqual({ 'set-cookie': 'a=1, b=2', 'content-type': 'text/plain' })
+  })
+
+  it('lowercases names, so a mixed-case Content-Type is found', () => {
+    expect(
+      flattenInterceptedHeaders({ 'Content-Type': 'application/json' })
+    ).toEqual({ 'content-type': 'application/json' })
+  })
+
+  it('returns an empty map for absent headers', () => {
+    expect(flattenInterceptedHeaders(undefined)).toEqual({})
+  })
+})
+
+describe('interceptedBodyText', () => {
+  it('keeps a string, serializes a parsed body, and reads empty as absent', () => {
+    expect(interceptedBodyText('raw')).toBe('raw')
+    expect(interceptedBodyText({ a: 1 })).toBe('{"a":1}')
+    expect(interceptedBodyText('')).toBeUndefined()
+    expect(interceptedBodyText(null)).toBeUndefined()
+    expect(interceptedBodyText(undefined)).toBeUndefined()
+  })
+})
+
+describe('interceptedToNetworkRequest', () => {
+  it('maps an entry onto a network row stamped when it was observed', () => {
+    const row = interceptedToNetworkRequest(
+      entry({ requestHeaders: { accept: '*/*' }, requestBody: 'q=1' }),
+      'interceptor-1-0',
+      { timestamp: 1000, startTime: 50 }
+    )
+    expect(row).toMatchObject({
+      id: 'interceptor-1-0',
+      url: 'https://api.example.com/items',
+      method: 'GET',
+      status: 200,
+      type: 'fetch',
+      timestamp: 1000,
+      startTime: 50,
+      endTime: 50,
+      requestHeaders: { accept: '*/*' },
+      requestBody: 'q=1',
+      responseBody: '{"ok":true}',
+      size: 11,
+      response: {
+        fromCache: false,
+        mimeType: 'application/json',
+        status: 200
+      }
+    })
+  })
+
+  it('measures size in bytes, not characters', () => {
+    const row = interceptedToNetworkRequest(
+      entry({ responseBody: 'é' }),
+      'id',
+      { timestamp: 0, startTime: 0 }
+    )
+    expect(row.size).toBe(2)
+  })
+
+  it('leaves mime type and size empty when the response carries neither', () => {
+    const row = interceptedToNetworkRequest(
+      entry({ responseHeaders: undefined, responseBody: undefined }),
+      'id',
+      { timestamp: 0, startTime: 0 }
+    )
+    expect(row.response?.mimeType).toBe('')
+    expect(row.size).toBeUndefined()
+  })
+})
+
+describe('InterceptorNetworkPoller', () => {
+  beforeEach(() => {
+    execute.mockReset()
+  })
+
+  it('starts a listener and reads only the entries it has not listed yet', async () => {
+    const poller = new InterceptorNetworkPoller(execute)
+    execute.mockResolvedValueOnce('L1')
+    expect(await poller.start()).toBe(true)
+    expect(execute).toHaveBeenLastCalledWith('interceptor: startListening', [
+      {}
+    ])
+
+    execute.mockResolvedValueOnce([entry()])
+    const first = await poller.drain()
+    expect(first.map((r) => r.id)).toEqual(['interceptor-L1-0'])
+    expect(execute).toHaveBeenLastCalledWith(
+      'interceptor: getInterceptedData',
+      [{ id: 'L1' }]
+    )
+
+    execute.mockResolvedValueOnce([entry(), entry({ method: 'POST' })])
+    const second = await poller.drain()
+    expect(second.map((r) => [r.id, r.method])).toEqual([
+      ['interceptor-L1-1', 'POST']
+    ])
+
+    execute.mockResolvedValueOnce([entry(), entry({ method: 'POST' })])
+    expect(await poller.drain()).toEqual([])
+  })
+
+  it('does not list an entry twice when drains overlap', async () => {
+    const poller = new InterceptorNetworkPoller(execute)
+    execute.mockResolvedValueOnce('L1')
+    await poller.start()
+    execute.mockResolvedValue([entry()])
+    const [a, b] = await Promise.all([poller.drain(), poller.drain()])
+    expect([...a, ...b]).toHaveLength(1)
+  })
+
+  it('reports a failed start and then reads nothing', async () => {
+    const poller = new InterceptorNetworkPoller(execute)
+    execute.mockResolvedValueOnce(null)
+    expect(await poller.start()).toBe(false)
+    expect(await poller.drain()).toEqual([])
+    expect(execute).toHaveBeenCalledTimes(1)
+  })
+
+  // A transport that resolves null on failure must surface as a failed read, not as "nothing new".
+  it('rejects a failed or malformed read and keeps its position', async () => {
+    const poller = new InterceptorNetworkPoller(execute)
+    execute.mockResolvedValueOnce('L1')
+    await poller.start()
+    execute.mockResolvedValueOnce(null)
+    await expect(poller.drain()).rejects.toThrow('no intercepted data')
+    execute.mockResolvedValueOnce({ error: 'x' })
+    await expect(poller.drain()).rejects.toThrow('no intercepted data')
+    execute.mockResolvedValueOnce([entry()])
+    expect((await poller.drain()).map((r) => r.id)).toEqual([
+      'interceptor-L1-0'
+    ])
+  })
+})
+
+describe('InterceptorNetworkPoller after a failed read', () => {
+  it('reports the failure to that drain and still reads on the next one', async () => {
+    const poller = new InterceptorNetworkPoller(execute)
+    execute.mockReset()
+    execute.mockResolvedValueOnce('L1')
+    await poller.start()
+    execute.mockRejectedValueOnce(new Error('socket hang up'))
+    await expect(poller.drain()).rejects.toThrow('socket hang up')
+    execute.mockResolvedValueOnce([entry()])
+    expect((await poller.drain()).map((r) => r.id)).toEqual([
+      'interceptor-L1-0'
+    ])
+  })
+})
+
+describe('requestsInterceptedNetwork', () => {
+  it('reads flat, W3C-wrapped and map-style capabilities', () => {
+    expect(requestsInterceptedNetwork({ 'appium:intercept': true })).toBe(true)
+    expect(
+      requestsInterceptedNetwork({ alwaysMatch: { 'appium:intercept': true } })
+    ).toBe(true)
+    expect(
+      requestsInterceptedNetwork({
+        firstMatch: [{}, { 'appium:intercept': true }]
+      })
+    ).toBe(true)
+    const map = new Map<string, unknown>([['appium:intercept', true]])
+    expect(requestsInterceptedNetwork(map)).toBe(true)
+  })
+
+  // Appium merges `appium:options` into the session, so the plugin starts its proxy for these too.
+  it('reads intercept grouped under appium:options, at every level', () => {
+    const grouped = { 'appium:options': { intercept: true } }
+    expect(requestsInterceptedNetwork(grouped)).toBe(true)
+    expect(requestsInterceptedNetwork({ alwaysMatch: grouped })).toBe(true)
+    expect(requestsInterceptedNetwork({ firstMatch: [{}, grouped] })).toBe(true)
+    expect(
+      requestsInterceptedNetwork(
+        new Map<string, unknown>([['appium:options', { intercept: true }]])
+      )
+    ).toBe(true)
+    expect(
+      requestsInterceptedNetwork({ 'appium:options': { intercept: 'yes' } })
+    ).toBe(false)
+  })
+
+  it('needs the literal true', () => {
+    expect(requestsInterceptedNetwork({ 'appium:intercept': 'yes' })).toBe(
+      false
+    )
+    expect(requestsInterceptedNetwork({})).toBe(false)
+    expect(requestsInterceptedNetwork(undefined)).toBe(false)
+  })
+})
+
+describe('NativeNetworkCapture', () => {
+  const NATIVE = {
+    platformName: 'Android',
+    'appium:automationName': 'UiAutomator2'
+  }
+  const sink = () => ({ recordNetworkRequests: vi.fn() })
+  const onLog = vi.fn()
+
+  beforeEach(() => {
+    execute.mockReset()
+    onLog.mockReset()
+  })
+
+  it('starts a listener for a native session that asked for the proxy and records what it drains', async () => {
+    const target = sink()
+    const capture = new NativeNetworkCapture(target, onLog)
+    execute.mockResolvedValueOnce('L1')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    execute.mockResolvedValueOnce([entry()]).mockResolvedValueOnce([entry()])
+    await capture.forward()
+    await capture.forward()
+    expect(target.recordNetworkRequests).toHaveBeenCalledTimes(1)
+    expect(onLog).toHaveBeenCalledWith('info', expect.any(String))
+  })
+
+  it('stays off unless asked, for a session with a document, and with no target', async () => {
+    const capture = new NativeNetworkCapture(sink(), onLog)
+    await capture.start({ requested: false, capabilities: NATIVE, execute })
+    await capture.start({
+      requested: true,
+      capabilities: { platformName: 'Android', browserName: 'chrome' },
+      execute
+    })
+    await capture.start(undefined)
+    expect(execute).not.toHaveBeenCalled()
+  })
+
+  it('warns in neutral terms, or with the reason the adapter gives, when the driver is not reachable', async () => {
+    const capture = new NativeNetworkCapture(sink(), onLog)
+    await capture.start({
+      requested: true,
+      capabilities: NATIVE,
+      execute: undefined
+    })
+    expect(onLog).toHaveBeenLastCalledWith(
+      'warn',
+      expect.stringContaining('cannot be reached outside the command path')
+    )
+    await capture.start({
+      requested: true,
+      capabilities: NATIVE,
+      execute: undefined,
+      unavailableReason: 'executeScript was not patched'
+    })
+    expect(onLog).toHaveBeenLastCalledWith(
+      'warn',
+      expect.stringContaining('executeScript was not patched')
+    )
+  })
+
+  it('logs a read the transport answered with null', async () => {
+    const capture = new NativeNetworkCapture(sink(), onLog)
+    execute.mockResolvedValueOnce('L1')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    execute.mockResolvedValueOnce(null)
+    await capture.forward()
+    expect(onLog).toHaveBeenLastCalledWith(
+      'warn',
+      expect.stringContaining('no intercepted data')
+    )
+  })
+
+  it('throttles per-command drains but never a boundary drain', async () => {
+    let now = 10_000
+    const capture = new NativeNetworkCapture(sink(), onLog, () => now)
+    execute.mockResolvedValueOnce('L1')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    execute.mockResolvedValue([])
+    const reads = () =>
+      execute.mock.calls.filter(
+        ([script]) => script === 'interceptor: getInterceptedData'
+      ).length
+
+    await capture.forward({ throttle: true })
+    expect(reads()).toBe(1)
+    now += NATIVE_NETWORK_DRAIN_INTERVAL_MS - 1
+    await capture.forward({ throttle: true })
+    expect(reads()).toBe(1)
+    await capture.forward()
+    expect(reads()).toBe(2)
+    now += NATIVE_NETWORK_DRAIN_INTERVAL_MS
+    await capture.forward({ throttle: true })
+    expect(reads()).toBe(3)
+  })
+
+  it('does not throttle the first drain of a new session', async () => {
+    const now = 10_000
+    const capture = new NativeNetworkCapture(sink(), onLog, () => now)
+    execute.mockResolvedValueOnce('L1')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    execute.mockResolvedValue([])
+    await capture.forward({ throttle: true })
+    execute.mockResolvedValueOnce('L2')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    await capture.forward({ throttle: true })
+    expect(
+      execute.mock.calls.filter(
+        ([script]) => script === 'interceptor: getInterceptedData'
+      )
+    ).toHaveLength(2)
+  })
+
+  it('warns and forwards nothing when the plugin does not answer or the start throws', async () => {
+    const target = sink()
+    const capture = new NativeNetworkCapture(target, onLog)
+    execute.mockResolvedValueOnce(null)
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    execute.mockRejectedValueOnce(new Error('unknown command'))
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    await capture.forward()
+    expect(execute).toHaveBeenCalledTimes(2)
+    expect(target.recordNetworkRequests).not.toHaveBeenCalled()
+    expect(onLog).toHaveBeenCalledWith(
+      'warn',
+      expect.stringContaining('--use-plugins=appium-interceptor')
+    )
+    expect(onLog).toHaveBeenCalledWith(
+      'warn',
+      expect.stringContaining('unknown command')
+    )
+  })
+
+  it('logs a failed read instead of rejecting', async () => {
+    const capture = new NativeNetworkCapture(sink(), onLog)
+    execute.mockResolvedValueOnce('L1')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    execute.mockRejectedValueOnce(new Error('session gone'))
+    await expect(capture.forward()).resolves.toBeUndefined()
+    expect(onLog).toHaveBeenCalledWith(
+      'warn',
+      expect.stringContaining('session gone')
+    )
+  })
+
+  it('drops the old listener for a replaced session, and a start that a later one superseded', async () => {
+    const capture = new NativeNetworkCapture(sink(), onLog)
+    let answerFirst: (id: string) => void = () => {}
+    execute.mockReturnValueOnce(
+      new Promise((resolve) => {
+        answerFirst = resolve
+      })
+    )
+    const first = capture.start({
+      requested: true,
+      capabilities: NATIVE,
+      execute
+    })
+    const second = vi.fn().mockResolvedValueOnce(null)
+    await capture.start({
+      requested: true,
+      capabilities: NATIVE,
+      execute: second
+    })
+    answerFirst('L1')
+    await first
+    await capture.forward()
+    expect(execute).toHaveBeenCalledTimes(1)
+  })
+
+  it('reads nothing once stopped', async () => {
+    const capture = new NativeNetworkCapture(sink(), onLog)
+    execute.mockResolvedValueOnce('L1')
+    await capture.start({ requested: true, capabilities: NATIVE, execute })
+    capture.stop()
+    await capture.forward()
+    expect(execute).toHaveBeenCalledTimes(1)
+  })
+})
diff --git a/packages/core/tests/session-capturer-base.test.ts b/packages/core/tests/session-capturer-base.test.ts
index 646690dc..67e1fabb 100644
--- a/packages/core/tests/session-capturer-base.test.ts
+++ b/packages/core/tests/session-capturer-base.test.ts
@@ -437,6 +437,39 @@ describe('processTracePayload network ids', () => {
   })
 })
 
+describe('recordNetworkRequests', () => {
+  const row = (id: string, status?: number) => ({
+    id,
+    url: 'https://x/a',
+    method: 'GET',
+    type: 'fetch' as const,
+    timestamp: 1,
+    startTime: 1,
+    ...(status ? { status } : {})
+  })
+
+  it('replaces a row re-sent under the same id instead of listing it twice', () => {
+    cap.recordNetworkRequests([row('r1')])
+    cap.recordNetworkRequests([row('r1', 200), row('r2')])
+    expect(cap.networkRequests.map((r) => [r.id, r.status])).toEqual([
+      ['r1', 200],
+      ['r2', undefined]
+    ])
+  })
+
+  it('sends one upstream message per call', () => {
+    const rows = [row('r1'), row('r2')]
+    cap.recordNetworkRequests(rows)
+    expect(cap.upstream).toEqual([{ scope: 'networkRequests', data: rows }])
+  })
+
+  it('sends nothing for an empty batch', () => {
+    cap.recordNetworkRequests([])
+    expect(cap.upstream).toEqual([])
+    expect(cap.networkRequests).toEqual([])
+  })
+})
+
 describe('processTracePayload network ids at scale', () => {
   it('keeps each request one row without rescanning the recording', () => {
     class Capturer extends SessionCapturerBase {
diff --git a/packages/nightwatch-devtools/README.md b/packages/nightwatch-devtools/README.md
index 34c4f31f..12305c8d 100644
--- a/packages/nightwatch-devtools/README.md
+++ b/packages/nightwatch-devtools/README.md
@@ -287,6 +287,31 @@ module.exports = {
 
 Native sessions record commands, screenshots and element data from the platform's XML tree, with no DOM. A hybrid app's webview is captured as if it were still native. From the root of a built checkout of this repo (`pnpm install && pnpm build`), try it with `pnpm demo:nightwatch:mobile`; prerequisites are in [`examples/MOBILE.md`](../../examples/MOBILE.md), and the full reference is the [Mobile Testing page](https://webdriver.io/docs/devtools/mobile).
 
+### Native app network
+
+A native app has no document and Appium serves no BiDi network events, so by default the Network tab of a native session is empty. The plugin can read the app's traffic from the [appium-interceptor](https://github.com/AppiumTestDistribution/appium-interceptor-plugin) plugin instead, which routes the device through a proxy of its own:
+
+```sh
+appium plugin install --source=npm appium-interceptor
+appium --use-plugins=appium-interceptor
+```
+
+Then ask for the proxy in `desiredCapabilities`:
+
+```js
+desiredCapabilities: {
+  browserName: null,
+  platformName: 'Android',
+  'appium:automationName': 'UiAutomator2',
+  'appium:app': '/path/to/app.apk',
+  'appium:intercept': true
+}
+```
+
+The plugin starts a listener for each session, including one that replaces it mid-run, and reads what the proxy recorded at most every 2 seconds between commands (each read returns the whole history), and always before each test's trace slice is written, in both live and trace mode. Each request shows its method, URL, status, headers and bodies. The plugin records no timing, so a request is placed at the moment it was first read and shows no duration. Without the plugin a warning is logged and the session runs as usual. This applies to native sessions only; a mobile browser session already captures its network through the page.
+
+The plugin supports Android. For HTTPS traffic the app has to trust the certificate the plugin installs on the device, which a release build usually does not (see Android's network security configuration).
+
 ## Examples
 
 Working examples live in the repo's top-level [`examples/nightwatch/`](../../examples/nightwatch). Build the workspace once (`pnpm install && pnpm build`), then run from the repo root:
diff --git a/packages/nightwatch-devtools/src/cucumber-lifecycle.ts b/packages/nightwatch-devtools/src/cucumber-lifecycle.ts
index 60633da2..0c7f6fc3 100644
--- a/packages/nightwatch-devtools/src/cucumber-lifecycle.ts
+++ b/packages/nightwatch-devtools/src/cucumber-lifecycle.ts
@@ -281,9 +281,7 @@ export async function captureCucumberScenarioBeforeQuit(
         assertCalls
       )
     }
-    // Anchored: a scenario ending on a navigation leaves the destination's
-    // async initial anchor unrun, so its DOM would be absent from the slice.
-    await ctx.sessionCapturer.captureTrace(browser, true)
+    await ctx.sessionCapturer.captureTraceAtTestEnd(browser)
     flushTestSlice(ctx)
     await ctx.emitTestArtifacts(
       scenario?.uid,
diff --git a/packages/nightwatch-devtools/src/helpers/browserProxy.ts b/packages/nightwatch-devtools/src/helpers/browserProxy.ts
index 49a48e51..8d9706cd 100644
--- a/packages/nightwatch-devtools/src/helpers/browserProxy.ts
+++ b/packages/nightwatch-devtools/src/helpers/browserProxy.ts
@@ -346,6 +346,10 @@ export class BrowserProxy {
       this.attachScreenshot(browser, lastCommand, methodName)
     }
     this.anchorDestination(browser, methodName)
+    // A DOM-mutating command already drains it through `captureTrace`.
+    if (!this.isDomMutating(methodName)) {
+      void this.sessionCapturer.nativeNetwork.forward({ throttle: true })
+    }
   }
 
   private isDomMutating(methodName: string): boolean {
diff --git a/packages/nightwatch-devtools/src/helpers/webdriverHttp.ts b/packages/nightwatch-devtools/src/helpers/webdriverHttp.ts
index 8b830451..4379a8fc 100644
--- a/packages/nightwatch-devtools/src/helpers/webdriverHttp.ts
+++ b/packages/nightwatch-devtools/src/helpers/webdriverHttp.ts
@@ -9,8 +9,10 @@
 
 import logger from '@wdio/logger'
 import {
+  webdriverExecute as coreExecute,
   webdriverGet as coreGet,
   webdriverPost as corePost,
+  type InterceptorExecute,
   type WebDriverAddress
 } from '@wdio/devtools-core'
 import type { NightwatchBrowser } from '../types.js'
@@ -39,14 +41,11 @@ const firstDefined = (obj: unknown, ...keys: string[]): unknown => {
   return undefined
 }
 
-/**
- * Walks Nightwatch's internal config (transport / queue.transport /
- * nightwatchInstance — none of which are on the public NightwatchBrowser type)
- * to find the underlying WebDriver host+port.
- */
-export function resolveWebDriverAddress(browser: NightwatchBrowser): {
-  driverHost: string
-  driverPort: number
+/** The driver settings as configured, with no defaults applied. */
+function configuredWebDriver(browser: NightwatchBrowser): {
+  host: unknown
+  port: unknown
+  pathPrefix: unknown
 } {
   const transportSettings =
     getPath(browser, 'transport', 'settings', 'webdriver') ||
@@ -64,19 +63,34 @@ export function resolveWebDriverAddress(browser: NightwatchBrowser): {
     getPath(browser, 'nightwatchInstance', 'settings') ||
     getPath(browser, 'globals', 'nightwatchInstance', 'settings') ||
     {}
-  const driverHost = String(
-    firstDefined(transportSettings, 'host', 'server_address') ||
-      firstDefined(getProp(opts, 'webdriver'), 'host') ||
-      firstDefined(getProp(nightwatchSettings, 'webdriver'), 'host') ||
-      'localhost'
-  )
-  const driverPort = Number(
-    firstDefined(transportSettings, 'port') ||
-      firstDefined(getProp(opts, 'webdriver'), 'port') ||
-      firstDefined(getProp(nightwatchSettings, 'webdriver'), 'port') ||
-      9515
-  )
-  return { driverHost, driverPort }
+  const sources = [
+    transportSettings,
+    getProp(opts, 'webdriver'),
+    getProp(nightwatchSettings, 'webdriver')
+  ]
+  const pick = (...keys: string[]) =>
+    sources.map((source) => firstDefined(source, ...keys)).find(Boolean)
+  return {
+    host: pick('host', 'server_address'),
+    port: pick('port'),
+    pathPrefix: pick('default_path_prefix')
+  }
+}
+
+/**
+ * Walks Nightwatch's internal config (transport / queue.transport /
+ * nightwatchInstance, none of which are on the public NightwatchBrowser type)
+ * to find the underlying WebDriver host+port.
+ */
+export function resolveWebDriverAddress(browser: NightwatchBrowser): {
+  driverHost: string
+  driverPort: number
+} {
+  const { host, port } = configuredWebDriver(browser)
+  return {
+    driverHost: String(host || 'localhost'),
+    driverPort: Number(port || 9515)
+  }
 }
 
 /** The core transport's address, plus this adapter's logger. Nightwatch always
@@ -115,6 +129,25 @@ export function webdriverPost<T>(
     : Promise.resolve(null)
 }
 
+/** Appium execute methods for the session current now: a listener belongs to one session, so a later read never reaches its replacement. */
+export function sessionExecutor(
+  browser: NightwatchBrowser
+): InterceptorExecute | undefined {
+  const id = sessionId(browser)
+  const { host, port, pathPrefix } = configuredWebDriver(browser)
+  // No default port here: guessing chromedriver's would blame a missing plugin for a wrong address.
+  if (!id || !port) {
+    return undefined
+  }
+  const at: WebDriverAddress = {
+    hostname: String(host || 'localhost'),
+    port: Number(port),
+    path: typeof pathPrefix === 'string' ? pathPrefix : undefined,
+    onWarn: (message: string) => log.warn(message)
+  }
+  return (script, args) => coreExecute(at, id, script, args)
+}
+
 /** Run a script in the page, outside the command queue. `body` is a function
  *  body, matching what `browser.execute` accepts. */
 export function webdriverExecute<T>(
diff --git a/packages/nightwatch-devtools/src/index.ts b/packages/nightwatch-devtools/src/index.ts
index 9c64dea0..2df0b4b0 100644
--- a/packages/nightwatch-devtools/src/index.ts
+++ b/packages/nightwatch-devtools/src/index.ts
@@ -569,9 +569,7 @@ class NightwatchDevToolsPlugin {
             this.browserProxy.drainNativeAssertCalls()
           )
         }
-        // Anchored: a test ending on a navigation leaves the destination's
-        // async initial anchor unrun, so its DOM would be absent from the slice.
-        await this.sessionCapturer.captureTrace(browser, true)
+        await this.sessionCapturer.captureTraceAtTestEnd(browser)
         // Flush this test's slice before the next test overwrites its outcome.
         flushTestSlice(this.#getInternals())
         const results = (browser.currentTest as NightwatchCurrentTest)?.results
@@ -626,6 +624,8 @@ class NightwatchDevToolsPlugin {
   }
 
   async after(browser?: NightwatchBrowser) {
+    // The session is over, so a late drain must not reach it.
+    this.sessionCapturer?.nativeNetwork.stop()
     // Drain first: a rotation still in flight would otherwise start its new
     // recorder after this finalize and strand the poll interval.
     await this.#screencast.rotation
diff --git a/packages/nightwatch-devtools/src/session-init.ts b/packages/nightwatch-devtools/src/session-init.ts
index 5b4d6a43..af5094e8 100644
--- a/packages/nightwatch-devtools/src/session-init.ts
+++ b/packages/nightwatch-devtools/src/session-init.ts
@@ -18,6 +18,7 @@ import logger from '@wdio/logger'
 import { resolveViewport, restoreBrowserFocus } from '@wdio/devtools-core'
 import {
   resolveWebDriverAddress,
+  sessionExecutor,
   webdriverExecute,
   webdriverGet
 } from './helpers/webdriverHttp.js'
@@ -25,6 +26,7 @@ import {
   errorMessage,
   finalizeScreencast,
   registerCollectorPreload,
+  requestsInterceptedNetwork,
   resolveAdapterOutputDir
 } from '@wdio/devtools-core'
 import { TraceType } from './types.js'
@@ -294,6 +296,13 @@ async function armCaptureForSession(
   ctx.armedSessionId = browser.sessionId ?? undefined
   await tryRegisterPreload(ctx, browser)
   await tryAttachBidi(ctx, browser)
+  await ctx.sessionCapturer.nativeNetwork.start({
+    requested: requestsInterceptedNetwork(browser.desiredCapabilities),
+    capabilities: browser.capabilities,
+    execute: sessionExecutor(browser),
+    unavailableReason:
+      'the session id or the configured WebDriver port is unknown'
+  })
 }
 
 /**
@@ -341,6 +350,8 @@ async function armReplacedSession(
   ctx.sessionCapturer.bidiActive = false
   ctx.sessionCapturer.bidiNetworkObserved = false
   ctx.sessionCapturer.loadsListedByLog = []
+  // The old listener's session is gone; the new one starts only after the BiDi handshake.
+  ctx.sessionCapturer.nativeNetwork.stop()
   log.info(`Session replaced — re-arming capture for ${sessionId}`)
   await armCaptureForSession(ctx, browser)
 }
diff --git a/packages/nightwatch-devtools/src/session.ts b/packages/nightwatch-devtools/src/session.ts
index ea5a488b..1abb199b 100644
--- a/packages/nightwatch-devtools/src/session.ts
+++ b/packages/nightwatch-devtools/src/session.ts
@@ -10,6 +10,7 @@ import {
   loadInjectableScript,
   mapChromeBrowserLogs,
   mapCommandToAction,
+  NativeNetworkCapture,
   pollUntilReady,
   serializeError,
   upsertRichestSnapshot,
@@ -90,6 +91,10 @@ export class SessionCapturer extends SessionCapturerBase {
   runner: TestRunnerId = nightwatchRunnerId(false)
   readonly actionSnapshots: ActionSnapshot[] = []
   readonly snapshotCaptures: Promise<void>[] = []
+  /** Native app network from appium-interceptor's proxy; started per session. */
+  readonly nativeNetwork = new NativeNetworkCapture(this, (level, message) =>
+    log[level](message)
+  )
 
   constructor(
     devtoolsOptions: SessionCapturerOptions = {},
@@ -543,10 +548,10 @@ export class SessionCapturer extends SessionCapturerBase {
     forceAnchor = false,
     anchorTimestamp?: number
   ) {
-    // A native app has no document to drain. Inside the method, because four
-    // call sites reach it — the command hook, the navigation proxy, the
-    // cucumber pre-quit hook and finalize.
+    // A native app has no document to drain, only the proxy's requests. Inside
+    // the method, because the command hook and the navigation proxy both reach it.
     if (this.isNativeAppSession) {
+      await this.nativeNetwork.forward({ throttle: true })
       return
     }
     // Performance logs only accumulate, so they lose nothing by waiting, while
@@ -555,6 +560,17 @@ export class SessionCapturer extends SessionCapturerBase {
     await this.captureNetworkFromPerformanceLogs(browser)
   }
 
+  /** At a test or scenario end, before its slice is written: the proxy drain there is never throttled. */
+  async captureTraceAtTestEnd(browser: NightwatchBrowser) {
+    if (this.isNativeAppSession) {
+      await this.nativeNetwork.forward()
+      return
+    }
+    // Anchored: a test ending on a navigation leaves the destination's async
+    // initial anchor unrun, so its DOM would be absent from the slice.
+    await this.captureTrace(browser, true)
+  }
+
   private async drainCollector(
     browser: NightwatchBrowser,
     forceAnchor: boolean,
diff --git a/packages/nightwatch-devtools/tests/browserProxy.test.ts b/packages/nightwatch-devtools/tests/browserProxy.test.ts
index bb2fc257..ef78a2a6 100644
--- a/packages/nightwatch-devtools/tests/browserProxy.test.ts
+++ b/packages/nightwatch-devtools/tests/browserProxy.test.ts
@@ -64,7 +64,8 @@ function makeCapturer() {
     captureTrace: vi.fn(async () => {}),
     injectScript: vi.fn(async () => {}),
     anchorAfterNavigation: vi.fn(async () => {}),
-    snapshotCaptures: [] as Promise<void>[]
+    snapshotCaptures: [] as Promise<void>[],
+    nativeNetwork: { forward: vi.fn(async () => {}) }
   } as unknown as SessionCapturer
   return { capturer, commandsLog, captureCommand, captureAssertCommand }
 }
@@ -825,4 +826,17 @@ describe('BrowserProxy capture ordering after a DOM-mutating command', () => {
     expect(capturer.captureTrace).not.toHaveBeenCalled()
     expect(capturer.anchorAfterNavigation).not.toHaveBeenCalled()
   })
+  // A native app's requests sit in the proxy until drained; a DOM-mutating
+  // command drains them through `captureTrace`, every other command here.
+  it('drains native network after a command that cannot change the DOM', () => {
+    const { capturer } = run('getTitle')
+    expect(capturer.nativeNetwork.forward).toHaveBeenCalledWith({
+      throttle: true
+    })
+  })
+
+  it('leaves the native drain to captureTrace for a DOM-mutating command', () => {
+    const { capturer } = run('click')
+    expect(capturer.nativeNetwork.forward).not.toHaveBeenCalled()
+  })
 })
diff --git a/packages/nightwatch-devtools/tests/native-session.test.ts b/packages/nightwatch-devtools/tests/native-session.test.ts
index 1a99aa20..160568f8 100644
--- a/packages/nightwatch-devtools/tests/native-session.test.ts
+++ b/packages/nightwatch-devtools/tests/native-session.test.ts
@@ -60,6 +60,27 @@ describe('nightwatch SessionCapturer on a native session', () => {
     expect(webdriverExecute).not.toHaveBeenCalled()
   })
 
+  it('drains the native network proxy from captureTrace, throttled per command', async () => {
+    const cap = capturerWith(NATIVE)
+    const forward = vi.spyOn(cap.nativeNetwork, 'forward')
+    await cap.captureTrace(browser(), true)
+    expect(forward).toHaveBeenCalledWith({ throttle: true })
+  })
+
+  // afterEach and the cucumber pre-quit hook run this before the slice flush.
+  it('drains unthrottled at a test end, ahead of the test slice', async () => {
+    const cap = capturerWith(NATIVE)
+    const forward = vi.spyOn(cap.nativeNetwork, 'forward')
+    await cap.captureTraceAtTestEnd(browser())
+    expect(forward).toHaveBeenCalledWith()
+    expect(webdriverExecute).not.toHaveBeenCalled()
+  })
+
+  it('drains the page with an anchor at a web test end', async () => {
+    await capturerWith(MOBILE_WEB).captureTraceAtTestEnd(browser())
+    expect(webdriverExecute).toHaveBeenCalled()
+  })
+
   it('makes none from injectScript or anchorAfterNavigation', async () => {
     const cap = capturerWith(NATIVE)
     const b = browser()
diff --git a/packages/nightwatch-devtools/tests/sessionExecutor.test.ts b/packages/nightwatch-devtools/tests/sessionExecutor.test.ts
new file mode 100644
index 00000000..a07f9529
--- /dev/null
+++ b/packages/nightwatch-devtools/tests/sessionExecutor.test.ts
@@ -0,0 +1,54 @@
+import { describe, it, expect, vi } from 'vitest'
+import type * as DevToolsCore from '@wdio/devtools-core'
+import { sessionExecutor } from '../src/helpers/webdriverHttp.js'
+import type { NightwatchBrowser } from '../src/types.js'
+
+const webdriverExecute = vi.hoisted(() => vi.fn(async () => 'L1'))
+
+vi.mock('@wdio/devtools-core', async (importOriginal) => ({
+  ...(await importOriginal<typeof DevToolsCore>()),
+  webdriverExecute
+}))
+
+const browser = (
+  sessionId: string | undefined,
+  webdriver: Record<string, unknown> = { host: '127.0.0.1', port: 4724 }
+) => ({ sessionId, options: { webdriver } }) as unknown as NightwatchBrowser
+
+describe('sessionExecutor', () => {
+  it('posts an Appium execute method to the session current when it was built', async () => {
+    const b = browser('sess-1')
+    const execute = sessionExecutor(b)!
+    ;(b as { sessionId?: string }).sessionId = 'sess-2'
+
+    await execute('interceptor: startListening', [{}])
+
+    expect(webdriverExecute).toHaveBeenCalledWith(
+      expect.objectContaining({ hostname: '127.0.0.1', port: 4724 }),
+      'sess-1',
+      'interceptor: startListening',
+      [{}]
+    )
+  })
+
+  it('carries the configured path prefix', async () => {
+    await sessionExecutor(
+      browser('sess-1', { port: 4723, default_path_prefix: '/wd/hub' })
+    )!('interceptor: startListening', [{}])
+    expect(webdriverExecute).toHaveBeenLastCalledWith(
+      expect.objectContaining({ port: 4723, path: '/wd/hub' }),
+      'sess-1',
+      'interceptor: startListening',
+      [{}]
+    )
+  })
+
+  it('is undefined without a session', () => {
+    expect(sessionExecutor(browser(undefined))).toBeUndefined()
+  })
+
+  // Regression: a missing port defaulted to chromedriver's 9515 and blamed the plugin.
+  it('is undefined when no WebDriver port is configured', () => {
+    expect(sessionExecutor(browser('sess-1', {}))).toBeUndefined()
+  })
+})
diff --git a/packages/nightwatch-devtools/tests/sessionRearm.test.ts b/packages/nightwatch-devtools/tests/sessionRearm.test.ts
index 493304f7..6c4fa5ff 100644
--- a/packages/nightwatch-devtools/tests/sessionRearm.test.ts
+++ b/packages/nightwatch-devtools/tests/sessionRearm.test.ts
@@ -38,6 +38,10 @@ type RearmCtx = Pick<
 
 /** Stands in for the accumulated trace state a re-arm must leave alone. */
 interface FakeCapturer {
+  nativeNetwork: {
+    start: ReturnType<typeof vi.fn>
+    stop: ReturnType<typeof vi.fn>
+  }
   preloadRegistered: boolean
   bidiActive: boolean
   loadsListedByLog: unknown[]
@@ -50,6 +54,7 @@ function makeCtx(
   overrides: { armedSessionId?: string; bidiEnabled?: boolean } = {}
 ) {
   const capturer: FakeCapturer = {
+    nativeNetwork: { start: vi.fn(async () => {}), stop: vi.fn() },
     preloadRegistered: true,
     bidiActive: true,
     loadsListedByLog: ['load-1'],
@@ -156,6 +161,34 @@ describe('rearmCaptureForSession', () => {
     expect(ctx.armedSessionId).toBe('session-2')
   })
 
+  // A listener belongs to the session that started it, so the replacement needs its own.
+  it('starts a native network listener for the new session', async () => {
+    const { ctx, capturer } = makeCtx()
+    const browser = {
+      ...fakeBrowser('session-2'),
+      capabilities: { platformName: 'Android' },
+      desiredCapabilities: { 'appium:intercept': true }
+    } as unknown as NightwatchBrowser
+
+    rearm(ctx, browser)
+    await settle()
+
+    expect(capturer.nativeNetwork.start).toHaveBeenCalledWith(
+      expect.objectContaining({
+        requested: true,
+        capabilities: { platformName: 'Android' }
+      })
+    )
+  })
+
+  // Regression: a command on the new session drained the old listener through the dead session.
+  it('stops the old listener at detection, before the handshake', () => {
+    const { ctx, capturer } = makeCtx()
+    rearm(ctx, fakeBrowser('session-2'))
+    expect(capturer.nativeNetwork.stop).toHaveBeenCalledTimes(1)
+    expect(capturer.nativeNetwork.start).not.toHaveBeenCalled()
+  })
+
   it('leaves the accumulated capture state untouched', async () => {
     // The re-arm swaps the session's capture channels; the run's capturer and
     // everything it has accumulated outlive every session it is pointed at.
diff --git a/packages/nightwatch-devtools/tests/sessionRebind.test.ts b/packages/nightwatch-devtools/tests/sessionRebind.test.ts
index bb0660fb..48d6dab0 100644
--- a/packages/nightwatch-devtools/tests/sessionRebind.test.ts
+++ b/packages/nightwatch-devtools/tests/sessionRebind.test.ts
@@ -54,7 +54,8 @@ function makeCapturer() {
     mutations: ['mut-1'],
     traceLogs: [],
     actionSnapshots: ['snap-1'],
-    sendUpstream: vi.fn()
+    sendUpstream: vi.fn(),
+    nativeNetwork: { start: vi.fn(async () => {}), stop: vi.fn() }
   }
 }
 
diff --git a/packages/selenium-devtools-py/README.md b/packages/selenium-devtools-py/README.md
index 4beefb86..8bdf2c41 100644
--- a/packages/selenium-devtools-py/README.md
+++ b/packages/selenium-devtools-py/README.md
@@ -231,6 +231,7 @@ is on `PATH`; otherwise keep it current (`brew upgrade chromedriver`).
 | Test / suite tree | pytest plugin (`pytest_runtest_logreport` / `sessionfinish`) | `suites` | both |
 | Browser console + JS errors | Selenium **BiDi** (`driver.script` handlers) | `consoleLogs` | both |
 | Network requests | Selenium **BiDi** (`Network.add_event_handler`, observe-only, never an intercept, which would pause every request); without BiDi, the page collector's `fetch` / `XMLHttpRequest` patch and Resource Timing buffer, drained beside the mutations | `networkRequests` | both |
+| Native app network | the appium-interceptor plugin's proxy, when the session sets `appium:intercept: True` | `networkRequests` | both |
 | Assertions | pytest hooks under pytest; line tracing for a plain script | `commands` | both |
 | DOM snapshot / time-travel | register `packages/script` as a BiDi document-start preload, drain mutations with a forced document anchor (per-document `<script>` injection when BiDi is absent) | `mutations` | both |
 | Navigation + resource timing | read from the page after a navigation, then the command row is re-sent with it | `commands` | both |
@@ -387,6 +388,29 @@ DEVTOOLS_MOBILE_PLATFORM=ios pnpm demo:python:mobile # iOS Settings
 
 `DEVTOOLS_MOBILE=web` drives the device's own browser instead. Prerequisites and the `DEVTOOLS_MOBILE` / `APPIUM_APP` switches are in [`examples/MOBILE.md`](../../examples/MOBILE.md); the full reference is the [Mobile Testing page](https://webdriver.io/docs/devtools/mobile).
 
+### Native app network
+
+A native app has no document and Appium serves no BiDi network events, so by default the Network tab of a native session is empty. The adapter can read the app's traffic from the [appium-interceptor](https://github.com/AppiumTestDistribution/appium-interceptor-plugin) plugin instead, which routes the device through a proxy of its own:
+
+```sh
+appium plugin install --source=npm appium-interceptor
+appium --use-plugins=appium-interceptor
+```
+
+Then ask for the proxy in the session's capabilities:
+
+```python
+from appium.options.android import UiAutomator2Options
+
+options = UiAutomator2Options()
+options.app = "/path/to/app.apk"
+options.set_capability("appium:intercept", True)
+```
+
+The adapter starts a listener when the session starts and reads what the proxy recorded at most every 2 seconds between commands (each read returns the whole history), and always when each pytest test reports its outcome and when the driver quits, in both live and trace mode. Each request shows its method, URL, status, headers and bodies. The plugin records no timing, so a request is placed at the moment it was first read and shows no duration. Without the plugin the adapter logs a warning and the session runs as usual. This applies to native sessions only; a mobile browser session already captures its network through the page.
+
+The plugin supports Android. For HTTPS traffic the app has to trust the certificate the plugin installs on the device, which a release build usually does not (see Android's network security configuration).
+
 ## Trace mode
 
 Instead of a live dashboard, write the run to a portable archive — the same
@@ -460,6 +484,7 @@ src/selenium_devtools/
   performance.py      navigation + resource timing for a navigation command
   bidi.py             BiDi console/JS-error + network capture (pure mapping + wiring)
   bidi_preload.py     document-start registration of the collector (BiDi preload)
+  interceptor_network.py native app network from the appium-interceptor plugin's proxy
   snapshot.py         mutation drain, and the <script> injection used without BiDi
   collector_source.py where the page-side collector's source comes from
   element_scripts.py  the backend-served a11y/element scripts the A11y tab needs
diff --git a/packages/selenium-devtools-py/changes/native-network-interceptor.md b/packages/selenium-devtools-py/changes/native-network-interceptor.md
new file mode 100644
index 00000000..d9c3a995
--- /dev/null
+++ b/packages/selenium-devtools-py/changes/native-network-interceptor.md
@@ -0,0 +1,5 @@
+---
+minor
+---
+
+Capture a native app's network requests on Appium, through the [appium-interceptor](https://github.com/AppiumTestDistribution/appium-interceptor-plugin) plugin. A native app has no document and Appium serves no BiDi network events, so the Network tab of a native session was always empty. When Appium runs with `--use-plugins=appium-interceptor` and the session sets `appium:intercept: True`, the adapter starts a listener on the plugin's proxy and lists what it recorded after every command, when each test reports its outcome, and when the driver quits. Each request carries its method, URL, status, headers and bodies; the plugin records no timing, so a request is placed when it was first read. Without the plugin the adapter logs a warning and the session runs as usual. The plugin supports Android.
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/constants.py b/packages/selenium-devtools-py/src/selenium_devtools/constants.py
index 4bc6fb0f..c1e24f21 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/constants.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/constants.py
@@ -40,6 +40,8 @@
 # not transient and is never retried.
 COLLECTOR_RETRY_LIMIT = 3
 COLLECTOR_RETRY_COOLDOWN_S = 2.0
+# Per-command proxy drains are throttled to this; test and session boundaries always drain.
+NATIVE_NETWORK_DRAIN_INTERVAL_S = 2.0
 
 # ── Logging ──────────────────────────────────────────────────────────────────
 # The adapter's logger name. Every operational message logs under this or a
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
index 5c4bf049..1e57cd83 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/instrumentation.py
@@ -19,7 +19,15 @@
 import weakref
 from typing import Any, List, Optional
 
-from . import assertions, bidi, bidi_preload, element_locators, frames, performance
+from . import (
+    assertions,
+    bidi,
+    bidi_preload,
+    element_locators,
+    frames,
+    interceptor_network,
+    performance,
+)
 from .assert_tracer import ScriptAssertionTracer
 from .capturer import SessionCapturer
 from .cdp_screencast import start_push_screencast
@@ -550,13 +558,17 @@ def _enable_bidi_capability(params: Any) -> None:
         pass
 
 
-def _close_entry(capturer: SessionCapturer, entry: dict) -> None:
+def _close_entry(capturer: SessionCapturer, entry: dict, replaced: bool = False) -> None:
     """Drain and encode one driver's capture, attributed to the session it was
     recorded under."""
     # Before the encode: a frame still arriving would land after the buffer was
     # read, and the browser keeps sending until it is told to stop.
     _stop_push_screencast(entry)
     _flush_page_collector(capturer, entry)
+    poller = entry.pop("interceptor", None)
+    # A replaced session's driver now addresses the new one, which knows nothing of the old listener.
+    if not replaced:
+        interceptor_network.forward(capturer, poller)
     entry["snapshot"] = None
     _finalize_screencast(capturer, entry["session_id"], entry)
     summary = bidi.network_summary(entry.get("network"))
@@ -696,7 +708,7 @@ def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> Optional[di
     if entry is not None:
         if entry["session_id"] == session_id:
             return entry
-        _close_entry(capturer, entry)  # same driver, new session
+        _close_entry(capturer, entry, replaced=True)  # same driver, new session
     entry = {
         "session_id": session_id, "screencast": None, "snapshot": None,
         "preloaded": False,
@@ -723,6 +735,9 @@ def _ensure_session_setup(driver: Any, capturer: SessionCapturer) -> Optional[di
             _log.info("BiDi attached — capturing console + network")
     except Exception as exc:  # noqa: BLE001 — capture must never break the test
         _log.warning("BiDi attach threw: %s", exc)
+    entry["interceptor"] = interceptor_network.start(
+        driver, _guarded_execute_script(driver)
+    )
     try:
         if _state["trace"] and not _state["filmstrip"]:
             # With no filmstrip the archive's frames are the per-command
@@ -893,6 +908,12 @@ def record_live_failure() -> None:
         mark_run_failed()
 
 
+def drain_native_network(capturer: SessionCapturer) -> None:
+    """List native requests made since each session's last command, before a test's outcome closes its window."""
+    for entry in list(_state["sessions"].values()):
+        interceptor_network.forward(capturer, entry.get("interceptor"))
+
+
 def mark_run_failed() -> None:
     """Record that an exception reached top level (see `enable`'s excepthook)."""
     _state["run_failed"] = True
@@ -961,6 +982,7 @@ def patched_execute(self, driver_command: str, params: Any = None):  # noqa: ANN
                 _on_quit(capturer, self)
             elif driver_command == "newSession":
                 _enable_bidi_capability(params)  # request BiDi before the session opens
+                interceptor_network.note_session_request(self, params)
             return orig_execute(self, driver_command, params)
 
         # First real command: the driver is now fully initialized — set up
@@ -1012,6 +1034,9 @@ def patched_execute(self, driver_command: str, params: Any = None):  # noqa: ANN
         # navigate too, not just get/back/…), re-injecting if the page changed.
         if entry is not None:
             _refresh_snapshot(capturer, entry)
+            interceptor_network.forward(
+                capturer, entry.get("interceptor"), throttle=True
+            )
             # After the command, not before: until it returns, the page is
             # whatever preceded it — on the first command of a fresh driver
             # that is an unpainted about:blank, and a pushed stream sends
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/interceptor_network.py b/packages/selenium-devtools-py/src/selenium_devtools/interceptor_network.py
new file mode 100644
index 00000000..6b673bfa
--- /dev/null
+++ b/packages/selenium-devtools-py/src/selenium_devtools/interceptor_network.py
@@ -0,0 +1,203 @@
+"""Native app network from the appium-interceptor plugin's proxy, mirroring core's ``appium-interceptor.ts``."""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import threading
+import time
+import weakref
+from typing import Any, Callable, Dict, List, Optional
+
+from ._wire_types import NetworkRequest
+from .bidi import request_type_for
+from .constants import LOGGER_NAME, NATIVE_NETWORK_DRAIN_INTERVAL_S
+from .device import driver_is_native_app
+from .utils import now_ms
+
+_log = logging.getLogger(f"{LOGGER_NAME}.interceptor")
+
+INTERCEPT_CAPABILITY = "appium:intercept"
+APPIUM_OPTIONS = "appium:options"
+
+# Whether each driver's newSession asked for the proxy; the returned capabilities need not echo it.
+_requested: "weakref.WeakKeyDictionary[Any, bool]" = weakref.WeakKeyDictionary()
+
+_WHITESPACE = re.compile(r"\s")
+
+
+def _asks_for_intercept(caps: Any) -> bool:
+    """Whether one capability set asks for the proxy, directly or grouped under ``appium:options``."""
+    if not isinstance(caps, dict):
+        return False
+    if caps.get(INTERCEPT_CAPABILITY) is True:
+        return True
+    grouped = caps.get(APPIUM_OPTIONS)
+    return isinstance(grouped, dict) and grouped.get("intercept") is True
+
+
+def note_session_request(driver: Any, params: Any) -> None:
+    """Record whether this driver's newSession set ``appium:intercept: True``."""
+    caps = params.get("capabilities") if isinstance(params, dict) else None
+    if not isinstance(caps, dict):
+        return
+    candidates = [caps.get("alwaysMatch")]
+    first = caps.get("firstMatch")
+    if isinstance(first, list):
+        candidates.extend(first)
+    asked = any(_asks_for_intercept(c) for c in candidates)
+    try:
+        _requested[driver] = asked
+    except TypeError:
+        if asked:
+            _log.warning(
+                "%s is set but this driver cannot be tracked; native network is not captured",
+                INTERCEPT_CAPABILITY,
+            )
+
+
+def flatten_headers(headers: Any) -> Dict[str, str]:
+    flat: Dict[str, str] = {}
+    if not isinstance(headers, dict):
+        return flat
+    for name, value in headers.items():
+        # The plugin files the response status line as a header ("http/1.1 201 created").
+        if not isinstance(name, str) or _WHITESPACE.search(name):
+            continue
+        flat[name.lower()] = (
+            ", ".join(str(v) for v in value) if isinstance(value, list) else str(value)
+        )
+    return flat
+
+
+def body_text(body: Any) -> Optional[str]:
+    if body is None or body == "":
+        return None
+    if isinstance(body, str):
+        return body
+    return json.dumps(body, separators=(",", ":"), ensure_ascii=False)
+
+
+def to_network_request(entry: Dict[str, Any], row_id: str, observed_at: int) -> NetworkRequest:
+    """The plugin records no timing, so a row is stamped when it was first read."""
+    url = str(entry.get("url") or "")
+    status = entry.get("statusCode")
+    response_headers = flatten_headers(entry.get("responseHeaders"))
+    content_type = response_headers.get("content-type")
+    response_body = body_text(entry.get("responseBody"))
+    row: NetworkRequest = {
+        "id": row_id,
+        "url": url,
+        "method": str(entry.get("method") or ""),
+        "type": request_type_for(url, content_type),  # type: ignore[typeddict-item]
+        "timestamp": observed_at,
+        "startTime": observed_at,
+        "endTime": observed_at,
+        "requestHeaders": flatten_headers(entry.get("requestHeaders")),
+        "responseHeaders": response_headers,
+    }
+    if isinstance(status, (int, float)) and not isinstance(status, bool):
+        row["status"] = status
+        row["response"] = {
+            "fromCache": False,
+            "headers": response_headers,
+            "mimeType": (content_type or "").split(";")[0].strip(),
+            "status": status,
+        }
+    request_body = body_text(entry.get("requestBody"))
+    if request_body is not None:
+        row["requestBody"] = request_body
+    if response_body is not None:
+        row["responseBody"] = response_body
+        row["size"] = len(response_body.encode("utf-8"))
+    return row
+
+
+class InterceptorPoller:
+    """One listener on one session, read incrementally."""
+
+    def __init__(
+        self, execute: Callable[..., Any], clock: Callable[[], float] = time.monotonic
+    ) -> None:
+        self._execute = execute
+        self._clock = clock
+        self._last_drain: Optional[float] = None
+        self._listener: Optional[str] = None
+        self._seen = 0
+        # Drains run on the test thread and at teardown, possibly another thread; two at once would list the same entries.
+        self._lock = threading.Lock()
+
+    def start(self) -> bool:
+        listener = self._execute("interceptor: startListening", {})
+        self._listener = listener if isinstance(listener, str) else None
+        return self._listener is not None
+
+    def drain(self, throttle: bool = False) -> List[NetworkRequest]:
+        """A per-command caller passes ``throttle``: each read returns the listener's whole history, bodies included."""
+        with self._lock:
+            listener = self._listener
+            if listener is None:
+                return []
+            now = self._clock()
+            if (
+                throttle
+                and self._last_drain is not None
+                and now - self._last_drain < NATIVE_NETWORK_DRAIN_INTERVAL_S
+            ):
+                return []
+            self._last_drain = now
+            data = self._execute("interceptor: getInterceptedData", {"id": listener})
+            # A failed call that answers None would otherwise read as "nothing new".
+            if not isinstance(data, list):
+                raise RuntimeError("the plugin returned no intercepted data")
+            if len(data) <= self._seen:
+                return []
+            observed_at = now_ms()
+            fresh = [
+                to_network_request(entry, f"interceptor-{listener}-{index}", observed_at)
+                for index, entry in enumerate(data)
+                if index >= self._seen and isinstance(entry, dict)
+            ]
+            self._seen = len(data)
+            return fresh
+
+
+def start(driver: Any, execute: Callable[..., Any]) -> Optional[InterceptorPoller]:
+    """A started poller for a native session that asked for the proxy, else None."""
+    try:
+        asked = _requested.get(driver, False)
+    except TypeError:
+        asked = False
+    if not asked or not driver_is_native_app(driver):
+        return None
+    poller = InterceptorPoller(execute)
+    try:
+        started = poller.start()
+    except Exception as exc:  # noqa: BLE001 - capture must never break the test
+        _log.warning("appium-interceptor listener failed to start: %s", exc)
+        started = False
+    if not started:
+        _log.warning(
+            "%s is set but the interceptor listener did not start; is Appium "
+            "running with --use-plugins=appium-interceptor?",
+            INTERCEPT_CAPABILITY,
+        )
+        return None
+    _log.info("native network capture active via appium-interceptor")
+    return poller
+
+
+def forward(
+    capturer: Any, poller: Optional[InterceptorPoller], throttle: bool = False
+) -> None:
+    """List what the proxy recorded since the last call. A test or session boundary must not pass ``throttle``."""
+    if poller is None:
+        return
+    try:
+        requests = poller.drain(throttle)
+    except Exception as exc:  # noqa: BLE001 - capture must never break the test
+        _log.warning("appium-interceptor read failed: %s", exc)
+        return
+    if requests:
+        capturer.send_network_requests(requests)
diff --git a/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py b/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py
index dbbbb94d..2a9eac8c 100644
--- a/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py
+++ b/packages/selenium-devtools-py/src/selenium_devtools/pytest_plugin.py
@@ -583,6 +583,9 @@ def pytest_runtest_logreport(report) -> None:  # noqa: ANN001
     state = reported_state(report)
     if state is None:
         return
+    from . import instrumentation
+
+    instrumentation.drain_native_network(capturer)
     file, line, name = report.location
     _registry.record(
         report.nodeid, file, name, line or 0, state, abs_file=_absolute(file)
diff --git a/packages/selenium-devtools-py/tests/test_interceptor_network.py b/packages/selenium-devtools-py/tests/test_interceptor_network.py
new file mode 100644
index 00000000..c1a1a798
--- /dev/null
+++ b/packages/selenium-devtools-py/tests/test_interceptor_network.py
@@ -0,0 +1,328 @@
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+from selenium_devtools import instrumentation, interceptor_network
+from selenium_devtools import pytest_plugin as plugin
+from selenium_devtools.capturer import SessionCapturer
+
+from wire_contract import RecordingTransport
+
+NATIVE_CAPS = {"platformName": "Android", "appium:automationName": "UiAutomator2"}
+
+
+def _new_session(intercept):
+    always = dict(NATIVE_CAPS)
+    if intercept is not None:
+        always["appium:intercept"] = intercept
+    return {"capabilities": {"alwaysMatch": always, "firstMatch": [{}]}}
+
+
+class FakePlugin:
+    """The appium-interceptor execute scripts, answering from a list the test fills."""
+
+    def __init__(self, listener="L1"):
+        self.listener = listener
+        self.data = []
+        self.calls = []
+
+    def __call__(self, script, *args):
+        self.calls.append((script, args))
+        if script == "interceptor: startListening":
+            return self.listener
+        if script == "interceptor: getInterceptedData":
+            return list(self.data)
+        raise AssertionError(f"unexpected script {script}")
+
+
+class NativeDriver:
+    """Stand-in for an Appium driver: execute() is the chokepoint, execute_script the plugin."""
+
+    def __init__(self, plugin_fake=None, caps=None):
+        self.session_id = None
+        self.capabilities = dict(caps or NATIVE_CAPS)
+        self.caps = self.capabilities
+        self.plugin = plugin_fake or FakePlugin()
+        self.next_session = "sess-1"
+
+    def execute(self, command, params=None):
+        if command == "newSession":
+            self.session_id = self.next_session
+        return {"value": f"ok:{command}"}
+
+    def execute_script(self, script, *args):
+        return self.plugin(script, *args)
+
+
+def _entry(url="https://api.example/x", **over):
+    entry = {
+        "url": url,
+        "method": "GET",
+        "statusCode": 200,
+        "requestHeaders": {"Accept": "*/*"},
+        "responseHeaders": {"content-type": "application/json"},
+        "responseBody": {"ok": True},
+    }
+    entry.update(over)
+    return entry
+
+
+class TestMapping(unittest.TestCase):
+    def test_lowercases_names_drops_status_line_and_joins_repeats(self):
+        self.assertEqual(
+            interceptor_network.flatten_headers(
+                {
+                    "http/1.1 201 created": "",
+                    "Set-Cookie": ["a=1", "b=2"],
+                    "Content-Type": "text/plain",
+                }
+            ),
+            {"set-cookie": "a=1, b=2", "content-type": "text/plain"},
+        )
+
+    def test_mixed_case_content_type_sets_type_and_mime(self):
+        row = interceptor_network.to_network_request(
+            _entry(responseHeaders={"Content-Type": "application/json; charset=utf-8"}),
+            "interceptor-L1-0",
+            1000,
+        )
+        self.assertEqual(row["type"], "fetch")
+        self.assertEqual(row["response"]["mimeType"], "application/json")
+
+    def test_row_shape(self):
+        row = interceptor_network.to_network_request(
+            _entry(requestBody="q=1"), "interceptor-L1-0", 1000
+        )
+        self.assertEqual(row["id"], "interceptor-L1-0")
+        self.assertEqual(row["method"], "GET")
+        self.assertEqual(row["status"], 200)
+        self.assertEqual(row["timestamp"], 1000)
+        self.assertEqual(row["startTime"], 1000)
+        self.assertEqual(row["endTime"], 1000)
+        self.assertEqual(row["requestHeaders"], {"accept": "*/*"})
+        self.assertEqual(row["requestBody"], "q=1")
+        self.assertEqual(row["responseBody"], '{"ok":true}')
+
+    def test_size_counts_bytes_not_characters(self):
+        row = interceptor_network.to_network_request(
+            _entry(responseBody="héllo"), "id", 1
+        )
+        self.assertEqual(row["size"], 6)
+
+    def test_empty_bodies_are_absent(self):
+        row = interceptor_network.to_network_request(
+            _entry(responseBody="", requestBody=None), "id", 1
+        )
+        self.assertNotIn("responseBody", row)
+        self.assertNotIn("requestBody", row)
+        self.assertNotIn("size", row)
+
+
+class TestPoller(unittest.TestCase):
+    def test_reads_only_entries_past_the_last_drain(self):
+        fake = FakePlugin()
+        poller = interceptor_network.InterceptorPoller(fake)
+        self.assertTrue(poller.start())
+        fake.data = [_entry("https://a/1"), _entry("https://a/2")]
+        first = poller.drain()
+        fake.data.append(_entry("https://a/3"))
+        second = poller.drain()
+        third = poller.drain()
+        self.assertEqual([r["id"] for r in first], ["interceptor-L1-0", "interceptor-L1-1"])
+        self.assertEqual([r["url"] for r in second], ["https://a/3"])
+        self.assertEqual(second[0]["id"], "interceptor-L1-2")
+        self.assertEqual(third, [])
+        self.assertEqual(
+            fake.calls[-1], ("interceptor: getInterceptedData", ({"id": "L1"},))
+        )
+
+    def test_throttles_per_command_drains_but_never_a_boundary_drain(self):
+        fake = FakePlugin()
+        now = [100.0]
+        poller = interceptor_network.InterceptorPoller(fake, clock=lambda: now[0])
+        poller.start()
+        reads = lambda: sum(1 for c in fake.calls if c[0] == "interceptor: getInterceptedData")
+        poller.drain(throttle=True)
+        self.assertEqual(reads(), 1)
+        now[0] += interceptor_network.NATIVE_NETWORK_DRAIN_INTERVAL_S - 0.001
+        poller.drain(throttle=True)
+        self.assertEqual(reads(), 1)
+        poller.drain()
+        self.assertEqual(reads(), 2)
+        now[0] += interceptor_network.NATIVE_NETWORK_DRAIN_INTERVAL_S
+        poller.drain(throttle=True)
+        self.assertEqual(reads(), 3)
+
+    # A failed call that answers None must surface as a failed read, not as "nothing new".
+    def test_a_none_read_is_logged_as_a_failure(self):
+        poller = interceptor_network.InterceptorPoller(
+            lambda script, *args: "L1" if script == "interceptor: startListening" else None
+        )
+        poller.start()
+        capturer = mock.Mock()
+        with self.assertLogs("selenium_devtools.interceptor", "WARNING") as logs:
+            interceptor_network.forward(capturer, poller)
+        self.assertIn("no intercepted data", logs.output[0])
+        capturer.send_network_requests.assert_not_called()
+
+    def test_unstarted_poller_reads_nothing(self):
+        fake = FakePlugin(listener=None)
+        poller = interceptor_network.InterceptorPoller(fake)
+        self.assertFalse(poller.start())
+        self.assertEqual(poller.drain(), [])
+        self.assertEqual(len(fake.calls), 1)
+
+
+class TestGate(unittest.TestCase):
+    def test_starts_for_a_native_session_that_asked(self):
+        driver = NativeDriver()
+        interceptor_network.note_session_request(driver, _new_session(True))
+        self.assertIsNotNone(interceptor_network.start(driver, driver.execute_script))
+
+    def test_firstmatch_request_counts(self):
+        driver = NativeDriver()
+        params = {"capabilities": {"alwaysMatch": {}, "firstMatch": [{"appium:intercept": True}]}}
+        interceptor_network.note_session_request(driver, params)
+        self.assertIsNotNone(interceptor_network.start(driver, driver.execute_script))
+
+    # Appium merges appium:options into the session, so the plugin starts its proxy for these too.
+    def test_appium_options_grouping_counts_at_every_level(self):
+        grouped = {"appium:options": {"intercept": True}}
+        for caps in (
+            {"alwaysMatch": grouped},
+            {"alwaysMatch": {}, "firstMatch": [{}, grouped]},
+        ):
+            driver = NativeDriver()
+            interceptor_network.note_session_request(driver, {"capabilities": caps})
+            self.assertIsNotNone(interceptor_network.start(driver, driver.execute_script))
+        driver = NativeDriver()
+        interceptor_network.note_session_request(
+            driver, {"capabilities": {"alwaysMatch": {"appium:options": {"intercept": "yes"}}}}
+        )
+        self.assertIsNone(interceptor_network.start(driver, driver.execute_script))
+
+    def test_warns_when_an_asking_driver_cannot_be_tracked(self):
+        class Untrackable:
+            __slots__ = ()
+
+        with self.assertLogs("selenium_devtools.interceptor", "WARNING") as logs:
+            interceptor_network.note_session_request(Untrackable(), _new_session(True))
+        self.assertIn("cannot be tracked", logs.output[0])
+
+    def test_stays_off_unless_asked_with_true(self):
+        for value in (None, "yes", False):
+            driver = NativeDriver()
+            interceptor_network.note_session_request(driver, _new_session(value))
+            self.assertIsNone(interceptor_network.start(driver, driver.execute_script))
+            self.assertEqual(driver.plugin.calls, [])
+
+    def test_stays_off_for_a_session_with_a_document(self):
+        driver = NativeDriver(caps={**NATIVE_CAPS, "browserName": "chrome"})
+        interceptor_network.note_session_request(driver, _new_session(True))
+        self.assertIsNone(interceptor_network.start(driver, driver.execute_script))
+        self.assertEqual(driver.plugin.calls, [])
+
+    def test_warns_when_the_plugin_does_not_answer(self):
+        driver = NativeDriver(FakePlugin(listener=None))
+        interceptor_network.note_session_request(driver, _new_session(True))
+        with self.assertLogs("selenium_devtools.interceptor", "WARNING") as logs:
+            self.assertIsNone(interceptor_network.start(driver, driver.execute_script))
+        self.assertIn("--use-plugins=appium-interceptor", logs.output[0])
+
+    def test_warns_when_the_plugin_command_throws(self):
+        driver = NativeDriver()
+
+        def boom(script, *args):
+            raise RuntimeError("unknown command")
+
+        interceptor_network.note_session_request(driver, _new_session(True))
+        with self.assertLogs("selenium_devtools.interceptor", "WARNING"):
+            self.assertIsNone(interceptor_network.start(driver, boom))
+
+
+class TestInstrumentationWiring(unittest.TestCase):
+    def setUp(self):
+        instrumentation.uninstall()
+        self.tx = RecordingTransport()
+        self.cap = SessionCapturer(self.tx)
+        instrumentation.install(self.cap, NativeDriver)
+        self.driver = NativeDriver()
+        self.driver.execute("newSession", _new_session(True))
+
+    def tearDown(self):
+        instrumentation.uninstall()
+
+    def _rows(self):
+        return [r for s, d in self.tx.sent if s == "networkRequests" for r in d]
+
+    def test_lists_requests_after_each_command_without_duplicates(self):
+        self.driver.plugin.data = [_entry("https://a/1")]
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        self.driver.execute("findElement", {"using": "id", "value": "y"})
+        self.assertEqual([r["url"] for r in self._rows()], ["https://a/1"])
+
+    def test_plugin_calls_are_not_action_rows(self):
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        commands = [d[0]["command"] for s, d in self.tx.sent if s == "commands"]
+        self.assertEqual(commands, ["findElement"])
+
+    def test_quit_lists_what_arrived_after_the_last_command(self):
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        self.driver.plugin.data = [_entry("https://a/late")]
+        self.driver.execute("quit")
+        self.assertEqual([r["url"] for r in self._rows()], ["https://a/late"])
+
+    def test_test_end_drain_lists_what_arrived_after_the_last_command(self):
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        self.driver.plugin.data = [_entry("https://a/late")]
+        instrumentation.drain_native_network(self.cap)
+        self.assertEqual([r["url"] for r in self._rows()], ["https://a/late"])
+
+    # Regression: the old listener was read through the driver, which by then addressed the new session.
+    def test_a_replaced_session_never_reads_the_old_listener_through_the_new_one(self):
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        before = len(self.driver.plugin.calls)
+        self.driver.plugin.listener = "L2"
+        self.driver.next_session = "sess-2"
+        self.driver.execute("newSession", _new_session(True))
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        old_reads = [
+            c for c in self.driver.plugin.calls[before:]
+            if c == ("interceptor: getInterceptedData", ({"id": "L1"},))
+        ]
+        self.assertEqual(old_reads, [])
+
+    def test_a_replaced_session_gets_a_fresh_listener(self):
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        self.driver.plugin.listener = "L2"
+        self.driver.plugin.data = []
+        self.driver.next_session = "sess-2"
+        self.driver.execute("newSession", _new_session(True))
+        self.driver.plugin.data = [_entry("https://b/1")]
+        self.driver.execute("findElement", {"using": "id", "value": "x"})
+        starts = [c for c in self.driver.plugin.calls if c[0] == "interceptor: startListening"]
+        self.assertEqual(len(starts), 2)
+        self.assertEqual(self._rows()[-1]["id"], "interceptor-L2-0")
+
+
+class TestPytestHook(unittest.TestCase):
+    def test_drains_before_the_outcome_is_recorded(self):
+        order = []
+        report = SimpleNamespace(
+            when="call", skipped=False, passed=True, failed=False,
+            nodeid="t.py::test_a", location=("t.py", 1, "test_a"),
+        )
+        with mock.patch.object(plugin, "_opted_in", return_value=True), mock.patch.object(
+            plugin.devtools, "get_capturer", return_value=object()
+        ), mock.patch.object(
+            instrumentation, "drain_native_network",
+            side_effect=lambda c: order.append("drain"),
+        ), mock.patch.object(
+            plugin._registry, "record", side_effect=lambda *a, **k: order.append("record")
+        ), mock.patch.object(plugin, "_publish"):
+            plugin.pytest_runtest_logreport(report)
+        self.assertEqual(order, ["drain", "record"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/packages/selenium-devtools/README.md b/packages/selenium-devtools/README.md
index 05e9f1e8..b26b4794 100644
--- a/packages/selenium-devtools/README.md
+++ b/packages/selenium-devtools/README.md
@@ -463,6 +463,33 @@ Appium sessions are captured in live and trace mode. A mobile-browser session is
 
 From the root of a built checkout of this repo (`pnpm install && pnpm build`), try it with `pnpm demo:selenium:mobile`, which drives the Android Clock app; `DEVTOOLS_MOBILE_PLATFORM=ios` runs the iOS spec against Settings, and `DEVTOOLS_MOBILE=web` drives the device's own browser. Prerequisites and the `DEVTOOLS_MOBILE` / `APPIUM_APP` switches are in [`examples/MOBILE.md`](../../examples/MOBILE.md); the full reference is the [Mobile Testing page](https://webdriver.io/docs/devtools/mobile).
 
+### Native app network
+
+A native app has no document and Appium serves no BiDi network events, so by default the Network tab of a native session is empty. The plugin can read the app's traffic from the [appium-interceptor](https://github.com/AppiumTestDistribution/appium-interceptor-plugin) plugin instead, which routes the device through a proxy of its own:
+
+```sh
+appium plugin install --source=npm appium-interceptor
+appium --use-plugins=appium-interceptor
+```
+
+Then ask for the proxy in the capabilities the `Builder` sends:
+
+```js
+const driver = await new Builder()
+  .usingServer('http://127.0.0.1:4723')
+  .withCapabilities({
+    platformName: 'Android',
+    'appium:automationName': 'UiAutomator2',
+    'appium:app': '/path/to/app.apk',
+    'appium:intercept': true
+  })
+  .build()
+```
+
+The plugin starts a listener for each driver and reads what the proxy recorded at most every 2 seconds between commands (each read returns the whole history), and always before each test's trace slice is written and before the driver quits, in both live and trace mode. Each request shows its method, URL, status, headers and bodies. The plugin records no timing, so a request is placed at the moment it was first read and shows no duration. Without the plugin a warning is logged and the session runs as usual. This applies to native sessions only; a mobile browser session already captures its network through the page.
+
+The plugin supports Android. For HTTPS traffic the app has to trust the certificate the plugin installs on the device, which a release build usually does not (see Android's network security configuration).
+
 ## How It Works
 
 The plugin patches `selenium-webdriver`'s `Builder`, `WebDriver`, and `WebElement` prototypes at import time:
diff --git a/packages/selenium-devtools/src/helpers/commandPostActions.ts b/packages/selenium-devtools/src/helpers/commandPostActions.ts
index 6acbd500..f761c20a 100644
--- a/packages/selenium-devtools/src/helpers/commandPostActions.ts
+++ b/packages/selenium-devtools/src/helpers/commandPostActions.ts
@@ -344,6 +344,7 @@ export async function handleOnCommand(
   maybeDrainAfterDomCommand(ctx, capturer, cmd)
   maybeDrainAfterLiveCommand(ctx, capturer, cmd)
   queueActionSnapshot(ctx, capturer, cmd, entry.timestamp, error)
+  void capturer.nativeNetwork.forward({ throttle: true })
 }
 
 /** Fire-and-forget post-action snapshot, drained at finalize. Stamped with the
diff --git a/packages/selenium-devtools/src/index.ts b/packages/selenium-devtools/src/index.ts
index 3d59ae85..53b2cebf 100644
--- a/packages/selenium-devtools/src/index.ts
+++ b/packages/selenium-devtools/src/index.ts
@@ -52,6 +52,7 @@ import {
 import {
   RetryTracker,
   errorMessage,
+  requestsInterceptedNetwork,
   tracePolicyModeWarning
 } from '@wdio/devtools-core'
 import { SeleniumTestArtifacts } from './test-artifacts.js'
@@ -363,6 +364,8 @@ class SeleniumDevToolsPlugin {
       get retryTracker() {
         return self.#retryTracker
       },
+      interceptRequestedFor: (driver: SeleniumDriverLike) =>
+        self.#interceptByDriver.get(driver) ?? false,
       get driver() {
         return self.#driver
       },
@@ -530,6 +533,8 @@ class SeleniumDevToolsPlugin {
   }
 
   async onDriverCreated(driver: SeleniumDriverLike) {
+    // Before any await: build() notes its builder and creates its driver in one synchronous turn.
+    this.#interceptByDriver.set(driver, this.#pendingIntercept)
     await sessionOnDriverCreated(this.#getInternals(), driver)
   }
 
@@ -591,12 +596,15 @@ class SeleniumDevToolsPlugin {
 
   /** The browser the next `build()` asks for, read before the dashboard opens. */
   #pendingBrowserName: string | undefined
+  #pendingIntercept = false
+  #interceptByDriver = new WeakMap<SeleniumDriverLike, boolean>()
   noteBuilder(builder: unknown) {
     const caps = (
       builder as { getCapabilities?: () => { get?: (k: string) => unknown } }
     ).getCapabilities?.()
     const name = caps?.get?.('browserName')
     this.#pendingBrowserName = typeof name === 'string' ? name : undefined
+    this.#pendingIntercept = requestsInterceptedNetwork(caps)
   }
 
   #finalized = false
diff --git a/packages/selenium-devtools/src/plugin-internals.ts b/packages/selenium-devtools/src/plugin-internals.ts
index 7e9f01f6..4efd86cb 100644
--- a/packages/selenium-devtools/src/plugin-internals.ts
+++ b/packages/selenium-devtools/src/plugin-internals.ts
@@ -46,6 +46,8 @@ export interface PluginInternals {
   readonly isReuse: boolean
   readonly finalized: boolean
   readonly retryTracker: RetryTracker
+  /** Whether the `build()` that created this driver asked for appium-interceptor's proxy. */
+  interceptRequestedFor(driver: SeleniumDriverLike): boolean
 
   // Mutable runtime instances
   driver: SeleniumDriverLike | undefined
diff --git a/packages/selenium-devtools/src/runnerHooks/cucumber.ts b/packages/selenium-devtools/src/runnerHooks/cucumber.ts
index 31febfc6..ba7e56c0 100644
--- a/packages/selenium-devtools/src/runnerHooks/cucumber.ts
+++ b/packages/selenium-devtools/src/runnerHooks/cucumber.ts
@@ -16,7 +16,7 @@ type CucumberModule = Record<string, unknown> & {
   BeforeAll?: (fn: () => void) => void
   AfterAll?: (fn: () => void) => void
   BeforeStep?: (fn: (arg: unknown) => void) => void
-  AfterStep?: (fn: (arg: unknown) => void) => void
+  AfterStep?: (fn: (arg: unknown) => void | Promise<void>) => void
 }
 
 interface StepDefinition {
@@ -390,10 +390,11 @@ function registerStepHooks(
     })
   }
   if (typeof AfterStep === 'function') {
-    AfterStep(function (raw) {
+    // Awaited so the step's slice is written, native requests included, before the next step records its boundary.
+    AfterStep(async function (raw) {
       const arg = raw as CucumberStepArg
       const state = mapCucumberStatus(String(arg?.result?.status ?? ''))
-      callbacks.onTestEnd(state)
+      await callbacks.onTestEnd(state)
     })
   }
 }
diff --git a/packages/selenium-devtools/src/session-lifecycle.ts b/packages/selenium-devtools/src/session-lifecycle.ts
index f9ae39e2..d353cb1c 100644
--- a/packages/selenium-devtools/src/session-lifecycle.ts
+++ b/packages/selenium-devtools/src/session-lifecycle.ts
@@ -32,6 +32,7 @@ import { SuiteManager } from './helpers/suiteManager.js'
 import { ScreencastRecorder } from './screencast.js'
 import { buildDriverMetadata } from './helpers/driverMetadata.js'
 import { attachBidiHandlers, buildBidiSinks } from './bidi.js'
+import { getDriverOriginals } from './driverPatcher.js'
 import { gracefulShutdown } from './helpers/processHooks.js'
 import type {
   ActionSnapshot,
@@ -72,6 +73,7 @@ export interface SessionLifecycleCtx {
   readonly launchCommand: string | undefined
   readonly isReuse: boolean
   readonly finalized: boolean
+  interceptRequestedFor(driver: SeleniumDriverLike): boolean
 
   driver: SeleniumDriverLike | undefined
   sessionCapturer: SessionCapturer | undefined
@@ -221,10 +223,26 @@ async function initPerDriverCapture(
   await Promise.all([
     screencastPromise,
     attachBidi(ctx, driver),
-    registerPreload(ctx, driver)
+    registerPreload(ctx, driver),
+    startNativeNetwork(ctx, ctx.sessionCapturer!, driver)
   ])
 }
 
+/** Through the unpatched `executeScript`, so the plugin's commands never render as action rows. */
+function startNativeNetwork(
+  ctx: SessionLifecycleCtx,
+  capturer: SessionCapturer,
+  driver: SeleniumDriverLike
+): Promise<void> {
+  const exec = getDriverOriginals().executeScript
+  return capturer.nativeNetwork.start({
+    requested: ctx.interceptRequestedFor(driver),
+    capabilities: capturer.metadata?.capabilities,
+    execute: exec && ((script, args) => exec(driver, script, ...args)),
+    unavailableReason: "selenium-webdriver's executeScript was not patched"
+  })
+}
+
 /** Route this driver's console/network through BiDi, which suppresses the
  *  script-injected equivalents so neither stream double-reports. */
 async function attachBidi(
@@ -277,6 +295,7 @@ export async function onDriverEnd(ctx: SessionLifecycleCtx): Promise<void> {
   // live mode, before the dashboard loses the session).
   // Gated on ctx.driver so the post-quit onDriverEnd call skips a dead session.
   if (ctx.driver && ctx.sessionCapturer) {
+    await ctx.sessionCapturer.nativeNetwork.forward()
     // Anchored: a run ending on a navigation (logout → /login as the last
     // action) leaves the destination's async initial anchor unrun at teardown.
     // Live mode goes through the post-command queue instead, so the drains
@@ -311,6 +330,7 @@ export async function onDriverEnd(ctx: SessionLifecycleCtx): Promise<void> {
   if (ctx.options.filmstrip && ctx.screencast) {
     ctx.filmstripFrames.push(...ctx.screencast.frames)
   }
+  ctx.sessionCapturer?.nativeNetwork.stop()
   ctx.driver = undefined
   ctx.screencast = undefined
   ctx.sessionId = undefined
@@ -531,23 +551,29 @@ function recordTestBoundary(
 export function flushCurrentTestTrace(
   ctx: SessionLifecycleCtx
 ): Promise<TraceArtifact | undefined> {
-  if (ctx.options.traceGranularity !== 'test' || !ctx.sessionCapturer) {
+  const capturer = ctx.sessionCapturer
+  if (!capturer) {
     return Promise.resolve(undefined)
   }
-  const sessionId = ctx.sessionCapturer.metadata?.sessionId
+  if (ctx.options.traceGranularity !== 'test') {
+    return capturer.nativeNetwork.forward().then(() => undefined)
+  }
+  const sessionId = capturer.metadata?.sessionId
+  // Chosen before any await: an unawaited cucumber AfterStep lets the next step record its boundary meanwhile.
   const currentRange = findFlushableRange(ctx.specRanges)
   if (!sessionId || currentRange?.testUid === undefined) {
     return Promise.resolve(undefined)
   }
-  const flush = flushRangeLogged(
-    buildTraceExportContext(
-      ctx,
-      ctx.sessionCapturer,
-      sessionId,
-      ctx.testFilePath
-    ),
-    currentRange
-  )
+  const testFilePath = ctx.testFilePath
+  const write = () =>
+    flushRangeLogged(
+      buildTraceExportContext(ctx, capturer, sessionId, testFilePath),
+      currentRange
+    )
+  // A request the app made after the test's last command is only in the proxy until drained.
+  const flush = capturer.nativeNetwork.active
+    ? capturer.nativeNetwork.forward().then(write)
+    : write()
   ctx.traceFlushes.push(flush)
   return flush
 }
diff --git a/packages/selenium-devtools/src/session.ts b/packages/selenium-devtools/src/session.ts
index 6450cce8..ea9e25d4 100644
--- a/packages/selenium-devtools/src/session.ts
+++ b/packages/selenium-devtools/src/session.ts
@@ -9,6 +9,7 @@ import {
   errorMessage,
   isSessionGoneError,
   loadInjectableScript,
+  NativeNetworkCapture,
   mapChromeBrowserLogs,
   pollUntilReady,
   serializeError
@@ -39,6 +40,11 @@ export class SessionCapturer extends SessionCapturerBase {
    *  post-navigation re-injection) unreachable rather than merely redundant. */
   preloadRegistered = false
 
+  /** Native app network from appium-interceptor's proxy; started per driver. */
+  readonly nativeNetwork = new NativeNetworkCapture(this, (level, message) =>
+    log[level](message)
+  )
+
   #clientConnected = false
   #clientConnectedWaiters: Array<() => void> = []
   #onClientDisconnected?: () => void
diff --git a/packages/selenium-devtools/src/test-artifacts.ts b/packages/selenium-devtools/src/test-artifacts.ts
index 04a69585..e4be621c 100644
--- a/packages/selenium-devtools/src/test-artifacts.ts
+++ b/packages/selenium-devtools/src/test-artifacts.ts
@@ -3,9 +3,9 @@
 // Selenium-only concerns: a live Allure attach sink (nightwatch is produce-only)
 // and the just-flushed trace slice threaded through the input bag.
 //
-// Selenium's runner hooks fire-and-forget some emits (Gherkin `AfterStep`), so
-// the mutable capture arrays are copied synchronously before the first await —
-// the next test would otherwise overwrite their backing arrays mid-flight. Each
+// The mutable capture arrays are copied synchronously before the first await,
+// so a runner that does not await the emit cannot have the next test overwrite
+// their backing arrays mid-flight. Each
 // core call self-gates on mode + `test` granularity + policy, so a non-trace run
 // or a coarser granularity is a no-op.
 
diff --git a/packages/selenium-devtools/tests/collector-preload.test.ts b/packages/selenium-devtools/tests/collector-preload.test.ts
index a7116bd5..7f9ba2a1 100644
--- a/packages/selenium-devtools/tests/collector-preload.test.ts
+++ b/packages/selenium-devtools/tests/collector-preload.test.ts
@@ -197,6 +197,7 @@ describe('per-driver preload registration', () => {
       screencastOptions: {},
       detectedRunner: 'mocha',
       finalized: false,
+      interceptRequestedFor: () => false,
       driver: undefined,
       sessionCapturer: undefined,
       actionSnapshots: [],
diff --git a/packages/selenium-devtools/tests/cucumber-step-hooks.test.ts b/packages/selenium-devtools/tests/cucumber-step-hooks.test.ts
new file mode 100644
index 00000000..da550fad
--- /dev/null
+++ b/packages/selenium-devtools/tests/cucumber-step-hooks.test.ts
@@ -0,0 +1,46 @@
+import { describe, expect, it, vi } from 'vitest'
+
+const hooks = vi.hoisted(() => ({
+  afterStep: undefined as ((arg: unknown) => void | Promise<void>) | undefined
+}))
+
+vi.mock('node:module', async (importOriginal) => ({
+  ...(await importOriginal<object>()),
+  createRequire: () => () => ({
+    Before: vi.fn(),
+    After: vi.fn(),
+    BeforeAll: vi.fn(),
+    AfterAll: vi.fn(),
+    BeforeStep: vi.fn(),
+    AfterStep: (fn: (arg: unknown) => void | Promise<void>) => {
+      hooks.afterStep = fn
+    }
+  })
+}))
+
+import { tryRegisterCucumberHooks } from '../src/runnerHooks/cucumber.js'
+
+describe('cucumber AfterStep', () => {
+  // Regression: unawaited, the next step recorded its trace boundary while this step's native drain was in flight.
+  it('waits for the step end before cucumber moves on to the next step', async () => {
+    let release: () => void = () => {}
+    const onTestEnd = vi.fn(
+      () => new Promise<void>((resolve) => (release = resolve))
+    )
+    expect(tryRegisterCucumberHooks({ onTestStart: vi.fn(), onTestEnd })).toBe(
+      true
+    )
+
+    let settled = false
+    const done = Promise.resolve(
+      hooks.afterStep?.({ result: { status: 'PASSED' } })
+    ).then(() => (settled = true))
+    await Promise.resolve()
+    expect(onTestEnd).toHaveBeenCalledWith('passed')
+    expect(settled).toBe(false)
+
+    release()
+    await done
+    expect(settled).toBe(true)
+  })
+})
diff --git a/packages/selenium-devtools/tests/native-network.test.ts b/packages/selenium-devtools/tests/native-network.test.ts
new file mode 100644
index 00000000..c69b5037
--- /dev/null
+++ b/packages/selenium-devtools/tests/native-network.test.ts
@@ -0,0 +1,239 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import { RetryTracker } from '@wdio/devtools-core'
+import type * as DevToolsCore from '@wdio/devtools-core'
+import type { ActionSnapshot } from '@wdio/devtools-shared'
+import { getDriverOriginals } from '../src/driverPatcher.js'
+import { resetSignatureCounters } from '../src/helpers/utils.js'
+import { TestManager } from '../src/helpers/testManager.js'
+import { SuiteManager } from '../src/helpers/suiteManager.js'
+import { TestReporter } from '../src/reporter.js'
+import {
+  handleOnCommand,
+  type OnCommandCtx
+} from '../src/helpers/commandPostActions.js'
+import {
+  flushCurrentTestTrace,
+  onDriverCreated,
+  onDriverEnd,
+  type SessionLifecycleCtx
+} from '../src/session-lifecycle.js'
+import type { CapturedCommand, SeleniumDriverLike } from '../src/types.js'
+
+const NATIVE = {
+  platformName: 'Android',
+  'appium:automationName': 'UiAutomator2'
+}
+
+const { capabilities } = vi.hoisted(() => ({
+  capabilities: { current: {} as Record<string, unknown> }
+}))
+
+const { flushed } = vi.hoisted(() => ({ flushed: [] as unknown[] }))
+
+vi.mock('@wdio/devtools-core', async (importOriginal) => ({
+  ...(await importOriginal<typeof DevToolsCore>()),
+  registerCollectorPreload: vi.fn(async () => false),
+  flushRangeLogged: vi.fn(async (_ctx: unknown, range: unknown) => {
+    flushed.push(range)
+    return undefined
+  })
+}))
+
+vi.mock('../src/bidi.js', async (importOriginal) => ({
+  ...(await importOriginal<object>()),
+  attachBidiHandlers: vi.fn(async () => false)
+}))
+
+vi.mock('../src/helpers/driverMetadata.js', () => ({
+  buildDriverMetadata: vi.fn(async () => ({
+    sessionId: 'sess',
+    metadata: { sessionId: 'sess', capabilities: capabilities.current }
+  }))
+}))
+
+const ENTRY = {
+  url: 'https://api.example.com/items',
+  method: 'GET',
+  statusCode: 200
+}
+
+const exec =
+  vi.fn<
+    (driver: unknown, script: string, ...args: unknown[]) => Promise<unknown>
+  >()
+
+const driver = (id: string) => ({ id }) as unknown as SeleniumDriverLike
+
+function makeCtx(interceptRequested: boolean): SessionLifecycleCtx {
+  return {
+    options: { hostname: 'localhost', port: 0, mode: 'trace' },
+    screencastOptions: {},
+    detectedRunner: 'mocha',
+    interceptRequestedFor: () => interceptRequested,
+    finalized: false,
+    driver: undefined,
+    sessionCapturer: undefined,
+    actionSnapshots: [],
+    snapshotCaptures: [],
+    specRanges: [],
+    flushedSpecs: new Set<string>(),
+    traceFlushes: [],
+    artifacts: [],
+    filmstripFrames: [],
+    setFinalized: () => {},
+    ensureBackendStarted: async () => {},
+    flushPendingTestActions: () => {},
+    resetRetryTracker: () => {},
+    clearKeepAlive: () => {}
+  } as unknown as SessionLifecycleCtx
+}
+
+const scripts = () =>
+  exec.mock.calls.map(([d, script]) => [(d as { id: string }).id, script])
+
+let ctx: SessionLifecycleCtx
+
+beforeEach(() => {
+  flushed.length = 0
+  capabilities.current = NATIVE
+  exec.mockReset()
+  getDriverOriginals().executeScript = exec as ReturnType<
+    typeof getDriverOriginals
+  >['executeScript']
+})
+
+afterEach(() => {
+  ctx?.sessionCapturer?.cleanup()
+  delete getDriverOriginals().executeScript
+})
+
+describe('selenium native network capture', () => {
+  it('starts a listener through the unpatched executeScript for a native session that asked for it', async () => {
+    exec.mockResolvedValueOnce('L1')
+    ctx = makeCtx(true)
+    await onDriverCreated(ctx, driver('d1'))
+    expect(exec).toHaveBeenCalledWith(
+      expect.objectContaining({ id: 'd1' }),
+      'interceptor: startListening',
+      {}
+    )
+    expect(ctx.sessionCapturer?.nativeNetwork.active).toBe(true)
+  })
+
+  it('stays off when the build did not ask, and for a session with a document', async () => {
+    ctx = makeCtx(false)
+    await onDriverCreated(ctx, driver('d1'))
+    ctx.sessionCapturer?.cleanup()
+    capabilities.current = { platformName: 'Android', browserName: 'chrome' }
+    ctx = makeCtx(true)
+    await onDriverCreated(ctx, driver('d1'))
+    expect(exec).not.toHaveBeenCalled()
+  })
+
+  it('records what the proxy holds after each command', async () => {
+    exec.mockResolvedValueOnce('L1').mockResolvedValue([ENTRY])
+    ctx = makeCtx(true)
+    await onDriverCreated(ctx, driver('d1'))
+    resetSignatureCounters()
+    const reporter = new TestReporter(vi.fn())
+    const suiteManager = new SuiteManager(reporter)
+    const root = suiteManager.getOrCreateRootSuite('app.spec.ts', 'Suite')
+    const commandCtx = {
+      sessionCapturer: ctx.sessionCapturer,
+      testManager: new TestManager(root, reporter, suiteManager),
+      retryTracker: new RetryTracker(),
+      options: { captureScreenshots: false, mode: 'live' },
+      finalized: false,
+      driver: undefined,
+      actionSnapshots: [] as ActionSnapshot[],
+      snapshotCaptures: []
+    } as unknown as OnCommandCtx
+    const cmd: CapturedCommand = {
+      command: 'getTitle',
+      fromElement: false,
+      args: [],
+      timestamp: 0,
+      startTime: 0,
+      result: 'x',
+      error: undefined,
+      callSource: undefined
+    }
+    await handleOnCommand(commandCtx, cmd)
+    await vi.waitFor(() =>
+      expect(ctx.sessionCapturer?.networkRequests).toHaveLength(1)
+    )
+    expect(ctx.sessionCapturer?.networkRequests[0]?.id).toBe('interceptor-L1-0')
+  })
+
+  it('drains before the per-test slice is written', async () => {
+    exec.mockResolvedValueOnce('L1').mockResolvedValue([ENTRY])
+    ctx = makeCtx(true)
+    await onDriverCreated(ctx, driver('d1'))
+    await flushCurrentTestTrace(ctx)
+    expect(ctx.sessionCapturer?.networkRequests).toHaveLength(1)
+  })
+
+  // Regression: the slice was chosen after the drain, so a boundary recorded meanwhile stole it.
+  it('flushes the ended test even when the next boundary lands during the drain', async () => {
+    exec.mockResolvedValueOnce('L1')
+    ctx = makeCtx(true)
+    ;(ctx.options as { traceGranularity?: string }).traceGranularity = 'test'
+    await onDriverCreated(ctx, driver('d1'))
+    const range = (testUid: string) => ({
+      specFile: 'app.spec.ts',
+      key: testUid,
+      testUid,
+      commandStartIdx: 0,
+      consoleStartIdx: 0,
+      networkStartIdx: 0,
+      mutationStartIdx: 0
+    })
+    ctx.specRanges.push(range('test-a') as (typeof ctx.specRanges)[number])
+    let answer!: (entries: unknown) => void
+    exec.mockReturnValueOnce(new Promise((resolve) => (answer = resolve)))
+    const pending = flushCurrentTestTrace(ctx)
+    expect(ctx.traceFlushes).toHaveLength(1)
+    ctx.specRanges.push(range('test-b') as (typeof ctx.specRanges)[number])
+    answer([ENTRY])
+    await pending
+    expect(flushed).toEqual([expect.objectContaining({ testUid: 'test-a' })])
+    expect(ctx.sessionCapturer?.networkRequests).toHaveLength(1)
+  })
+
+  it('reads the intercept flag the build of this driver asked for', async () => {
+    exec.mockResolvedValueOnce('L1')
+    ctx = makeCtx(false)
+    const asked = driver('d1')
+    ctx.interceptRequestedFor = (d) => d === asked
+    await onDriverCreated(ctx, asked)
+    expect(ctx.sessionCapturer?.nativeNetwork.active).toBe(true)
+  })
+
+  it('drains at driver end and reads nothing after the driver quit', async () => {
+    exec.mockResolvedValueOnce('L1').mockResolvedValue([ENTRY])
+    ctx = makeCtx(true)
+    await onDriverCreated(ctx, driver('d1'))
+    vi.spyOn(ctx.sessionCapturer!, 'captureTrace').mockResolvedValue()
+    await onDriverEnd(ctx)
+    expect(ctx.sessionCapturer?.networkRequests).toHaveLength(1)
+    expect(ctx.sessionCapturer?.nativeNetwork.active).toBe(false)
+  })
+
+  it('starts a new listener on the next driver', async () => {
+    exec.mockResolvedValueOnce('L1')
+    ctx = makeCtx(true)
+    await onDriverCreated(ctx, driver('d1'))
+    vi.spyOn(ctx.sessionCapturer!, 'captureTrace').mockResolvedValue()
+    exec.mockResolvedValueOnce([])
+    await onDriverEnd(ctx)
+    exec.mockResolvedValueOnce('L2')
+    await onDriverCreated(ctx, driver('d2'))
+    expect(scripts()).toEqual([
+      ['d1', 'interceptor: startListening'],
+      ['d1', 'interceptor: getInterceptedData'],
+      ['d2', 'interceptor: startListening']
+    ])
+    expect(ctx.sessionCapturer?.nativeNetwork.active).toBe(true)
+  })
+})
diff --git a/packages/service/README.md b/packages/service/README.md
index 75de9051..86b1d029 100644
--- a/packages/service/README.md
+++ b/packages/service/README.md
@@ -223,6 +223,30 @@ No configuration change is needed to switch modes — the service detects browse
 
 After every navigation command (`url`, `navigateTo`, etc.), the service runs the shared `CAPTURE_PERFORMANCE_SCRIPT` from `@wdio/devtools-core` to read `window.performance.getEntriesByType('navigation' | 'resource')`, cookies, and document info. The result is attached to the command entry in the Actions tab so you see `loadTime` / `domReady` / `responseTime` / resource counts per navigation. Same script and `applyPerformanceData` post-processing used by selenium-devtools and nightwatch-devtools — uniform dashboard fields across all three adapters.
 
+## Native app network (Appium)
+
+A native app has no document and Appium serves no BiDi network events, so by default the Network tab of a native session is empty. The service can read the app's traffic from the [appium-interceptor](https://github.com/AppiumTestDistribution/appium-interceptor-plugin) plugin instead, which routes the device through a proxy of its own:
+
+```sh
+appium plugin install --source=npm appium-interceptor
+appium --use-plugins=appium-interceptor
+```
+
+Then ask for the proxy in the session's capabilities:
+
+```ts
+capabilities: [{
+    platformName: 'Android',
+    'appium:automationName': 'UiAutomator2',
+    'appium:app': '/path/to/app.apk',
+    'appium:intercept': true
+}]
+```
+
+The service starts a listener when the session starts and reads what the proxy recorded at most every 2 seconds between commands (each read returns the whole history), and always before each test's trace slice is written and at the end of the session, in both live and trace mode. Each request shows its method, URL, status, headers and bodies. The plugin records no timing, so a request is placed at the moment it was first read and shows no duration. Without the plugin the service logs a warning and the session runs as usual. This applies to native sessions only; a mobile browser session already captures its network through the page.
+
+The plugin supports Android. For HTTPS traffic the app has to trust the certificate the plugin installs on the device, which a release build usually does not (see Android's network security configuration).
+
 ## Reference
 
 ```ts
diff --git a/packages/service/src/index.ts b/packages/service/src/index.ts
index 07ee2321..38b9c7d5 100644
--- a/packages/service/src/index.ts
+++ b/packages/service/src/index.ts
@@ -62,6 +62,7 @@ import { inPageProbesDeadlock, isAppiumSession } from './mobile.js'
 import { resolveSessionMetadata } from './session-metadata.js'
 import { stampRunnerMetadata } from './wdio-runner-id.js'
 import { publishConfigFile } from './standalone.js'
+import { InterceptedNetwork } from './intercepted-network.js'
 
 export * from './types.js'
 export const launcher = DevToolsAppLauncher
@@ -89,6 +90,7 @@ export default class DevToolsHookService implements Services.ServiceInstance {
    *  millisecond, so a scan mistook it for this slot and skipped the only
    *  capture of the last action's result. */
   #finalizedSlot?: { command: string; timestamp: number }
+  #nativeNetwork?: InterceptedNetwork
   #assertionTracker: AssertionTracker
   #screencast: ScreencastLifecycle
   #slices: TraceSliceTracker
@@ -281,6 +283,8 @@ export default class DevToolsHookService implements Services.ServiceInstance {
     }
 
     await this.#screencast.start(browser)
+    this.#nativeNetwork = new InterceptedNetwork(caps, this.#sessionCapturer)
+    await this.#nativeNetwork.start(browser)
 
     /**
      * Propagate session metadata at the beginning of the session. Not awaited
@@ -512,6 +516,8 @@ export default class DevToolsHookService implements Services.ServiceInstance {
     uid: string | undefined,
     failed: boolean
   ): Promise<void> {
+    // Requests the app made after the test's last command are only in the proxy until drained.
+    await this.#nativeNetwork?.forward()
     const attach = await getAllureSink()
     const onLog = (level: 'info' | 'warn', msg: string) => log[level](msg)
     if (uid) {
@@ -806,6 +812,7 @@ export default class DevToolsHookService implements Services.ServiceInstance {
         if (this.#options.mode !== 'trace') {
           await this.#drainAfterLiveCommand(command)
         }
+        await this.#nativeNetwork?.forward({ throttle: true })
         return captured
       }
     }
@@ -853,6 +860,7 @@ export default class DevToolsHookService implements Services.ServiceInstance {
     // finalize on — would lose it. A framework run has already finalized per
     // test; there this only re-captures the same timestamp and merges.
     await this.#finalizePerScenario()
+    await this.#nativeNetwork?.forward()
 
     // Stop and encode the screencast for the current session.
     await this.#screencast.finalize(this.#browser.sessionId)
@@ -876,6 +884,7 @@ export default class DevToolsHookService implements Services.ServiceInstance {
     // independent of screencast, so it runs before that early-returns.
     this.#sessionCapturer.resetScriptInjection()
     await this.#ensureInjected('reloadSession')
+    await this.#nativeNetwork?.start(this.#browser)
 
     await this.#screencast.handleReload(oldSessionId)
   }
diff --git a/packages/service/src/intercepted-network.ts b/packages/service/src/intercepted-network.ts
new file mode 100644
index 00000000..bf20a0dc
--- /dev/null
+++ b/packages/service/src/intercepted-network.ts
@@ -0,0 +1,56 @@
+import logger from '@wdio/logger'
+import {
+  NativeNetworkCapture,
+  requestsInterceptedNetwork,
+  webdriverExecute,
+  type InterceptorExecute
+} from '@wdio/devtools-core'
+import type { Capabilities } from '@wdio/types'
+import type { SessionCapturer } from './session.js'
+import { resolveWebDriverAddress } from './webdriver-address.js'
+
+const log = logger('@wdio/devtools-service:interceptedNetwork')
+
+/** WDIO glue over core's native network capture, for a session that asked for it with `appium:intercept`. */
+export class InterceptedNetwork {
+  #requested: boolean
+  #capture: NativeNetworkCapture
+
+  constructor(
+    requestedCaps: Capabilities.RequestedStandaloneCapabilities | undefined,
+    capturer: SessionCapturer
+  ) {
+    this.#requested = requestsInterceptedNetwork(requestedCaps)
+    this.#capture = new NativeNetworkCapture(capturer, (level, message) =>
+      log[level](message)
+    )
+  }
+
+  /** Called per session: a listener dies with its session, so `reloadSession` needs a new one. */
+  start(browser: WebdriverIO.Browser | undefined): Promise<void> {
+    return this.#capture.start(
+      browser && {
+        requested: this.#requested,
+        capabilities: browser.capabilities,
+        execute: executorFor(browser),
+        unavailableReason: 'the Appium address is not knowable'
+      }
+    )
+  }
+
+  forward(options?: { throttle?: boolean }): Promise<void> {
+    return this.#capture.forward(options)
+  }
+}
+
+/** Raw WebDriver HTTP, so the plugin's commands never re-enter WDIO's hooks or render as action rows. */
+function executorFor(
+  browser: WebdriverIO.Browser
+): InterceptorExecute | undefined {
+  const address = resolveWebDriverAddress(browser)
+  const sessionId = browser.sessionId
+  if (!address || !sessionId) {
+    return undefined
+  }
+  return (script, args) => webdriverExecute(address, sessionId, script, args)
+}
diff --git a/packages/service/src/session.ts b/packages/service/src/session.ts
index 0dc9846f..0052ef8e 100644
--- a/packages/service/src/session.ts
+++ b/packages/service/src/session.ts
@@ -670,8 +670,7 @@ export class SessionCapturer extends SessionCapturerBase {
         responseHeaders,
         size: response.bytesReceived
       }
-      this.networkRequests.push(networkRequest)
-      this.sendUpstream('networkRequests', [networkRequest])
+      this.recordNetworkRequests([networkRequest])
     } catch (err) {
       log.error(`handleNetworkResponseCompleted error: ${err}`)
     }
@@ -701,8 +700,7 @@ export class SessionCapturer extends SessionCapturerBase {
         time: endTime - pending.startTime,
         requestHeaders: pending.requestHeaders
       }
-      this.networkRequests.push(networkRequest)
-      this.sendUpstream('networkRequests', [networkRequest])
+      this.recordNetworkRequests([networkRequest])
     }
   }
 }
diff --git a/packages/service/tests/index.test.ts b/packages/service/tests/index.test.ts
index 10913a5a..dbf60511 100644
--- a/packages/service/tests/index.test.ts
+++ b/packages/service/tests/index.test.ts
@@ -89,6 +89,17 @@ vi.mock('@wdio/devtools-core', async (importOriginal) => {
   }
 })
 
+const mockNativeNetwork = {
+  start: vi.fn().mockResolvedValue(undefined),
+  forward: vi.fn().mockResolvedValue(undefined)
+}
+
+vi.mock('../src/intercepted-network.js', () => ({
+  InterceptedNetwork: vi.fn(function () {
+    return mockNativeNetwork
+  })
+}))
+
 vi.mock('node:fs/promises', () => ({
   default: { writeFile: vi.fn().mockResolvedValue(undefined) }
 }))
@@ -144,6 +155,24 @@ describe('DevtoolsService - Internal Command Filtering', () => {
     })
   })
 
+  // Regression: a request the app made after the test's last command was left out of that test's trace.
+  describe('native network at the test boundary', () => {
+    beforeEach(async () => {
+      await service.before({} as any, [], mockBrowser)
+      vi.clearAllMocks()
+    })
+
+    it('drains the proxy when a test ends', async () => {
+      await service.afterTest({ file: '/specs/a.ts', title: 't' })
+      expect(mockNativeNetwork.forward).toHaveBeenCalledTimes(1)
+    })
+
+    it('drains the proxy when a scenario ends', async () => {
+      await service.afterScenario({ pickle: { uri: 'a.feature', name: 's' } })
+      expect(mockNativeNetwork.forward).toHaveBeenCalledTimes(1)
+    })
+  })
+
   describe('config frame', () => {
     // Regression: sent from beforeSession on a capturer with no socket, so it was dropped every run.
     it('sends the config file on the capturer before() creates', async () => {
diff --git a/packages/service/tests/intercepted-network.test.ts b/packages/service/tests/intercepted-network.test.ts
new file mode 100644
index 00000000..144deda3
--- /dev/null
+++ b/packages/service/tests/intercepted-network.test.ts
@@ -0,0 +1,134 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockLogger, execute } = vi.hoisted(() => ({
+  mockLogger: { info: vi.fn(), warn: vi.fn(), debug: vi.fn(), error: vi.fn() },
+  execute: vi.fn()
+}))
+
+vi.mock('@wdio/logger', () => ({ default: vi.fn(() => mockLogger) }))
+
+vi.mock('@wdio/devtools-core', async (importOriginal) => ({
+  ...(await importOriginal<object>()),
+  webdriverExecute: execute
+}))
+
+import { InterceptedNetwork } from '../src/intercepted-network.js'
+import type { SessionCapturer } from '../src/session.js'
+
+const NATIVE_CAPS = {
+  platformName: 'Android',
+  'appium:automationName': 'UiAutomator2'
+}
+const REQUESTED = { ...NATIVE_CAPS, 'appium:intercept': true }
+
+const browser = (over: Record<string, unknown> = {}) =>
+  ({
+    sessionId: 'sess',
+    capabilities: NATIVE_CAPS,
+    options: { hostname: '127.0.0.1', port: 4723, path: '/' },
+    ...over
+  }) as unknown as WebdriverIO.Browser
+
+const capturer = () =>
+  ({
+    recordNetworkRequests: vi.fn()
+  }) as unknown as SessionCapturer & {
+    recordNetworkRequests: ReturnType<typeof vi.fn>
+  }
+
+const ENTRY = {
+  url: 'https://api.example.com/items',
+  method: 'GET',
+  statusCode: 200
+}
+const START = ['interceptor: startListening', [{}]]
+
+describe('InterceptedNetwork', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    execute.mockReset()
+  })
+
+  it('starts a listener over raw WebDriver HTTP for a native session that asked for the proxy', async () => {
+    execute.mockResolvedValueOnce('L1')
+    await new InterceptedNetwork(REQUESTED, capturer()).start(browser())
+    expect(execute).toHaveBeenCalledWith(
+      expect.objectContaining({ hostname: '127.0.0.1', port: 4723 }),
+      'sess',
+      ...START
+    )
+  })
+
+  it('stays off when the session did not ask for the proxy', async () => {
+    await new InterceptedNetwork(NATIVE_CAPS, capturer()).start(browser())
+    const notTrue = { ...NATIVE_CAPS, 'appium:intercept': 'yes' }
+    await new InterceptedNetwork(notTrue, capturer()).start(browser())
+    expect(execute).not.toHaveBeenCalled()
+  })
+
+  it('stays off when there is no browser to start on', async () => {
+    await new InterceptedNetwork(REQUESTED, capturer()).start(undefined)
+    expect(execute).not.toHaveBeenCalled()
+  })
+
+  it('stays off for a session with a document, whose network the page already captures', async () => {
+    const web = browser({
+      capabilities: { platformName: 'Android', browserName: 'chrome' }
+    })
+    await new InterceptedNetwork(REQUESTED, capturer()).start(web)
+    expect(execute).not.toHaveBeenCalled()
+  })
+
+  it('warns and stays off when the driver address is not knowable', async () => {
+    await new InterceptedNetwork(REQUESTED, capturer()).start(
+      browser({ options: {} })
+    )
+    expect(execute).not.toHaveBeenCalled()
+    expect(mockLogger.warn).toHaveBeenCalledWith(
+      expect.stringContaining('address is not knowable')
+    )
+  })
+
+  it('warns and forwards nothing when the plugin does not answer', async () => {
+    execute.mockResolvedValueOnce(null)
+    const target = capturer()
+    const network = new InterceptedNetwork(REQUESTED, target)
+    await network.start(browser())
+    await network.forward()
+    expect(mockLogger.warn).toHaveBeenCalledWith(
+      expect.stringContaining('appium-interceptor')
+    )
+    expect(execute).toHaveBeenCalledTimes(1)
+    expect(target.recordNetworkRequests).not.toHaveBeenCalled()
+  })
+
+  it('records and streams the rows a drain returns, and nothing for an empty one', async () => {
+    execute
+      .mockResolvedValueOnce('L1')
+      .mockResolvedValueOnce([ENTRY])
+      .mockResolvedValueOnce([ENTRY])
+    const target = capturer()
+    const network = new InterceptedNetwork(REQUESTED, target)
+    await network.start(browser())
+    await network.forward()
+    await network.forward()
+    expect(target.recordNetworkRequests).toHaveBeenCalledTimes(1)
+    expect(target.recordNetworkRequests).toHaveBeenCalledWith([
+      expect.objectContaining({ id: 'interceptor-L1-0', url: ENTRY.url })
+    ])
+  })
+
+  it('starts a fresh listener for a reloaded session and drops the old one', async () => {
+    execute.mockResolvedValueOnce('L1').mockResolvedValueOnce(null)
+    const network = new InterceptedNetwork(REQUESTED, capturer())
+    await network.start(browser())
+    await network.start(browser({ sessionId: 'sess-2' }))
+    expect(execute).toHaveBeenLastCalledWith(
+      expect.anything(),
+      'sess-2',
+      ...START
+    )
+    await network.forward()
+    expect(execute).toHaveBeenCalledTimes(2)
+  })
+})