udpscope/docs/superpowers/plans/2026-11-02-flat-udp-source.md

Flat UDP Source 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 flatudp source kind: a receiver for the MARTe2 UDPSender datagram (signals concatenated in declaration order, no header), configured by a hand-entered signal list.

Architecture: Three new units under src/source/flatudp/: a GTK-free signal-list codec, a DataSource that owns a non-blocking UDP socket and does all its work inside poll() on the producer thread, and a SourceKind with a row-editor connect form. The existing pipeline stamps the frames from the declared rate; the source never reports loss.

Tech Stack: C++20, GTK4/libadwaita (>= 1.8), POSIX sockets, Python 3 for the mock sender. No unit-test suite: the contract is the application's own headless self-check (xvfb-run, env-scheduled probes in src/ui_probe.cpp).

Spec: docs/superpowers/specs/2026-11-02-flat-udp-source-design.md. Branch: feature/source-flatudp, cut from feature/source-plugins (do not push or merge).

Global Constraints

• Kind id flatudp; new files src/source/flatudp/flatudp_{signals,source,kind}.{hpp,cpp}, tools/mock_flatudp.py.
• Every source file carries the SPDX MIT header (// SPDX-License-Identifier: MIT / // SPDX-FileCopyrightText: 2026 Martino Ferrari; # form in Python).
• Config keys: port (default 44488), bind (empty), multicast (empty), iface (empty), rate (none, required, Hz), endian (little default or big), signals (name:type[count],...).
• Types: uint8/16/32/64, int8/16/32/64, float32, float64. Count 1 is a scalar signal, N a vector signal.
• Validity (form and --source-opt alike): at least one signal; names non-empty and unique; count 1-4096; total at most 65507 bytes; rate finite and positive.
• One datagram is one cycle. A datagram of any other size than the expected one is dropped, never decoded.
• Batching: all rows scalar means block = clamp(round(rate / 100), 1, 1024) datagrams per Frame, a partial block flushed after 100 ms. Any vector row means one datagram per Frame (num_samples = 1).
• Frame: counter a running frame count, lost = 0, late = false, recv_time the wall-clock epoch seconds at delivery, publish_mode = 1.
• on_config is sent from the first poll(): one SignalInfo per row (rows = count, cols = 1, time_mode = kTimePacket, sampling_rate = rate).
• Connected on the first valid datagram, Disconnected after 2 s of silence, Connected again on the next valid datagram. Each distinct wrong size: one Error ("datagram of N bytes, expected M, check the signal list"), one stderr line, stale incremented per wrong datagram.
• has_datagrams() is true and set_datagram_tap is honoured for every datagram, valid or not.
• Threading unchanged: the source lives on the producer thread; the probe reads only atomics.
• GTK traps: notify handlers are (GObject*, GParamSpec*, gpointer), clicked is (GtkButton*, gpointer), changed is (GtkEditable*, gpointer), spin value-changed is (GtkSpinButton*, gpointer); contexts are freed by the destroy notify of g_signal_connect_data; rows never parented are g_object_ref_sink + g_object_unref in probes; no theme icons (text buttons only).
• Baselines of every run: a clean build has exactly 3 warnings (an incremental one shows the 2 pre-existing ui_probe.cpp MathNode::source ones; none from the new files), the exit dump reads gtk warnings: 0, no Gtk-CRITICAL.
• Keep verification runs short and serial; kill every mock afterwards.
• Never push, never merge. Commit per task.

File Structure

File	Responsibility
src/source/flatudp/flatudp_signals.{hpp,cpp} (new)	Type/Row, the signals string codec (parse, format), problem, size arithmetic, decode_datagram. No GTK, no sockets.
src/source/flatudp/flatudp_source.{hpp,cpp} (new)	FlatUdpSettings (config parsing and validity, valid_ipv4, valid_multicast) and FlatUdpSource (socket, batching, events).
src/source/flatudp/flatudp_kind.cpp (new)	FlatUdpKind, the connect form, make_flatudp_kind().
src/source/registry.{hpp,cpp}	Declare and register make_flatudp_kind() after e1608.
CMakeLists.txt	The five new .cpp files.
src/ui_probe.cpp, src/ui_app.hpp, src/ui.cpp	The UDPSCOPE_PROBE_FLATUDP probe family and the [source-kinds] sample.
tools/mock_flatudp.py (new)	A UDPSender-shaped packet source for a signal spec.
README.md, docs/claude/{build-and-run,verification,probe-readings,architecture}.md	The kind, its limits, the run recipes and the probe readings.

Task ordering

1. Codec (pure, checked by a hand-computed probe line).
2. Settings, source, kind with an interim form, mock, stream probes, docs (end to end against the mock).
3. Row-editor form, its probe lines, README and architecture docs.

---

Task 1: The signal-list codec

Files:

• Create: src/source/flatudp/flatudp_signals.hpp, src/source/flatudp/flatudp_signals.cpp
• Modify: CMakeLists.txt (source list, after src/source/e1608/e1608_panels.cpp)
• Modify: src/ui_probe.cpp (include, flatudp_codec_check() and flatudp_probe() after e1608_probe_reconnect)
• Modify: src/ui_app.hpp (declaration after e1608_probe_resumed), src/ui.cpp (env hook after UDPSCOPE_PROBE_E1608_RECONNECT)

Interfaces:

• Consumes: nothing.

• Produces (namespace udpscope::flatudp, header source/flatudp/flatudp_signals.hpp):

  • enum class Type : std::uint8_t { U8 = 0, I8, U16, I16, U32, I32, U64, I64, F32, F64 } -- the numeric value is the SignalInfo::type_code (UDPS_T_* order).
  • constexpr std::uint32_t kMaxCount = 4096; constexpr std::size_t kMaxDatagram = 65507;
  • struct Row { std::string name; Type type = Type::F32; std::uint32_t count = 1; bool operator==(const Row&) const = default; };
  • std::size_t type_size(Type), const char* type_name(Type), bool parse_type(const std::string&, Type&)
  • std::size_t datagram_size(const std::vector<Row>&), bool all_scalar(const std::vector<Row>&)
  • std::string format(const std::vector<Row>&); std::string parse(const std::string& text, std::vector<Row>& rows) (returns an error text, empty on success; rows is cleared on failure)
  • std::string problem(const std::vector<Row>&) (first validity failure, empty when valid)
  • void decode_datagram(const std::vector<Row>&, const std::uint8_t* data, bool big_endian, std::vector<std::vector<double>>& cols) -- appends row s's count values to cols[s] (growing cols to rows.size()); the caller guarantees datagram_size(rows) bytes at data.
  • In ui_app.hpp: gboolean flatudp_probe(gpointer user);

• Step 1: Create the branch

git checkout feature/source-plugins && git checkout -b feature/source-flatudp

• Step 2: Write the header

Create src/source/flatudp/flatudp_signals.hpp:

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#pragma once

#include <cstddef>
#include <cstdint>
#include <string>
#include <vector>

namespace udpscope::flatudp {

/// The numeric value is the `SignalInfo::type_code` (the order of the client library's `UDPS_T_*`).
enum class Type : std::uint8_t { U8 = 0, I8, U16, I16, U32, I32, U64, I64, F32, F64 };

constexpr std::uint32_t kMaxCount = 4096;
constexpr std::size_t kMaxDatagram = 65507;  ///< the largest UDP payload over IPv4

/// One signal of the sender's datagram: `count` consecutive values of `type`.
struct Row {
  std::string name;
  Type type = Type::F32;
  std::uint32_t count = 1;
  bool operator==(const Row&) const = default;
};

std::size_t type_size(Type t);
const char* type_name(Type t);  ///< "uint8" ... "float64"
bool parse_type(const std::string& s, Type& out);

/// The bytes one datagram must have: the sum of `type_size * count`.
std::size_t datagram_size(const std::vector<Row>& rows);
/// True when no row is a vector. Such a list can be batched, several datagrams to a frame.
bool all_scalar(const std::vector<Row>& rows);

/// `name:type[,name:type[count]...]`; a count of 1 is written without brackets.
std::string format(const std::vector<Row>& rows);
/// The inverse of `format`. Whitespace around names, types and counts is ignored. An empty text
/// is an empty list (and not an error: `problem` reports it). Returns an error in words, empty
/// on success; `rows` is empty after a failure.
std::string parse(const std::string& text, std::vector<Row>& rows);

/// The first reason this list cannot describe a datagram, in words for the user; empty when valid.
std::string problem(const std::vector<Row>& rows);

/// Decodes one datagram of `datagram_size(rows)` bytes to doubles, row by row, byte-swapping when
/// `big_endian` differs from the host. Appends to `cols[s]`; the caller clears them between uses.
void decode_datagram(const std::vector<Row>& rows, const std::uint8_t* data, bool big_endian,
                     std::vector<std::vector<double>>& cols);

}  // namespace udpscope::flatudp

• Step 3: Write the implementation

Create src/source/flatudp/flatudp_signals.cpp:

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#include "source/flatudp/flatudp_signals.hpp"

#include <algorithm>
#include <bit>
#include <cctype>
#include <cstring>
#include <set>

namespace udpscope::flatudp {
namespace {

constexpr const char* kNames[] = {"uint8",  "int8",  "uint16", "int16",   "uint32",
                                  "int32",  "uint64", "int64", "float32", "float64"};
constexpr std::size_t kSizes[] = {1, 1, 2, 2, 4, 4, 8, 8, 4, 8};

std::string trim(const std::string& s) {
  std::size_t a = 0, b = s.size();
  while (a < b && std::isspace(static_cast<unsigned char>(s[a]))) ++a;
  while (b > a && std::isspace(static_cast<unsigned char>(s[b - 1]))) --b;
  return s.substr(a, b - a);
}

/// A name must survive `format` and `parse`: none of the separators, no edge whitespace.
bool valid_name(const std::string& n) {
  if (n.empty()) return false;
  if (std::isspace(static_cast<unsigned char>(n.front())) ||
      std::isspace(static_cast<unsigned char>(n.back())))
    return false;
  for (char c : n) {
    if (c == ',' || c == ':' || c == '[' || c == ']' || static_cast<unsigned char>(c) < 0x20)
      return false;
  }
  return true;
}

template <class T>
double load(const std::uint8_t* p, bool swap) {
  unsigned char b[sizeof(T)];
  std::memcpy(b, p, sizeof(T));
  if (swap) std::reverse(b, b + sizeof(T));
  T v;
  std::memcpy(&v, b, sizeof(T));
  return static_cast<double>(v);
}

double load_typed(Type t, const std::uint8_t* p, bool swap) {
  switch (t) {
    case Type::U8: return load<std::uint8_t>(p, swap);
    case Type::I8: return load<std::int8_t>(p, swap);
    case Type::U16: return load<std::uint16_t>(p, swap);
    case Type::I16: return load<std::int16_t>(p, swap);
    case Type::U32: return load<std::uint32_t>(p, swap);
    case Type::I32: return load<std::int32_t>(p, swap);
    case Type::U64: return load<std::uint64_t>(p, swap);
    case Type::I64: return load<std::int64_t>(p, swap);
    case Type::F32: return load<float>(p, swap);
    case Type::F64: return load<double>(p, swap);
  }
  return 0.0;
}

}  // namespace

std::size_t type_size(Type t) { return kSizes[static_cast<std::size_t>(t)]; }
const char* type_name(Type t) { return kNames[static_cast<std::size_t>(t)]; }

bool parse_type(const std::string& s, Type& out) {
  for (std::size_t i = 0; i < 10; ++i) {
    if (s == kNames[i]) {
      out = static_cast<Type>(i);
      return true;
    }
  }
  return false;
}

std::size_t datagram_size(const std::vector<Row>& rows) {
  std::size_t n = 0;
  for (const Row& r : rows) n += type_size(r.type) * static_cast<std::size_t>(r.count);
  return n;
}

bool all_scalar(const std::vector<Row>& rows) {
  return std::all_of(rows.begin(), rows.end(), [](const Row& r) { return r.count == 1; });
}

std::string format(const std::vector<Row>& rows) {
  std::string out;
  for (const Row& r : rows) {
    if (!out.empty()) out += ',';
    out += r.name + ':' + type_name(r.type);
    if (r.count != 1) out += '[' + std::to_string(r.count) + ']';
  }
  return out;
}

std::string parse(const std::string& text, std::vector<Row>& rows) {
  rows.clear();
  if (trim(text).empty()) return {};
  std::size_t pos = 0;
  int index = 0;
  for (;;) {
    std::size_t end = text.find(',', pos);
    if (end == std::string::npos) end = text.size();
    ++index;
    const std::string where = "signal " + std::to_string(index) + ": ";
    const std::string tok = trim(text.substr(pos, end - pos));
    const std::size_t colon = tok.find(':');
    if (colon == std::string::npos) {
      rows.clear();
      return where + "expected name:type";
    }
    Row r;
    r.name = trim(tok.substr(0, colon));
    std::string ty = trim(tok.substr(colon + 1));
    const std::size_t br = ty.find('[');
    if (br != std::string::npos) {
      if (ty.back() != ']') {
        rows.clear();
        return where + "missing ']' after the count";
      }
      const std::string n = trim(ty.substr(br + 1, ty.size() - br - 2));
      const bool digits = !n.empty() && n.size() <= 6 &&
                          std::all_of(n.begin(), n.end(), [](char c) { return c >= '0' && c <= '9'; });
      if (!digits) {
        rows.clear();
        return where + "the count must be a whole number";
      }
      r.count = static_cast<std::uint32_t>(std::stoul(n));
      ty = trim(ty.substr(0, br));
    }
    if (!parse_type(ty, r.type)) {
      rows.clear();
      return where + "unknown type '" + ty + "'";
    }
    rows.push_back(r);
    if (end >= text.size()) break;
    pos = end + 1;
  }
  return {};
}

std::string problem(const std::vector<Row>& rows) {
  if (rows.empty()) return "add at least one signal";
  std::set<std::string> seen;
  for (std::size_t i = 0; i < rows.size(); ++i) {
    const Row& r = rows[i];
    if (r.name.empty()) return "signal " + std::to_string(i + 1) + " has no name";
    if (!valid_name(r.name))
      return "'" + r.name + "': a name cannot hold , : [ ] or edge spaces";
    if (!seen.insert(r.name).second) return "the name '" + r.name + "' is used twice";
    if (r.count < 1 || r.count > kMaxCount)
      return "'" + r.name + "': the count must be 1 to " + std::to_string(kMaxCount);
  }
  const std::size_t n = datagram_size(rows);
  if (n > kMaxDatagram)
    return std::to_string(n) + " bytes per datagram exceed the " + std::to_string(kMaxDatagram) +
           " a UDP datagram can carry";
  return {};
}

void decode_datagram(const std::vector<Row>& rows, const std::uint8_t* data, bool big_endian,
                     std::vector<std::vector<double>>& cols) {
  if (cols.size() < rows.size()) cols.resize(rows.size());
  const bool swap = big_endian != (std::endian::native == std::endian::big);
  std::size_t off = 0;
  for (std::size_t s = 0; s < rows.size(); ++s) {
    const std::size_t w = type_size(rows[s].type);
    for (std::uint32_t k = 0; k < rows[s].count; ++k) {
      cols[s].push_back(load_typed(rows[s].type, data + off, swap));
      off += w;
    }
  }
}

}  // namespace udpscope::flatudp

• Step 4: Register in CMake

In CMakeLists.txt, after the line src/source/e1608/e1608_panels.cpp add:

  src/source/flatudp/flatudp_signals.cpp

• Step 5: Write the failing probe (codec check)

In src/ui_probe.cpp add the include after #include "source/e1608/e1608_source.hpp":

#include "source/flatudp/flatudp_signals.hpp"

In the anonymous namespace, directly after e1608_decode_check()'s closing brace and before the namespace's closing } // namespace, add:

