udpsclient: size the TCP receive buffer from max_packet_bytes so large MARTe frames are not dropped

A TCP frame is never fragmented; a real TCPStreamer cycle (240 KB) exceeded the UDP datagram
limit and made the client reconnect forever, seeing CONFIG but never DATA.

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

CMakeLists.txt

@@ -241,6 +241,7 @@ add_executable(udpscope
   src/source/registry.cpp
   src/source/udps_source.cpp
   src/source/udps_kind.cpp
+  src/source/tcps_kind.cpp
   src/source/probe_source.cpp
   src/source/e1608/e1608_link.cpp
   src/source/e1608/e1608_device.cpp

README.md

@@ -327,6 +327,27 @@ delivers one frame per datagram. The source is connected on the first valid data
 was verified on a loopback group (`239.255.42.99` joined on `127.0.0.1`, although this machine's `lo` does not
 carry the `MULTICAST` flag); a real network, its switches and IGMP were not exercised.
 
+The fourth built-in kind is **TCPStreamer** (`--source tcps`), a receiver for the MARTe2 `TCPStreamer`
+datasource, standalone or through a shared `TCPStreamService`. Both speak the UDPStreamer framing over a TCP
+connection, so it is the UDPS client in a TCP mode: the same CONFIG and DATA, the same channels, calibration and
+timing. The form takes the server (default 127.0.0.1), the port (44500) and an optional **stream name**: leave
+it empty for a standalone `TCPStreamer`, which starts sending on connect; name the stream for a
+`TCPStreamService`, which wants a `SUBSCRIBE` first (the client sends one only when a name is set). On the
+command line these are `--source tcps --source-opt server=... --source-opt port=... --source-opt stream=...`.
+The source is disconnected after 2 s of silence or a closed connection and reconnects every 0.5 s; a stream
+whose frames are more than 2 s apart therefore reconnects continually, as UDPS does. It is verified against
+`tools/mock_tcpstreamer.py`, not against a MARTe2 application. Its limits:
+
+- **One stream per connection.** A multiplexed `SUBSCRIBE` list (`a,b,c`) is refused by the form, because a
+  source is single-timebase; merge streams on the server with `TCPStreamJoin` and name the result.
+- **No packet capture.** The capture is of the UDP socket; the button stays disabled.
+- **A gap is reported at once.** TCP is ordered, so there is no reorder window: a counter gap is a server-side
+  drop and is counted as lost immediately.
+- **A frame is at most `UDPS_MAX_DATAGRAM_BYTES` (about 65 KB)**, which the server's `MaxPayloadSize` default of
+  60000 satisfies.
+- **After a reconnect to a server whose time base restarted** the app prints `the clock stalled at ...`
+  warnings, exactly as for a restarted UDPS stream.
+
 A session file written before kinds existed has no `source_kind`, and is read as UDPS; a UDPS file is
 written exactly as before. A kind's capability panels (for instance a source's own controls) appear as
 pages of the side stack, and are insensitive rather than hidden while no source is connected.
@@ -3299,6 +3320,10 @@ silence timeout and then resumes, which is the EMI case end to end.
   exactly faithful in `strict` mode, and the only reading of the wire that does
   not invent samples in `accumulate` mode. The info popup and the CONFIG dump on
   stderr report the element counts.
+- **The TCPStreamer source takes one stream per connection and has no packet capture.** A multiplexed
+  `SUBSCRIBE` list is not supported (merge on the server with `TCPStreamJoin`), the capture button stays
+  disabled, a gap is reported at once (no reorder window) and a frame is limited to about 65 KB; see "Data
+  sources".
 - **The event window can be shorter than the buffer.** The trigger latches a window
   of the event-span selection; the ring holds far more, so with trigger mode off the
   rest is scrollable history.

deps/udpsclient/udps_client.c

