Merge branch 'refactor/json-stream-events'

Replace the stdout pipe with iperf3's json_callback hook and consume the
resulting --json-stream events as structured data: interval samples with
retransmits and per-stream detail, and an end summary with sent/received
totals and CPU utilisation.

Deletes the pipe, the reader thread and the join-ordering and ownership
constraints that came with them.

Verified on an XQ-AS72 against a local iperf3 server: 4 streams at 889
Mbits/sec, matching the server's own 887 Mbits/sec receiver figure. Across
four consecutive runs thread count returned to baseline and fd count returned
exactly to baseline, with no ANR or native abort.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

android/app/src/main/cpp/iperf_jni.c

@@ -5,16 +5,25 @@
  * the Flutter app to run iperf3 tests natively inside the app process
  * (no Process.start needed, no SELinux restrictions).
  *
+ * OUTPUT NOTE:
+ * The test runs in --json-stream mode with a json_callback installed, so
+ * iperf3 hands each event ("start", "interval", "end", "error") straight to
+ * us as a JSON string and never touches test->outfile. That is what the hook
+ * exists for -- iperf.h calls it "allow user apps to receive the JSON
+ * strings, instead of writing them to the output file".
+ *
+ * An earlier version piped test->outfile to a reader thread instead. That
+ * required closing the pipe's write end before joining the reader (which
+ * blocks in fgets() until EOF) and single ownership of the shared state
+ * struct. Both were gotten wrong and deadlocked every test. Reading events
+ * from the callback removes the pipe, the reader thread, and that whole class
+ * of ordering bug.
+ *
  * THREADING NOTE:
  * runIperfLive() BLOCKS for the whole duration of the test. Kotlin must
  * call it from a worker thread, never from the Flutter platform thread.
- *
+ * json_callback fires on whichever thread iperf3 reports from, so it resolves
- * PIPE / SHUTDOWN NOTE:
+ * a JNIEnv for the calling thread rather than reusing a stored one.
- * iperf's output goes to a pipe drained by a reader thread that blocks in
- * fgets(). The reader only sees EOF once the write end is closed, so the
- * write end MUST be closed before pthread_join() — joining first deadlocks.
- * The state struct is owned solely by the JNI caller, which frees it after
- * joining the reader; the reader must never free it.
  *
  * SIGNAL HANDLING NOTE:
  * iperf3's run() function (called from iperf_run_client()) uses
@@ -61,21 +70,22 @@ static pthread_mutex_t g_test_mutex = PTHREAD_MUTEX_INITIALIZER;
 static volatile int g_stop = 0;
 static char g_tmp_template[1024] = "/data/local/tmp/iperf3.XXXXXX";
 
-/* Per-call state shared between the JNI caller and the reader thread.
+/* Per-call state, reachable from the json_callback.
- * We store the callback object (global ref) and method IDs.
  * Callbacks are invoked via CallVoidMethod, NOT as function pointers. */
 struct iperf_run_state {
     jobject callback_ref;       /* Global ref to the callback object */
     jmethodID mid_output;       /* Method ID for onOutput(String) */
     jmethodID mid_error;        /* Method ID for onError(String) */
     jmethodID mid_complete;     /* Method ID for onComplete() */
-    FILE *outfile;              /* Pipe write end; NULL once closed */
-    int pipe_fd[2];
     JavaVM *jvm;
-    pthread_t reader_thread;
-    int reader_started;         /* Only join if pthread_create succeeded */
 };
 