/// The `signals` codec and the datagram decode against values worked out by hand. The row set
/// mixes every width and a vector: A:uint8, B:int16, C:float32, D:uint64[2], E:int8, 24 bytes.
/// A = 254, B = -2, C = 1.5, D = {1, 4294967298}, E = -1.
bool flatudp_codec_check() {
  using namespace flatudp;
  std::vector<Row> rows;
  if (!parse(" A:uint8, B:int16,C : float32 ,D:uint64[2],E:int8", rows).empty()) return false;
  const std::vector<Row> want = {{"A", Type::U8, 1},  {"B", Type::I16, 1}, {"C", Type::F32, 1},
                                 {"D", Type::U64, 2}, {"E", Type::I8, 1}};
  if (rows != want) return false;
  if (format(rows) != "A:uint8,B:int16,C:float32,D:uint64[2],E:int8") return false;
  if (datagram_size(rows) != 24 || all_scalar(rows) || !problem(rows).empty()) return false;
  // The enum value is the type code the model keeps.
  if (static_cast<int>(Type::F64) != 9 || static_cast<int>(Type::U64) != 6) return false;

  const std::uint8_t le[24] = {0xFE,                    // A
                               0xFE, 0xFF,              // B
                               0x00, 0x00, 0xC0, 0x3F,  // C
                               0x01, 0, 0, 0, 0, 0, 0, 0,  // D[0]
                               0x02, 0, 0, 0, 0x01, 0, 0, 0,  // D[1]
                               0xFF};                   // E
  const std::uint8_t be[24] = {0xFE,
                               0xFF, 0xFE,
                               0x3F, 0xC0, 0x00, 0x00,
                               0, 0, 0, 0, 0, 0, 0, 0x01,
                               0, 0, 0, 0x01, 0, 0, 0, 0x02,
                               0xFF};
  const double expect[5][2] = {{254, 0}, {-2, 0}, {1.5, 0}, {1, 4294967298.0}, {-1, 0}};
  const std::size_t counts[5] = {1, 1, 1, 2, 1};
  for (int pass = 0; pass < 2; ++pass) {
    std::vector<std::vector<double>> cols;
    decode_datagram(rows, pass == 0 ? le : be, pass == 1, cols);
    if (cols.size() != 5) return false;
    for (std::size_t s = 0; s < 5; ++s) {
      if (cols[s].size() != counts[s]) return false;
      for (std::size_t k = 0; k < counts[s]; ++k)
        if (cols[s][k] != expect[s][k]) return false;
    }
  }

  // Parse failures name the signal; a bad list never half-parses.
  std::vector<Row> bad;
  if (parse("A:uint8,B", bad).find("signal 2") == std::string::npos || !bad.empty()) return false;
  if (parse("A:uint7", bad).find("uint7") == std::string::npos) return false;
  if (parse("A:uint8[x]", bad).empty() || parse("A:uint8[3", bad).empty()) return false;
  if (!parse("", bad).empty() || !bad.empty() || problem(bad).empty()) return false;

  // The validity rules.
  const auto prob = [](std::vector<Row> r) { return problem(r); };
  if (prob({{"A", Type::U8, 1}, {"A", Type::U8, 1}}).find("twice") == std::string::npos) return false;
  if (prob({{"", Type::U8, 1}}).empty() || prob({{"a,b", Type::U8, 1}}).empty()) return false;
  if (prob({{"A", Type::U8, 0}}).empty() || prob({{"A", Type::U8, 4097}}).empty()) return false;
  if (!prob({{"A", Type::U8, 4096}}).empty()) return false;
  // 2 x 32768 = 65536 bytes is over the datagram limit; 32768 + 16384 is not.
  if (prob({{"A", Type::F64, 4096}, {"B", Type::F64, 4096}}).find("65536") == std::string::npos)
    return false;
  if (!prob({{"A", Type::F64, 4096}, {"B", Type::F32, 4096}}).empty()) return false;
  return true;
}

Then, after the function e1608_probe_reconnect (outside the anonymous namespace, directly before gboolean pan_probe), add:

/// The flat UDP source's probe. For now only the pure codec check; it needs no flatudp source.
gboolean flatudp_probe(gpointer user) {
  App* a = static_cast<App*>(user);
  if (a->window == nullptr) return G_SOURCE_REMOVE;
  std::fprintf(stderr, "[flatudp] codec: ok=%d\n", flatudp_codec_check() ? 1 : 0);
  std::fflush(stderr);
  return G_SOURCE_REMOVE;
}

In src/ui_app.hpp, after gboolean e1608_probe_resumed(gpointer user); add:

/// The flat UDP (MARTe2 UDPSender) source: the signal-list codec and a hand-computed decode, then,
/// against `tools/mock_flatudp.py`, the stream, the wrong-size accounting and the silence timeout.
gboolean flatudp_probe(gpointer user);

In src/ui.cpp, after the UDPSCOPE_PROBE_E1608_RECONNECT block add:

  if (const char* env = g_getenv("UDPSCOPE_PROBE_FLATUDP")) {
    g_timeout_add(static_cast<guint>(std::atoi(env)), flatudp_probe, &app);
  }

• Step 6: Build and run the check

This is the first run of the check (the implementation is written in step 3). Reconfigure (new source file) and build:

cmake -S . -B build >/dev/null && cmake --build build -j8 2>&1 | grep -E "warning|error|flatudp" | head -20

Expected: no error; no warning mentioning flatudp.

UDPSCOPE_PROBE_SOURCE_KINDS=60000 UDPSCOPE_PROBE_FLATUDP=1500 UDPSCOPE_EXIT_AFTER_MS=3000 \
  timeout 30 xvfb-run -a build/udpscope --source probe --source-opt rate=500 2>&1 | grep -E "\[flatudp\]|Gtk-CRITICAL|gtk warnings"

Expected: [flatudp] codec: ok=1 and gtk warnings: 0, no Gtk-CRITICAL. If ok=0, bisect the check by returning early at each stage; every expected value above is hand-computed, so a mismatch is a codec defect, not a test defect (re-derive the byte arrays before touching the codec).

• Step 7: Commit

git add CMakeLists.txt src/source/flatudp/flatudp_signals.hpp src/source/flatudp/flatudp_signals.cpp \
  src/ui_probe.cpp src/ui_app.hpp src/ui.cpp
git commit -m "flatudp: the signal-list codec, with a hand-computed decode check"

---

Task 2: The settings, the source, the kind, the mock and the stream probes

Files:

• Create: src/source/flatudp/flatudp_source.hpp, src/source/flatudp/flatudp_source.cpp, src/source/flatudp/flatudp_kind.cpp, tools/mock_flatudp.py
• Modify: CMakeLists.txt, src/source/registry.hpp, src/source/registry.cpp
• Modify: src/ui_probe.cpp (include, anonymous-namespace helpers, flatudp_probe rewritten, three new phase functions, the [source-kinds] sample), src/ui_app.hpp, src/ui.cpp
• Modify: docs/claude/build-and-run.md, docs/claude/verification.md, docs/claude/probe-readings.md

Interfaces:

• Consumes (Task 1): everything in flatudp_signals.hpp.

• Produces (namespace udpscope::flatudp, header source/flatudp/flatudp_source.hpp):

  • constexpr int kDefaultPort = 44488;
  • bool valid_ipv4(const std::string&), bool valid_multicast(const std::string&) (224.0.0.0/4)
  • struct FlatUdpSettings { int port; std::string bind, multicast, iface; double rate; bool big_endian; bool endian_ok; std::vector<Row> rows; std::string signals_error; static FlatUdpSettings from(const SourceConfig&); std::string problem() const; std::string identity() const; std::size_t block() const; }
  • class FlatUdpSource final : public DataSource with explicit FlatUdpSource(const SourceConfig&), const FlatUdpSettings& settings() const, std::uint64_t valid_datagrams() const, std::uint64_t wrong_size_datagrams() const, std::string last_message() const.
  • std::unique_ptr<SourceKind> make_flatudp_kind(); (namespace udpscope, declared in registry.hpp).
  • In ui_app.hpp: flatudp_probe, flatudp_probe_report, flatudp_probe_gap, flatudp_probe_resumed.

• Step 1: Write the settings and source header

Create src/source/flatudp/flatudp_source.hpp:

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#pragma once

#include "source/data_source.hpp"
#include "source/flatudp/flatudp_signals.hpp"
#include "source/source_config.hpp"

#include <atomic>
#include <chrono>
#include <cstdint>
#include <mutex>
#include <set>
#include <string>
#include <vector>

namespace udpscope::flatudp {

constexpr int kDefaultPort = 44488;  ///< `UDPSender`'s own default

bool valid_ipv4(const std::string& s);
/// An IPv4 address in 224.0.0.0/4.
bool valid_multicast(const std::string& s);

/// Everything a configured flat UDP source needs, parsed once from the flat `SourceConfig`.
struct FlatUdpSettings {
  int port = kDefaultPort;
  std::string bind;       ///< local address; empty is any
  std::string multicast;  ///< group to join; empty is plain unicast
  std::string iface;      ///< local interface address for the join; empty is the system's choice
  double rate = 0.0;      ///< the sender's cycle rate in Hz; 0 means not given
  bool big_endian = false;
  bool endian_ok = true;  ///< false when `endian` is neither "little" nor "big"
  std::vector<Row> rows;
  std::string signals_error;  ///< why the `signals` text did not parse; `rows` is empty then

  static FlatUdpSettings from(const SourceConfig& cfg);
  /// The first reason this cannot connect, in words for the user; empty when it can.
  std::string problem() const;
  /// `[multicast:]port`; names the session file.
  std::string identity() const;
  /// Datagrams per frame: 1 for a list with a vector, else `clamp(round(rate / 100), 1, 1024)`.
  std::size_t block() const;
};

class FlatUdpSource final : public DataSource {
 public:
  explicit FlatUdpSource(const SourceConfig& cfg);
  ~FlatUdpSource() override;