@@ -174,6 +174,7 @@ struct udps_client {
     char server_addr[128];
     char mcast_group[128];
     char iface_addr[128];
+    char stream_name[256];
 
     struct sockaddr_in server_sa;
     int    udp_fd;   /**< Unicast receive socket, or joined multicast socket. */
@@ -197,6 +198,7 @@ struct udps_client {
     size_t                vals_cap;
 
     uint8_t *rxbuf;
+    size_t   rxbuf_cap;   /**< bytes allocated for rxbuf */
 
     udps_held_t held[UDPS_REORDER_SLOTS];
     size_t      held_bytes;    /**< payload bytes currently held */
@@ -1149,9 +1151,50 @@ static int connect_multicast(udps_client_t *c) {
     return 0;
 }
 
+/** TCP transport: connect, then SUBSCRIBE when a stream name is set. */
+static int connect_tcp(udps_client_t *c) {
+    struct timeval tv;
+    size_t         name_len = strlen(c->stream_name);
+
+    c->tcp_fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (c->tcp_fd < 0) {
+        return fail(c, "TCP socket() failed: %s", strerror(errno));
+    }
+    set_recv_buffer(c->tcp_fd, c->cfg.recv_buffer_bytes);
+    /* Linux bounds a blocking connect() with SO_SNDTIMEO, so an unreachable host
+     * cannot hold the polling thread for minutes. */
+    tv.tv_sec  = 3;
+    tv.tv_usec = 0;
+    (void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, (socklen_t)sizeof tv);
+    if (connect(c->tcp_fd, (struct sockaddr *)&c->server_sa,
+                (socklen_t)sizeof c->server_sa) != 0) {
+        int e = errno;
+        close_sockets(c);
+        return fail(c, "TCP connect to %s:%u failed: %s", c->server_addr,
+                    (unsigned)c->cfg.server_port, strerror(e));
+    }
+    /* A half-received frame must not wedge the poll loop (read_exact_tcp blocks). */
+    tv.tv_sec = 1;
+    (void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, (socklen_t)sizeof tv);
+
+    if (name_len > 0u) {
+        uint8_t pkt[UDPS_HEADER_SIZE + 255u];
+        build_header(pkt, UDPS_PKT_SUBSCRIBE, 0u, 0u, 1u, (uint32_t)name_len);
+        memcpy(pkt + UDPS_HEADER_SIZE, c->stream_name, name_len);
+        if (send(c->tcp_fd, pkt, UDPS_HEADER_SIZE + name_len, MSG_NOSIGNAL) !=
+            (ssize_t)(UDPS_HEADER_SIZE + name_len)) {
+            int e = errno;
+            close_sockets(c);
+            return fail(c, "SUBSCRIBE failed: %s", strerror(e));
+        }
+    }
+    return 0;
+}
+
 static int do_connect(udps_client_t *c) {
-    int rc = (c->mcast_group[0] != '\0') ? connect_multicast(c)
+    int rc = c->cfg.use_tcp ? connect_tcp(c)
-                                         : connect_unicast(c);
+             : (c->mcast_group[0] != '\0') ? connect_multicast(c)
+                                           : connect_unicast(c);
     if (rc != 0) {
         c->disconnect_t = now_mono();
         return rc;
@@ -1174,7 +1217,10 @@ static void do_disconnect(udps_client_t *c, const char *why) {
     if (!c->connected) {
         return;
     }
-    (void)send_control(c, UDPS_PKT_DISCONNECT);
+    if (!c->cfg.use_tcp) {
+        /* TCP: a write to a peer that closed would raise SIGPIPE; the server sees the close itself. */
+        (void)send_control(c, UDPS_PKT_DISCONNECT);
+    }
     close_sockets(c);
     c->connected    = 0;
     c->disconnect_t = now_mono();
@@ -1211,7 +1257,7 @@ static int read_tcp_frame(udps_client_t *c) {
     if (udps_parse_header(c->rxbuf, UDPS_HEADER_SIZE, &h) != 0) {
         return -1; /* desynchronised: the stream cannot be resynced */
     }
-    if (h.payload_bytes > (uint32_t)(UDPS_RX_BUF_BYTES - UDPS_HEADER_SIZE)) {
+    if (h.payload_bytes > c->rxbuf_cap - UDPS_HEADER_SIZE) {
         return -1;
     }
     if (h.payload_bytes > 0u &&
@@ -1267,17 +1313,28 @@ udps_client_t *udps_client_create(const udps_client_config_t *cfg) {
     if (c == NULL) {
         return NULL;
     }
+    if (cfg->use_tcp && cfg->stream_name != NULL && strlen(cfg->stream_name) > 255u) {
+        free(c);
+        return NULL;
+    }
     c->cfg    = *cfg;
     c->udp_fd = -1;
     c->tcp_fd = -1;
     copy_str(c->server_addr, sizeof c->server_addr, cfg->server_addr);
     copy_str(c->mcast_group, sizeof c->mcast_group, cfg->multicast_group);
     copy_str(c->iface_addr, sizeof c->iface_addr, cfg->interface_addr);
+    copy_str(c->stream_name, sizeof c->stream_name, cfg->stream_name);
     /* The config's string pointers must not be used again: they belong to the
      * caller and the copies above are what the client works from. */
     c->cfg.server_addr     = c->server_addr;
     c->cfg.multicast_group = (c->mcast_group[0] != '\0') ? c->mcast_group : NULL;
     c->cfg.interface_addr  = (c->iface_addr[0] != '\0') ? c->iface_addr : NULL;
+    c->cfg.stream_name     = c->stream_name;
+    if (c->cfg.use_tcp) {
+        c->cfg.multicast_group  = NULL;
+        c->mcast_group[0]       = '\0';
+        c->cfg.reorder_window_s = 0.0;
+    }
 
     if (c->cfg.max_packet_bytes == 0u) {
         c->cfg.max_packet_bytes = 1u * 1024u * 1024u;
@@ -1294,7 +1351,13 @@ udps_client_t *udps_client_create(const udps_client_config_t *cfg) {
         return NULL;
     }
 
-    c->rxbuf = (uint8_t *)malloc(UDPS_RX_BUF_BYTES);
+    /* A UDP datagram is at most UDPS_MAX_DATAGRAM_BYTES; a TCP frame is not fragmented
+     * and may be as large as max_packet_bytes (MARTe's TCPStreamer allows 1 MiB). */
+    c->rxbuf_cap = UDPS_RX_BUF_BYTES;
+    if (c->cfg.use_tcp && (size_t)c->cfg.max_packet_bytes + UDPS_HEADER_SIZE > c->rxbuf_cap) {
+        c->rxbuf_cap = (size_t)c->cfg.max_packet_bytes + UDPS_HEADER_SIZE;
+    }
+    c->rxbuf = (uint8_t *)malloc(c->rxbuf_cap);
     if (c->rxbuf == NULL) {
         free(c);
         return NULL;

deps/udpsclient/udps_client.h

@@ -52,7 +52,8 @@ enum {
     UDPS_PKT_CONFIG     = 1, /**< Server -> client: signal metadata. */
     UDPS_PKT_ACK        = 2, /**< Client -> server: keepalive. */
     UDPS_PKT_CONNECT    = 3, /**< Client -> server: open a session. */
-    UDPS_PKT_DISCONNECT = 4  /**< Either direction: close a session. */
+    UDPS_PKT_DISCONNECT = 4, /**< Either direction: close a session. */
+    UDPS_PKT_SUBSCRIBE  = 5  /**< Client -> server (TCP stream service): payload = stream name. */
 };
 
 /** Sample type codes (udps_signal_t::type_code). */
@@ -279,6 +280,20 @@ typedef struct {
      * the drop-on-late behaviour. Default 0.020.
      */
     double      reorder_window_s;
+    /**
+     * Non-zero selects the TCP transport (a MARTe2 TCPStreamer / TCPStreamService):
+     * @c server_port is the TCP port, no CONNECT or DISCONNECT frame is sent, the
+     * server streams CONFIG then DATA on connect, and the reorder window is forced
+     * to 0 (TCP is ordered, so a counter gap is a server-side drop). @c multicast_group,
+     * @c interface_addr and @c data_port are ignored.
+     */
+    int         use_tcp;
+    /**
+     * TCP mode only: when non-empty (1..255 bytes) a SUBSCRIBE frame naming this stream is
+     * sent right after every connect, as a TCPStreamService requires. A plain
+     * TCPStreamer ignores it. NULL or "" sends nothing.
+     */
+    const char *stream_name;
 } udps_client_config_t;
 
 /** Fills @p cfg with the defaults documented in Docs/UDPS-C-Client.md. */

docs/claude/architecture.md

@@ -32,8 +32,16 @@ What feeds `Acquisition` is a `DataSource`, one of the registered *kinds*.
   source-neutral, and its pointers are valid **only during `on_data`**: copy what is kept.
 - The GTK thread reaches a source **only** through `DataSource::post(Command)`; the queue is drained
   at the top of `poll()`. A panel never calls a source directly.
-- `UdpsSource` wraps `deps/udpsclient` unchanged; `udps_*` names appear only in
+- `UdpsSource` wraps `deps/udpsclient` and takes a `Transport` (UDP or TCP); `udps_*` names appear only in
   `src/source/udps_*.cpp` (`grep -rn "udps_" src` outside those files is a leak of the old coupling).
+- **The TCPStreamer kind (`tcps`) is `UdpsSource` with `Transport::Tcp`**, not a new `DataSource`: the
+  transport argument sets `use_tcp` / `stream_name` in the client config, defaults the port to 44500 and
+  ignores multicast and interface. `src/source/tcps_kind.cpp` is only the form and the identity (`host:port` or
+  `host:port/stream`). One stream per connection (no multiplexed `SUBSCRIBE` list: a source is
+  single-timebase; merge on the server with `TCPStreamJoin`), and the form refuses a name with a comma.
+  `has_datagrams()` is false, so the capture button stays disabled. `Disconnected` follows the client's 2 s
+  silence timeout or a closed connection, and it reconnects every 0.5 s; a stream whose frames are further
+  apart than 2 s therefore reconnects continually, as UDPS does.
 - The frame is **single-timebase**. A multi-timebase source (ABScope) will change the interface and
   the data path together; do not bend one frame to carry two clocks.
 - A kind's capability panels are pages of the slide-over (`slideover_add_page`), named `source-<id>`. A disconnected source

docs/claude/build-and-run.md

@@ -164,6 +164,16 @@ python3 tools/mock_flatudp.py --port 44588 --rate 1000 --signals 'Trigger:uint8,
 (Quote the `signals` value: zsh globs the `[4]`. The mock takes the same text with or without the trailing `w`;
 the datagram is the same, `w` only says how the app shows it.)
 
+`--source tcps` receives a MARTe2 `TCPStreamer` (standalone) or `TCPStreamService` (named stream) over
+TCP: `server` (default 127.0.0.1), `port` 44500, and `stream` (optional; a `SUBSCRIBE` frame is sent only when it is
+set). `tools/mock_tcpstreamer.py` serves one client with the `mock_udpstreamer.py` signals (`--port`, `--stream`
+to require a `SUBSCRIBE`, `--rate`, `--batch`, `--freq`, `--close-after S`, `--drop-every N`).
+
+```sh
+python3 tools/mock_tcpstreamer.py --port 44500 --stream fast &
+./build/udpscope --source tcps --source-opt port=44500 --source-opt stream=fast
+```
+
 `tools/mock_udpstreamer.py` has waveform-injection options used by the probes:
 
 - `--square-harmonics N` shapes `square`: `N` odd harmonics of the ideal step's Fourier
@@ -201,7 +211,8 @@ form for a new host. Every probe below passes `--server`, so none of them sees i
 
 `deps/plotter` is built from source with only the GTK4 adapter enabled
 (`PLOT_BUILD_GTK4`); its tests and examples are forced off. `deps/udpsclient` is a
-vendored C99 source pair compiled into a static lib. `deps/udpsclient` gained one
+vendored C99 source pair compiled into a static lib. `deps/udpsclient` gained a TCP
+transport mode (item 6 below) and one
 additive entry point for the packet capture — `udps_client_set_datagram_cb()`, firing
 inside `udps_client_poll()` immediately after the UDP `recv()` returns, before the
 magic check or reassembly — plus the public `UDPS_MAX_DATAGRAM_BYTES` the capture
@@ -227,7 +238,7 @@ handler clears either pointer without knowing which dialog it is. Presenting eit
 app's plain `GtkApplicationWindow` is supported and warning-free; it was checked.
 
 `deps/udpsclient` is a vendored C99 source pair and its diff from the upstream source
-is intentional. Five things were changed:
+is intentional. Six things were changed:
 
 1. **A reorder ring of reassembled payloads.** When a packet arrives with a higher counter
    than expected, the raw datagram is held in a fixed-size ring rather than decoded immediately.
@@ -258,6 +269,17 @@ is intentional. Five things were changed:
    would be the same silent loss by another route. `stats.reorder_overflows` counts the event,
    so an undersized ring is a reading rather than a mystery; it is surfaced in the exit dump and
    the `[reseq-probe]` line as `overflows=`.
+6. **A TCP transport mode.** `udps_client_config_t::use_tcp` / `stream_name` select a TCP
+   connection to a MARTe2 `TCPStreamer` or `TCPStreamService`: `connect_tcp()` (a 3 s connect
+   bound through `SO_SNDTIMEO`, a 1 s receive bound, an optional `SUBSCRIBE` frame,
+   `UDPS_PKT_SUBSCRIBE = 5`, sent only when `stream_name` is set), the existing
+   `read_tcp_frame()` for the frames, the reorder window forced to 0 (TCP is ordered, so a
+   counter gap is a server-side drop and is reported at once), multicast forced off, no
+   DISCONNECT frame on close (a write to a closed peer would raise SIGPIPE), no keepalive ACK
+   and no datagram tap. The receive buffer is `max_packet_bytes + header` (1 MiB by default,
+   matching the server's frame ceiling) in TCP mode, since a TCP frame is never fragmented and
+   a MARTe cycle of large arrays is one frame; it is `UDPS_MAX_DATAGRAM_BYTES` only for UDP. A
+   stream name over 255 bytes is refused at create.
 
 `deps/plotter` is upstream's own repository and this app's figure support was added
 there rather than worked around here: `Figure::set_min_y_gutter()` (a floor under the

docs/claude/probe-readings.md

@@ -2117,8 +2117,9 @@ and `regained` reads 400 ms later.
 ## `[source-kinds]`
 
 ```
-[source-kinds] kinds=udps,e1608,flatudp,probe
+[source-kinds] kinds=udps,tcps,e1608,flatudp,probe
 [source-kinds] form udps: round_trip=1 valid=1
+[source-kinds] form tcps: round_trip=1 valid=1
 [source-kinds] form e1608: round_trip=1 valid=1
 [source-kinds] form flatudp: round_trip=1 valid=1
 [source-kinds] form probe: round_trip=1 valid=1
@@ -2132,10 +2133,10 @@ and `regained` reads 400 ms later.
 source-plugin path from the registry to the producer thread. The variable also registers the test-only `probe`
 kind, so a run without it never shows that kind. The `form` line appears once per registered kind.
 
-- `kinds=udps,e1608,flatudp,probe`: a missing kind means `register_builtin_sources()` was not called before `parse_args()`,
+- `kinds=udps,tcps,e1608,flatudp,probe`: a missing kind means `register_builtin_sources()` was not called before `parse_args()`,
   or the env var gate is wrong.
 - `round_trip=0`: a form's `write` and `read` disagree about a key -- a renamed `SourceConfig` key, or an
-  absent key that was not defaulted. `valid=0` is a default configuration the form refuses. The e1608 form's sample sets address, port, scan_rate, mode, ai1, trigger, dir1, plot0 and gen_rate; a mismatch names one of those. The flatudp sample sets port, bind, multicast, iface, rate, endian and signals.
+  absent key that was not defaulted. `valid=0` is a default configuration the form refuses. The e1608 form's sample sets address, port, scan_rate, mode, ai1, trigger, dir1, plot0 and gen_rate; a mismatch names one of those. The flatudp sample sets port, bind, multicast, iface, rate, endian and signals. The tcps sample sets server, port and stream.
 - `panel disconnected: sensitive=0 present=1`: **disable, don't hide.** `present=0` is a panel that was hidden
   or never added; `sensitive=1` is a panel that stays live with no source behind it.
 - `panel connected: sensitive=1`: anything else means `apply_source_panels` acted on the wrong widget (it must
@@ -2302,6 +2303,43 @@ same reason.
 - `resumed: ok=1`: `Connected` was raised again and frames resumed. `connected=0` here means the source never
   recovered after the sender came back.
 
+## `[tcps]`
+
+```
+[tcps] stream: connected=1 signals=5 names=time,sine,square,ramp,noise frames=373 ok=1
+[tcps] capture button: present=1 sensitive=0
+[tcps] reconnect: connected=1 frames_before=211 after=301 reconnects_before=0 after=1 ok=1
+```
+
+`UDPSCOPE_PROBE_TCPS=1500` (the first two lines; counts vary) and `UDPSCOPE_PROBE_TCPS_RECONNECT=2000` (the last)
+run as `--source tcps --source-opt port=44500` against `tools/mock_tcpstreamer.py` with the options in
+`verification.md`. A service run (`--stream fast` on both sides) reads the same two lines; the mock logs
+`SUBSCRIBE 'fast'`. With no mock the stream line reads `connected=0 signals=0 names= frames=0 ok=0`: a reading,
+not a failure (the retries are quiet on stderr).
+A mock run with `--rate 1000000 --batch 20000` sends frames of about 560 KB; the stream line must still
+read `frames>0 ok=1`. `frames=0` with the mock logging a new client every few hundred ms is the signature
+of a receive buffer smaller than the server's frame (the client drops the connection, reconnects, and
+gets CONFIG again, never DATA).
+
+- `stream: ok=1`: connected, the five names the mock sends (its `time` master channel is a signal of its own, so
+  the list is `time,sine,square,ramp,noise`, not four), and frames counted. `connected=0` with a mock running is
+  the connect or the `SUBSCRIBE` failing -- a mock started with `--stream` and an app without `stream=` (or a
+  different name) is closed by the mock (at once on a wrong name, after 5 s on none) and the app reconnects
+  forever; the mock's log says `no valid SUBSCRIBE`. `names` wrong or `signals=0` with `connected=1` is a CONFIG that was not decoded.
+- `capture button: present=1 sensitive=0`: the capture is of the UDP socket and a TCP source reports no
+  datagrams (`has_datagrams()` false), so it is disabled, not hidden. `sensitive=1` is the button enabled for a
+  kind that has nothing to capture.
+- `reconnect: ok=1`: after the mock closed the connection (`--close-after 3`) the client came back
+  (`connected=1`), frames advanced past `frames_before` and the `reconnects` stat advanced past
+  `reconnects_before`. The probe polls every 100 ms for up to 6 s, so it reads the first moment all three hold.
+  `ok=0` is a client that did not come back after the close, or `Acquisition` not mapping `reconnects`;
+  `reconnects_before=1` is the probe's baseline falling **after** the close (`UDPSCOPE_PROBE_TCPS_RECONNECT` must
+  be earlier than the mock's `--close-after`).
+- **Not a defect:** after the reconnect the mock restarts its time base at 0 and the app prints repeating
+  `warning: the clock stalled at ... samples stamped [...] are all behind it`. That is `Acquisition`'s behaviour
+  for any stream whose clock goes backwards, identical to restarting the UDPS mock; the line is a reading, not a
+  failure of this probe.
+
 ## `[prefs-probe]`
 
 ```

docs/claude/verification.md

@@ -576,7 +576,7 @@ precedent, not an exception.
 - `UDPSCOPE_PROBE_SOURCE_KINDS=2000` with `UDPSCOPE_EXIT_AFTER_MS=5000`, run as `--source probe --source-opt rate=500`
   -- the source-plugin path end to end. The variable's presence also registers the test-only `probe` kind (a
   synthetic sine with one panel), so it is the only way a run sees that kind; its value is the delay in ms.
-  Nine `[source-kinds]` lines (one `form` line per registered kind: `udps`, `e1608`, `flatudp`, `probe`); the
+  Ten `[source-kinds]` lines (one `form` line per registered kind: `udps`, `tcps`, `e1608`, `flatudp`, `probe`); the
   reading's own entry is in `probe-readings.md`.
 - `UDPSCOPE_PROBE_E1608=1500` with `UDPSCOPE_EXIT_AFTER_MS`, run as `--source e1608` against
   `tools/mock_e1608.py` -- the MCC E-1608 end to end: panels present and sensitive, a DAC write, a
@@ -637,6 +637,22 @@ precedent, not an exception.
   - **no sender:** `connected=0` throughout, a reading and not a failure.
 - `UDPSCOPE_PROBE_FLATUDP_GAP=1500` -- starts the silence poll described above (2 s after the source's last
   datagram it must be disconnected; then it must reconnect and the frame count move on).
+- `UDPSCOPE_PROBE_TCPS=1500` with `UDPSCOPE_EXIT_AFTER_MS`, run as `--source tcps --source-opt port=44500`
+  against `tools/mock_tcpstreamer.py --port 44500` -- the TCPStreamer source end to end: the connection, the
+  five channel names the mock sends (`time,sine,square,ramp,noise`), frames flowing and the capture button
+  present but disabled. It enables the signals itself, so do **not** add `UDPSCOPE_ENABLE_ALL_MS`. Runs, each
+  short and serial, with the mock killed afterwards:
+  - **standalone:** the mock with no `--stream`, `UDPSCOPE_PROBE_TCPS=1500 UDPSCOPE_EXIT_AFTER_MS=4000`;
+  - **service:** mock `--stream fast`, app `--source-opt stream=fast`, the same environment; the mock logs
+    `SUBSCRIBE 'fast'`;
+  - **reconnect:** mock `--close-after 3`, `UDPSCOPE_PROBE_TCPS=1500 UDPSCOPE_PROBE_TCPS_RECONNECT=2000
+    UDPSCOPE_EXIT_AFTER_MS=8000`; stderr also carries `the clock stalled at ...` warnings after the
+    reconnect (see `probe-readings.md`);
+  - **no mock:** `connected=0` throughout, a reading and not a failure, with no retry flood on stderr.
+- `UDPSCOPE_PROBE_TCPS_RECONNECT=2000` -- records the frame and reconnect counts, then polls every 100 ms
+  (up to 6 s) until connected, frames and reconnects have all advanced, and prints. **It must fall before the
+  mock's `--close-after`** (the baseline is read before the close); the poll, not a fixed delay, absorbs
+  when the close lands.
 - `UDPSCOPE_PROBE_SEARCH_MS` — runs a buffer search for the armed condition, cross-checks the live
   trigger's own `t0` against the nearest hit (`agree=1` is the load-bearing reading), and jumps to
   hit 0. Reports the resulting `mode=`, `axis=` and `marks placed=`.

docs/superpowers/plans/2026-11-06-tcp-streamer-source.md

@@ -0,0 +1,592 @@
+# TCPStreamer Source Kind Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add a `tcps` source kind that receives a MARTe2 `TCPStreamer` DataSource, standalone or through a `TCPStreamService`.
+
+**Architecture:** `deps/udpsclient` gets an additive TCP transport mode (connect, optional SUBSCRIBE, framed reads reusing `read_tcp_frame`). `UdpsSource` takes a transport argument so the TCP kind reuses its callbacks. A new `tcps_kind.cpp` supplies the form and registration. The headless self-check gets a `[tcps]` probe against a new Python mock server.
+
+**Tech Stack:** C99 (`deps/udpsclient`), C++17 + GTK4/libadwaita (`src/`), Python 3 + numpy (mock), CMake.
+
+Spec: `docs/superpowers/specs/2026-11-06-tcp-streamer-source-design.md`.
+
+## Global Constraints
+
+- There is no unit test suite; verification is the app's own headless self-check under `xvfb-run` against a mock. Read `docs/claude/verification.md` (baselines) before Task 3 and run probes **short and serial**, killing the mock afterwards.
+- Baselines for every run: exactly **one** build warning (the pre-existing one; record the count before starting), and no `Gtk-CRITICAL`.
+- Every new source file starts with the two-line SPDX header: `// SPDX-License-Identifier: MIT` / `// SPDX-FileCopyrightText: 2026 Martino Ferrari` (`#` comments in Python, after the shebang).
+- The GTK thread reaches a source only through `DataSource::post()`; `udps_*` names stay in `src/source/udps_*.cpp` and the new `tcps` files may include `udps_source.hpp` only.
+- One connection carries one stream; multiplexed `SUBSCRIBE` lists are out of scope.
+- Session naming for a non-UDPS kind is automatic (`session_host` -> `tcps-<identity>`, port 1, `source_kind=tcps`); do not touch `registry.cpp`'s session functions.
+- A kind id is `[a-z0-9]+`: `tcps`. Display name: `TCPStreamer`. Default port 44500.
+- Do not commit unless the task step says so.
+
+## File Structure
+
+| File | Change | Responsibility |
+| --- | --- | --- |
+| `deps/udpsclient/udps_client.h` | modify | `use_tcp`, `stream_name` config fields |
+| `deps/udpsclient/udps_client.c` | modify | `connect_tcp()`, dispatch, no DISCONNECT frame in TCP mode, reorder window 0 |
+| `tools/mock_tcpstreamer.py` | create | Mock TCPStreamer / TCPStreamService server, reuses `mock_udpstreamer.py` frame builders |
+| `src/source/udps_source.hpp/.cpp` | modify | `Transport` enum, constructor argument, accessor |
+| `src/source/tcps_kind.cpp` | create | `tcps` `SourceKind` and its form |
+| `src/source/registry.hpp/.cpp` | modify | `make_tcps_kind()`, registration |
+| `CMakeLists.txt` | modify | add `tcps_kind.cpp` |
+| `src/ui_probe.cpp`, `src/ui.cpp` | modify | `[source-kinds]` sample, `[tcps]` probes and their env scheduling |
+| `README.md`, `docs/claude/*.md` | modify | document the kind, the upstream change, the probes |
+
+---
+
+### Task 1: TCP transport in `deps/udpsclient`, and the mock server
+
+**Files:**
+- Modify: `deps/udpsclient/udps_client.h` (struct `udps_client_config_t`, around line 263-282)
+- Modify: `deps/udpsclient/udps_client.c` (struct `udps_client` ~line 166, `send_control` ~1021, `do_connect` ~1076, `do_disconnect` ~1093, `udps_client_create` ~1260)
+- Create: `tools/mock_tcpstreamer.py`
+
+**Interfaces:**
+- Produces: `udps_client_config_t::use_tcp` (`int`, 0 = UDP as before) and `::stream_name` (`const char *`, NULL or empty = no SUBSCRIBE). Both default to zero/NULL through `udps_client_config_init()` (it memsets). `server_port` is the TCP port in TCP mode.
+- Produces: `tools/mock_tcpstreamer.py --port P [--stream NAME] [--rate R] [--batch B] [--close-after S] [--drop-every N]`.
+
+- [ ] **Step 1: Add the config fields** at the end of `udps_client_config_t` in `udps_client.h`, after `reorder_window_s`:
+
+```c
+    /**
+     * Non-zero selects the TCP transport (a MARTe2 TCPStreamer / TCPStreamService):
+     * @c server_port is the TCP port, no CONNECT or DISCONNECT frame is sent, the
+     * server streams CONFIG then DATA on connect, and the reorder window is forced
+     * to 0 (TCP is ordered, so a counter gap is a server-side drop). @c multicast_group,
+     * @c interface_addr and @c data_port are ignored.
+     */
+    int         use_tcp;
+    /**
+     * TCP mode only: when non-empty (1..255 bytes) a SUBSCRIBE frame naming this stream is
+     * sent right after every connect, as a TCPStreamService requires. A plain
+     * TCPStreamer ignores it. NULL or "" sends nothing.
+     */
+    const char *stream_name;
+```
+
+- [ ] **Step 2: Own a copy of the name.** In `struct udps_client`, after `char iface_addr[128];` add `char stream_name[256];`. In `udps_client_create`, next to the other `copy_str` calls add `copy_str(c->stream_name, sizeof c->stream_name, cfg->stream_name);` and after the `c->cfg.interface_addr = ...` line add:
+
+```c
+    c->cfg.stream_name = c->stream_name;
+    if (c->cfg.use_tcp) {
+        c->cfg.multicast_group  = NULL;
+        c->mcast_group[0]       = '\0';
+        c->cfg.reorder_window_s = 0.0;
+    }
+```
+
+Also reject a name over 255 bytes: before `copy_str`, `if (cfg->use_tcp && cfg->stream_name != NULL && strlen(cfg->stream_name) > 255u) { free(c); return NULL; }` (place right after the `calloc` succeeds).
+
+- [ ] **Step 3: Add `connect_tcp()`** after `connect_multicast` (before `do_connect`):
+
+```c
+/** TCP transport: connect, then SUBSCRIBE when a stream name is set. */
+static int connect_tcp(udps_client_t *c) {
+    struct timeval tv;
+    size_t         name_len = strlen(c->stream_name);
+
+    c->tcp_fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (c->tcp_fd < 0) {
+        return fail(c, "TCP socket() failed: %s", strerror(errno));
+    }
+    set_recv_buffer(c->tcp_fd, c->cfg.recv_buffer_bytes);
+    /* Linux bounds a blocking connect() with SO_SNDTIMEO, so an unreachable host
+     * cannot hold the polling thread for minutes. */
+    tv.tv_sec  = 3;
+    tv.tv_usec = 0;
+    (void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, (socklen_t)sizeof tv);
+    if (connect(c->tcp_fd, (struct sockaddr *)&c->server_sa,
+                (socklen_t)sizeof c->server_sa) != 0) {
+        int e = errno;
+        close_sockets(c);
+        return fail(c, "TCP connect to %s:%u failed: %s", c->server_addr,
+                    (unsigned)c->cfg.server_port, strerror(e));
+    }
+    /* A half-received frame must not wedge the poll loop (read_exact_tcp blocks). */
+    tv.tv_sec = 1;
+    (void)setsockopt(c->tcp_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, (socklen_t)sizeof tv);
+
+    if (name_len > 0u) {
+        uint8_t pkt[UDPS_HEADER_SIZE + 255u];
+        build_header(pkt, UDPS_PKT_SUBSCRIBE, 0u, 0u, 1u, (uint32_t)name_len);
+        memcpy(pkt + UDPS_HEADER_SIZE, c->stream_name, name_len);
+        if (send(c->tcp_fd, pkt, UDPS_HEADER_SIZE + name_len, MSG_NOSIGNAL) !=
+            (ssize_t)(UDPS_HEADER_SIZE + name_len)) {
+            int e = errno;
+            close_sockets(c);
+            return fail(c, "SUBSCRIBE failed: %s", strerror(e));
+        }
+    }
+    return 0;
+}
+```
+
+Add `UDPS_PKT_SUBSCRIBE = 5 /**< Client -> server (TCP stream service): payload = stream name. */` to the packet-type enum in `udps_client.h` (add a comma after the `UDPS_PKT_DISCONNECT = 4` line).
+
+- [ ] **Step 4: Dispatch and disconnect.** Replace the first lines of `do_connect`:
+
+```c
+    int rc = c->cfg.use_tcp ? connect_tcp(c)
+             : (c->mcast_group[0] != '\0') ? connect_multicast(c)
+                                           : connect_unicast(c);
+```
+
+In `do_disconnect`, replace `(void)send_control(c, UDPS_PKT_DISCONNECT);` with `if (!c->cfg.use_tcp) { (void)send_control(c, UDPS_PKT_DISCONNECT); }` (a write to a peer that closed would raise SIGPIPE and kill the application; the server detects the close itself).
+
+- [ ] **Step 5: Confirm the other TCP-path assumptions by reading, not guessing.** Read `deliver_payload` (~line 848) and confirm that with `reorder_window_s == 0` a counter jump delivers the frame with `lost > 0` and `stats.counter_gaps` incremented, and that `read_tcp_frame` -> `handle_packet` needs no change. If either is false, stop and report; do not patch around it.
+
+- [ ] **Step 6: Write the mock** `tools/mock_tcpstreamer.py` (`chmod +x`):
+
+```python
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# SPDX-FileCopyrightText: 2026 Martino Ferrari
+"""A minimal MARTe2 TCPStreamer / TCPStreamService producer, for exercising udpscope.
+
+Listens on `--port`, serves one client at a time with the same frames as
+mock_udpstreamer.py (CONFIG, then Accumulate DATA packets of sine/square/ramp/noise), each
+framed UDPS-over-TCP. Without `--stream` it behaves like a standalone TCPStreamer: the
+stream starts at once and anything the client sends is ignored. With `--stream NAME` it
+behaves like a TCPStreamService: the client must open with a SUBSCRIBE frame naming NAME
+within 5 s, otherwise the connection is closed.
+
+    ./tools/mock_tcpstreamer.py --port 44500 --stream fast --close-after 6
+"""
+
+import argparse
+import os
+import select
+import socket
+import struct
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import mock_udpstreamer as m  # noqa: E402
+
+PKT_SUBSCRIBE = 5
+HEADER = 17
+
+
+def read_exact(conn, n, deadline):
+    buf = b""
+    while len(buf) < n:
+        left = deadline - time.monotonic()
+        if left <= 0:
+            return None
+        r, _, _ = select.select([conn], [], [], left)
+        if not r:
+            return None
+        chunk = conn.recv(n - len(buf))
+        if not chunk:
+            return None
+        buf += chunk
+    return buf
+
+
+def subscribed_to(conn, want):
+    """True once a SUBSCRIBE frame naming `want` arrived within 5 s."""
+    deadline = time.monotonic() + 5.0
+    head = read_exact(conn, HEADER, deadline)
+    if head is None:
+        return False
+    magic, ptype, _, _, _, plen = struct.unpack("<IBIHHI", head)
+    if magic != m.MAGIC or ptype != PKT_SUBSCRIBE or not 0 < plen <= 4096:
+        return False
+    name = read_exact(conn, plen, deadline)
+    return name is not None and name.decode(errors="replace") == want
+
+
+def serve(conn, args):
+    dt = 1.0 / args.rate
+    period = args.batch / args.rate
+    conn.sendall(m.build_config(args.rate, m.TIME_FULL_ARRAY, "V"))
+    counter, t_stream = 1, 0.0
+    started = time.monotonic()
+    next_send = started
+    while True:
+        now = time.monotonic()
+        if args.close_after > 0 and now - started >= args.close_after:
+            print("mock tcpstreamer: closing the connection (--close-after)", flush=True)
+            return
+        if now < next_send:
+            time.sleep(min(next_send - now, 0.002))
+            continue
+        pkt = m.build_data(counter, t_stream, dt, args.batch, args.freq)
+        if not (args.drop_every > 0 and counter % args.drop_every == 0):
+            conn.sendall(pkt)
+        counter += 1
+        t_stream += args.batch * dt
+        next_send += period
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--port", type=int, default=44500)
+    ap.add_argument("--stream", default="", help="TCPStreamService mode: require SUBSCRIBE of this name")
+    ap.add_argument("--rate", type=float, default=10000.0, help="samples per second")
+    ap.add_argument("--batch", type=int, default=100, help="samples per packet")
+    ap.add_argument("--freq", type=float, default=5.0, help="waveform frequency, Hz")
+    ap.add_argument("--close-after", type=float, default=0.0, metavar="S",
+                    help="close each connection after S seconds (the client must reconnect); 0 = never")
+    ap.add_argument("--drop-every", type=int, default=0, metavar="N",
+                    help="skip sending every Nth packet but advance the counter (a server-side drop); 0 = never")
+    args = ap.parse_args()
+
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    srv.bind(("0.0.0.0", args.port))
+    srv.listen(4)
+    print("mock tcpstreamer: listening on :%d (%s)" % (
+        args.port, "service stream '%s'" % args.stream if args.stream else "standalone"), flush=True)
+    while True:
+        conn, addr = srv.accept()
+        print("mock tcpstreamer: client %s:%d" % addr, flush=True)
+        try:
+            if args.stream and not subscribed_to(conn, args.stream):
+                print("mock tcpstreamer: no valid SUBSCRIBE for '%s'; closing" % args.stream, flush=True)
+                continue
+            if args.stream:
+                print("mock tcpstreamer: SUBSCRIBE '%s'" % args.stream, flush=True)
+            serve(conn, args)
+        except (BrokenPipeError, ConnectionResetError):
+            print("mock tcpstreamer: client went away", flush=True)
+        finally:
+            conn.close()
+
+
+if __name__ == "__main__":
+    main()
+```
+
+- [ ] **Step 7: Build the dependency and the app** with the project's usual command (see `docs/claude/build-and-run.md`; normally `cmake --build build -j`). Expected: it builds; the warning count equals the count recorded before the task. Run `python3 tools/mock_tcpstreamer.py --help` and `python3 -c "import ast,sys;ast.parse(open('tools/mock_tcpstreamer.py').read())"`: both succeed.
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add deps/udpsclient/udps_client.h deps/udpsclient/udps_client.c tools/mock_tcpstreamer.py
+git commit -m "udpsclient: a TCP transport mode (SUBSCRIBE, no CONNECT) and a TCP mock server"
+```
+
+---
+
+### Task 2: The `tcps` source kind
+
+**Files:**
+- Modify: `src/source/udps_source.hpp`, `src/source/udps_source.cpp`
+- Create: `src/source/tcps_kind.cpp`
+- Modify: `src/source/registry.hpp`, `src/source/registry.cpp`, `CMakeLists.txt` (source list near line 242-243)
+
+**Interfaces:**
+- Consumes: `udps_client_config_t::use_tcp` / `::stream_name` from Task 1.
+- Produces: `enum class UdpsSource::Transport { Udp, Tcp }`; `UdpsSource(const SourceConfig&, Transport = Transport::Udp)`; `Transport UdpsSource::transport() const`; `std::unique_ptr<SourceKind> make_tcps_kind()`; kind id `tcps` with config keys `server` (default `127.0.0.1`), `port` (default `44500`), `stream` (optional).
+
+- [ ] **Step 1: `udps_source.hpp`.** Inside `class UdpsSource`, add before the constructor:
+
+```cpp
+  enum class Transport { Udp, Tcp };
+```
+
+change the constructor to `explicit UdpsSource(const SourceConfig& cfg, Transport transport = Transport::Udp);`, add `Transport transport() const { return transport_; }` in the public section, and members `Transport transport_ = Transport::Udp;` and `std::string stream_;`.
+
+- [ ] **Step 2: `udps_source.cpp`.** Replace the constructor:
+
+```cpp
+UdpsSource::UdpsSource(const SourceConfig& cfg, Transport transport)
+    : transport_(transport),
+      server_(cfg.get("server", "127.0.0.1")),
+      port_(static_cast<std::uint16_t>(
+          cfg.get_int("port", transport == Transport::Tcp ? 44500 : 8000))),
+      multicast_(transport == Transport::Tcp ? std::string() : cfg.get("multicast")),
+      interface_(transport == Transport::Tcp ? std::string() : cfg.get("interface")),
+      stream_(transport == Transport::Tcp ? cfg.get("stream") : std::string()),
+      recv_buffer_(static_cast<int>(cfg.get_int("recv_buffer", 8 * 1024 * 1024))) {}
+```
+
+Member declaration order must match the initialiser order: in the header declare `transport_` before `server_`, and `stream_` after `interface_` and before `recv_buffer_` (an out-of-order initialiser is a `-Wreorder` warning, which breaks the one-warning baseline). In `open()`, after `c.keepalive_interval_s = 1.0;` add:
+
+```cpp
+  if (transport_ == Transport::Tcp) {
+    c.use_tcp = 1;
+    c.stream_name = stream_.empty() ? nullptr : stream_.c_str();
+  }
+```
+
+`datagram_cb` registration stays; in TCP mode the client never fires it. Leave `set_datagram_tap` as is.
+
+- [ ] **Step 3: Create `src/source/tcps_kind.cpp`.** Copy the helper block of `src/source/udps_kind.cpp` (`ChangeCb`, `entry_row`, `on_edited`, `on_value`, `hook_value`, `hook`, `text_of`; they are in an anonymous namespace, so the copy is local, as three kinds already do) and add the kind:
+
+```cpp
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/registry.hpp"
+#include "source/udps_source.hpp"
+
+#include <adwaita.h>
+
+#include <memory>
+
+namespace udpscope {
+namespace {
+
+/* ChangeCb, entry_row, on_edited, on_value, hook_value, hook, text_of:
+   the same six helpers as udps_kind.cpp, verbatim. */
+
+class TcpsKind final : public SourceKind {
+ public:
+  std::string id() const override { return "tcps"; }
+  std::string display_name() const override { return "TCPStreamer"; }
+
+  std::string identity(const SourceConfig& cfg) const override {
+    std::string s = cfg.get("server", "127.0.0.1") + ":" + std::to_string(cfg.get_int("port", 44500));
+    if (!cfg.get("stream").empty()) s += "/" + cfg.get("stream");
+    return s;
+  }
+
+  std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
+    return std::make_unique<UdpsSource>(cfg, UdpsSource::Transport::Tcp);
+  }
+
+  ConfigForm make_config_form() const override {
+    GtkWidget* server = entry_row("_Server");
+    GtkWidget* port = GTK_WIDGET(adw_spin_row_new_with_range(1, 65535, 1));
+    adw_preferences_row_set_title(ADW_PREFERENCES_ROW(port), "_Port");
+    adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(port), TRUE);
+    adw_spin_row_set_value(ADW_SPIN_ROW(port), 44500);
+    GtkWidget* stream = entry_row("S_tream name");
+
+    ChangeCb cb = std::make_shared<std::function<void()>>();
+    hook(server, cb);
+    hook_value(port, cb);
+    hook(stream, cb);
+
+    ConfigForm f;
+    f.rows = {server, port, stream};
+    f.description =
+        "Leave the stream name empty for a standalone TCPStreamer; name the stream for a TCPStreamService";
+    f.focus = server;
+    f.write = [=](const SourceConfig& c) {
+      gtk_editable_set_text(GTK_EDITABLE(server), c.get("server", "127.0.0.1").c_str());
+      adw_spin_row_set_value(ADW_SPIN_ROW(port), static_cast<double>(c.get_int("port", 44500)));
+      gtk_editable_set_text(GTK_EDITABLE(stream), c.get("stream").c_str());
+    };
+    f.read = [=]() {
+      SourceConfig c;
+      c.set("server", text_of(server));
+      c.set("port", std::to_string(static_cast<long>(adw_spin_row_get_value(ADW_SPIN_ROW(port)))));
+      if (!text_of(stream).empty()) c.set("stream", text_of(stream));
+      return c;
+    };
+    // A stream name is 1..255 bytes and, being the SUBSCRIBE payload, may not hold a comma
+    // (that would select a multiplexed list, which this kind does not support).
+    f.valid = [=]() {
+      const std::string s = text_of(stream);
+      return !text_of(server).empty() && s.size() <= 255 && s.find(',') == std::string::npos;
+    };
+    f.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
+    return f;
+  }
+};
+
+}  // namespace
+
+std::unique_ptr<SourceKind> make_tcps_kind() { return std::make_unique<TcpsKind>(); }
+
+}  // namespace udpscope
+```
+
+When writing the file, paste the six real helper definitions in place of the comment block. Check `udps_kind.cpp` includes `<functional>` transitively (it does through `source_kind.hpp`).
+
+- [ ] **Step 4: Register.** `registry.hpp`: add `std::unique_ptr<SourceKind> make_tcps_kind();` after `make_udps_kind()` and add `tcps_kind.cpp` to the comment's file list. `registry.cpp`: in `register_builtin_sources()` add `register_kind(make_tcps_kind());` right after `make_udps_kind()` (so the chooser lists the two network kinds together). `CMakeLists.txt`: add `src/source/tcps_kind.cpp` after `src/source/udps_kind.cpp`.
+
+- [ ] **Step 5: Build.** `cmake --build build -j`. Expected: builds, warning count unchanged from the baseline.
+
+- [ ] **Step 6: Smoke-run against the mock** (serial, kill the mock after):
+
+```bash
+python3 tools/mock_tcpstreamer.py --port 44500 &
+MOCK=$!
+UDPSCOPE_EXIT_AFTER_MS=4000 xvfb-run -a build/udpscope --source tcps --source-opt server=127.0.0.1 --source-opt port=44500 2>&1 | tail -20
+kill $MOCK
+```
+
+Expected: no `Gtk-CRITICAL`; the mock prints `client 127.0.0.1:...`. (If the app's CLI spells the options differently, take the spelling from `docs/claude/build-and-run.md`.) Then the same with `--stream fast` on the mock and `--source-opt stream=fast` on the app: the mock prints `SUBSCRIBE 'fast'`. Then once with the wrong name (`stream=slow`): the mock prints `no valid SUBSCRIBE`, and the app keeps retrying without a stderr flood.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add src/source/udps_source.hpp src/source/udps_source.cpp src/source/tcps_kind.cpp src/source/registry.hpp src/source/registry.cpp CMakeLists.txt
+git commit -m "sources: a TCPStreamer kind (standalone or TCPStreamService) on the UDPS client's TCP mode"
+```
+
+---
+
+### Task 3: Probes
+
+**Files:**
+- Modify: `src/ui_probe.cpp` (sample at ~line 2683 `const auto sample`; new `[tcps]` functions after `flatudp_probe`, ~line 3703), `src/ui.cpp` (env scheduling after the `UDPSCOPE_PROBE_FLATUDP_GAP` block, ~line 7657)
+- Declarations: wherever `flatudp_probe` / `flatudp_probe_gap` are declared (grep `flatudp_probe_gap` in `src/*.hpp`; add the two new functions beside them)
+
+**Interfaces:**
+- Consumes: `UdpsSource::transport()`, kind `tcps`, `a->model->stats()` fields `connected`, `frames`, `reconnects`, `a->model->signals()`, `a->capture_button`.
+- Produces: env vars `UDPSCOPE_PROBE_TCPS=<ms>` and `UDPSCOPE_PROBE_TCPS_RECONNECT=<ms>`; stderr lines `[tcps] stream:`, `[tcps] capture button:`, `[tcps] reconnect:`; a `[source-kinds] form tcps:` line.
+
+- [ ] **Step 1: Form round trip.** In `source_kinds_probe`'s `sample` lambda add a branch:
+
+```cpp
+    } else if (id == "tcps") {
+      c.set("server", "10.0.0.7");
+      c.set("port", "44501");
+      c.set("stream", "fast");
+```
+
+Build and run `UDPSCOPE_PROBE_SOURCE_KINDS=2000 UDPSCOPE_EXIT_AFTER_MS=5000 xvfb-run -a build/udpscope --source probe --source-opt rate=500 2>&1 | grep source-kinds`. Expected: `kinds=udps,tcps,e1608,flatudp,probe` and `form tcps: round_trip=1 valid=1`. The `kinds=` line changes for every run from now on, so update `docs/claude/probe-readings.md` (~line 2120-2123, 2135) and `docs/claude/verification.md` (~line 579, "Nine `[source-kinds]` lines" -> ten, "one `form` line per registered kind: `udps`, `tcps`, ...") in Task 4.
+
+- [ ] **Step 2: Add the source accessor and stream probe** in `ui_probe.cpp`, after `flatudp_probe`:
+
+```cpp
+namespace {
+
+UdpsSource* tcps_source(App* a) {
+  UdpsSource* s = a->acq != nullptr ? dynamic_cast<UdpsSource*>(a->acq->source()) : nullptr;
+  return s != nullptr && s->transport() == UdpsSource::Transport::Tcp ? s : nullptr;
+}
+
+std::uint64_t g_tcps_frames_before = 0;
+std::uint64_t g_tcps_reconnects_before = 0;
+
+}  // namespace
+
+/// Phase 2, 1 s after the signals were enabled: connected, the mock's four channels, frames
+/// flowing, and the capture button present but disabled (the capture is of the UDP socket).
+gboolean tcps_probe_report(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  const bool up = a->model->stats().connected.load(std::memory_order_relaxed);
+  std::string names;
+  for (const SignalInfo& s : a->model->signals()) names += (names.empty() ? "" : ",") + s.name;
+  const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
+  const bool ok = up && names == "sine,square,ramp,noise" && frames > 0;
+  std::fprintf(stderr,
+               "[tcps] stream: connected=%d signals=%zu names=%s frames=%" G_GUINT64_FORMAT " ok=%d\n",
+               up ? 1 : 0, a->model->signals().size(), names.c_str(), frames, ok ? 1 : 0);
+  std::fprintf(stderr, "[tcps] capture button: present=%d sensitive=%d\n",
+               a->capture_button != nullptr ? 1 : 0,
+               a->capture_button != nullptr && gtk_widget_get_sensitive(a->capture_button) ? 1 : 0);
+  std::fflush(stderr);
+  return G_SOURCE_REMOVE;
+}
+
+/// Phase 1: enable every signal in one structural change, report a second later.
+gboolean tcps_probe(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  if (tcps_source(a) == nullptr) {
+    std::fprintf(stderr, "[tcps] no tcps source (run with --source tcps)\n");
+    std::fflush(stderr);
+    return G_SOURCE_REMOVE;
+  }
+  std::vector<std::size_t> all(a->model->signals().size());
+  for (std::size_t i = 0; i < all.size(); ++i) all[i] = i;
+  if (!all.empty()) a->model->set_signals_enabled(all, true);
+  g_timeout_add(1000, tcps_probe_report, a);
+  return G_SOURCE_REMOVE;
+}
+
+/// After the mock closed the connection (`--close-after`): the client must have reconnected and
+/// the frame count moved on. Phase 2 reads 1.5 s after phase 1 recorded the baseline.
+gboolean tcps_probe_reconnect_report(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  const bool up = a->model->stats().connected.load(std::memory_order_relaxed);
+  const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
+  const std::uint64_t rc = a->model->stats().reconnects.load(std::memory_order_relaxed);
+  const bool ok = up && frames > g_tcps_frames_before && rc > g_tcps_reconnects_before;
+  std::fprintf(stderr,
+               "[tcps] reconnect: connected=%d frames_before=%" G_GUINT64_FORMAT " after=%" G_GUINT64_FORMAT
+               " reconnects_before=%" G_GUINT64_FORMAT " after=%" G_GUINT64_FORMAT " ok=%d\n",
+               up ? 1 : 0, g_tcps_frames_before, frames, g_tcps_reconnects_before, rc, ok ? 1 : 0);
+  std::fflush(stderr);
+  return G_SOURCE_REMOVE;
+}
+
+gboolean tcps_probe_reconnect(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  g_tcps_frames_before = a->model->stats().frames.load(std::memory_order_relaxed);
+  g_tcps_reconnects_before = a->model->stats().reconnects.load(std::memory_order_relaxed);
+  g_timeout_add(1500, tcps_probe_reconnect_report, a);
+  return G_SOURCE_REMOVE;
+}
+```
+
+Make sure `ui_probe.cpp` includes `source/udps_source.hpp` (add it beside the flatudp includes at line 11-12 if absent). The `reconnects` stat is fed from `SourceStats::reconnects` by `Acquisition`; if after Task 3 Step 4 it reads 0 while the client demonstrably reconnected, `Acquisition` does not map it for this path: report that rather than weakening the check.
+
+- [ ] **Step 3: Schedule and declare.** Add the two declarations next to `flatudp_probe_gap`'s, and in `ui.cpp` after the `UDPSCOPE_PROBE_FLATUDP_GAP` block:
+
+```cpp
+  if (const char* env = g_getenv("UDPSCOPE_PROBE_TCPS")) {
+    g_timeout_add(static_cast<guint>(std::atoi(env)), tcps_probe, &app);
+  }
+  if (const char* env = g_getenv("UDPSCOPE_PROBE_TCPS_RECONNECT")) {
+    g_timeout_add(static_cast<guint>(std::atoi(env)), tcps_probe_reconnect, &app);
+  }
+```
+
+- [ ] **Step 4: Run the three scenarios**, each short and serial, mock killed after each, stderr filtered with `grep -E '^\[tcps\]|Gtk-CRITICAL|warning'`:
+
+1. Standalone: mock `--port 44500`; app `--source tcps --source-opt port=44500`, `UDPSCOPE_PROBE_TCPS=2500 UDPSCOPE_EXIT_AFTER_MS=5000`. Expected `[tcps] stream: connected=1 signals=4 names=sine,square,ramp,noise frames=<n> ok=1` (n > 0) and `[tcps] capture button: present=1 sensitive=0`.
+2. Service: mock `--port 44500 --stream fast`; app additionally `--source-opt stream=fast`. Same expectations.
+3. Reconnect: mock `--port 44500 --close-after 3`; app `UDPSCOPE_PROBE_TCPS=2500 UDPSCOPE_PROBE_TCPS_RECONNECT=4500 UDPSCOPE_EXIT_AFTER_MS=8000`. Expected `[tcps] reconnect: connected=1 ... ok=1`.
+
+Also check the no-server case: no mock, `UDPSCOPE_PROBE_TCPS=2500`: `connected=0 ... ok=0` is a reading, not a crash; no stderr flood (a retry every 0.5 s is acceptable, more than a few lines per second is not).
+
+- [ ] **Step 5: Baselines.** A full rebuild shows the same warning count as before the plan, and none of the runs printed `Gtk-CRITICAL`. Then re-run the source-kinds probe (Step 1) and confirm the existing `udps`/`e1608`/`flatudp`/`probe` form lines still read `round_trip=1 valid=1`.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/ui_probe.cpp src/ui.cpp
+git commit -m "probes: the TCPStreamer kind end to end (standalone, service, reconnect)"
+```
+
+(Include the header that carries the probe declarations in `git add` if Step 3 touched one.)
+
+---
+
+### Task 4: Documentation
+
+**Files:**
+- Modify: `README.md`, `docs/claude/architecture.md`, `docs/claude/build-and-run.md`, `docs/claude/verification.md`, `docs/claude/probe-readings.md`, `CLAUDE.md` (only its intro sentence if it enumerates the source kinds)
+
+- [ ] **Step 1: `docs/claude/build-and-run.md`.** The `deps/udpsclient` section says "Five things were changed". Make it six and add item 6: "**A TCP transport mode.** `udps_client_config_t::use_tcp` / `stream_name` select a TCP connection to a MARTe2 `TCPStreamer` or `TCPStreamService`: `connect_tcp()` (3 s connect bound via `SO_SNDTIMEO`, 1 s receive bound, an optional `SUBSCRIBE` frame), the existing `read_tcp_frame()` for the frames, the reorder window forced to 0, no DISCONNECT frame on close (a write to a closed peer would raise SIGPIPE), and no datagram tap. A frame must fit `UDPS_MAX_DATAGRAM_BYTES` (the server's `MaxPayloadSize` default of 60000 does)." Also update the earlier sentence "`deps/udpsclient` gained one additive entry point" to mention the TCP mode.
+
+- [ ] **Step 2: `docs/claude/architecture.md` (Sources section).** Add a bullet: "**The TCPStreamer kind (`tcps`) is `UdpsSource` with `Transport::Tcp`** ... `src/source/tcps_kind.cpp` is only the form and identity (`host:port` or `host:port/stream`); there is no new `DataSource`. One stream per connection (no multiplexed `SUBSCRIBE` list: a source is single-timebase; merge on the server with `TCPStreamJoin`). `has_datagrams()` is false, so the capture button stays disabled. `Disconnected` follows the client's 2 s silence timeout or a closed connection, and it reconnects every 0.5 s. A stream whose frames are further apart than 2 s therefore reconnects continually, as UDPS does." Fix the bullet that says `UdpsSource` "wraps `deps/udpsclient` unchanged".
+
+- [ ] **Step 3: `docs/claude/verification.md` and `probe-readings.md`.** Verification: update "Nine `[source-kinds]` lines (... `udps`, `e1608`, `flatudp`, `probe`)" to include `tcps`; add `UDPSCOPE_PROBE_TCPS` and `UDPSCOPE_PROBE_TCPS_RECONNECT` entries beside the FLATUDP ones, with the three scenarios from Task 3 Step 4 (commands, mock options, exit times). Probe readings: update `kinds=` (line ~2120 and ~2135) to `udps,tcps,e1608,flatudp,probe`, add `[source-kinds] form tcps: round_trip=1 valid=1` (and the form sample's keys: server, port, stream), and add a `## [tcps]` section after `[flatudp]` with the sample output of the three lines and what a wrong reading means (`connected=0` with a mock running: SUBSCRIBE or connect failing; `names` wrong: CONFIG not decoded; `sensitive=1`: the capture button was enabled for a kind with no datagrams; `reconnect ... ok=0`: the client did not come back after the close, or `Acquisition` does not map `reconnects`).
+
+- [ ] **Step 4: `README.md`.** Wherever it lists the source kinds or the command-line `--source` values, add `tcps` with its options (`server`, `port` default 44500, `stream` optional), one line saying it receives a MARTe2 `TCPStreamer`/`TCPStreamService`, and under known limitations: single stream per connection, no packet capture over TCP, frames up to 65 KB (the server's `MaxPayloadSize`). If `CLAUDE.md`'s intro sentence enumerates the kinds, add `tcps` there too.
+
+- [ ] **Step 5: Read-through.** `grep -rn "Nine \`\[source-kinds\]\|Five things\|unchanged" README.md CLAUDE.md docs/claude` and fix every statement the change made false.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add README.md CLAUDE.md docs/claude
+git commit -m "docs: the TCPStreamer source kind, the client's TCP mode, and its probes"
+```
+
+---
+
+## Self-review (against the spec)
+
+- Wire/scope (single stream, SUBSCRIBE only with a name): Task 1 Steps 1-4, Task 2 Step 3 (`valid` rejects a comma).
+- `udpsclient` TCP mode (use_tcp, stream_name, connect_tcp, reorder 0, no tap, additive, recorded as a sixth change): Task 1; recording: Task 4 Step 1.
+- Source adapter reuse via transport argument; form rows, description, identity, session naming (automatic), registration and CMake: Task 2.
+- Verification (fake server, probe lines for signals/frames/reconnect, baselines, real smoke test): Task 3 uses a Python mock rather than the in-process fake server the spec named, because the existing probes all use external mocks under `tools/`. A smoke test against a real MARTe `TCPStreamer` is not in the plan; do it only if one is runnable.
+- Docs: Task 4.
+- Types: `UdpsSource::Transport`, `transport()`, `make_tcps_kind()`, `tcps_source()`, env names are used identically across tasks.

docs/superpowers/specs/2026-11-06-tcp-streamer-source-design.md

@@ -0,0 +1,51 @@
+# TCPStreamer source kind
+
+A new source kind `tcps` ("TCPStreamer") that receives a MARTe2 `TCPStreamer` DataSource, either
+standalone or through a shared `TCPStreamService`.
+
+## Wire and scope
+
+- Both transports are plain UDPS framing over TCP. A standalone `TCPStreamer` (TCPSServer) sends
+  CONFIG then DATA on connect. A `TCPStreamService` additionally requires a first client frame,
+  `SUBSCRIBE` (type 5, payload = stream name, 1..255 bytes).
+- A plain server ignores `SUBSCRIBE`, so the client sends it only when a stream name is set.
+- One connection carries one stream. Multiplexed lists (`"a,b,c"`, stream index in `fragmentIdx`)
+  are out of scope: a udpscope source is single-timebase. Several streams are merged server-side
+  with `TCPStreamJoin`, which publishes an ordinary named stream.
+
+## `deps/udpsclient`: a TCP transport mode
+
+- `udps_client_config_t` gains `use_tcp` and `stream_name`.
+- New `connect_tcp()`: open the TCP socket, send `SUBSCRIBE` if a stream name is set, send no
+  `CONNECT`. No UDP socket, no keepalive ACK.
+- `udps_client_poll` reuses `read_tcp_frame` / `handle_packet`.
+- `reorder_window_s` is forced to 0 in TCP mode: TCP is ordered, so a counter gap is a server-side
+  drop and is reported at once.
+- Silence timeout and reconnect delay behave as in UDP (dead server, or a stalled client the server
+  evicted, reconnects).
+- The datagram tap does not fire; the packet capture stays UDP-only (`has_datagrams()` false).
+- Additive. Recorded in `docs/claude/build-and-run.md` as a sixth intentional upstream change.
+
+## Source adapter and kind
+
+- `UdpsSource` takes a transport argument (`UdpsSource(cfg, Transport::Tcp)`) instead of being
+  copied: its callbacks are identical. The TCP variant reads `server`, `port` (default 44500) and
+  `stream`.
+- `src/source/tcps_kind.cpp`: kind `tcps`, display name "TCPStreamer". Form rows: Server, Port,
+  Stream name (optional). Description: "Leave empty for a standalone TCPStreamer; name the stream
+  for a TCPStreamService".
+- Identity: `host:port` or `host:port/stream`. Session naming follows the existing non-UDPS rule
+  (`tcps-<identity>`, port 1, `source_kind=tcps`, `source_opt_*`).
+- Registered in `register_builtin_sources()` and `CMakeLists.txt`.
+
+## Verification
+
+No unit suite. Extend the self-check in `src/ui_probe.cpp` with an in-process fake TCP server that
+sends CONFIG then DATA, and requires `SUBSCRIBE` in one mode. Probe lines assert the signals, the
+frame count, and the reconnect after the server closes. Baselines: exactly one build warning, no
+`Gtk-CRITICAL`. A smoke test against a real MARTe `TCPStreamer` if one is runnable.
+
+## Docs
+
+README and `docs/claude/architecture.md` (Sources) describe the kind; `docs/claude/probe-readings.md`
+gets the new probe lines.

src/source/registry.cpp

@@ -50,6 +50,7 @@ void register_builtin_sources() {
   if (done) return;
   done = true;
   register_kind(make_udps_kind());
+  register_kind(make_tcps_kind());
   register_kind(make_e1608_kind());
   register_kind(make_flatudp_kind());
   if (std::getenv("UDPSCOPE_PROBE_SOURCE_KINDS") != nullptr) register_kind(make_probe_kind());

src/source/registry.hpp

@@ -21,8 +21,9 @@ const std::vector<SourceKind*>& kinds();
 /// may drop. A later `dlopen` loader would only call `register_kind()`.
 void register_builtin_sources();
 
-/// The built-in kinds, each defined in its own file (`udps_kind.cpp`, `probe_source.cpp`, `e1608/e1608_kind.cpp`, `flatudp/flatudp_kind.cpp`).
+/// The built-in kinds, each defined in its own file (`udps_kind.cpp`, `tcps_kind.cpp`, `probe_source.cpp`, `e1608/e1608_kind.cpp`, `flatudp/flatudp_kind.cpp`).
 std::unique_ptr<SourceKind> make_udps_kind();
+std::unique_ptr<SourceKind> make_tcps_kind();
 std::unique_ptr<SourceKind> make_probe_kind();
 std::unique_ptr<SourceKind> make_e1608_kind();
 std::unique_ptr<SourceKind> make_flatudp_kind();

src/source/tcps_kind.cpp

@@ -0,0 +1,107 @@
+// SPDX-License-Identifier: MIT
+// SPDX-FileCopyrightText: 2026 Martino Ferrari
+
+#include "source/registry.hpp"
+#include "source/udps_source.hpp"
+
+#include <adwaita.h>
+
+#include <memory>
+
+namespace udpscope {
+namespace {
+
+using ChangeCb = std::shared_ptr<std::function<void()>>;
+
+GtkWidget* entry_row(const char* title) {
+  GtkWidget* r = adw_entry_row_new();
+  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
+  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
+  return r;
+}
+
+void on_edited(GtkEditable*, gpointer user) {
+  const auto& cb = *static_cast<ChangeCb*>(user);
+  if (cb && *cb) (*cb)();
+}
+
+void on_value(GObject*, GParamSpec*, gpointer user) {
+  const auto& cb = *static_cast<ChangeCb*>(user);
+  if (cb && *cb) (*cb)();
+}
+
+void hook_value(GtkWidget* spin, const ChangeCb& cb) {
+  g_signal_connect_data(spin, "notify::value", G_CALLBACK(on_value), new ChangeCb(cb),
+                        +[](gpointer d, GClosure*) { delete static_cast<ChangeCb*>(d); },
+                        static_cast<GConnectFlags>(0));
+}
+
+void hook(GtkWidget* entry, const ChangeCb& cb) {
+  g_signal_connect_data(entry, "changed", G_CALLBACK(on_edited), new ChangeCb(cb),
+                        +[](gpointer d, GClosure*) { delete static_cast<ChangeCb*>(d); },
+                        static_cast<GConnectFlags>(0));
+}
+
+std::string text_of(GtkWidget* w) { return gtk_editable_get_text(GTK_EDITABLE(w)); }
+
+class TcpsKind final : public SourceKind {
+ public:
+  std::string id() const override { return "tcps"; }
+  std::string display_name() const override { return "TCPStreamer"; }
+
+  std::string identity(const SourceConfig& cfg) const override {
+    std::string s = cfg.get("server", "127.0.0.1") + ":" + std::to_string(cfg.get_int("port", 44500));
+    if (!cfg.get("stream").empty()) s += "/" + cfg.get("stream");
+    return s;
+  }
+
+  std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
+    return std::make_unique<UdpsSource>(cfg, UdpsSource::Transport::Tcp);
+  }
+
+  ConfigForm make_config_form() const override {
+    GtkWidget* server = entry_row("_Server");
+    GtkWidget* port = GTK_WIDGET(adw_spin_row_new_with_range(1, 65535, 1));
+    adw_preferences_row_set_title(ADW_PREFERENCES_ROW(port), "_Port");
+    adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(port), TRUE);
+    adw_spin_row_set_value(ADW_SPIN_ROW(port), 44500);
+    GtkWidget* stream = entry_row("S_tream name");
+
+    ChangeCb cb = std::make_shared<std::function<void()>>();
+    hook(server, cb);
+    hook_value(port, cb);
+    hook(stream, cb);
+
+    ConfigForm f;
+    f.rows = {server, port, stream};
+    f.description =
+        "Leave the stream name empty for a standalone TCPStreamer; name the stream for a TCPStreamService";
+    f.focus = server;
+    f.write = [=](const SourceConfig& c) {
+      gtk_editable_set_text(GTK_EDITABLE(server), c.get("server", "127.0.0.1").c_str());
+      adw_spin_row_set_value(ADW_SPIN_ROW(port), static_cast<double>(c.get_int("port", 44500)));
+      gtk_editable_set_text(GTK_EDITABLE(stream), c.get("stream").c_str());
+    };
+    f.read = [=]() {
+      SourceConfig c;
+      c.set("server", text_of(server));
+      c.set("port", std::to_string(static_cast<long>(adw_spin_row_get_value(ADW_SPIN_ROW(port)))));
+      if (!text_of(stream).empty()) c.set("stream", text_of(stream));
+      return c;
+    };
+    // A stream name is 1..255 bytes and, being the SUBSCRIBE payload, may not hold a comma
+    // (that would select a multiplexed list, which this kind does not support).
+    f.valid = [=]() {
+      const std::string s = text_of(stream);
+      return !text_of(server).empty() && s.size() <= 255 && s.find(',') == std::string::npos;
+    };
+    f.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
+    return f;
+  }
+};
+
+}  // namespace
+
+std::unique_ptr<SourceKind> make_tcps_kind() { return std::make_unique<TcpsKind>(); }
+
+}  // namespace udpscope

src/source/udps_source.cpp

@@ -40,11 +40,14 @@ SignalInfo to_info(const udps_signal_t& s) {
 
 }  // namespace
 
-UdpsSource::UdpsSource(const SourceConfig& cfg)
+UdpsSource::UdpsSource(const SourceConfig& cfg, Transport transport)
-    : server_(cfg.get("server", "127.0.0.1")),
+    : transport_(transport),
-      port_(static_cast<std::uint16_t>(cfg.get_int("port", 8000))),
+      server_(cfg.get("server", "127.0.0.1")),
-      multicast_(cfg.get("multicast")),
+      port_(static_cast<std::uint16_t>(
-      interface_(cfg.get("interface")),
+          cfg.get_int("port", transport == Transport::Tcp ? 44500 : 8000))),
+      multicast_(transport == Transport::Tcp ? std::string() : cfg.get("multicast")),
+      interface_(transport == Transport::Tcp ? std::string() : cfg.get("interface")),
+      stream_(transport == Transport::Tcp ? cfg.get("stream") : std::string()),
       recv_buffer_(static_cast<int>(cfg.get_int("recv_buffer", 8 * 1024 * 1024))) {}
 
 UdpsSource::~UdpsSource() { close(); }
@@ -62,6 +65,10 @@ void UdpsSource::open(SourceSink& sink) {
   c.silence_timeout_s = 2.0;
   c.reconnect_delay_s = 0.5;
   c.keepalive_interval_s = 1.0;
+  if (transport_ == Transport::Tcp) {
+    c.use_tcp = 1;
+    c.stream_name = stream_.empty() ? nullptr : stream_.c_str();
+  }
 
   client_ = udps_client_create(&c);
   if (client_ == nullptr) {

src/source/udps_source.hpp

@@ -18,7 +18,9 @@ namespace udpscope {
 /// all protocol work stays in `deps/udpsclient`.
 class UdpsSource final : public DataSource {
  public:
-  explicit UdpsSource(const SourceConfig& cfg);
+  enum class Transport { Udp, Tcp };
+
+  explicit UdpsSource(const SourceConfig& cfg, Transport transport = Transport::Udp);
   ~UdpsSource() override;
 
   void open(SourceSink& sink) override;
@@ -27,6 +29,7 @@ class UdpsSource final : public DataSource {
   SourceStats stats() const override;
   bool connected() const override;
   void set_datagram_tap(DatagramTap tap) override { tap_ = std::move(tap); }
+  Transport transport() const { return transport_; }
 
  private:
   static void config_cb(const udps_signal_t* signals, std::uint32_t count,
@@ -36,10 +39,12 @@ class UdpsSource final : public DataSource {
   static void datagram_cb(const std::uint8_t* bytes, std::size_t len, double recv_time,
                           void* user);
 
+  Transport transport_ = Transport::Udp;
   std::string server_;
   std::uint16_t port_ = 8000;
   std::string multicast_;
   std::string interface_;
+  std::string stream_;
   int recv_buffer_ = 8 * 1024 * 1024;
 
   udps_client_t* client_ = nullptr;

src/ui.cpp

@@ -7655,6 +7655,12 @@ int run_ui(int argc, char** argv, const Config& cfg, ScopeModel& model, Acquisit
   if (const char* env = g_getenv("UDPSCOPE_PROBE_FLATUDP_GAP")) {
     g_timeout_add(static_cast<guint>(std::atoi(env)), flatudp_probe_gap, &app);
   }
+  if (const char* env = g_getenv("UDPSCOPE_PROBE_TCPS")) {
+    g_timeout_add(static_cast<guint>(std::atoi(env)), tcps_probe, &app);
+  }
+  if (const char* env = g_getenv("UDPSCOPE_PROBE_TCPS_RECONNECT")) {
+    g_timeout_add(static_cast<guint>(std::atoi(env)), tcps_probe_reconnect, &app);
+  }
 
   // Our own options were consumed by parse_args(); GTK would reject them, so only
   // the program name is handed to g_application_run().

src/ui_app.hpp

@@ -2308,6 +2308,13 @@ gboolean flatudp_probe(gpointer user);
 gboolean flatudp_probe_report(gpointer user);
 gboolean flatudp_probe_gap(gpointer user);
 gboolean flatudp_probe_resumed(gpointer user);
+/// The TCPStreamer kind: `tcps_probe` enables every signal, `tcps_probe_report` reads the stream
+/// and the capture button a second later; `tcps_probe_reconnect` records a baseline and
+/// `tcps_probe_reconnect_report` checks a reconnect moved the counters on.
+gboolean tcps_probe(gpointer user);
+gboolean tcps_probe_report(gpointer user);
+gboolean tcps_probe_reconnect(gpointer user);
+gboolean tcps_probe_reconnect_report(gpointer user);
 gboolean pan_probe(gpointer user);
 gboolean arm_probe(gpointer user);
 gboolean trigrenum_probe_step3(gpointer user);

src/ui_probe.cpp

@@ -11,6 +11,7 @@
 #include "source/flatudp/flatudp_signals.hpp"
 #include "source/flatudp/flatudp_source.hpp"
 #include "source/registry.hpp"
+#include "source/udps_source.hpp"
 #include "ui_app.hpp"
 
 #include <sys/stat.h>
@@ -2711,6 +2712,10 @@ gboolean source_kinds_probe(gpointer user) {
       c.set("rate", "2500");
       c.set("endian", "big");
       c.set("signals", "Trigger:uint8,Ia:float32,Wave:int16[4]");
+    } else if (id == "tcps") {
+      c.set("server", "10.0.0.7");
+      c.set("port", "44501");
+      c.set("stream", "fast");
     }
     return c;
   };
@@ -3696,6 +3701,84 @@ gboolean flatudp_probe(gpointer user) {
   return G_SOURCE_REMOVE;
 }
 
+namespace {
+
+UdpsSource* tcps_source(App* a) {
+  UdpsSource* s = a->acq != nullptr ? dynamic_cast<UdpsSource*>(a->acq->source()) : nullptr;
+  return s != nullptr && s->transport() == UdpsSource::Transport::Tcp ? s : nullptr;
+}
+
+std::uint64_t g_tcps_frames_before = 0;
+std::uint64_t g_tcps_reconnects_before = 0;
+
+}  // namespace
+
+/// Phase 2, 1 s after the signals were enabled: connected, the mock's time master and four channels, frames
+/// flowing, and the capture button present but disabled (the capture is of the UDP socket).
+gboolean tcps_probe_report(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  const bool up = a->model->stats().connected.load(std::memory_order_relaxed);
+  std::string names;
+  for (const SignalInfo& s : a->model->signals()) names += (names.empty() ? "" : ",") + s.name;
+  const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
+  const bool ok = up && names == "time,sine,square,ramp,noise" && frames > 0;
+  std::fprintf(stderr,
+               "[tcps] stream: connected=%d signals=%zu names=%s frames=%" G_GUINT64_FORMAT " ok=%d\n",
+               up ? 1 : 0, a->model->signals().size(), names.c_str(), frames, ok ? 1 : 0);
+  std::fprintf(stderr, "[tcps] capture button: present=%d sensitive=%d\n",
+               a->capture_button != nullptr ? 1 : 0,
+               a->capture_button != nullptr && gtk_widget_get_sensitive(a->capture_button) ? 1 : 0);
+  std::fflush(stderr);
+  return G_SOURCE_REMOVE;
+}
+
+/// Phase 1: enable every signal in one structural change, report a second later.
+gboolean tcps_probe(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  if (tcps_source(a) == nullptr) {
+    std::fprintf(stderr, "[tcps] no tcps source (run with --source tcps)\n");
+    std::fflush(stderr);
+    return G_SOURCE_REMOVE;
+  }
+  std::vector<std::size_t> all(a->model->signals().size());
+  for (std::size_t i = 0; i < all.size(); ++i) all[i] = i;
+  if (!all.empty()) a->model->set_signals_enabled(all, true);
+  g_timeout_add(1000, tcps_probe_report, a);
+  return G_SOURCE_REMOVE;
+}
+
+/// After the mock closed the connection (`--close-after`): the client must have reconnected and
+/// the frame count moved on. Phase 1 records the baseline; phase 2 polls every 100 ms until both
+/// have happened (the close falls whenever the mock's timer says, so a fixed delay would race it)
+/// and reports the last reading after 6 s.
+gboolean tcps_probe_reconnect_report(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  const bool up = a->model->stats().connected.load(std::memory_order_relaxed);
+  const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
+  const std::uint64_t rc = a->model->stats().reconnects.load(std::memory_order_relaxed);
+  const bool ok = up && frames > g_tcps_frames_before && rc > g_tcps_reconnects_before;
+  static int ticks = 0;
+  if (!ok && ++ticks < 60) return G_SOURCE_CONTINUE;
+  std::fprintf(stderr,
+               "[tcps] reconnect: connected=%d frames_before=%" G_GUINT64_FORMAT " after=%" G_GUINT64_FORMAT
+               " reconnects_before=%" G_GUINT64_FORMAT " after=%" G_GUINT64_FORMAT " ok=%d\n",
+               up ? 1 : 0, g_tcps_frames_before, frames, g_tcps_reconnects_before, rc, ok ? 1 : 0);
+  std::fflush(stderr);
+  return G_SOURCE_REMOVE;
+}
+
+gboolean tcps_probe_reconnect(gpointer user) {
+  App* a = static_cast<App*>(user);
+  if (a->window == nullptr) return G_SOURCE_REMOVE;
+  g_tcps_frames_before = a->model->stats().frames.load(std::memory_order_relaxed);
+  g_tcps_reconnects_before = a->model->stats().reconnects.load(std::memory_order_relaxed);
+  g_timeout_add(100, tcps_probe_reconnect_report, a);
+  return G_SOURCE_REMOVE;
+}
+
 gboolean pan_probe(gpointer user) {
   App* a = static_cast<App*>(user);
   if (a->window == nullptr || a->rows->row_count() <= 0) return G_SOURCE_REMOVE;

tools/mock_tcpstreamer.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# SPDX-FileCopyrightText: 2026 Martino Ferrari
+"""A minimal MARTe2 TCPStreamer / TCPStreamService producer, for exercising udpscope.
+
+Listens on `--port`, serves one client at a time with the same frames as
+mock_udpstreamer.py (CONFIG, then Accumulate DATA packets of sine/square/ramp/noise), each
+framed UDPS-over-TCP. Without `--stream` it behaves like a standalone TCPStreamer: the
+stream starts at once and anything the client sends is ignored. With `--stream NAME` it
+behaves like a TCPStreamService: the client must open with a SUBSCRIBE frame naming NAME
+within 5 s, otherwise the connection is closed.
+
+    ./tools/mock_tcpstreamer.py --port 44500 --stream fast --close-after 6
+"""
+
+import argparse
+import os
+import select
+import socket
+import struct
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import mock_udpstreamer as m  # noqa: E402
+
+PKT_SUBSCRIBE = 5
+HEADER = 17
+
+
+def read_exact(conn, n, deadline):
+    buf = b""
+    while len(buf) < n:
+        left = deadline - time.monotonic()
+        if left <= 0:
+            return None
+        r, _, _ = select.select([conn], [], [], left)
+        if not r:
+            return None
+        chunk = conn.recv(n - len(buf))
+        if not chunk:
+            return None
+        buf += chunk
+    return buf
+
+
+def subscribed_to(conn, want):
+    """True once a SUBSCRIBE frame naming `want` arrived within 5 s."""
+    deadline = time.monotonic() + 5.0
+    head = read_exact(conn, HEADER, deadline)
+    if head is None:
+        return False
+    magic, ptype, _, _, _, plen = struct.unpack("<IBIHHI", head)
+    if magic != m.MAGIC or ptype != PKT_SUBSCRIBE or not 0 < plen <= 4096:
+        return False
+    name = read_exact(conn, plen, deadline)
+    return name is not None and name.decode(errors="replace") == want
+
+
+def serve(conn, args):
+    dt = 1.0 / args.rate
+    period = args.batch / args.rate
+    conn.sendall(m.build_config(args.rate, m.TIME_FULL_ARRAY, "V"))
+    counter, t_stream = 1, 0.0
+    started = time.monotonic()
+    next_send = started
+    while True:
+        now = time.monotonic()
+        if args.close_after > 0 and now - started >= args.close_after:
+            print("mock tcpstreamer: closing the connection (--close-after)", flush=True)
+            return
+        if now < next_send:
+            time.sleep(min(next_send - now, 0.002))
+            continue
+        pkt = m.build_data(counter, t_stream, dt, args.batch, args.freq)
+        if not (args.drop_every > 0 and counter % args.drop_every == 0):
+            conn.sendall(pkt)
+        counter += 1
+        t_stream += args.batch * dt
+        next_send += period
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--port", type=int, default=44500)
+    ap.add_argument("--stream", default="", help="TCPStreamService mode: require SUBSCRIBE of this name")
+    ap.add_argument("--rate", type=float, default=10000.0, help="samples per second")
+    ap.add_argument("--batch", type=int, default=100, help="samples per packet")
+    ap.add_argument("--freq", type=float, default=5.0, help="waveform frequency, Hz")
+    ap.add_argument("--close-after", type=float, default=0.0, metavar="S",
+                    help="close each connection after S seconds (the client must reconnect); 0 = never")
+    ap.add_argument("--drop-every", type=int, default=0, metavar="N",
+                    help="skip sending every Nth packet but advance the counter (a server-side drop); 0 = never")
+    args = ap.parse_args()
+
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    srv.bind(("0.0.0.0", args.port))
+    srv.listen(4)
+    print("mock tcpstreamer: listening on :%d (%s)" % (
+        args.port, "service stream '%s'" % args.stream if args.stream else "standalone"), flush=True)
+    while True:
+        conn, addr = srv.accept()
+        print("mock tcpstreamer: client %s:%d" % addr, flush=True)
+        try:
+            if args.stream and not subscribed_to(conn, args.stream):
+                print("mock tcpstreamer: no valid SUBSCRIBE for '%s'; closing" % args.stream, flush=True)
+                continue
+            if args.stream:
+                print("mock tcpstreamer: SUBSCRIBE '%s'" % args.stream, flush=True)
+            serve(conn, args)
+        except (BrokenPipeError, ConnectionResetError):
+            print("mock tcpstreamer: client went away", flush=True)
+        finally:
+            conn.close()
+
+
+if __name__ == "__main__":
+    main()