+/* iperf3's json_callback signature carries no user-data pointer, so the
+ * current run's state has to be reachable from a global. Only one test runs
+ * at a time (enforced in runIperfLive), and this is published before
+ * iperf_run_client() and cleared after it returns. */
+static struct iperf_run_state *g_current_state = NULL;
+
 /* Helper: invoke the output callback via CallVoidMethod */
 static void signal_output(JNIEnv *env, struct iperf_run_state *state,
                           const char *msg) {
@@ -108,50 +118,36 @@ static void signal_complete(JNIEnv *env, struct iperf_run_state *state) {
 }
 
 /* ------------------------------------------------------------------ */
-/* Output reader thread — reads from iperf's output pipe and forwards */
+/* json_callback — receives one JSON event string per report from      */
-/* each line to the Java callback via JNI.                            */
+/* iperf3 and forwards it to Java verbatim. Parsing happens in Dart.   */
 /*                                                                    */
-/* Exits when the write end of the pipe is closed (EOF). Does NOT own */
+/* Runs on iperf3's reporting thread. In the client that is the thread */
-/* `state`: the JNI caller frees it after joining this thread.        */
+/* that called iperf_run_client() (already attached, since Kotlin's    */
+/* executor thread is a Java thread), but GetEnv is checked rather     */
+/* than assumed, with an attach as fallback.                          */
 /* ------------------------------------------------------------------ */
-static void *reader_thread_func(void *arg) {
+static void json_event_callback(struct iperf_test *test, char *json) {
-    struct iperf_run_state *state = (struct iperf_run_state *) arg;
+    struct iperf_run_state *state = g_current_state;
-    JNIEnv *env;
+    if (!state || !json) return;
-    int ret;
 
-    /* Attach this native thread to the Java VM */
+    JNIEnv *env = NULL;
-    ret = (*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL);
+    int attached = 0;
-    if (ret != JNI_OK) {
-        LOGE("AttachCurrentThread failed: %d", ret);
-        /* Still drain the pipe so writers never block on a full pipe. */
-        char drain[1024];
-        while (read(state->pipe_fd[0], drain, sizeof(drain)) > 0) { }
-        close(state->pipe_fd[0]);
-        return NULL;
-    }
 
-    char buf[1024];
+    if ((*state->jvm)->GetEnv(state->jvm, (void **) &env, JNI_VERSION_1_6)
-    FILE *fp = fdopen(state->pipe_fd[0], "r");
+            != JNI_OK) {
-    if (fp) {
+        if ((*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL)
-        while (fgets(buf, sizeof(buf), fp) != NULL) {
+                != JNI_OK) {
-            /* Skip empty lines */
+            LOGE("json_event_callback: could not reach a JNIEnv");
-            if (buf[0] == '\n' || buf[0] == '\0')
+            return;
-                continue;
-            jstring line = (*env)->NewStringUTF(env, buf);
-            if (line != NULL) {
-                (*env)->CallVoidMethod(env, state->callback_ref,
-                                       state->mid_output, line);
-                (*env)->DeleteLocalRef(env, line);
-            }
         }
-        fclose(fp); /* also closes pipe_fd[0] */
+        attached = 1;
-    } else {
-        close(state->pipe_fd[0]);
     }
 
-    /* Detach this native thread */
+    signal_output(env, state, json);
-    (*state->jvm)->DetachCurrentThread(state->jvm);
+
-    return NULL;
+    if (attached) {
+        (*state->jvm)->DetachCurrentThread(state->jvm);
+    }
 }
 
 /* ------------------------------------------------------------------ */
@@ -198,30 +194,12 @@ static void free_argv(char **argv, int argc) {
     free(argv);
 }
 
-/* ------------------------------------------------------------------ */
-/* Close the pipe write end. Idempotent — both the normal and the      */
-/* longjmp path call it, and the second call must be a no-op.          */
-/*                                                                    */
-/* Closing this is what makes the reader thread see EOF and exit, so   */
-/* it must happen before any pthread_join().                           */
-/* ------------------------------------------------------------------ */
-static void close_outfile(struct iperf_run_state *state) {
-    if (state && state->outfile) {
-        fflush(state->outfile);
-        fclose(state->outfile);
-        state->outfile = NULL;
-    }
-}
-
 /* Retire the global test pointer and free it, under the mutex. */
 static void retire_test(void) {
     pthread_mutex_lock(&g_test_mutex);
     struct iperf_test *test = g_test;
     g_test = NULL;
     if (test) {
-        /* outfile is a closed FILE* by now — point it somewhere harmless so
-         * nothing in the teardown path can write to a dead stream. */
-        test->outfile = stdout;
         iperf_free_test(test);
     }
     pthread_mutex_unlock(&g_test_mutex);
@@ -236,14 +214,8 @@ static void iperf_cleanup_and_signal(JNIEnv *env,
                                      struct iperf_run_state *state,
                                      char **argv, int argc,
                                      int ret, int stopped) {
-    /* Close the write end FIRST: the reader thread is blocked in fgets()
+    /* No more events can be delivered once the run is over. */
-     * and only unblocks on EOF. Joining before this would deadlock. */
+    g_current_state = NULL;
-    close_outfile(state);
-
-    if (state->reader_started) {
-        pthread_join(state->reader_thread, NULL);
-        state->reader_started = 0;
-    }
 
     /* Restore default signal handlers before g_sigend_jmp_buf goes stale. */
     iperf_catch_sigend(SIG_DFL);
@@ -259,15 +231,11 @@ static void iperf_cleanup_and_signal(JNIEnv *env,
 
     /* Signal result to Java */
     if (user_stopped) {
-        signal_output(env, state, "[iPerf JNI] Test was stopped by user.\n");
+        signal_output(env, state, "{\"event\":\"jni_stopped\",\"data\":{}}");
     } else if (ret < 0) {
         char errbuf[256];
-        snprintf(errbuf, sizeof(errbuf),
+        snprintf(errbuf, sizeof(errbuf), "%s", iperf_strerror(i_errno));
-                 "[iPerf JNI] Test failed: %s\n",
-                 iperf_strerror(i_errno));
         signal_error(env, state, errbuf);
-    } else {
-        signal_output(env, state, "[iPerf JNI] Test completed successfully.\n");
     }
 
     /* Signal completion to Java */
@@ -325,20 +293,20 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
         if (mid_complete) (*env)->CallVoidMethod(env, callback, mid_complete); \
     } while (0)
 
-    /* Refuse to start a second concurrent test — the globals below hold
+    /* Refuse to start a second concurrent test — the globals above hold
      * exactly one test's worth of state. */
     pthread_mutex_lock(&g_test_mutex);
     int busy = (g_test != NULL);
     pthread_mutex_unlock(&g_test_mutex);
     if (busy) {
-        FAIL_EARLY("[iPerf JNI] A test is already running.");
+        FAIL_EARLY("A test is already running.");
         return;
     }
 
     /* Create the iperf test object */
     struct iperf_test *test = iperf_new_test();
     if (!test) {
-        FAIL_EARLY("[iPerf JNI] Failed to create iperf test");
+        FAIL_EARLY("Failed to create iperf test");
         return;
     }
 
@@ -352,7 +320,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
     char **argv = build_argv(env, arguments, &total_argc);
     if (!argv) {
         retire_test();
-        FAIL_EARLY("[iPerf JNI] Failed to allocate argv");
+        FAIL_EARLY("Failed to allocate argv");
         return;
     }
 
@@ -365,19 +333,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
     if (iperf_parse_arguments(test, total_argc, argv) < 0) {
         char errbuf[256];
         snprintf(errbuf, sizeof(errbuf),
-                 "[iPerf JNI] Argument parse error: %s", iperf_strerror(i_errno));
+                 "Argument parse error: %s", iperf_strerror(i_errno));
-        free_argv(argv, total_argc);
-        retire_test();
-        FAIL_EARLY(errbuf);
-        return;
-    }
-
-    /* Create a pipe to capture iperf's output */
-    int pipefd[2];
-    if (pipe(pipefd) != 0) {
-        char errbuf[256];
-        snprintf(errbuf, sizeof(errbuf),
-                 "[iPerf JNI] Failed to create output pipe: %s", strerror(errno));
         free_argv(argv, total_argc);
         retire_test();
         FAIL_EARLY(errbuf);
@@ -389,25 +345,11 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
             calloc(1, sizeof(struct iperf_run_state));
     if (!state) {
         free_argv(argv, total_argc);
-        close(pipefd[0]);
-        close(pipefd[1]);
         retire_test();
-        FAIL_EARLY("[iPerf JNI] Failed to allocate run state");
+        FAIL_EARLY("Failed to allocate run state");
         return;
     }
 
-    FILE *outfile = fdopen(pipefd[1], "w");
-    if (!outfile) {
-        free(state);
-        free_argv(argv, total_argc);
-        close(pipefd[0]);
-        close(pipefd[1]);
-        retire_test();
-        FAIL_EARLY("[iPerf JNI] Failed to open output stream");
-        return;
-    }
-
-    /* Get JavaVM for reader-thread attachment */
     JavaVM *jvm;
     (*env)->GetJavaVM(env, &jvm);
 
@@ -415,14 +357,16 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
     state->mid_error = mid_error;
     state->mid_complete = mid_complete;
     state->callback_ref = (*env)->NewGlobalRef(env, callback);
-    state->outfile = outfile;
-    state->pipe_fd[0] = pipefd[0];
-    state->pipe_fd[1] = pipefd[1];
     state->jvm = jvm;
 
-    /* Redirect iperf's output to our pipe */
+    /* Deliver results as a stream of JSON events straight to
-    test->outfile = state->outfile;
+     * json_event_callback, rather than as text through test->outfile.
-    setvbuf(test->outfile, NULL, _IOLBF, 0); /* Line-buffered */
+     * --json-stream sets both of these flags, so setting json_stream alone
+     * would leave iperf3 printing human-readable text as well. */
+    iperf_set_test_json_output(test, 1);
+    iperf_set_test_json_stream(test, 1);
+    iperf_set_test_json_callback(test, json_event_callback);
+    g_current_state = state;
 
     /* Set temp directory for stream buffers (mkstemp in iperf_new_stream).
      * On Android, /data/local/tmp is NOT writable from the app sandbox —
@@ -430,22 +374,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
      * the strdup internally. */
     iperf_set_test_template(test, g_tmp_template);
 
-    /* Spawn reader thread to forward iperf output to Java. Started before
-     * the test so early output is not lost, and before setjmp so the
-     * longjmp path can rely on reader_started being accurate. */
-    if (pthread_create(&state->reader_thread, NULL,
-                       reader_thread_func, state) != 0) {
-        close_outfile(state);
-        close(pipefd[0]);
-        free_argv(argv, total_argc);
-        retire_test();
-        (*env)->DeleteGlobalRef(env, state->callback_ref);
-        free(state);
-        FAIL_EARLY("[iPerf JNI] Failed to start output reader thread");
-        return;
-    }
-    state->reader_started = 1;
-
     #undef FAIL_EARLY
 
     /*
@@ -462,8 +390,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
         signal(SIGPIPE, SIG_IGN);  /* Ignore SIGPIPE for socket errors */
         iperf_catch_sigend(sigend_handler);
 
-        signal_output(env, state, "[iPerf JNI] Starting iPerf3 client...\n");
-
         g_stop = 0;
         int ret = iperf_run_client(test);
         LOGI("iperf_run_client returned: %d i_errno=%d strerror=%s",
@@ -478,8 +404,8 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive(
     } else {
         /*
          * We got here via longjmp() from sigend_handler.
-         * iperf_cleanup_and_signal() closes the pipe, joins the reader and
+         * iperf_cleanup_and_signal() frees everything, so there is nothing
-         * frees everything, so there is nothing to unwind by hand here.
+         * to unwind by hand here.
          */
         LOGI("Test interrupted by signal %d", g_signed_sig);
         g_stop = 1;
@@ -515,19 +441,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_forceStopIperfTest(
     pthread_mutex_unlock(&g_test_mutex);
 
     if (!requested) return;
-
+    LOGI("Graceful stop requested");
-    /* Notify Java */
-    jclass callback_cls = (*env)->GetObjectClass(env, callback);
-    jmethodID mid_output = (*env)->GetMethodID(
-            env, callback_cls, "onOutput", "(Ljava/lang/String;)V");
-    if (mid_output) {
-        jstring msg = (*env)->NewStringUTF(env,
-                                           "[iPerf JNI] Requested graceful stop.\n");
-        if (msg) {
-            (*env)->CallVoidMethod(env, callback, mid_output, msg);
-            (*env)->DeleteLocalRef(env, msg);
-        }
-    }
 }
 
 /* ------------------------------------------------------------------ */

docs/README.md

@@ -42,8 +42,9 @@ User taps GO
   → Kotlin IperfRunner: submits to the iperf3-run worker thread, returns at once
   → JNI C: build_argv() (prepends argv[0]) → iperf_parse_arguments()
   → iperf_run_client(test)             [blocks on the worker thread]
-  → Reader thread pipes output → CallVoidMethod → main-thread hop → Dart onOutput
+  → iperf3 json_callback → one JSON event per report → CallVoidMethod
-  → Dart streams each line to the UI via Iperf3Service.outputLines
+  → main-thread hop → Dart onOutput → Iperf3Event.parse()
+  → Iperf3Service.events → view model: transcript + interval series + summary
   → On completion: Dart Completer resolves with Iperf3Result
 ```
 
@@ -61,16 +62,51 @@ to the main looper — stalled every callback until the test was already over.
 ### Callback Mechanism
 The JNI bridge uses `CallVoidMethod` on stored `jmethodID`s to invoke the Kotlin `IperfCallback` interface, NOT function pointer casts. Direct function pointer casts caused SIGBUS crashes on ARM due to JNINativeInterface method ID layout differences.
 
-### Output Streaming
+### Output via json_callback, not a pipe
-iperf3 output is redirected via pipe to a background reader thread that attaches
+The test runs with `json_output` and `json_stream` set and a `json_callback`
-to the Java VM via `AttachCurrentThread` and invokes callbacks via
+installed, so iperf3 hands each report to the bridge as a JSON string and never
-`CallVoidMethod`. Dart re-publishes each line on `Iperf3Service.outputLines`, so
+writes to `test->outfile`. `iperf.h` describes the hook as "allow user apps to
-the transcript grows while the test runs.
+receive the JSON strings, instead of writing them to the output file" — this is
+its intended use.
 
-The reader blocks in `fgets()` and only unblocks on EOF, so the pipe's write end
+An earlier design piped `test->outfile` to a reader thread. That imposed two
-must be closed **before** `pthread_join()` — joining first deadlocks the worker
+constraints, both of which were gotten wrong and deadlocked every test: the
-thread. The JNI caller solely owns the run-state struct and frees it after the
+pipe's write end had to be closed *before* joining a reader blocked in
-join; the reader must never free it.
+`fgets()`, and exactly one side could own the shared state struct. Reading from
+the callback removes the pipe, the reader thread and both constraints. Don't
+reintroduce it.
+
+`json_callback` carries no user-data pointer, so the run state is reached
+through a global published before `iperf_run_client()` and cleared afterwards;
+only one test may run at a time, which `runIperfLive` enforces. The callback
+fires on iperf3's reporting thread, so it resolves a `JNIEnv` for the calling
+thread rather than reusing a stored one.
+
+Dart parses each event into a sealed `Iperf3Event`; unrecognised output becomes
+an `Iperf3LogLine` rather than being discarded. The UI rebuilds iperf3's text
+layout from the events and keeps the interval series as data.
+
+### Embedding a CLI in a long-lived process
+iperf3's `main()` runs one test and exits, so anything it leaves behind is
+reclaimed by the OS. Linked into an app process that runs tests repeatedly for
+hours, every such leftover becomes this app's problem. What that costs in
+practice:
+
+- **Signal handlers are process-global.** `iperf_catch_sigend()` and
+  `signal(SIGPIPE, SIG_IGN)` change behaviour for the whole process, Flutter and
+  ART included. Handlers are restored to `SIG_DFL` when a test ends.
+- **`g_sigend_jmp_buf` outlives its frame.** Harmless in a process about to
+  exit; a jump into a dead stack frame otherwise. Same fix as above.
+- **Globals mean one test at a time.** `i_errno` and the bridge's own statics
+  hold a single test's worth of state, so concurrent runs are refused.
+
+What iperf3 itself handles correctly, verified rather than assumed: per-stream
+threads are joined (`iperf_client_api.c`), stream buffer fds are closed, and the
+`mkstemp` temp file is `unlink`ed immediately after creation
+(`iperf_api.c:4652`), so nothing accumulates in the cache dir. Measured across
+four consecutive 4-stream runs on device, thread count returned to baseline
+(33 → 40 mid-run → 34) and fd count returned exactly to baseline (175 → 183 →
+175).
 
 ### Stop Handling
 A stop shuts down the control socket, so `iperf_run_client()` returns -1 with

lib/data/services/iperf3_service.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 
 import 'package:flutter/foundation.dart';
 import 'package:flutter/services.dart';
+import 'package:iperf3gui/domain/models/iperf3_event.dart';
 import 'package:iperf3gui/domain/models/iperf3_result.dart';
 import 'package:iperf3gui/domain/models/test_configuration.dart';
 
@@ -10,9 +11,11 @@ import 'package:iperf3gui/domain/models/test_configuration.dart';
 /// On Android, iperf3 is compiled into the app as a shared library (.so).
 /// This avoids SELinux restrictions on executing ELF binaries from assets.
 ///
-/// The service talks to the native layer over a MethodChannel. Native output
+/// The service talks to the native layer over a MethodChannel. The test runs
-/// arrives line by line on [outputLines] while the test runs; [run] resolves
+/// in `--json-stream` mode, so each native report arrives as one JSON object;
-/// with the accumulated transcript once the native layer reports completion.
+/// they are parsed and published on [events] while the test runs. [run]
+/// resolves once the native layer reports completion, carrying the raw JSON
+/// lines as the result's output.
 class Iperf3Service {
   Iperf3Service() {
     _channel.setMethodCallHandler(_handleNativeCall);
@@ -33,16 +36,15 @@ class Iperf3Service {
   /// Exit code reported when the native layer never signalled completion.
   static const int exitCodeTimeout = -3;
 
-  /// Lines emitted by the native layer during the current run.
+  /// Events parsed from the native layer during the current run.
-  final StreamController<String> _outputController =
+  final StreamController<Iperf3Event> _eventController =
-      StreamController<String>.broadcast();
+      StreamController<Iperf3Event>.broadcast();
 
-  /// Live output, one line per event. Use this to render progress; [run]'s
+  /// Live reports, one event per native report. Use this to render progress.
-  /// result carries the same text joined together.
+  Stream<Iperf3Event> get events => _eventController.stream;
-  Stream<String> get outputLines => _outputController.stream;
 
-  /// Transcript of the current (or most recent) run.
+  /// Raw JSON lines received during the current (or most recent) run.
-  final List<String> _outputBuffer = [];
+  final List<String> _rawLines = [];
 
   /// Completer that resolves when the native iperf3 test finishes.
   Completer<Iperf3Result>? _runCompleter;
@@ -71,7 +73,7 @@ class Iperf3Service {
   /// Release the stream controller. Call from the owner's dispose().
   Future<void> dispose() async {
     _channel.setMethodCallHandler(null);
-    await _outputController.close();
+    await _eventController.close();
   }
 
   /// Run iperf3 with the given configuration.
@@ -88,7 +90,7 @@ class Iperf3Service {
     _isRunning = true;
     _sawError = false;
     _stopRequested = false;
-    _outputBuffer.clear();
+    _rawLines.clear();
 
     final args = config.toCliArgs();
     debugPrint('Iperf3Service: running: ${args.join(" ")}');
@@ -121,9 +123,9 @@ class Iperf3Service {
             debugPrint('Iperf3Service: error stopping after timeout: $e');
             return null;
           });
-          _emit('[ERROR] Timed out waiting for iperf3 to finish.');
+          _publish(const Iperf3Error('Timed out waiting for iperf3 to finish.'));
           return Iperf3Result(
-            rawOutput: _transcript(),
+            rawOutput: _rawOutput(),
             exitCode: exitCodeTimeout,
           );
         },
@@ -145,7 +147,12 @@ class Iperf3Service {
       case 'onOutput':
         final output = call.arguments['output'] as String?;
         if (output != null) {
-          _emit(output);
+          _rawLines.add(output);
+          final event = Iperf3Event.parse(output);
+          // iperf3 reports a failed test as an `error` event rather than a
+          // non-zero return, so this is where most failures surface.
+          if (event is Iperf3Error) _sawError = true;
+          _publish(event);
         }
       case 'onError':
         final error = call.arguments['error'] as String?;
@@ -153,7 +160,7 @@ class Iperf3Service {
           // Not terminal on its own — iperf3 reports the failure reason and
           // then completes, so wait for onComplete before resolving.
           _sawError = true;
-          _emit('[ERROR] $error');
+          _publish(Iperf3Error(error));
         }
       case 'onComplete':
         debugPrint('Iperf3Service: test completed');
@@ -161,7 +168,7 @@ class Iperf3Service {
         if (completer != null && !completer.isCompleted) {
           completer.complete(
             Iperf3Result(
-              rawOutput: _transcript(),
+              rawOutput: _rawOutput(),
               // A user-requested stop is a normal outcome, not a failure.
               exitCode:
                   (_sawError && !_stopRequested) ? exitCodeNativeError : 0,
@@ -174,17 +181,14 @@ class Iperf3Service {
     }
   }
 
-  void _emit(String line) {
+  void _publish(Iperf3Event event) {
-    _outputBuffer.add(line);
+    if (!_eventController.isClosed) {
-    if (!_outputController.isClosed) {
+      _eventController.add(event);
-      _outputController.add(line);
     }
-    debugPrint('Iperf3Service output: ${line.trimRight()}');
   }
 
-  /// Native output already carries its own newlines; keep it verbatim except
+  /// The raw JSON stream as received, one event per line.
-  /// for lines we synthesised ourselves.
+  String _rawOutput() => _rawLines
-  String _transcript() => _outputBuffer
       .map((line) => line.endsWith('\n') ? line : '$line\n')
       .join();
 

lib/domain/models/iperf3_event.dart

@@ -0,0 +1,238 @@
+import 'dart:convert';
+
+/// One report from iperf3's `--json-stream` output.
+///
+/// iperf3 emits `{"event": "<name>", "data": {...}}` per report; the events a
+/// client run produces are `start`, `interval`, `end` and `error`. The bridge
+/// adds `jni_stopped` when a run ended because the user stopped it.
+sealed class Iperf3Event {
+  const Iperf3Event();
+
+  /// Parse one line of `--json-stream` output.
+  ///
+  /// Returns an [Iperf3LogLine] for anything that is not a recognised JSON
+  /// event, so unexpected output is surfaced rather than dropped.
+  static Iperf3Event parse(String line) {
+    final trimmed = line.trim();
+    if (trimmed.isEmpty) return const Iperf3LogLine('');
+    if (!trimmed.startsWith('{')) return Iperf3LogLine(trimmed);
+
+    final Object? decoded;
+    try {
+      decoded = jsonDecode(trimmed);
+    } on FormatException {
+      return Iperf3LogLine(trimmed);
+    }
+    if (decoded is! Map<String, dynamic>) return Iperf3LogLine(trimmed);
+
+    final event = decoded['event'];
+    final data = decoded['data'];
+    if (event is! String) return Iperf3LogLine(trimmed);
+    final map = data is Map<String, dynamic> ? data : const <String, dynamic>{};
+
+    return switch (event) {
+      'start' => Iperf3Start.fromJson(map),
+      'interval' => Iperf3Interval.fromJson(map),
+      'end' => Iperf3End.fromJson(map),
+      'error' => Iperf3Error(_stringOf(data) ?? 'unknown error'),
+      'jni_stopped' => const Iperf3Stopped(),
+      _ => Iperf3LogLine(trimmed),
+    };
+  }
+
+  static String? _stringOf(Object? value) {
+    if (value is String) return value;
+    if (value is Map<String, dynamic>) {
+      final error = value['error'];
+      if (error is String) return error;
+    }
+    return null;
+  }
+}
+
+/// Connection details, emitted once at the start of a test.
+class Iperf3Start extends Iperf3Event {
+  final String? targetHost;
+  final int? targetPort;
+  final int? streamCount;
+  final int? durationSeconds;
+  final String? protocol;
+
+  const Iperf3Start({
+    this.targetHost,
+    this.targetPort,
+    this.streamCount,
+    this.durationSeconds,
+    this.protocol,
+  });
+
+  factory Iperf3Start.fromJson(Map<String, dynamic> data) {
+    final connecting = data['connecting_to'];
+    final testStart = data['test_start'];
+    return Iperf3Start(
+      targetHost: connecting is Map ? connecting['host'] as String? : null,
+      targetPort: connecting is Map ? _asInt(connecting['port']) : null,
+      streamCount: testStart is Map ? _asInt(testStart['num_streams']) : null,
+      durationSeconds: testStart is Map ? _asInt(testStart['duration']) : null,
+      protocol: testStart is Map ? testStart['protocol'] as String? : null,
+    );
+  }
+}
+
+/// One reporting interval, summed across all streams.
+class Iperf3Interval extends Iperf3Event {
+  final double start;
+  final double end;
+  final int bytes;
+  final double bitsPerSecond;
+  final int? retransmits;
+  final bool omitted;
+
+  /// Per-stream figures, when more than one stream is running.
+  final List<Iperf3StreamInterval> streams;
+
+  const Iperf3Interval({
+    required this.start,
+    required this.end,
+    required this.bytes,
+    required this.bitsPerSecond,
+    this.retransmits,
+    this.omitted = false,
+    this.streams = const [],
+  });
+
+  factory Iperf3Interval.fromJson(Map<String, dynamic> data) {
+    // Non-bidirectional runs use "sum"; bidir adds "sum_bidir_reverse".
+    final sum = data['sum'] ?? data['sum_bidir_reverse'];
+    final sumMap = sum is Map<String, dynamic> ? sum : const <String, dynamic>{};
+    final rawStreams = data['streams'];
+
+    return Iperf3Interval(
+      start: _asDouble(sumMap['start']) ?? 0,
+      end: _asDouble(sumMap['end']) ?? 0,
+      bytes: _asInt(sumMap['bytes']) ?? 0,
+      bitsPerSecond: _asDouble(sumMap['bits_per_second']) ?? 0,
+      retransmits: _asInt(sumMap['retransmits']),
+      omitted: sumMap['omitted'] == true,
+      streams: rawStreams is List
+          ? rawStreams
+              .whereType<Map<String, dynamic>>()
+              .map(Iperf3StreamInterval.fromJson)
+              .toList(growable: false)
+          : const [],
+    );
+  }
+}
+
+/// One stream's contribution to an interval.
+class Iperf3StreamInterval {
+  final int? socket;
+  final int bytes;
+  final double bitsPerSecond;
+  final int? retransmits;
+  final int? sendCwnd;
+  final int? rtt;
+
+  const Iperf3StreamInterval({
+    this.socket,
+    required this.bytes,
+    required this.bitsPerSecond,
+    this.retransmits,
+    this.sendCwnd,
+    this.rtt,
+  });
+
+  factory Iperf3StreamInterval.fromJson(Map<String, dynamic> data) {
+    return Iperf3StreamInterval(
+      socket: _asInt(data['socket']),
+      bytes: _asInt(data['bytes']) ?? 0,
+      bitsPerSecond: _asDouble(data['bits_per_second']) ?? 0,
+      retransmits: _asInt(data['retransmits']),
+      sendCwnd: _asInt(data['snd_cwnd']),
+      rtt: _asInt(data['rtt']),
+    );
+  }
+}
+
+/// Totals for the whole test, emitted once at the end.
+class Iperf3End extends Iperf3Event {
+  final Iperf3Total? sent;
+  final Iperf3Total? received;
+  final double? hostCpuPercent;
+  final double? remoteCpuPercent;
+
+  const Iperf3End({
+    this.sent,
+    this.received,
+    this.hostCpuPercent,
+    this.remoteCpuPercent,
+  });
+
+  factory Iperf3End.fromJson(Map<String, dynamic> data) {
+    final cpu = data['cpu_utilization_percent'];
+    return Iperf3End(
+      sent: Iperf3Total.fromJson(
+          data['sum_sent'] ?? data['sum_sent_bidir_reverse']),
+      received: Iperf3Total.fromJson(
+          data['sum_received'] ?? data['sum_received_bidir_reverse']),
+      hostCpuPercent: cpu is Map ? _asDouble(cpu['host_total']) : null,
+      remoteCpuPercent: cpu is Map ? _asDouble(cpu['remote_total']) : null,
+    );
+  }
+}
+
+/// A sent or received total from the `end` event.
+class Iperf3Total {
+  final double seconds;
+  final int bytes;
+  final double bitsPerSecond;
+  final int? retransmits;
+
+  const Iperf3Total({
+    required this.seconds,
+    required this.bytes,
+    required this.bitsPerSecond,
+    this.retransmits,
+  });
+
+  static Iperf3Total? fromJson(Object? data) {
+    if (data is! Map<String, dynamic>) return null;
+    return Iperf3Total(
+      seconds: _asDouble(data['seconds']) ?? 0,
+      bytes: _asInt(data['bytes']) ?? 0,
+      bitsPerSecond: _asDouble(data['bits_per_second']) ?? 0,
+      retransmits: _asInt(data['retransmits']),
+    );
+  }
+}
+
+/// iperf3 reported an error for this test.
+class Iperf3Error extends Iperf3Event {
+  final String message;
+  const Iperf3Error(this.message);
+}
+
+/// The run ended because the user stopped it.
+class Iperf3Stopped extends Iperf3Event {
+  const Iperf3Stopped();
+}
+
+/// Output that was not a recognised JSON event.
+class Iperf3LogLine extends Iperf3Event {
+  final String text;
+  const Iperf3LogLine(this.text);
+}
+
+int? _asInt(Object? value) => switch (value) {
+      int v => v,
+      double v => v.round(),
+      String v => int.tryParse(v),
+      _ => null,
+    };
+
+double? _asDouble(Object? value) => switch (value) {
+      double v => v,
+      int v => v.toDouble(),
+      String v => double.tryParse(v),
+      _ => null,
+    };

lib/ui/features/test_run/iperf3_event_formatter.dart

@@ -0,0 +1,95 @@
+import 'package:iperf3gui/domain/models/iperf3_event.dart';
+
+/// Renders parsed iperf3 events as the kind of line iperf3 itself would print.
+///
+/// The app consumes `--json-stream`, so the familiar text layout has to be
+/// rebuilt here. Keeping it in one pure function makes it testable and leaves
+/// the structured events untouched for anything else that wants them.
+String? formatIperf3Event(Iperf3Event event) {
+  return switch (event) {
+    Iperf3Start e => _formatStart(e),
+    Iperf3Interval e => _formatInterval(e),
+    Iperf3End e => _formatEnd(e),
+    Iperf3Error e => 'iperf3 error: ${e.message}',
+    Iperf3Stopped _ => 'Stopped by user.',
+    Iperf3LogLine e => e.text.isEmpty ? null : e.text,
+  };
+}
+
+String _formatStart(Iperf3Start e) {
+  final parts = <String>[];
+  if (e.targetHost != null) {
+    final port = e.targetPort != null ? ', port ${e.targetPort}' : '';
+    parts.add('Connecting to host ${e.targetHost}$port');
+  }
+  final detail = <String>[
+    if (e.protocol != null) e.protocol!,
+    if (e.streamCount != null)
+      '${e.streamCount} stream${e.streamCount == 1 ? '' : 's'}',
+    if (e.durationSeconds != null) '${e.durationSeconds}s',
+  ];
+  if (detail.isNotEmpty) parts.add(detail.join(' · '));
+  return parts.isEmpty ? '' : parts.join('\n');
+}
+
+String _formatInterval(Iperf3Interval e) {
+  final window = '${_seconds(e.start)}-${_seconds(e.end)} sec';
+  final line = StringBuffer('[SUM]  $window   '
+      '${formatBytes(e.bytes)}   ${formatBitrate(e.bitsPerSecond)}');
+  if (e.retransmits != null) {
+    line.write('   ${e.retransmits} retr');
+  }
+  if (e.omitted) line.write('   (omitted)');
+  return line.toString();
+}
+
+String _formatEnd(Iperf3End e) {
+  final lines = <String>['- ' * 20];
+  final sent = e.sent;
+  final received = e.received;
+  if (sent != null) {
+    lines.add('[SUM]  0.00-${_seconds(sent.seconds)} sec   '
+        '${formatBytes(sent.bytes)}   ${formatBitrate(sent.bitsPerSecond)}'
+        '${sent.retransmits != null ? '   ${sent.retransmits} retr' : ''}'
+        '   sender');
+  }
+  if (received != null) {
+    lines.add('[SUM]  0.00-${_seconds(received.seconds)} sec   '
+        '${formatBytes(received.bytes)}   '
+        '${formatBitrate(received.bitsPerSecond)}   receiver');
+  }
+  if (e.hostCpuPercent != null) {
+    lines.add('CPU: local ${e.hostCpuPercent!.toStringAsFixed(1)}%'
+        '${e.remoteCpuPercent != null ? ', remote ${e.remoteCpuPercent!.toStringAsFixed(1)}%' : ''}');
+  }
+  return lines.join('\n');
+}
+
+String _seconds(double value) => value.toStringAsFixed(2);
+
+/// Byte totals the way iperf3 prints them — `93.8 MBytes`.
+String formatBytes(int bytes) {
+  const units = ['Bytes', 'KBytes', 'MBytes', 'GBytes', 'TBytes'];
+  var value = bytes.toDouble();
+  var unit = 0;
+  while (value >= 1024 && unit < units.length - 1) {
+    value /= 1024;
+    unit++;
+  }
+  final precision = value >= 100 ? 0 : (value >= 10 ? 1 : 2);
+  return '${value.toStringAsFixed(precision)} ${units[unit]}';
+}
+
+/// Bitrates the way iperf3 prints them — `786 Mbits/sec`. Decimal units, since
+/// network rates are powers of ten while transfer totals are powers of two.
+String formatBitrate(double bitsPerSecond) {
+  const units = ['bits/sec', 'Kbits/sec', 'Mbits/sec', 'Gbits/sec'];
+  var value = bitsPerSecond;
+  var unit = 0;
+  while (value >= 1000 && unit < units.length - 1) {
+    value /= 1000;
+    unit++;
+  }
+  final precision = value >= 100 ? 0 : (value >= 10 ? 1 : 2);
+  return '${value.toStringAsFixed(precision)} ${units[unit]}';
+}

lib/ui/features/test_run/test_run_screen.dart

@@ -3,6 +3,7 @@ import 'package:flutter/services.dart';
 import 'package:provider/provider.dart';
 import 'package:iperf3gui/data/services/iperf3_service.dart';
 import 'package:iperf3gui/domain/models/test_configuration.dart';
+import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart';
 import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart';
 
 /// Main test run screen.
@@ -193,6 +194,35 @@ class _TestRunContentState extends State<_TestRunContent> {
             ),
             const SizedBox(height: 12),
 
+            // Live throughput figures, derived from the parsed interval series
+            Consumer<TestRunViewModel>(
+              builder: (context, vm, _) {
+                final intervals = vm.intervals;
+                if (intervals.isEmpty) return const SizedBox.shrink();
+                final rates =
+                    intervals.map((i) => i.bitsPerSecond).toList(growable: false);
+                final latest = rates.last;
+                final peak = rates.reduce((a, b) => a > b ? a : b);
+                final mean = rates.reduce((a, b) => a + b) / rates.length;
+                final summary = vm.summary?.sent;
+                return Padding(
+                  padding: const EdgeInsets.only(bottom: 12),
+                  child: Row(
+                    mainAxisAlignment: MainAxisAlignment.spaceBetween,
+                    children: [
+                      _Stat(
+                        label: summary != null ? 'total' : 'now',
+                        value: formatBitrate(summary?.bitsPerSecond ?? latest),
+                      ),
+                      _Stat(label: 'avg', value: formatBitrate(mean)),
+                      _Stat(label: 'peak', value: formatBitrate(peak)),
+                      _Stat(label: 'samples', value: '${rates.length}'),
+                    ],
+                  ),
+                );
+              },
+            ),
+
             // Live output
             Expanded(
               child: Container(
@@ -229,3 +259,32 @@ class _TestRunContentState extends State<_TestRunContent> {
     );
   }
 }
+
+
+/// One figure in the throughput row.
+class _Stat extends StatelessWidget {
+  const _Stat({required this.label, required this.value});
+
+  final String label;
+  final String value;
+
+  @override
+  Widget build(BuildContext context) {
+    final theme = Theme.of(context);
+    return Column(
+      crossAxisAlignment: CrossAxisAlignment.start,
+      children: [
+        Text(
+          label,
+          style: theme.textTheme.labelSmall
+              ?.copyWith(color: theme.colorScheme.onSurfaceVariant),
+        ),
+        Text(
+          value,
+          style: theme.textTheme.titleSmall
+              ?.copyWith(fontFeatures: const [FontFeature.tabularFigures()]),
+        ),
+      ],
+    );
+  }
+}

lib/ui/features/test_run/test_run_view_model.dart

@@ -2,24 +2,33 @@ import 'dart:async';
 
 import 'package:flutter/foundation.dart';
 import 'package:iperf3gui/data/services/iperf3_service.dart';
+import 'package:iperf3gui/domain/models/iperf3_event.dart';
 import 'package:iperf3gui/domain/models/iperf3_result.dart';
 import 'package:iperf3gui/domain/models/test_configuration.dart';
+import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart';
 
 /// State of the test run.
 enum TestRunState { idle, running, completed, error }
 
 /// ViewModel for managing test run state.
 ///
-/// Subscribes to [Iperf3Service.outputLines] so the transcript grows while the
+/// Subscribes to [Iperf3Service.events] so the transcript and the interval
-/// test runs rather than appearing all at once at the end.
+/// series both grow while the test runs, rather than appearing at the end.
 class TestRunViewModel extends ChangeNotifier {
   final Iperf3Service iperf3Service;
 
+  /// How often to notify listeners while output is arriving. iperf3 can report
+  /// several times a second across many streams, and each rebuild re-renders
+  /// the whole transcript — so repaints are coalesced instead of one per event.
+  static const Duration displayInterval = Duration(milliseconds: 250);
+
   TestRunViewModel({required this.iperf3Service}) {
-    _outputSubscription = iperf3Service.outputLines.listen(_onOutputLine);
+    _eventSubscription = iperf3Service.events.listen(_onEvent);
   }
 
-  StreamSubscription<String>? _outputSubscription;
+  StreamSubscription<Iperf3Event>? _eventSubscription;
+  Timer? _repaintTimer;
+  bool _dirty = false;
 
   TestRunState _state = TestRunState.idle;
   TestRunState get state => _state;
@@ -30,11 +39,26 @@ class TestRunViewModel extends ChangeNotifier {
   TestConfiguration? _currentConfig;
   TestConfiguration? get currentConfig => _currentConfig;
 
-  final StringBuffer _liveOutput = StringBuffer();
+  final StringBuffer _transcript = StringBuffer();
 
-  /// Output accumulated so far. Grows line by line during a run, and holds the
+  /// Human-readable output accumulated so far.
-  /// finished transcript once the test completes.
+  String get output => _transcript.toString();
-  String get output => _liveOutput.toString();
+
+  final List<Iperf3Interval> _intervals = [];
+
+  /// Per-interval throughput samples, in arrival order. Structured, so a chart
+  /// can be driven from this without re-parsing text.
+  List<Iperf3Interval> get intervals => List.unmodifiable(_intervals);
+
+  Iperf3Start? _start;
+
+  /// Connection details reported at the start of the run.
+  Iperf3Start? get start => _start;
+
+  Iperf3End? _summary;
+
+  /// Totals reported at the end of the run.
+  Iperf3End? get summary => _summary;
 
   /// Whether a test is currently in flight.
   bool get isRunning => _state == TestRunState.running;
@@ -47,35 +71,32 @@ class TestRunViewModel extends ChangeNotifier {
     if (_state == TestRunState.running) return;
 
     if (!config.isValid) {
-      _liveOutput
+      _resetRunData();
-        ..clear()
+      _transcript
-        ..writeln('Invalid configuration: check the server address and port.');
+          .writeln('Invalid configuration: check the server address and port.');
       _state = TestRunState.error;
-      _result = null;
       notifyListeners();
       return;
     }
 
+    _resetRunData();
     _state = TestRunState.running;
     _currentConfig = config;
-    _result = null;
-    _liveOutput.clear();
     notifyListeners();
 
     try {
       final result = await iperf3Service.run(config);
       _result = result;
-      // Prefer the streamed transcript — it is what the user watched build up.
+      if (_transcript.isEmpty && result.rawOutput.isNotEmpty) {
-      // Fall back to the service's copy if nothing streamed through.
+        _transcript.write(result.rawOutput);
-      if (_liveOutput.isEmpty && result.rawOutput.isNotEmpty) {
-        _liveOutput.write(result.rawOutput);
       }
       _state =
           result.exitCode == 0 ? TestRunState.completed : TestRunState.error;
     } catch (e) {
-      _liveOutput.writeln('Test failed: $e');
+      _transcript.writeln('Test failed: $e');
       _state = TestRunState.error;
     } finally {
+      _stopRepaintTimer();
       notifyListeners();
     }
   }
@@ -88,21 +109,63 @@ class TestRunViewModel extends ChangeNotifier {
 
   /// Reset to idle state.
   void reset() {
+    _resetRunData();
     _state = TestRunState.idle;
-    _result = null;
     _currentConfig = null;
-    _liveOutput.clear();
     notifyListeners();
   }
 
-  void _onOutputLine(String line) {
+  void _resetRunData() {
-    _liveOutput.write(line.endsWith('\n') ? line : '$line\n');
+    _result = null;
-    notifyListeners();
+    _transcript.clear();
+    _intervals.clear();
+    _start = null;
+    _summary = null;
+    _dirty = false;
+  }
+
+  void _onEvent(Iperf3Event event) {
+    switch (event) {
+      case Iperf3Start e:
+        _start = e;
+      case Iperf3Interval e:
+        // Omitted intervals are warm-up (-O) and are not part of the series.
+        if (!e.omitted) _intervals.add(e);
+      case Iperf3End e:
+        _summary = e;
+      case Iperf3Error _:
+      case Iperf3Stopped _:
+      case Iperf3LogLine _:
+        break;
+    }
+
+    final line = formatIperf3Event(event);
+    if (line != null && line.isNotEmpty) {
+      _transcript.writeln(line);
+    }
+    _scheduleRepaint();
+  }
+
+  /// Coalesce repaints: mark dirty and let a timer flush at [displayInterval].
+  void _scheduleRepaint() {
+    _dirty = true;
+    _repaintTimer ??= Timer.periodic(displayInterval, (_) {
+      if (!_dirty) return;
+      _dirty = false;
+      notifyListeners();
+    });
+  }
+
+  void _stopRepaintTimer() {
+    _repaintTimer?.cancel();
+    _repaintTimer = null;
+    _dirty = false;
   }
 
   @override
   void dispose() {
-    _outputSubscription?.cancel();
+    _stopRepaintTimer();
+    _eventSubscription?.cancel();
     iperf3Service.dispose();
     super.dispose();
   }

test/domain/models/iperf3_event_test.dart

@@ -0,0 +1,107 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:iperf3gui/domain/models/iperf3_event.dart';
+
+void main() {
+  group('Iperf3Event.parse', () {
+    test('parses a start event', () {
+      const line = '{"event":"start","data":{'
+          '"connecting_to":{"host":"192.168.2.182","port":5201},'
+          '"test_start":{"protocol":"TCP","num_streams":4,"duration":10}}}';
+      final event = Iperf3Event.parse(line);
+      expect(event, isA<Iperf3Start>());
+      final start = event as Iperf3Start;
+      expect(start.targetHost, equals('192.168.2.182'));
+      expect(start.targetPort, equals(5201));
+      expect(start.streamCount, equals(4));
+      expect(start.durationSeconds, equals(10));
+      expect(start.protocol, equals('TCP'));
+    });
+
+    test('parses an interval event with per-stream detail', () {
+      const line = '{"event":"interval","data":{'
+          '"streams":[{"socket":143,"bytes":49283072,'
+          '"bits_per_second":394264576.0,"retransmits":0,"snd_cwnd":2023424,'
+          '"rtt":3521}],'
+          '"sum":{"start":0.0,"end":1.000151,"seconds":1.000151,'
+          '"bytes":98566144,"bits_per_second":788410352.0,"retransmits":0,'
+          '"omitted":false,"sender":true}}}';
+      final event = Iperf3Event.parse(line);
+      expect(event, isA<Iperf3Interval>());
+      final interval = event as Iperf3Interval;
+      expect(interval.bytes, equals(98566144));
+      expect(interval.bitsPerSecond, closeTo(788410352.0, 0.1));
+      expect(interval.retransmits, equals(0));
+      expect(interval.omitted, isFalse);
+      expect(interval.streams, hasLength(1));
+      expect(interval.streams.single.socket, equals(143));
+      expect(interval.streams.single.rtt, equals(3521));
+    });
+
+    test('marks omitted intervals', () {
+      const line = '{"event":"interval","data":{"streams":[],'
+          '"sum":{"start":0.0,"end":1.0,"bytes":1,"bits_per_second":8.0,'
+          '"omitted":true}}}';
+      final interval = Iperf3Event.parse(line) as Iperf3Interval;
+      expect(interval.omitted, isTrue);
+    });
+
+    test('parses an end event with sent and received totals', () {
+      const line = '{"event":"end","data":{'
+          '"sum_sent":{"seconds":10.0,"bytes":987654321,'
+          '"bits_per_second":790123456.0,"retransmits":7},'
+          '"sum_received":{"seconds":10.01,"bytes":987000000,'
+          '"bits_per_second":789000000.0},'
+          '"cpu_utilization_percent":{"host_total":12.5,"remote_total":3.25}}}';
+      final end = Iperf3Event.parse(line) as Iperf3End;
+      expect(end.sent!.bytes, equals(987654321));
+      expect(end.sent!.retransmits, equals(7));
+      expect(end.received!.bitsPerSecond, closeTo(789000000.0, 0.1));
+      expect(end.hostCpuPercent, closeTo(12.5, 0.01));
+      expect(end.remoteCpuPercent, closeTo(3.25, 0.01));
+    });
+
+    test('falls back to bidir-reverse keys when present', () {
+      const line = '{"event":"interval","data":{"streams":[],'
+          '"sum_bidir_reverse":{"start":0.0,"end":1.0,"bytes":42,'
+          '"bits_per_second":336.0}}}';
+      final interval = Iperf3Event.parse(line) as Iperf3Interval;
+      expect(interval.bytes, equals(42));
+    });
+
+    test('parses an error event', () {
+      const line = '{"event":"error","data":"unable to connect to server"}';
+      final event = Iperf3Event.parse(line);
+      expect(event, isA<Iperf3Error>());
+      expect((event as Iperf3Error).message,
+          equals('unable to connect to server'));
+    });
+
+    test('parses the bridge stop marker', () {
+      final event = Iperf3Event.parse('{"event":"jni_stopped","data":{}}');
+      expect(event, isA<Iperf3Stopped>());
+    });
+
+    test('surfaces non-JSON output as a log line rather than dropping it', () {
+      final event = Iperf3Event.parse('iperf3: something unexpected');
+      expect(event, isA<Iperf3LogLine>());
+      expect((event as Iperf3LogLine).text,
+          equals('iperf3: something unexpected'));
+    });
+
+    test('surfaces malformed JSON as a log line', () {
+      final event = Iperf3Event.parse('{"event":"interval",');
+      expect(event, isA<Iperf3LogLine>());
+    });
+
+    test('treats an unknown event name as a log line', () {
+      final event = Iperf3Event.parse('{"event":"future_thing","data":{}}');
+      expect(event, isA<Iperf3LogLine>());
+    });
+
+    test('tolerates a missing data object', () {
+      final event = Iperf3Event.parse('{"event":"interval"}');
+      expect(event, isA<Iperf3Interval>());
+      expect((event as Iperf3Interval).bytes, equals(0));
+    });
+  });
+}

test/ui/features/test_run/iperf3_event_formatter_test.dart

@@ -0,0 +1,87 @@
+import 'package:flutter_test/flutter_test.dart';
+import 'package:iperf3gui/domain/models/iperf3_event.dart';
+import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart';
+
+void main() {
+  group('formatBytes', () {
+    test('scales in binary units the way iperf3 prints totals', () {
+      expect(formatBytes(0), equals('0.00 Bytes'));
+      expect(formatBytes(1024), equals('1.00 KBytes'));
+      expect(formatBytes(98566144), equals('94.0 MBytes'));
+      expect(formatBytes(1073741824), equals('1.00 GBytes'));
+    });
+  });
+
+  group('formatBitrate', () {
+    test('scales in decimal units, as network rates are quoted', () {
+      expect(formatBitrate(0), equals('0.00 bits/sec'));
+      expect(formatBitrate(8000), equals('8.00 Kbits/sec'));
+      expect(formatBitrate(790000000), equals('790 Mbits/sec'));
+      expect(formatBitrate(2500000000), equals('2.50 Gbits/sec'));
+    });
+  });
+
+  group('formatIperf3Event', () {
+    test('renders a start event', () {
+      const event = Iperf3Start(
+        targetHost: '192.168.2.182',
+        targetPort: 5201,
+        protocol: 'TCP',
+        streamCount: 1,
+        durationSeconds: 10,
+      );
+      final text = formatIperf3Event(event)!;
+      expect(text, contains('Connecting to host 192.168.2.182, port 5201'));
+      expect(text, contains('TCP'));
+      expect(text, contains('1 stream'));
+    });
+
+    test('renders an interval with retransmits', () {
+      const event = Iperf3Interval(
+        start: 0,
+        end: 1,
+        bytes: 98566144,
+        bitsPerSecond: 788410352,
+        retransmits: 3,
+      );
+      final text = formatIperf3Event(event)!;
+      expect(text, contains('0.00-1.00 sec'));
+      expect(text, contains('94.0 MBytes'));
+      expect(text, contains('788 Mbits/sec'));
+      expect(text, contains('3 retr'));
+    });
+
+    test('marks an omitted interval', () {
+      const event = Iperf3Interval(
+          start: 0, end: 1, bytes: 1, bitsPerSecond: 8, omitted: true);
+      expect(formatIperf3Event(event), contains('omitted'));
+    });
+
+    test('renders sender and receiver lines from an end event', () {
+      const event = Iperf3End(
+        sent: Iperf3Total(
+            seconds: 10, bytes: 987654321, bitsPerSecond: 790000000,
+            retransmits: 7),
+        received: Iperf3Total(
+            seconds: 10.01, bytes: 987000000, bitsPerSecond: 789000000),
+        hostCpuPercent: 12.5,
+      );
+      final text = formatIperf3Event(event)!;
+      expect(text, contains('sender'));
+      expect(text, contains('receiver'));
+      expect(text, contains('7 retr'));
+      expect(text, contains('CPU: local 12.5%'));
+    });
+
+    test('renders errors and stops', () {
+      expect(formatIperf3Event(const Iperf3Error('boom')),
+          equals('iperf3 error: boom'));
+      expect(formatIperf3Event(const Iperf3Stopped()),
+          equals('Stopped by user.'));
+    });
+
+    test('drops an empty log line', () {
+      expect(formatIperf3Event(const Iperf3LogLine('')), isNull);
+    });
+  });
+}

test/ui/features/test_run/test_run_view_model_test.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 
 import 'package:flutter_test/flutter_test.dart';
 import 'package:iperf3gui/data/services/iperf3_service.dart';
+import 'package:iperf3gui/domain/models/iperf3_event.dart';
 import 'package:iperf3gui/domain/models/iperf3_result.dart';
 import 'package:iperf3gui/domain/models/test_configuration.dart';
 import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart';
@@ -56,11 +57,16 @@ void main() {
       expect(vm.result!.exitCode, equals(1));
     });
 
-    test('streamed lines accumulate into output during the run', () async {
+    test('streamed events accumulate into output during the run', () async {
       final mock = _MockIperf3Service();
       mock.onRun = (_) async {
-        mock.emit('Connecting to host 192.168.1.1\n');
+        mock.emit(const Iperf3Start(targetHost: '192.168.1.1', targetPort: 5201));
-        mock.emit('[  5]   0.00-1.00   sec  11.2 MBytes  94.1 Mbits/sec\n');
+        mock.emit(const Iperf3Interval(
+          start: 0,
+          end: 1,
+          bytes: 11744051,
+          bitsPerSecond: 94100000,
+        ));
         // Let the broadcast stream deliver before the run resolves.
         await Future<void>.delayed(Duration.zero);
         return const Iperf3Result(rawOutput: 'ignored', exitCode: 0);
@@ -69,12 +75,83 @@ void main() {
       final vm = TestRunViewModel(iperf3Service: mock);
       await vm.runTest(const TestConfiguration(server: '192.168.1.1'));
 
-      expect(vm.output, contains('Connecting to host'));
+      expect(vm.output, contains('Connecting to host 192.168.1.1'));
       expect(vm.output, contains('94.1 Mbits/sec'));
       // The streamed transcript wins over the service's own copy.
       expect(vm.output, isNot(contains('ignored')));
     });
 
+    test('intervals are collected as structured samples', () async {
+      final mock = _MockIperf3Service();
+      mock.onRun = (_) async {
+        mock.emit(const Iperf3Interval(
+            start: 0, end: 1, bytes: 1000, bitsPerSecond: 8000));
+        mock.emit(const Iperf3Interval(
+            start: 1, end: 2, bytes: 2000, bitsPerSecond: 16000));
+        await Future<void>.delayed(Duration.zero);
+        return const Iperf3Result(rawOutput: '', exitCode: 0);
+      };
+
+      final vm = TestRunViewModel(iperf3Service: mock);
+      await vm.runTest(const TestConfiguration(server: '192.168.1.1'));
+
+      expect(vm.intervals, hasLength(2));
+      expect(vm.intervals[1].bitsPerSecond, equals(16000));
+    });
+
+    test('omitted warm-up intervals are excluded from the series', () async {
+      final mock = _MockIperf3Service();
+      mock.onRun = (_) async {
+        mock.emit(const Iperf3Interval(
+            start: 0, end: 1, bytes: 1, bitsPerSecond: 1, omitted: true));
+        mock.emit(const Iperf3Interval(
+            start: 1, end: 2, bytes: 2000, bitsPerSecond: 16000));
+        await Future<void>.delayed(Duration.zero);
+        return const Iperf3Result(rawOutput: '', exitCode: 0);
+      };
+
+      final vm = TestRunViewModel(iperf3Service: mock);
+      await vm.runTest(const TestConfiguration(server: '192.168.1.1'));
+
+      expect(vm.intervals, hasLength(1));
+      expect(vm.intervals.single.bitsPerSecond, equals(16000));
+    });
+
+    test('end event is retained as the summary', () async {
+      final mock = _MockIperf3Service();
+      mock.onRun = (_) async {
+        mock.emit(const Iperf3End(
+          sent: Iperf3Total(
+              seconds: 10, bytes: 987654321, bitsPerSecond: 790000000),
+        ));
+        await Future<void>.delayed(Duration.zero);
+        return const Iperf3Result(rawOutput: '', exitCode: 0);
+      };
+
+      final vm = TestRunViewModel(iperf3Service: mock);
+      await vm.runTest(const TestConfiguration(server: '192.168.1.1'));
+
+      expect(vm.summary?.sent?.bitsPerSecond, equals(790000000));
+      expect(vm.output, contains('790 Mbits/sec'));
+    });
+
+    test('a new run clears the previous interval series', () async {
+      final mock = _MockIperf3Service();
+      mock.onRun = (_) async {
+        mock.emit(const Iperf3Interval(
+            start: 0, end: 1, bytes: 1000, bitsPerSecond: 8000));
+        await Future<void>.delayed(Duration.zero);
+        return const Iperf3Result(rawOutput: '', exitCode: 0);
+      };
+
+      final vm = TestRunViewModel(iperf3Service: mock);
+      await vm.runTest(const TestConfiguration(server: '192.168.1.1'));
+      expect(vm.intervals, hasLength(1));
+
+      await vm.runTest(const TestConfiguration(server: '192.168.1.1'));
+      expect(vm.intervals, hasLength(1), reason: 'not appended to the old run');
+    });
+
     test('falls back to result output when nothing streamed', () async {
       final mock = _MockIperf3Service(
         onRun: (_) async =>
@@ -166,19 +243,19 @@ class _MockIperf3Service implements Iperf3Service {
 
   Future<Iperf3Result> Function(TestConfiguration)? onRun;
 
-  final StreamController<String> _controller =
+  final StreamController<Iperf3Event> _controller =
-      StreamController<String>.broadcast();
+      StreamController<Iperf3Event>.broadcast();
 
   int runCallCount = 0;
   int stopCallCount = 0;
   bool disposed = false;
   bool _isRunning = false;
 
-  /// Push a line as if the native layer had emitted it.
+  /// Push an event as if the native layer had reported it.
-  void emit(String line) => _controller.add(line);
+  void emit(Iperf3Event event) => _controller.add(event);
 
   @override
-  Stream<String> get outputLines => _controller.stream;
+  Stream<Iperf3Event> get events => _controller.stream;
 
   @override
   bool get isRunning => _isRunning;