  void open(SourceSink& sink) override;
  void poll(int timeout_ms) override;
  void close() override;
  SourceStats stats() const override;
  bool connected() const override { return connected_.load(std::memory_order_relaxed); }
  void set_datagram_tap(DatagramTap tap) override { tap_ = std::move(tap); }

  const FlatUdpSettings& settings() const { return cfg_; }
  /// Datagrams of the expected size / of any other size; for the probe, readable from any thread.
  std::uint64_t valid_datagrams() const { return valid_.load(std::memory_order_relaxed); }
  std::uint64_t wrong_size_datagrams() const { return wrong_.load(std::memory_order_relaxed); }
  std::string last_message() const;

 private:
  using Clock = std::chrono::steady_clock;

  void close_fd();
  void drain_socket();
  void accept(const std::uint8_t* data);
  void reject(std::size_t len);
  void flush();
  void note(const std::string& msg);

  FlatUdpSettings cfg_;
  SourceSink* sink_ = nullptr;
  DatagramTap tap_;
  int fd_ = -1;
  bool configured_ = false;
  bool ever_connected_ = false;
  std::size_t expected_ = 0;  ///< bytes of a valid datagram
  std::size_t block_ = 1;

  std::vector<SignalInfo> signals_;
  std::vector<std::uint8_t> rx_;
  std::vector<std::vector<double>> pending_;  ///< decoded, not yet framed: [signal][value]
  std::vector<SignalValues> views_;
  std::size_t pending_n_ = 0;  ///< datagrams in `pending_`
  Clock::time_point batch_start_;
  Clock::time_point last_valid_;
  std::set<std::size_t> reported_sizes_;

  std::atomic<bool> connected_{false};
  std::atomic<std::uint64_t> valid_{0};
  std::atomic<std::uint64_t> wrong_{0};
  mutable std::mutex msg_mu_;
  std::string last_message_;

  std::uint32_t frame_counter_ = 0;
  std::uint64_t frames_ = 0;
  std::uint64_t reconnects_ = 0;
};

}  // namespace udpscope::flatudp

• Step 2: Write the settings and source implementation

Create src/source/flatudp/flatudp_source.cpp:

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#include "source/flatudp/flatudp_source.hpp"

#include <arpa/inet.h>
#include <netinet/in.h>
#include <poll.h>
#include <sys/socket.h>
#include <unistd.h>

#include <algorithm>
#include <cctype>
#include <cerrno>
#include <cmath>
#include <cstdio>
#include <cstring>
#include <stdexcept>
#include <thread>

namespace udpscope::flatudp {
namespace {

using namespace std::chrono_literals;

std::string lower(std::string s) {
  for (char& c : s) c = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
  return s;
}

std::string trim(const std::string& s) {
  std::size_t a = 0, b = s.size();
  while (a < b && std::isspace(static_cast<unsigned char>(s[a]))) ++a;
  while (b > a && std::isspace(static_cast<unsigned char>(s[b - 1]))) --b;
  return s.substr(a, b - a);
}

double epoch_seconds() {
  return std::chrono::duration<double>(std::chrono::system_clock::now().time_since_epoch()).count();
}

constexpr std::size_t kMaxReportedSizes = 8;  ///< a sender of random sizes must not flood stderr

}  // namespace

bool valid_ipv4(const std::string& s) {
  in_addr a{};
  return inet_pton(AF_INET, s.c_str(), &a) == 1;
}

bool valid_multicast(const std::string& s) {
  in_addr a{};
  if (inet_pton(AF_INET, s.c_str(), &a) != 1) return false;
  return (ntohl(a.s_addr) >> 28) == 0xE;
}

// --- FlatUdpSettings ------------------------------------------------------------------------

FlatUdpSettings FlatUdpSettings::from(const SourceConfig& c) {
  FlatUdpSettings s;
  s.port = static_cast<int>(c.get_int("port", kDefaultPort));
  s.bind = trim(c.get("bind"));
  s.multicast = trim(c.get("multicast"));
  s.iface = trim(c.get("iface"));
  s.rate = c.get_double("rate", 0.0);
  const std::string e = lower(trim(c.get("endian", "little")));
  s.endian_ok = e == "little" || e == "big";
  s.big_endian = e == "big";
  s.signals_error = parse(c.get("signals"), s.rows);
  return s;
}

std::string FlatUdpSettings::problem() const {
  if (port < 1 || port > 65535) return "the port must be 1 to 65535";
  if (!signals_error.empty()) return signals_error;
  const std::string p = flatudp::problem(rows);
  if (!p.empty()) return p;
  if (!(rate > 0.0) || !std::isfinite(rate)) return "give the sender's cycle rate in Hz (rate)";
  if (!endian_ok) return "endian must be little or big";
  if (!bind.empty() && !valid_ipv4(bind)) return "the bind address '" + bind + "' is not an IPv4 address";
  if (!multicast.empty() && !valid_multicast(multicast))
    return "'" + multicast + "' is not a multicast group (224.0.0.0 to 239.255.255.255)";
  if (!iface.empty() && !valid_ipv4(iface))
    return "the interface address '" + iface + "' is not an IPv4 address";
  return {};
}

std::string FlatUdpSettings::identity() const {
  return (multicast.empty() ? std::string() : multicast + ":") + std::to_string(port);
}

std::size_t FlatUdpSettings::block() const {
  if (!all_scalar(rows) || !(rate > 0.0) || !std::isfinite(rate)) return 1;
  const long b = std::lround(rate / 100.0);
  return static_cast<std::size_t>(std::min(1024L, std::max(1L, b)));
}

// --- FlatUdpSource --------------------------------------------------------------------------

FlatUdpSource::FlatUdpSource(const SourceConfig& cfg) : cfg_(FlatUdpSettings::from(cfg)) {}

FlatUdpSource::~FlatUdpSource() { close_fd(); }

std::string FlatUdpSource::last_message() const {
  std::lock_guard<std::mutex> lk(msg_mu_);
  return last_message_;
}

void FlatUdpSource::note(const std::string& msg) {
  {
    std::lock_guard<std::mutex> lk(msg_mu_);
    last_message_ = msg;
  }
  std::fprintf(stderr, "[flatudp] %s\n", msg.c_str());
  std::fflush(stderr);
}

void FlatUdpSource::close_fd() {
  if (fd_ >= 0) {
    ::close(fd_);
    fd_ = -1;
  }
}

void FlatUdpSource::open(SourceSink& sink) {
  const std::string p = cfg_.problem();
  if (!p.empty()) throw std::runtime_error(p);
  sink_ = &sink;

  signals_.clear();
  for (const Row& r : cfg_.rows) {
    SignalInfo s;
    s.name = r.name;
    s.type_code = static_cast<std::uint8_t>(r.type);
    s.rows = r.count;
    s.cols = 1;
    s.time_mode = kTimePacket;
    s.sampling_rate = cfg_.rate;
    signals_.push_back(s);
  }
  expected_ = datagram_size(cfg_.rows);
  block_ = cfg_.block();
  pending_.assign(cfg_.rows.size(), {});
  pending_n_ = 0;
  rx_.assign(65536, 0);

  const auto fail = [this](const std::string& what) {
    const std::string msg = what + ": " + std::strerror(errno);
    close_fd();
    throw std::runtime_error(msg);
  };

  fd_ = ::socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
  if (fd_ < 0) fail("cannot create a UDP socket");
  // Receivers of one multicast group share its port; a unicast port is the one app's.
  if (!cfg_.multicast.empty()) {
    const int one = 1;
    ::setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
  }
  const int rcvbuf = 4 << 20;  // best effort: a 1 kHz sender at 64 KiB would not fit a default one
  ::setsockopt(fd_, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof rcvbuf);

  sockaddr_in addr{};
  addr.sin_family = AF_INET;
  addr.sin_port = htons(static_cast<std::uint16_t>(cfg_.port));
  addr.sin_addr.s_addr = htonl(INADDR_ANY);
  if (!cfg_.bind.empty()) inet_pton(AF_INET, cfg_.bind.c_str(), &addr.sin_addr);
  if (::bind(fd_, reinterpret_cast<const sockaddr*>(&addr), sizeof addr) < 0)
    fail("cannot bind UDP port " + std::to_string(cfg_.port));

  if (!cfg_.multicast.empty()) {
    ip_mreq mreq{};
    inet_pton(AF_INET, cfg_.multicast.c_str(), &mreq.imr_multiaddr);
    if (cfg_.iface.empty()) {
      mreq.imr_interface.s_addr = htonl(INADDR_ANY);
    } else {
      inet_pton(AF_INET, cfg_.iface.c_str(), &mreq.imr_interface);
    }
    if (::setsockopt(fd_, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof mreq) < 0)
      fail("cannot join multicast group " + cfg_.multicast);
  }
}

void FlatUdpSource::close() {
  close_fd();
  connected_.store(false, std::memory_order_relaxed);
}

SourceStats FlatUdpSource::stats() const {
  SourceStats s;
  s.packets = valid_.load(std::memory_order_relaxed) + wrong_.load(std::memory_order_relaxed);
  s.frames = frames_;
  s.stale = wrong_.load(std::memory_order_relaxed);
  s.reconnects = reconnects_;
  s.connected = connected_.load(std::memory_order_relaxed);
  return s;
}

void FlatUdpSource::poll(int timeout_ms) {
  drain_commands();
  // Never wait long: the partial-block flush and the silence timeout are checked on this cadence.
  const int wait = std::clamp(timeout_ms, 0, 20);
  if (fd_ < 0 || sink_ == nullptr) {
    std::this_thread::sleep_for(std::chrono::milliseconds(wait));
    return;
  }
  if (!configured_) {
    configured_ = true;
    sink_->on_config(signals_, 1);
  }
  pollfd p{};
  p.fd = fd_;
  p.events = POLLIN;
  if (::poll(&p, 1, wait) > 0 && (p.revents & POLLIN)) drain_socket();

  const auto now = Clock::now();
  if (pending_n_ > 0 && now - batch_start_ >= 100ms) flush();  // a stopped sender holds nothing back
  if (connected_.load(std::memory_order_relaxed) && now - last_valid_ >= 2s) {
    connected_.store(false, std::memory_order_relaxed);
    sink_->on_event(SourceEvent::Disconnected, "no datagram for 2 s");
  }
}

void FlatUdpSource::drain_socket() {
  for (int i = 0; i < 8192; ++i) {  // bounded, so a flood cannot starve the timers
    const ssize_t n = ::recv(fd_, rx_.data(), rx_.size(), 0);
    if (n < 0) {
      if (errno == EINTR) continue;
      break;  // EAGAIN: drained
    }
    const double t = epoch_seconds();
    const std::size_t len = static_cast<std::size_t>(n);
    if (tap_) tap_(rx_.data(), len, t);
    if (len == expected_) {
      accept(rx_.data());
    } else {
      reject(len);
    }
  }
}

void FlatUdpSource::accept(const std::uint8_t* data) {
  valid_.fetch_add(1, std::memory_order_relaxed);
  last_valid_ = Clock::now();
  if (!connected_.load(std::memory_order_relaxed)) {
    if (ever_connected_) ++reconnects_;
    ever_connected_ = true;
    connected_.store(true, std::memory_order_relaxed);
    sink_->on_event(SourceEvent::Connected, "");
  }
  decode_datagram(cfg_.rows, data, cfg_.big_endian, pending_);
  if (pending_n_ == 0) batch_start_ = last_valid_;
  if (++pending_n_ >= block_) flush();
}

void FlatUdpSource::reject(std::size_t len) {
  wrong_.fetch_add(1, std::memory_order_relaxed);
  if (reported_sizes_.size() >= kMaxReportedSizes || !reported_sizes_.insert(len).second) return;
  const std::string msg = "datagram of " + std::to_string(len) + " bytes, expected " +
                          std::to_string(expected_) + ", check the signal list";
  note(msg);
  sink_->on_event(SourceEvent::Error, msg.c_str());
}

void FlatUdpSource::flush() {
  const std::size_t n = signals_.size();
  views_.resize(n);
  for (std::size_t i = 0; i < n; ++i)
    views_[i] = SignalValues{pending_[i].data(), static_cast<std::uint32_t>(pending_[i].size())};
  Frame f;
  f.counter = frame_counter_++;
  f.lost = 0;
  f.late = false;
  f.recv_time = epoch_seconds();
  f.publish_mode = 1;
  f.num_samples = static_cast<std::uint32_t>(pending_n_);
  f.num_signals = static_cast<std::uint32_t>(n);
  f.signals = signals_.data();
  f.values = views_.data();
  ++frames_;
  sink_->on_data(f);
  for (auto& c : pending_) c.clear();
  pending_n_ = 0;
}

}  // namespace udpscope::flatudp

• Step 3: Write the kind with an interim form

The signals row is a plain text entry here; Task 3 replaces it with the row editor. Create src/source/flatudp/flatudp_kind.cpp:

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#include "source/flatudp/flatudp_source.hpp"
#include "source/registry.hpp"

#include <adwaita.h>

#include <cstdio>
#include <functional>
#include <memory>
#include <string>

namespace udpscope {
namespace {

using namespace flatudp;
using ChangeCb = std::shared_ptr<std::function<void()>>;

std::string num(double v) {
  char b[48];
  std::snprintf(b, sizeof b, "%.10g", v);
  return b;
}

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;
}

GtkWidget* spin_row(const char* title, double lo, double hi, double step, double def, int digits) {
  GtkWidget* r = GTK_WIDGET(adw_spin_row_new_with_range(lo, hi, step));
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
  adw_spin_row_set_digits(ADW_SPIN_ROW(r), static_cast<guint>(digits));
  adw_spin_row_set_value(ADW_SPIN_ROW(r), def);
  return r;
}

GtkWidget* combo_row(const char* title, const char* const* items) {
  GtkWidget* r = adw_combo_row_new();
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
  adw_combo_row_set_model(ADW_COMBO_ROW(r), G_LIST_MODEL(gtk_string_list_new(items)));
  return r;
}

GtkWidget* expander(const char* title, std::initializer_list<GtkWidget*> rows) {
  GtkWidget* e = adw_expander_row_new();
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(e), title);
  for (GtkWidget* r : rows) adw_expander_row_add_row(ADW_EXPANDER_ROW(e), r);
  return e;
}

std::string text_of(GtkWidget* w) { return gtk_editable_get_text(GTK_EDITABLE(w)); }
double value_of(GtkWidget* w) { return adw_spin_row_get_value(ADW_SPIN_ROW(w)); }
guint sel_of(GtkWidget* w) { return adw_combo_row_get_selected(ADW_COMBO_ROW(w)); }

void mark(GtkWidget* w, bool bad) {
  if (bad) {
    gtk_widget_add_css_class(w, "error");
  } else {
    gtk_widget_remove_css_class(w, "error");
  }
}

/// The form's widgets, shared by every signal closure and every `ConfigForm` closure.
struct Form {
  GtkWidget *port, *bind, *multicast, *iface, *rate, *endian, *signals;
  ChangeCb changed = std::make_shared<std::function<void()>>();
  bool updating = false;
  bool alive = true;  ///< cleared when the chooser destroys the rows; handlers then touch nothing

  /// A bad address is marked with the `error` style; the rest of the problem is `valid()`'s.
  void refresh() {
    mark(bind, !text_of(bind).empty() && !valid_ipv4(text_of(bind)));
    mark(multicast, !text_of(multicast).empty() && !valid_multicast(text_of(multicast)));
    mark(iface, !text_of(iface).empty() && !valid_ipv4(text_of(iface)));
  }

  SourceConfig read() const {
    SourceConfig c;
    c.set("port", num(value_of(port)));
    if (!text_of(bind).empty()) c.set("bind", text_of(bind));
    if (!text_of(multicast).empty()) c.set("multicast", text_of(multicast));
    if (!text_of(iface).empty()) c.set("iface", text_of(iface));
    c.set("rate", num(value_of(rate)));
    c.set("endian", sel_of(endian) == 1 ? "big" : "little");
    c.set("signals", text_of(signals));
    return c;
  }

  void write(const SourceConfig& cfg) {
    const FlatUdpSettings s = FlatUdpSettings::from(cfg);
    updating = true;
    adw_spin_row_set_value(ADW_SPIN_ROW(port), s.port);
    gtk_editable_set_text(GTK_EDITABLE(bind), s.bind.c_str());
    gtk_editable_set_text(GTK_EDITABLE(multicast), s.multicast.c_str());
    gtk_editable_set_text(GTK_EDITABLE(iface), s.iface.c_str());
    if (s.rate > 0.0) adw_spin_row_set_value(ADW_SPIN_ROW(rate), s.rate);
    adw_combo_row_set_selected(ADW_COMBO_ROW(endian), s.big_endian ? 1 : 0);
    gtk_editable_set_text(GTK_EDITABLE(signals), cfg.get("signals").c_str());
    updating = false;
    refresh();
  }

  bool valid() const { return FlatUdpSettings::from(read()).problem().empty(); }
};

using FormPtr = std::shared_ptr<Form>;

void on_notify(GObject*, GParamSpec*, gpointer user) {
  Form& f = **static_cast<FormPtr*>(user);
  if (!f.alive) return;
  f.refresh();
  if (!f.updating && f.changed && *f.changed) (*f.changed)();
}

void on_edited(GtkEditable*, gpointer user) {
  Form& f = **static_cast<FormPtr*>(user);
  if (!f.alive) return;
  f.refresh();
  if (!f.updating && f.changed && *f.changed) (*f.changed)();
}

void free_form(gpointer d, GClosure*) { delete static_cast<FormPtr*>(d); }

/// "destroy" on a row group: the siblings the handlers reach by raw pointer may already be gone.
void on_form_destroy(GtkWidget*, gpointer user) { (**static_cast<FormPtr*>(user)).alive = false; }

void hook_destroy(GtkWidget* w, const FormPtr& f) {
  g_signal_connect_data(w, "destroy", G_CALLBACK(on_form_destroy), new FormPtr(f), free_form,
                        static_cast<GConnectFlags>(0));
}

void hook_notify(GtkWidget* w, const char* signal, const FormPtr& f) {
  g_signal_connect_data(w, signal, G_CALLBACK(on_notify), new FormPtr(f), free_form,
                        static_cast<GConnectFlags>(0));
}

void hook_entry(GtkWidget* w, const FormPtr& f) {
  g_signal_connect_data(w, "changed", G_CALLBACK(on_edited), new FormPtr(f), free_form,
                        static_cast<GConnectFlags>(0));
}

class FlatUdpKind final : public SourceKind {
 public:
  std::string id() const override { return "flatudp"; }
  std::string display_name() const override { return "Flat UDP (MARTe2 UDPSender)"; }
  bool has_datagrams() const override { return true; }

  std::string identity(const SourceConfig& cfg) const override {
    return FlatUdpSettings::from(cfg).identity();
  }

  std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
    return std::make_unique<FlatUdpSource>(cfg);
  }

  ConfigForm make_config_form() const override {
    auto f = std::make_shared<Form>();
    static const char* const kEndian[] = {"Little endian", "Big endian", nullptr};

    f->port = spin_row("_Port", 1, 65535, 1, kDefaultPort, 0);
    f->bind = entry_row("_Bind address");
    f->multicast = entry_row("_Multicast group");
    f->iface = entry_row("Multicast _interface");
    f->rate = spin_row("Cycle _rate (Hz)", 0.001, 1000000, 1, 1000, 3);
    f->endian = combo_row("_Byte order", kEndian);
    f->signals = entry_row("_Signals");

    GtkWidget* net = expander("Network", {f->port, f->bind, f->multicast, f->iface});
    GtkWidget* timing = expander("Timing and format", {f->rate, f->endian});
    GtkWidget* sig = expander("Signals", {f->signals});
    adw_expander_row_set_expanded(ADW_EXPANDER_ROW(net), TRUE);
    adw_expander_row_set_expanded(ADW_EXPANDER_ROW(sig), TRUE);

    for (GtkWidget* w : {net, timing, sig}) hook_destroy(w, f);
    for (GtkWidget* w : {f->bind, f->multicast, f->iface, f->signals}) hook_entry(w, f);
    for (GtkWidget* w : {f->port, f->rate}) hook_notify(w, "notify::value", f);
    hook_notify(f->endian, "notify::selected", f);
    f->refresh();

    ConfigForm out;
    out.rows = {net, timing, sig};
    out.description =
        "The signal list must match the sender's exactly, in order: the datagrams carry no "
        "description. Write name:type[count], comma separated, e.g. Trigger:uint8,Ia:float32,"
        "Wave:float32[16]";
    out.focus = f->signals;
    out.write = [f](const SourceConfig& c) { f->write(c); };
    out.read = [f]() { return f->read(); };
    out.valid = [f]() { return f->valid(); };
    ChangeCb cb = f->changed;
    out.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
    return out;
  }
};

}  // namespace

std::unique_ptr<SourceKind> make_flatudp_kind() { return std::make_unique<FlatUdpKind>(); }

}  // namespace udpscope

• Step 4: Register the kind

CMakeLists.txt, after the flatudp_signals.cpp line add:

  src/source/flatudp/flatudp_source.cpp
  src/source/flatudp/flatudp_kind.cpp

src/source/registry.hpp: change the comment above the make_*_kind declarations to name flatudp/flatudp_kind.cpp too, and add after make_e1608_kind:

std::unique_ptr<SourceKind> make_flatudp_kind();

src/source/registry.cpp, in register_builtin_sources(), after register_kind(make_e1608_kind());:

  register_kind(make_flatudp_kind());

• Step 5: Write the mock sender

Create tools/mock_flatudp.py (then chmod +x):

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# SPDX-FileCopyrightText: 2026 Martino Ferrari
"""A stand-in for the MARTe2 UDPSender datasource, for the flatudp source's self-check.

One datagram is one cycle: the signals concatenated in declaration order, no padding, in the
chosen byte order. Packet i carries, for the signal of index k (from 0) and its element e,
the value (i * (k + 1) + e) % 100, so consecutive datagrams are predictable.

  --signals     name:type[count],...   default Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4]
  --rate        cycles per second
  --big-endian  pack big endian (default little)
  --gap S,D     send nothing from S to S+D seconds after the start (a sender that stops)
  --bad-every S one datagram one byte too long every S seconds (the wrong-size path)
  --multicast G --mcast-if A   send to group G out of interface A instead of --host
  --log FILE    one line per event
"""
import argparse
import socket
import struct
import sys
import time

CODES = {"uint8": "B", "int8": "b", "uint16": "H", "int16": "h", "uint32": "I", "int32": "i",
         "uint64": "Q", "int64": "q", "float32": "f", "float64": "d"}


def parse_signals(text):
    out = []
    for tok in text.split(","):
        name, _, ty = tok.strip().partition(":")
        count = 1
        if "[" in ty:
            ty, _, n = ty.partition("[")
            count = int(n.rstrip("]"))
        out.append((name.strip(), CODES[ty.strip()], count))
    return out


def make_packer(sigs, big):
    st = struct.Struct((">" if big else "<") + "".join(code * count for _, code, count in sigs))

    def pack(i):
        vals = []
        for k, (_, code, count) in enumerate(sigs):
            for e in range(count):
                v = (i * (k + 1) + e) % 100
                vals.append(float(v) if code in "fd" else v)
        return st.pack(*vals)

    return pack, st.size


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawTextHelpFormatter)
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=44488)
    ap.add_argument("--rate", type=float, default=1000.0)
    ap.add_argument("--signals", default="Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4]")
    ap.add_argument("--big-endian", action="store_true")
    ap.add_argument("--gap", default="")
    ap.add_argument("--bad-every", type=float, default=0.0)
    ap.add_argument("--multicast", default="")
    ap.add_argument("--mcast-if", default="127.0.0.1")
    ap.add_argument("--log", default="")
    args = ap.parse_args()

    logf = open(args.log, "w") if args.log else None

    def log(msg):
        if logf:
            logf.write("%.3f %s\n" % (time.monotonic() - t0, msg))
            logf.flush()

    sigs = parse_signals(args.signals)
    pack, size = make_packer(sigs, args.big_endian)
    gap = tuple(float(x) for x in args.gap.split(",")) if args.gap else None

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    dest = (args.host, args.port)
    if args.multicast:
        dest = (args.multicast, args.port)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(args.mcast_if))

    period = 1.0 / args.rate
    t0 = time.monotonic()
    log("start %s rate=%g size=%d %s" % (args.signals, args.rate, size, "big" if args.big_endian else "little"))
    next_t = t0
    next_bad = t0 + args.bad_every
    in_gap = False
    i = 0
    try:
        while True:
            now = time.monotonic()
            if now < next_t:
                time.sleep(min(next_t - now, 0.001))
                continue
            elapsed = now - t0
            if gap and gap[0] <= elapsed < gap[0] + gap[1]:
                if not in_gap:
                    in_gap = True
                    log("gap begins")
                next_t = now + period
                time.sleep(0.01)
                continue
            if in_gap:
                in_gap = False
                log("gap ends")
            if args.bad_every > 0 and now >= next_bad:
                next_bad = now + args.bad_every
                sock.sendto(pack(i) + b"\x00", dest)
                log("bad datagram of %d bytes" % (size + 1))
            burst = 0
            while next_t <= now and burst < 64:  # catch up a late timer, but bounded
                sock.sendto(pack(i), dest)
                i += 1
                next_t += period
                burst += 1
            if next_t < now - 0.5:
                next_t = now
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    sys.exit(main())

• Step 6: Replace the probe

In src/ui_probe.cpp:

1. Add the include #include "source/flatudp/flatudp_source.hpp" after the signals include.
2. In the anonymous namespace, directly after flatudp_codec_check() and before its closing } // namespace, add:

flatudp::FlatUdpSource* flatudp_source(App* a) {
  return a->acq != nullptr ? dynamic_cast<flatudp::FlatUdpSource*>(a->acq->source()) : nullptr;
}

std::uint64_t g_flatudp_frames_before = 0;

/// The mock's signal k carries (i * (k + 1) + element) % 100 in packet i, so two consecutive
/// samples of a scalar signal differ by k + 1 modulo 100. Returns the number of steps that do not,
/// over a window clear of the head the producer is writing; -1 when it cannot be read.
long flatudp_breaks(App* a, std::size_t channel, long step) {
  if (channel >= a->model->channels().size()) return -1;
  const ChannelBuffer& ch = a->model->live().channel(channel);
  const u64 n = ch.size();
  const u64 want = 400;
  if (n <= want + 50) return -1;
  std::vector<f32> v(static_cast<std::size_t>(want));
  ch.raw(n - want - 50, n - 50, v.data());
  long breaks = 0;
  for (std::size_t i = 1; i < v.size(); ++i) {
    const long d = std::lround(double(v[i])) - std::lround(double(v[i - 1]));
    if (((d % 100) + 100) % 100 != step) ++breaks;
  }
  return breaks;
}

3. Replace the whole flatudp_probe function (the T1 one) with:

/// Phase 2, 1 s after the signals were enabled: the stream, its channels, the value continuity, the
/// wrong-size accounting and the capture button.
gboolean flatudp_probe_report(gpointer user) {
  App* a = static_cast<App*>(user);
  if (a->window == nullptr) return G_SOURCE_REMOVE;
  flatudp::FlatUdpSource* src = flatudp_source(a);
  if (src == nullptr) return G_SOURCE_REMOVE;
  const flatudp::FlatUdpSettings& s = src->settings();
  const bool up = a->model->stats().connected.load(std::memory_order_relaxed);

  const std::vector<SignalInfo>& sigs = a->model->signals();
  std::string names, expect_names;
  std::size_t expect_channels = 0;
  for (const SignalInfo& si : sigs) names += (names.empty() ? "" : ",") + si.name;
  for (const flatudp::Row& r : s.rows) {
    expect_names += (expect_names.empty() ? "" : ",") + r.name;
    expect_channels += r.count;
  }
  const double rate = sigs.empty() ? 0.0 : sigs.front().sampling_rate;
  const std::uint64_t frames = a->model->stats().frames.load(std::memory_order_relaxed);
  const bool stream_ok = up && names == expect_names && std::fabs(rate - s.rate) < 1e-6 && frames > 0;
  std::fprintf(stderr,
               "[flatudp] stream: connected=%d signals=%zu names=%s expect=%s frames=%" G_GUINT64_FORMAT
               " rate=%.3f ok=%d\n",
               up ? 1 : 0, sigs.size(), names.c_str(), expect_names.c_str(), frames, rate,
               stream_ok ? 1 : 0);

  const std::size_t channels = a->model->channels().size();
  std::fprintf(stderr, "[flatudp] channels: count=%zu expect=%zu ok=%d\n", channels, expect_channels,
               channels == expect_channels ? 1 : 0);

  // Only the first two signals are checked, and only when they are scalars (the mock's default).
  long b0 = -1, b1 = -1;
  if (up && s.rows.size() >= 2 && s.rows[0].count == 1 && s.rows[1].count == 1) {
    b0 = flatudp_breaks(a, 0, 1);
    b1 = flatudp_breaks(a, 1, 2);
  }
  std::fprintf(stderr, "[flatudp] continuity: ch0_breaks=%ld ch1_breaks=%ld ok=%d\n", b0, b1,
               b0 == 0 && b1 == 0 ? 1 : 0);

  const std::uint64_t wrong = src->wrong_size_datagrams();
  const std::string msg = src->last_message();
  std::fprintf(stderr, "[flatudp] wrong size: count=%" G_GUINT64_FORMAT " message='%s' ok=%d\n", wrong,
               msg.c_str(), wrong >= 1 && msg.find("expected") != std::string::npos ? 1 : 0);

  std::fprintf(stderr, "[flatudp] 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::fprintf(stderr, "[flatudp] state: valid=%" G_GUINT64_FORMAT " wrong=%" G_GUINT64_FORMAT "\n",
               src->valid_datagrams(), wrong);
  std::fflush(stderr);
  return G_SOURCE_REMOVE;
}

/// Phase 1: the codec check (it needs no flatudp source), then every signal enabled in one structural
/// change, and the report a second later.
gboolean flatudp_probe(gpointer user) {
  App* a = static_cast<App*>(user);
  if (a->window == nullptr) return G_SOURCE_REMOVE;
  std::fprintf(stderr, "[flatudp] codec: ok=%d\n", flatudp_codec_check() ? 1 : 0);
  if (flatudp_source(a) == nullptr) {
    std::fprintf(stderr, "[flatudp] no flatudp source (run with --source flatudp)\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);
  std::fflush(stderr);
  g_timeout_add(1000, flatudp_probe_report, a);
  return G_SOURCE_REMOVE;
}

/// After the mock's gap: the stream's frame count moved on and the source is connected again.
gboolean flatudp_probe_resumed(gpointer user) {
  App* a = static_cast<App*>(user);
  if (a->window == nullptr) return G_SOURCE_REMOVE;
  flatudp::FlatUdpSource* src = flatudp_source(a);
  const std::uint64_t after = a->model->stats().frames.load(std::memory_order_relaxed);
  const bool up = src != nullptr && src->connected() &&
                  a->model->stats().connected.load(std::memory_order_relaxed);
  std::fprintf(stderr, "[flatudp] resumed: connected=%d frames_before=%" G_GUINT64_FORMAT
                       " frames_after=%" G_GUINT64_FORMAT " ok=%d\n",
               up ? 1 : 0, g_flatudp_frames_before, after,
               up && after > g_flatudp_frames_before ? 1 : 0);
  std::fflush(stderr);
  return G_SOURCE_REMOVE;
}

/// Inside the mock's gap: 2 s of silence must have raised `Disconnected`.
gboolean flatudp_probe_gap(gpointer user) {
  App* a = static_cast<App*>(user);
  if (a->window == nullptr) return G_SOURCE_REMOVE;
  flatudp::FlatUdpSource* src = flatudp_source(a);
  if (src == nullptr) return G_SOURCE_REMOVE;
  const bool src_up = src->connected();
  const bool model_up = a->model->stats().connected.load(std::memory_order_relaxed);
  g_flatudp_frames_before = a->model->stats().frames.load(std::memory_order_relaxed);
  std::fprintf(stderr, "[flatudp] silence: connected=%d model_connected=%d ok=%d\n", src_up ? 1 : 0,
               model_up ? 1 : 0, !src_up && !model_up ? 1 : 0);
  std::fflush(stderr);
  g_timeout_add(4000, flatudp_probe_resumed, a);
  return G_SOURCE_REMOVE;
}

The phase functions reference each other, so define them in this order (resumed, gap, report, probe) or forward-declare: they are all declared in ui_app.hpp (next step), so any order compiles.

4. In source_kinds_probe's sample lambda, after the e1608 branch, add:

    } else if (id == "flatudp") {
      c.set("port", "44600");
      c.set("bind", "10.0.0.8");
      c.set("multicast", "239.1.2.3");
      c.set("iface", "10.0.0.9");
      c.set("rate", "2500");
      c.set("endian", "big");
      c.set("signals", "Trigger:uint8,Ia:float32,Wave:int16[4]");
    }

(the existing e1608 branch's closing } moves after this one: the chain is if (udps) {...} else if (probe) {...} else if (e1608) {...} else if (flatudp) {...}).

5. src/ui_app.hpp: replace the T1 flatudp_probe declaration with:

/// The flat UDP (MARTe2 UDPSender) source: the signal-list codec and a hand-computed decode, then,
/// against `tools/mock_flatudp.py`, the stream, the value continuity, the wrong-size accounting and
/// the capture button (`_report`, a second later). `_gap` reads the source inside the mock's silence;
/// `_resumed` shows the stream came back.
gboolean flatudp_probe(gpointer user);
gboolean flatudp_probe_report(gpointer user);
gboolean flatudp_probe_gap(gpointer user);
gboolean flatudp_probe_resumed(gpointer user);

6. src/ui.cpp, after the UDPSCOPE_PROBE_FLATUDP block add:

  if (const char* env = g_getenv("UDPSCOPE_PROBE_FLATUDP_GAP")) {
    g_timeout_add(static_cast<guint>(std::atoi(env)), flatudp_probe_gap, &app);
  }

• Step 7: Build

cmake -S . -B build >/dev/null && cmake --build build -j8 2>&1 | grep -E "warning|error" | head -20

Expected: no errors; at most the 2 pre-existing MathNode::source warnings of ui_probe.cpp, nothing from flatudp. If ChannelBuffer::raw or model->channels() differ from what the e1608 probe calls (see e1608_probe_report), mirror that code.

• Step 8: Run against the mock (scalar and vector lists)

First the run with a vector signal in the list and the mock's gap and bad datagrams; the probe schedule is: phase 1 at 2.5 s, silence at 8 s, resumed at 12 s.

python3 tools/mock_flatudp.py --port 44588 --rate 1000 --signals Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] \
  --gap 6,5 --bad-every 1 --log /tmp/flatudp-mock.log &
MOCK=$!
sleep 1
UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_PROBE_FLATUDP_GAP=8000 UDPSCOPE_EXIT_AFTER_MS=14500 timeout 40 \
  xvfb-run -a build/udpscope --source flatudp --source-opt port=44588 --source-opt rate=1000 \
  --source-opt signals=Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] 2>&1 \
  | grep -E "\[flatudp\]|Gtk-CRITICAL|gtk warnings|buffer allocated"
kill $MOCK 2>/dev/null; wait $MOCK 2>/dev/null; head -5 /tmp/flatudp-mock.log

Expected, in order:

[flatudp] codec: ok=1
[flatudp] datagram of N bytes, expected M, check the signal list   <- exactly ONE such line (N = M + 1)
[flatudp] stream: connected=1 signals=4 names=Trigger,Ia,Cnt,Wave expect=Trigger,Ia,Cnt,Wave frames=(any N > 0) rate=1000.000 ok=1
[flatudp] channels: count=7 expect=7 ok=1
[flatudp] continuity: ch0_breaks=0 ch1_breaks=0 ok=1
[flatudp] wrong size: count=(any N >= 1) message='datagram of ... expected ...' ok=1
[flatudp] capture button: present=1 sensitive=1
[flatudp] state: valid=(any N > 0) wrong=(any N >= 1)
[flatudp] silence: connected=0 model_connected=0 ok=1
[flatudp] resumed: connected=1 frames_before=(N) frames_after=(a number larger than N) ok=1
gtk warnings: 0

Rough sizes: M = 1+4+8+8 = 21 bytes, N = 22. Notes for reading failures:

• stream rate is the declared rate (1000); names is the signal order, so swapped signals read ok=0.

• continuity nonzero: first check the source did not drop or reorder datagrams (valid against the mock's count), then whether the model's ingest decimation is 1:1 at 1 kHz (a decimated channel steps by more than k + 1). Do not weaken the check before one of those is shown.

• capture button sensitive=0: find where the button's sensitivity is derived from the active kind's has_datagrams() (grep capture_button); do not special-case the kind.

• Any Gtk-CRITICAL or a count other than one wrong-size stderr line is a defect.

• Step 9: Run the scalar list (batching) and the big-endian run

python3 tools/mock_flatudp.py --port 44589 --rate 1000 --signals Trigger:uint8,Ia:float32,Cnt:uint64 --big-endian &
MOCK=$!
sleep 1
UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000 timeout 30 \
  xvfb-run -a build/udpscope --source flatudp --source-opt port=44589 --source-opt rate=1000 \
  --source-opt endian=big --source-opt signals=Trigger:uint8,Ia:float32,Cnt:uint64 2>&1 \
  | grep -E "\[flatudp\]|Gtk-CRITICAL|gtk warnings"
kill $MOCK 2>/dev/null; wait $MOCK 2>/dev/null

Expected: stream ... names=Trigger,Ia,Cnt ... ok=1, channels: count=3 expect=3 ok=1, continuity: ch0_breaks=0 ch1_breaks=0 ok=1 (a byte-order or batching defect shows here), wrong size: count=0 ... ok=0 (no --bad-every in this run: a reading, not a failure), gtk warnings: 0.

Then the no-sender run (nothing listening on the mock side), 3 s:

UDPSCOPE_PROBE_FLATUDP=1500 UDPSCOPE_EXIT_AFTER_MS=3000 timeout 20 \
  xvfb-run -a build/udpscope --source flatudp --source-opt port=44590 --source-opt rate=1000 \
  --source-opt signals=Trigger:uint8,Ia:float32 2>&1 | grep -E "\[flatudp\]|Gtk-CRITICAL|gtk warnings"

Expected: stream: connected=0 ... frames=0 ... ok=0, continuity: ... -1 ... ok=0, no crash, gtk warnings: 0.

A --source flatudp run with no rate must fail to start with the message "give the sender's cycle rate in Hz (rate)" rather than silently using 1 kHz:

UDPSCOPE_EXIT_AFTER_MS=2000 timeout 20 xvfb-run -a build/udpscope --source flatudp \
  --source-opt signals=A:uint8 2>&1 | grep -iE "rate|error" | head -3

If the app starts a window with no error, look at how --source failures are reported for the e1608 (--source e1608 with no address) and match that behaviour rather than inventing one.

• Step 10: The [source-kinds] run

UDPSCOPE_PROBE_SOURCE_KINDS=2000 UDPSCOPE_EXIT_AFTER_MS=5000 timeout 30 \
  xvfb-run -a build/udpscope --source probe --source-opt rate=500 2>&1 | grep -E "\[source-kinds\]|gtk warnings|Gtk-CRITICAL"

Expected: kinds=udps,e1608,flatudp,probe, form flatudp: round_trip=1 valid=1, nine [source-kinds] lines in all, gtk warnings: 0. (valid=1 holds because the sample has a rate and a signal list.)

• Step 11: Documentation

1. docs/claude/build-and-run.md: after the e1608 sh block (before "tools/mock_udpstreamer.py has waveform-injection options"), add:

`--source flatudp` receives the MARTe2 `UDPSender` datasource (one datagram per cycle: the signals
concatenated in declaration order, no header, no padding). The datagram carries **no description**, so the
settings are the whole contract: `port` 44488, `bind`, `multicast` and `iface` (a group to join and the local
interface for it), `rate` (the sender's cycle rate in Hz; **required**, it sizes the ring and the ingest
decimation and spaces the samples within a frame), `endian` `little|big`, and `signals`
`name:type[count],...` with types `uint8/16/32/64`, `int8/16/32/64`, `float32`, `float64` (a count above 1
makes a vector signal), e.g. `Trigger:uint8,Ia:float32,Wave:float32[16]`. A datagram of any other size than
the list implies is dropped and counted as stale; each distinct wrong size is reported once on stderr as
`[flatudp] ...`. Loss is never reported: the stream has no counter. `tools/mock_flatudp.py` is a sender for
a signal spec (`--rate`, `--signals`, `--big-endian`, `--gap S,D`, `--bad-every S`, `--multicast G --mcast-if A`,
`--log FILE`).

```sh
python3 tools/mock_flatudp.py --port 44588 --rate 1000 --signals Trigger:uint8,Ia:float32,Wave:int16[4] &
./build/udpscope --source flatudp --source-opt port=44588 --source-opt rate=1000 \
  --source-opt signals=Trigger:uint8,Ia:float32,Wave:int16[4]
```

2. docs/claude/verification.md: in the UDPSCOPE_PROBE_SOURCE_KINDS entry change "Eight [source-kinds] lines (one form line per registered kind: udps, e1608, probe)" to "Nine [source-kinds] lines (one form line per registered kind: udps, e1608, flatudp, probe)". After the UDPSCOPE_PROBE_E1608_RECONNECT=8000 entry add:

- `UDPSCOPE_PROBE_FLATUDP=2500` with `UDPSCOPE_EXIT_AFTER_MS`, run as `--source flatudp` against
  `tools/mock_flatudp.py` -- the flat UDP source end to end: a hand-computed codec and decode check (it
  needs no source), the stream's signal names and declared rate, the channel count (vectors count per
  element), value continuity across the first two signals, the wrong-size accounting and the capture button.
  It enables the signals itself, so do **not** add `UDPSCOPE_ENABLE_ALL_MS`. Runs, each short and serial,
  with the mock killed afterwards:
  - **mixed list, gap and bad datagrams:** mock `--rate 1000 --signals Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4]
    --gap 6,5 --bad-every 1 --log /tmp/flatudp-mock.log`, the same `signals` as a `--source-opt`, with
    `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_PROBE_FLATUDP_GAP=8000 UDPSCOPE_EXIT_AFTER_MS=14500`; stderr carries
    exactly one `datagram of 22 bytes, expected 21` line;
  - **scalar list, big endian (batching):** mock `--signals Trigger:uint8,Ia:float32,Cnt:uint64 --big-endian`,
    app `--source-opt endian=big`, `UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000`;
  - **no sender:** `connected=0` throughout, a reading and not a failure.
- `UDPSCOPE_PROBE_FLATUDP_GAP=8000` -- reads the source inside the mock's silence (2 s after its last datagram
  the source is disconnected) and 4 s later checks it reconnected and the stream moved on.

3. docs/claude/probe-readings.md: in the [source-kinds] block change kinds=udps,e1608,probe to kinds=udps,e1608,flatudp,probe and insert [source-kinds] form flatudp: round_trip=1 valid=1 after the e1608 form line; in the bullet about the e1608 sample add "The flatudp sample sets port, bind, multicast, iface, rate, endian and signals." Add a new section after ## [e1608]'s last bullet (before the next ##):

## `[flatudp]`

```
[flatudp] codec: ok=1
[flatudp] datagram of 22 bytes, expected 21, check the signal list
[flatudp] stream: connected=1 signals=4 names=Trigger,Ia,Cnt,Wave expect=Trigger,Ia,Cnt,Wave frames=2410 rate=1000.000 ok=1
[flatudp] channels: count=7 expect=7 ok=1
[flatudp] continuity: ch0_breaks=0 ch1_breaks=0 ok=1
[flatudp] wrong size: count=3 message='datagram of 22 bytes, expected 21, check the signal list' ok=1
[flatudp] capture button: present=1 sensitive=1
[flatudp] state: valid=3511 wrong=3
[flatudp] silence: connected=0 model_connected=0 ok=1
[flatudp] resumed: connected=1 frames_before=3522 frames_after=4310 ok=1
```

`UDPSCOPE_PROBE_FLATUDP=2500` (the first lines; counts vary) and `UDPSCOPE_PROBE_FLATUDP_GAP=8000` (the last two)
run as `--source flatudp` against `tools/mock_flatudp.py` with the options in `verification.md`. The second line is
the source's own stderr message, printed once per distinct wrong size, not a probe line. With no sender the
stream lines read `connected=0 ... frames=0 ... ok=0` and `continuity` reads `-1`: a reading, not a failure. A
run without `--bad-every` reads `wrong size: count=0 ... ok=0` for the same reason.

- `codec: ok=1`: the `signals` text round-trips, a hand-built 24-byte datagram decodes to the hand-computed
  values in both byte orders (`uint8` 254, `int16` -2, `float32` 1.5, `uint64[2]` 1 and 4294967298, `int8` -1)
  and the validity rules refuse what they should. It needs no source, so `--source probe` shows it too.
  `ok=0` is a codec defect: a swapped byte order, a wrong width or a rule that does not fire.
- `stream: ... ok=1`: the model's signals are the list's rows in order, at the declared rate. `names` out of
  order is a decode that mixed rows; `rate` of 1000 against a different declared rate is a `sampling_rate`
  that did not reach the model (the ring and the decimation are then sized for the wrong cadence).
- `channels: count=7`: the sum of the rows' counts -- a vector signal is one channel per element.
- `continuity: ch0_breaks=0 ch1_breaks=0`: consecutive samples of the mock's first two signals step by 1 and
  by 2 modulo 100. A break is a lost or reordered datagram, a batch flushed with the wrong number of samples,
  or ingest decimation that is not 1:1.
- `wrong size: ok=1`: wrong-size datagrams were counted and the message names the received and expected size.
  The message must also be the only stderr line of its kind: one per distinct size, however many arrive.
- `capture button: present=1 sensitive=1`: the kind reports datagrams, so the capture UI is live (contrast
  `[source-kinds]`, where the probe kind leaves it disabled).
- `silence: connected=0 model_connected=0`: 2 s without a valid datagram raised `Disconnected`. `connected=1`
  is the timeout not firing, or wrong-size datagrams refreshing the silence clock (they must not).
- `resumed: ok=1`: `Connected` was raised again and frames resumed. `connected=0` here means the source never
  recovered after the sender came back.

• Step 12: Commit

git add CMakeLists.txt src/source/flatudp src/source/registry.hpp src/source/registry.cpp \
  src/ui_probe.cpp src/ui_app.hpp src/ui.cpp tools/mock_flatudp.py \
  docs/claude/build-and-run.md docs/claude/verification.md docs/claude/probe-readings.md
git commit -m "flatudp: the source, kind, mock sender and stream probes"

---

Task 3: The row-editor form, its probe lines, README and architecture

Amendment (read first). After Task 2 the vector semantics changed: see the "Amendment 2026-11-02" section of docs/superpowers/specs/2026-11-02-flat-udp-source-design.md. flatudp::Row now has a bool wave (default false) and the codec/source already implement channels vs waveform rows and the channel cap. This task's form must therefore: add a "Vector as" GtkDropDown (Channels / Waveform, widget name flatudp-vector) to each row, shown only while that row's count is above 1 (reset to Channels when the count returns to 1); write it into Row::wave; show the channel total in the subtitle (the first problem() text still wins); and extend the form probe with a waveform case (a waveform row beside a scalar row is invalid; two waveform rows of the same count are valid). Where the code below predates this, follow the amendment. README and the docs must describe both modes, the w suffix, the mixed-list rule and the 256 channel cap.

Files:

• Modify (full replacement): src/source/flatudp/flatudp_kind.cpp
• Modify: src/ui_probe.cpp (a form check, called from flatudp_probe)
• Modify: README.md, docs/claude/architecture.md, docs/claude/verification.md, docs/claude/probe-readings.md

Interfaces:

• Consumes: everything from Tasks 1 and 2 (flatudp::Row, Type, type_name, problem, format, datagram_size, FlatUdpSettings::from/problem, valid_ipv4, valid_multicast, kDefaultPort).

• Produces: the same make_flatudp_kind(); the widget names flatudp-add (the "Add signal" button), flatudp-remove (each row's button), which the probe finds with find_named.

• Step 1: Replace the kind with the row-editor form

Overwrite src/source/flatudp/flatudp_kind.cpp with:

// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2026 Martino Ferrari

#include "source/flatudp/flatudp_source.hpp"
#include "source/registry.hpp"

#include <adwaita.h>

#include <cctype>
#include <cstdio>
#include <functional>
#include <memory>
#include <string>
#include <vector>

namespace udpscope {
namespace {

using namespace flatudp;
using ChangeCb = std::shared_ptr<std::function<void()>>;

std::string num(double v) {
  char b[48];
  std::snprintf(b, sizeof b, "%.10g", v);
  return b;
}

std::string trim(const std::string& s) {
  std::size_t a = 0, b = s.size();
  while (a < b && std::isspace(static_cast<unsigned char>(s[a]))) ++a;
  while (b > a && std::isspace(static_cast<unsigned char>(s[b - 1]))) --b;
  return s.substr(a, b - a);
}

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;
}

GtkWidget* spin_row(const char* title, double lo, double hi, double step, double def, int digits) {
  GtkWidget* r = GTK_WIDGET(adw_spin_row_new_with_range(lo, hi, step));
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
  adw_spin_row_set_digits(ADW_SPIN_ROW(r), static_cast<guint>(digits));
  adw_spin_row_set_value(ADW_SPIN_ROW(r), def);
  return r;
}

GtkWidget* combo_row(const char* title, const char* const* items) {
  GtkWidget* r = adw_combo_row_new();
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(r), title);
  adw_preferences_row_set_use_underline(ADW_PREFERENCES_ROW(r), TRUE);
  adw_combo_row_set_model(ADW_COMBO_ROW(r), G_LIST_MODEL(gtk_string_list_new(items)));
  return r;
}

GtkWidget* expander(const char* title, std::initializer_list<GtkWidget*> rows) {
  GtkWidget* e = adw_expander_row_new();
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(e), title);
  for (GtkWidget* r : rows) adw_expander_row_add_row(ADW_EXPANDER_ROW(e), r);
  return e;
}

std::string text_of(GtkWidget* w) { return gtk_editable_get_text(GTK_EDITABLE(w)); }
double value_of(GtkWidget* w) { return adw_spin_row_get_value(ADW_SPIN_ROW(w)); }
guint sel_of(GtkWidget* w) { return adw_combo_row_get_selected(ADW_COMBO_ROW(w)); }

void mark(GtkWidget* w, bool bad) {
  if (bad) {
    gtk_widget_add_css_class(w, "error");
  } else {
    gtk_widget_remove_css_class(w, "error");
  }
}

/// One signal's three editing widgets; `row` is the `AdwEntryRow` that holds the other two.
struct SigRow {
  GtkWidget *row, *type, *count;
};

/// The form's widgets, shared by every signal closure and every `ConfigForm` closure. The raw
/// widget pointers are valid while the chooser keeps the rows.
struct Form : std::enable_shared_from_this<Form> {
  GtkWidget *port, *bind, *multicast, *iface, *rate, *endian, *group;
  std::vector<SigRow> sigs;
  ChangeCb changed = std::make_shared<std::function<void()>>();
  bool updating = false;
  bool alive = true;  ///< cleared when the chooser destroys the rows; handlers then touch nothing

  std::vector<Row> rows_now() const {
    std::vector<Row> rows;
    for (const SigRow& s : sigs) {
      Row r;
      r.name = trim(text_of(s.row));
      r.type = static_cast<Type>(std::min<guint>(gtk_drop_down_get_selected(GTK_DROP_DOWN(s.type)), 9));
      r.count = static_cast<std::uint32_t>(gtk_spin_button_get_value_as_int(GTK_SPIN_BUTTON(s.count)));
      rows.push_back(r);
    }
    return rows;
  }

  /// The first `S<n>` that no row uses.
  std::string next_name() const {
    const std::vector<Row> rows = rows_now();
    for (int n = 1;; ++n) {
      const std::string name = "S" + std::to_string(n);
      bool used = false;
      for (const Row& r : rows) used = used || r.name == name;
      if (!used) return name;
    }
  }

  /// Bad addresses are marked with the `error` style; the group's subtitle carries the signal list's
  /// own state, the first rule it breaks or the size it implies.
  void refresh() {
    mark(bind, !text_of(bind).empty() && !valid_ipv4(text_of(bind)));
    mark(multicast, !text_of(multicast).empty() && !valid_multicast(text_of(multicast)));
    mark(iface, !text_of(iface).empty() && !valid_ipv4(text_of(iface)));
    const std::vector<Row> rows = rows_now();
    const std::string p = flatudp::problem(rows);
    std::string sub = p;
    if (p.empty()) {
      sub = std::to_string(datagram_size(rows)) + " bytes per datagram";
    }
    adw_expander_row_set_subtitle(ADW_EXPANDER_ROW(group), sub.c_str());
  }

  void fire() {
    if (!updating && changed && *changed) (*changed)();
  }

  SourceConfig read() const {
    SourceConfig c;
    c.set("port", num(value_of(port)));
    if (!text_of(bind).empty()) c.set("bind", text_of(bind));
    if (!text_of(multicast).empty()) c.set("multicast", text_of(multicast));
    if (!text_of(iface).empty()) c.set("iface", text_of(iface));
    c.set("rate", num(value_of(rate)));
    c.set("endian", sel_of(endian) == 1 ? "big" : "little");
    c.set("signals", format(rows_now()));
    return c;
  }

  void add_row(const Row& r);
  void remove_row(GtkWidget* row);

  void write(const SourceConfig& cfg) {
    const FlatUdpSettings s = FlatUdpSettings::from(cfg);
    updating = true;
    adw_spin_row_set_value(ADW_SPIN_ROW(port), s.port);
    gtk_editable_set_text(GTK_EDITABLE(bind), s.bind.c_str());
    gtk_editable_set_text(GTK_EDITABLE(multicast), s.multicast.c_str());
    gtk_editable_set_text(GTK_EDITABLE(iface), s.iface.c_str());
    if (s.rate > 0.0) adw_spin_row_set_value(ADW_SPIN_ROW(rate), s.rate);
    adw_combo_row_set_selected(ADW_COMBO_ROW(endian), s.big_endian ? 1 : 0);
    while (!sigs.empty()) {
      GtkWidget* row = sigs.back().row;
      sigs.pop_back();
      adw_expander_row_remove(ADW_EXPANDER_ROW(group), row);
    }
    for (const Row& r : s.rows) add_row(r);
    updating = false;
    refresh();
  }

  bool valid() const { return FlatUdpSettings::from(read()).problem().empty(); }
};

using FormPtr = std::shared_ptr<Form>;

/// A remove button's context: which form, which row.
struct RowCtx {
  FormPtr form;
  GtkWidget* row;
};

void on_notify(GObject*, GParamSpec*, gpointer user) {
  Form& f = **static_cast<FormPtr*>(user);
  if (!f.alive) return;
  f.refresh();
  f.fire();
}

void on_edited(GtkEditable*, gpointer user) {
  Form& f = **static_cast<FormPtr*>(user);
  if (!f.alive) return;
  f.refresh();
  f.fire();
}

void on_spin(GtkSpinButton*, gpointer user) {
  Form& f = **static_cast<FormPtr*>(user);
  if (!f.alive) return;
  f.refresh();
  f.fire();
}

void on_add(GtkButton*, gpointer user) {
  FormPtr f = *static_cast<FormPtr*>(user);
  if (!f->alive) return;
  f->add_row(Row{f->next_name(), Type::F32, 1});
  adw_expander_row_set_expanded(ADW_EXPANDER_ROW(f->group), TRUE);
  f->refresh();
  f->fire();
}

void on_remove(GtkButton*, gpointer user) {
  // Copy first: removing the row finalizes this button, and with it this context.
  RowCtx* c = static_cast<RowCtx*>(user);
  FormPtr f = c->form;
  GtkWidget* row = c->row;
  if (!f->alive) return;
  f->remove_row(row);
}

void free_form(gpointer d, GClosure*) { delete static_cast<FormPtr*>(d); }
void free_ctx(gpointer d, GClosure*) { delete static_cast<RowCtx*>(d); }

/// "destroy" on a row group: the siblings the handlers reach by raw pointer may already be gone.
void on_form_destroy(GtkWidget*, gpointer user) { (**static_cast<FormPtr*>(user)).alive = false; }

void hook_destroy(GtkWidget* w, const FormPtr& f) {
  g_signal_connect_data(w, "destroy", G_CALLBACK(on_form_destroy), new FormPtr(f), free_form,
                        static_cast<GConnectFlags>(0));
}

void hook_notify(GtkWidget* w, const char* signal, const FormPtr& f) {
  g_signal_connect_data(w, signal, G_CALLBACK(on_notify), new FormPtr(f), free_form,
                        static_cast<GConnectFlags>(0));
}

void hook_entry(GtkWidget* w, const FormPtr& f) {
  g_signal_connect_data(w, "changed", G_CALLBACK(on_edited), new FormPtr(f), free_form,
                        static_cast<GConnectFlags>(0));
}

void Form::add_row(const Row& r) {
  FormPtr self = shared_from_this();
  GtkWidget* row = adw_entry_row_new();
  adw_preferences_row_set_title(ADW_PREFERENCES_ROW(row), "Name");
  gtk_editable_set_text(GTK_EDITABLE(row), r.name.c_str());

  const char* names[11];
  for (std::size_t i = 0; i < 10; ++i) names[i] = type_name(static_cast<Type>(i));
  names[10] = nullptr;
  GtkWidget* type = gtk_drop_down_new_from_strings(names);
  gtk_drop_down_set_selected(GTK_DROP_DOWN(type), static_cast<guint>(r.type));
  gtk_widget_set_valign(type, GTK_ALIGN_CENTER);
  gtk_widget_set_name(type, "flatudp-type");

  // The spin clamps to 1..4096, so a hand-edited session with a count outside it is repaired here.
  GtkWidget* count = gtk_spin_button_new_with_range(1, kMaxCount, 1);
  gtk_spin_button_set_value(GTK_SPIN_BUTTON(count), r.count);
  gtk_widget_set_valign(count, GTK_ALIGN_CENTER);
  gtk_widget_set_tooltip_text(count, "Elements: 1 is a scalar signal, more make a vector signal");
  gtk_widget_set_name(count, "flatudp-count");

  GtkWidget* rm = gtk_button_new_with_label("Remove");
  gtk_widget_set_valign(rm, GTK_ALIGN_CENTER);
  gtk_widget_set_name(rm, "flatudp-remove");

  adw_entry_row_add_suffix(ADW_ENTRY_ROW(row), type);
  adw_entry_row_add_suffix(ADW_ENTRY_ROW(row), count);
  adw_entry_row_add_suffix(ADW_ENTRY_ROW(row), rm);

  hook_entry(row, self);
  hook_notify(type, "notify::selected", self);
  g_signal_connect_data(count, "value-changed", G_CALLBACK(on_spin), new FormPtr(self), free_form,
                        static_cast<GConnectFlags>(0));
  g_signal_connect_data(rm, "clicked", G_CALLBACK(on_remove), new RowCtx{self, row}, free_ctx,
                        static_cast<GConnectFlags>(0));

  adw_expander_row_add_row(ADW_EXPANDER_ROW(group), row);
  sigs.push_back(SigRow{row, type, count});
}

void Form::remove_row(GtkWidget* row) {
  for (auto it = sigs.begin(); it != sigs.end(); ++it) {
    if (it->row == row) {
      sigs.erase(it);
      adw_expander_row_remove(ADW_EXPANDER_ROW(group), row);
      break;
    }
  }
  refresh();
  fire();
}

class FlatUdpKind final : public SourceKind {
 public:
  std::string id() const override { return "flatudp"; }
  std::string display_name() const override { return "Flat UDP (MARTe2 UDPSender)"; }
  bool has_datagrams() const override { return true; }

  std::string identity(const SourceConfig& cfg) const override {
    return FlatUdpSettings::from(cfg).identity();
  }

  std::unique_ptr<DataSource> make(const SourceConfig& cfg) const override {
    return std::make_unique<FlatUdpSource>(cfg);
  }

  ConfigForm make_config_form() const override {
    auto f = std::make_shared<Form>();
    static const char* const kEndian[] = {"Little endian", "Big endian", nullptr};

    f->port = spin_row("_Port", 1, 65535, 1, kDefaultPort, 0);
    f->bind = entry_row("_Bind address");
    f->multicast = entry_row("_Multicast group");
    f->iface = entry_row("Multicast _interface");
    f->rate = spin_row("Cycle _rate (Hz)", 0.001, 1000000, 1, 1000, 3);
    f->endian = combo_row("_Byte order", kEndian);

    GtkWidget* net = expander("Network", {f->port, f->bind, f->multicast, f->iface});
    GtkWidget* timing = expander("Timing and format", {f->rate, f->endian});
    f->group = adw_expander_row_new();
    adw_preferences_row_set_title(ADW_PREFERENCES_ROW(f->group), "Signals");
    adw_expander_row_set_expanded(ADW_EXPANDER_ROW(net), TRUE);
    adw_expander_row_set_expanded(ADW_EXPANDER_ROW(f->group), TRUE);

    GtkWidget* add = gtk_button_new_with_label("Add signal");
    gtk_widget_set_valign(add, GTK_ALIGN_CENTER);
    gtk_widget_set_name(add, "flatudp-add");
    adw_expander_row_add_suffix(ADW_EXPANDER_ROW(f->group), add);
    g_signal_connect_data(add, "clicked", G_CALLBACK(on_add), new FormPtr(f), free_form,
                          static_cast<GConnectFlags>(0));

    for (GtkWidget* w : {net, timing, f->group}) hook_destroy(w, f);
    for (GtkWidget* w : {f->bind, f->multicast, f->iface}) hook_entry(w, f);
    for (GtkWidget* w : {f->port, f->rate}) hook_notify(w, "notify::value", f);
    hook_notify(f->endian, "notify::selected", f);
    f->refresh();

    ConfigForm out;
    out.rows = {net, timing, f->group};
    out.description =
        "The signal list must match the sender's exactly, in order: the datagrams carry no "
        "description, and a datagram of any other size is dropped";
    out.focus = f->rate;
    out.write = [f](const SourceConfig& c) { f->write(c); };
    out.read = [f]() { return f->read(); };
    out.valid = [f]() { return f->valid(); };
    ChangeCb cb = f->changed;
    out.on_change = [cb](std::function<void()> fn) { *cb = std::move(fn); };
    return out;
  }
};

}  // namespace

std::unique_ptr<SourceKind> make_flatudp_kind() { return std::make_unique<FlatUdpKind>(); }

}  // namespace udpscope

Notes for the implementer: adw_expander_row_remove and adw_entry_row_add_suffix are in libadwaita 1.8. std::min needs <algorithm>; add it to the includes if the compiler asks. The remove handler removes the row from inside its own button's clicked emission: GTK keeps the emitting instance alive until the emission ends, which is why on_remove copies the form and the row out of its context first. If the probe in step 3 shows a Gtk-CRITICAL or a crash on remove, defer the removal with g_idle_add (carrying the same two values) instead of changing the ownership scheme.

• Step 2: Add the form check to the probe

In src/ui_probe.cpp, in the anonymous namespace after flatudp_breaks, add:

/// The connect form's validity rules and its signal-row editor, driven through the widgets: the
/// empty form is not connectable; a good list is, and its subtitle names the implied size; a
/// duplicate name and an over-size list are not; "Add signal" appends `S1`, `S2`; a row's
/// "Remove" button takes that row out.
void flatudp_form_probe() {
  SourceKind* k = find_kind("flatudp");
  if (k == nullptr) {
    std::fprintf(stderr, "[flatudp] form: no flatudp kind\n");
    return;
  }
  ConfigForm f = k->make_config_form();
  const bool initial = f.valid();
  const auto with = [&](const char* signals) {
    SourceConfig c;
    c.set("rate", "1000");
    c.set("signals", signals);
    f.write(c);
    return f.valid();
  };
  const auto subtitle = [&]() {
    const char* t = adw_expander_row_get_subtitle(ADW_EXPANDER_ROW(f.rows[2]));
    return std::string(t != nullptr ? t : "");
  };
  const bool good = with("A:uint8,B:float32,W:int16[4]");
  const bool size_ok = subtitle().find("13 bytes") != std::string::npos;
  const bool dup = with("A:uint8,A:float32");
  const bool dup_ok = subtitle().find("twice") != std::string::npos;
  const bool big = with("A:float64[4096],B:float64[4096]");
  with("A:uint8,B:float32");

  GtkWidget* add = find_named(f.rows[2], "flatudp-add");
  if (add != nullptr) {
    g_signal_emit_by_name(add, "clicked");
    g_signal_emit_by_name(add, "clicked");
  }
  const bool added = f.read().get("signals") == "A:uint8,B:float32,S1:float32,S2:float32";
  GtkWidget* rm = find_named(f.rows[2], "flatudp-remove");  // the first row's
  if (rm != nullptr) g_signal_emit_by_name(rm, "clicked");
  const bool removed = f.read().get("signals") == "B:float32,S1:float32,S2:float32";
  const bool ok = !initial && good && size_ok && !dup && dup_ok && !big && add != nullptr && added &&
                  rm != nullptr && removed && f.valid();
  std::fprintf(stderr,
               "[flatudp] form: initial_valid=%d good_valid=%d size_ok=%d dup_valid=%d dup_msg=%d "
               "oversize_valid=%d add=%d remove=%d ok=%d\n",
               initial ? 1 : 0, good ? 1 : 0, size_ok ? 1 : 0, dup ? 1 : 0, dup_ok ? 1 : 0,
               big ? 1 : 0, added ? 1 : 0, removed ? 1 : 0, ok ? 1 : 0);
  // The rows were never parented; sink and release them so the form leaves nothing behind.
  for (GtkWidget* row : f.rows) {
    g_object_ref_sink(row);
    g_object_unref(row);
  }
}

In flatudp_probe, directly after the [flatudp] codec: fprintf, add the call:

  flatudp_form_probe();

The anonymous-namespace helper find_named is defined earlier in the file (before source_panel_content), so it is in scope.

• Step 3: Build and run the form check (no sender needed)

cmake --build build -j8 2>&1 | grep -E "warning|error" | head
UDPSCOPE_PROBE_SOURCE_KINDS=60000 UDPSCOPE_PROBE_FLATUDP=1500 UDPSCOPE_EXIT_AFTER_MS=3000 \
  timeout 30 xvfb-run -a build/udpscope --source probe --source-opt rate=500 2>&1 \
  | grep -E "\[flatudp\]|Gtk-CRITICAL|Gtk-WARNING|gtk warnings"

Expected: no build warning from flatudp;

[flatudp] codec: ok=1
[flatudp] form: initial_valid=0 good_valid=1 size_ok=1 dup_valid=0 dup_msg=1 oversize_valid=0 add=1 remove=1 ok=1
[flatudp] no flatudp source (run with --source flatudp)
gtk warnings: 0

and no Gtk-CRITICAL. A Gtk-CRITICAL here is a widget used after removal or a row removed from the wrong container: fix it, do not filter it out.

Then the [source-kinds] round trip once more (the row editor changed how signals is written) and one full mock run (the vector list of Task 2, step 8, shortened is enough: UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000, no gap) to confirm the stream lines are unchanged:

UDPSCOPE_PROBE_SOURCE_KINDS=2000 UDPSCOPE_EXIT_AFTER_MS=5000 timeout 30 \
  xvfb-run -a build/udpscope --source probe --source-opt rate=500 2>&1 | grep -E "form flatudp|kinds=|gtk warnings|Gtk-CRITICAL"
python3 tools/mock_flatudp.py --port 44588 --rate 1000 --signals Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] --bad-every 1 &
MOCK=$!
sleep 1
UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000 timeout 30 xvfb-run -a build/udpscope --source flatudp \
  --source-opt port=44588 --source-opt rate=1000 --source-opt signals=Trigger:uint8,Ia:float32,Cnt:uint64,Wave:int16[4] 2>&1 \
  | grep -E "\[flatudp\]|Gtk-CRITICAL|gtk warnings"
kill $MOCK 2>/dev/null; wait $MOCK 2>/dev/null

Expected: form flatudp: round_trip=1 valid=1, kinds=udps,e1608,flatudp,probe; in the mock run every line ok=1 (the form: line included) and gtk warnings: 0.

• Step 4: Multicast loopback (if the sandbox allows it)

python3 tools/mock_flatudp.py --port 44591 --rate 1000 --signals Trigger:uint8,Ia:float32 \
  --multicast 239.255.42.99 --mcast-if 127.0.0.1 &
MOCK=$!
sleep 1
UDPSCOPE_PROBE_FLATUDP=2500 UDPSCOPE_EXIT_AFTER_MS=5000 timeout 30 xvfb-run -a build/udpscope --source flatudp \
  --source-opt port=44591 --source-opt rate=1000 --source-opt multicast=239.255.42.99 --source-opt iface=127.0.0.1 \
  --source-opt signals=Trigger:uint8,Ia:float32 2>&1 | grep -E "\[flatudp\]|Gtk-CRITICAL|gtk warnings"
kill $MOCK 2>/dev/null; wait $MOCK 2>/dev/null

Expected: stream: connected=1 ... ok=1 and continuity ... ok=1. If the sandbox or the lo interface refuses multicast (ip link show lo lacks MULTICAST, the join fails with "cannot join multicast group ..." or stream stays connected=0), record "multicast not verified here" in the commit message and in the README paragraph (step 5); do not claim it works.

• Step 5: README

In README.md's "Data sources" section, change the sentence "--help lists the registered kinds." paragraph's neighbours as follows. After the E-1608 paragraph that ends "...not why." (before "A session file written before kinds existed"), insert:

The third built-in kind is **Flat UDP**, a receiver for the MARTe2 `UDPSender` datasource: one datagram per
cycle, the signals concatenated in declaration order with no header, no padding and no timestamp, in the
sender's byte order. The datagram describes nothing, so **the signal list in the connection form is the
whole contract and is your responsibility**: one row per signal (a name, a type from `uint8` to `float64`,
and an element count; a count above 1 is a vector signal), in the sender's order, including the `Trigger`
byte an IndependentThread `UDPSender` puts first. The form shows the size the list implies ("41 bytes per
datagram"); a datagram of any other size is dropped, counted as stale, and reported once per distinct size
on stderr as `[flatudp] ...`. The form also takes the port (44488, the sender's own default), an optional bind
address, an optional multicast group with the interface to join it on, the byte order, and the sender's
**cycle rate in Hz, which is required**: it sizes the ring and the ingest decimation and spaces the samples
within a frame. Time is stamped by the existing pipeline from arrival and that declared rate (and
`--relative-time` gives a counted clock); there is no timestamp in the stream to prefer. **Loss is never
reported** -- with no counter in the datagram, a missed cycle is indistinguishable from a slow one -- and
neither is a changed sender: a list that does not match reads as "everything dropped". Source-address
filtering, detecting the list from the traffic and importing a MARTe2 `.cfg` are not provided. A list of
scalars is delivered in frames of about 10 ms (`round(rate / 100)` cycles, at most 1024; a partial frame is
flushed after 100 ms); a list with any vector signal delivers one frame per datagram. The source is
connected on the first valid datagram and disconnected after 2 s without one. It is verified against
`tools/mock_flatudp.py`, not against a MARTe2 application. *(Multicast: state here whether the loopback run of
the plan's step 4 passed or was not verifiable.)*

Replace the italic parenthetical in the inserted text with the real outcome of step 4 (either "Multicast was verified on a loopback group." or "Multicast joins were not verified here: the sandbox's loopback has no multicast route."). Then change the paragraph "Planned, each as its own sub-project: a flat UDP kind and ABScope. ABScope has more than one timebase, ..." to start "Planned, as its own sub-project: ABScope. It has more than one timebase, ...", keeping the rest of the sentence.

• Step 6: Architecture notes

In docs/claude/architecture.md, after the E-1608 bullet about E1608Readouts / stderr (the last E-1608 bullet in the source-plugins list), add:

- **The flat UDP kind is three files and no thread.** `flatudp_signals` is the `signals` string codec,
  the validity rules and `decode_datagram`, with no GTK and no sockets (the probe round-trips it);
  `flatudp_source` is `FlatUdpSettings` plus the `DataSource`; `flatudp_kind` is the form. The datagram has
  no description, so a list that is wrong is invisible except as dropped datagrams: the size check is the
  only defence, and it is why a wrong-size datagram is counted (`stale`) and reported once per size.
- Its frames are **fixed-size blocks** like the E-1608's for the same reason (`build_stamps` warns when the
  sample count varies): `round(rate / 100)` datagrams for an all-scalar list, a partial block flushed after
  100 ms. A vector signal carries one sample per `Frame` (`Frame::value` indexes by sample only when the
  value count equals `num_samples` and the signal has one element), so **one vector row forces
  one datagram per frame**. `rate` is required because `sampling_rate = 0` makes the model size the ring and
  the ingest decimation for 1 kHz; there is no time-base switch, since `acquisition.cpp` has no stamping
  path that ignores the declared cadence.
- `Disconnected` after 2 s without a *valid* datagram (wrong-size ones do not refresh the clock); it resets
  `Acquisition`'s per-session state, so the next datagram re-anchors rather than reading as a gap, and the
  history survives (`CONFIG` is sent once, from the first `poll()`).

• Step 7: Verification and probe-reading docs

docs/claude/verification.md: in the UDPSCOPE_PROBE_FLATUDP entry added in Task 2, change "a hand-computed codec and decode check (it needs no source)" to "a hand-computed codec and decode check and the connect form's validity rules and signal-row editor (both need no source)".

docs/claude/probe-readings.md: in the ## [flatudp] block add, after the codec line:

[flatudp] form: initial_valid=0 good_valid=1 size_ok=1 dup_valid=0 dup_msg=1 oversize_valid=0 add=1 remove=1 ok=1

and, after the codec: ok=1 bullet, this bullet:

- `form: ... ok=1`: driven through the widgets, not the settings. `initial_valid=0` -- an empty list is not
  connectable (`1` is a form that lets a source with no signals through). `good_valid=1 size_ok=1` -- a
  13-byte list is valid and its subtitle says `13 bytes per datagram`. `dup_valid=0 dup_msg=1` -- a duplicate
  name is refused *and the subtitle says why* ("used twice"). `oversize_valid=0` -- two `float64[4096]` rows
  (65536 bytes) exceed a datagram. `add=1` -- two clicks on "Add signal" append `S1` and `S2`; `remove=1` --
  the first row's "Remove" button takes that row out. A `0` on `add`/`remove` with a `Gtk-CRITICAL` is a row
  handled after it was removed. Like `codec`, it needs no flatudp source.

• Step 8: Final baselines and commit

A clean build must show exactly the 3 baseline warnings. Build from scratch in a scratch directory so the main build tree is untouched:

cmake -S . -B /tmp/flatudp-clean >/dev/null && cmake --build /tmp/flatudp-clean -j8 2>&1 | grep -c "warning:"
rm -rf /tmp/flatudp-clean

Expected: 3. Anything else, read the new warning: it comes from a flatudp file.

git add src/source/flatudp/flatudp_kind.cpp src/ui_probe.cpp README.md \
  docs/claude/architecture.md docs/claude/verification.md docs/claude/probe-readings.md
git commit -m "flatudp: the signal-row editor form, its probe lines, README and architecture notes"

Then ask the user whether to merge feature/source-flatudp into feature/source-plugins; do not merge or push on your own.