MARTe-Integrated-Components/docs/superpowers/plans/2026-06-25-streaming-chain-e2e.md

# Streaming-Chain E2E Suite Implementation Plan

> **For agentic workers:** Implement task-by-task. Steps use checkbox (`- [ ]`) syntax.

**Goal:** A curated full-chain E2E suite (MARTe2 UDPStreamer → StreamHub → client)
with waveform validation across a covering config matrix, all trigger/zoom/window
client checks, a Qt GUI smoke test, and one PDF + `results.json` report.

**Architecture:** New `Test/E2E/chain/`. Shell orchestrator drives a Python
matrix that, per scenario, generates `input.bin` + MARTe `.cfg` + StreamHub
`.cfg`, launches the two-process stack, runs a Go mock client, validates
waveforms, and aggregates a Typst PDF + JSON. Mirrors `Test/E2E/datasources/`.

**Tech Stack:** Python 3 (numpy, matplotlib, struct), Go (gorilla/websocket),
MARTe2 (MARTeApp + FileReader/IOGAM/LinuxTimer/FileWriter + UDPStreamer),
StreamHub.ex, Typst, Qt5/6 QTest.

## Global Constraints

- Artifacts go to `Build/x86-linux/E2E/chain/` and `/tmp/chain_e2e/`; **never**
  write generated cfg/data/plots into the source tree.
- Reuse `Test/E2E/datasources/validate_binary.py` read-helpers and the
  `run_recorder_e2e.sh` two-process orchestration pattern verbatim where possible.
- MARTe2 binary FileReader format: header `[u32 numSigs]` then per signal
  `[u16 TypeDescriptor.all][32B name][u32 numElements]`, then raw rows in signal
  order, each value little-endian in its native width.
- MARTe2 `TypeDescriptor.all` codes (verified needed values): uint8=0x0408,
  int8=0x0008, uint16=0x0410, int16=0x0010, uint32=0x0420, int32=0x0020,
  uint64=0x0440, int64=0x0040, float32=0x0808 (2056), float64=0x0810. Confirm
  each against a MARTe round-trip in Task 2.
- StreamHub config keys: `WSPort`, `MaxPoints`, `PushRate`, `MaxPushPoints`,
  `RingTemporal`, `RingScalar`, `Sources = { id = { Label Addr Port
  [MulticastGroup DataPort] } }`.
- UDPStreamer keys (verified): `Port`, `MulticastGroup`, `DataPort`,
  `MaxPayloadSize`, `PublishingMode ∈ {Strict,Accumulate,Decimate}`,
  `MinRefreshRate` (Accumulate), `Ratio` (Decimate); per-signal `Type`,
  `NumberOfDimensions`, `NumberOfElements`, `TimeMode ∈
  {PacketTime,FullArray,FirstSample,LastSample}`, `TimeSignal`, `SamplingRate`,
  `QuantizedType ∈ {none,uint8,int8,uint16,int16}`, `RangeMin`, `RangeMax`,
  `Unit`.
- Each scenario runs under `timeout`; always tear the stack down via a bash trap.
- Run `source env.sh` for all build/run steps; set `LD_LIBRARY_PATH` as in
  `run_recorder_e2e.sh`.

---

### Task 1: Scenario model + a 3-scenario starter set

**Files:**
- Create: `Test/E2E/chain/scenarios.py`

**Interfaces:**
- Produces: `SCENARIOS: list[dict]` and `validate_scenario(s) -> list[str]`
  (returns list of validity errors, empty if valid). Each scenario dict:
  ```python
  {
    "id": "s01_scalar_uint32",          # unique slug
    "network": "unicast"|"multicast",
    "publishing": "Strict"|"Accumulate"|"Decimate",
    "ratio": int|None,                   # Decimate
    "min_refresh_hz": float|None,        # Accumulate
    "max_payload": int,                  # bytes
    "ws_port": int, "udp_port": int, "data_port": int|None,
    "multicast_group": str|None,
    "sources": [                          # 1+ sources
      {"id": "src", "signals": [
        {"name","type","elements","time_mode","time_signal",
         "sampling_rate","quant","range_min","range_max","unit",
         "formula"}                       # formula: "sine"|"ramp"|"counter"
      ]}
    ],
    "oracle": "analytic"|"fed"|"both",
    "client_checks": ["live","zoom","window","trigger"],
  }
  ```

- [ ] **Step 1: Write `scenarios.py`** with the dict schema above, a
  `validate_scenario()` enforcing the rules (quant only on float32/float64;
  FullArray needs a matching-length `TimeSignal`; First/LastSample need a scalar
  `TimeSignal` + `sampling_rate>0`; Decimate needs `ratio>=1`; Accumulate needs
  `min_refresh_hz>0`; multicast needs `multicast_group`+`data_port`), and an
  initial 3 scenarios: `s01_scalar_uint32` (Strict unicast),
  `s02_array_float32_fullarray` (Strict unicast, 100-elem + uint64 ns time
  array), `s03_quant_uint16` (float32 quantised, RangeMin/Max). Give each a
  distinct `ws_port`/`udp_port` to allow isolation.

- [ ] **Step 2: Self-check the model**

Run: `python3 -c "import sys; sys.path.insert(0,'Test/E2E/chain'); import scenarios as S; [print(s['id'], S.validate_scenario(s)) for s in S.SCENARIOS]"`
Expected: each line prints the id and `[]` (no errors).

- [ ] **Step 3: Commit**
```bash
git add Test/E2E/chain/scenarios.py
git commit -m "test(e2e-chain): scenario model + starter scenarios"
```

---

### Task 2: Data generator (`gen_data.py`) — all types/shapes

**Files:**
- Create: `Test/E2E/chain/gen_data.py`
- Reference: `Test/E2E/datasources/gen_test_data.py` (format), `validate_binary.py`

**Interfaces:**
- Consumes: a scenario dict (Task 1).
- Produces CLI `python3 gen_data.py --scenario <id> --out <path>` writing
  `input_<id>.bin`; and a Python API
  `write_input(scenario, path) -> dict` returning, per `src:signal`, the
  ground-truth arrays `{"t":[...], "v":[...native...], "formula":..., "dt":...}`
  so the validator can reconstruct truth without re-deriving layout.
- Type code map `TYPE_CODES: dict[str,int]` and numpy dtype map `NP_DTYPE`.

- [ ] **Step 1: Write `gen_data.py`** — for the scenario's signals, build
  `NUM_ROWS` rows (default 200) of deterministic values:
  - `counter`: `r*elements + c` cast to the signal type.
  - `ramp`: linear in `(r,c)`.
  - `sine`: `A*sin(2π f (r*dt + c*dt_elem))` with per-signal frequency; for time
    signals (`unit ns/us`) emit the matching timestamp array/scalar.
  Write the MARTe binary: `[u32 numSigs]` + per-signal `[u16 TypeDescriptor.all]
  [32B name][u32 elements]` + rows, each value packed little-endian at native
  width via `numpy.tobytes()`. Return the ground-truth dict.

- [ ] **Step 2: Round-trip self-test**

Run: `python3 -c "import sys;sys.path.insert(0,'Test/E2E/chain');import scenarios as S,gen_data as G;G.write_input(S.SCENARIOS[0],'/tmp/chain_e2e/t.bin');import validate_binary" 2>&1; ls -l /tmp/chain_e2e/t.bin`
Expected: file exists, size = header + 200 rows × rowbytes.

- [ ] **Step 3: Verify a type round-trip through MARTe** — generate a uint32
  scalar file, point a tiny `FileReader→FileWriter` cfg at it (reuse datasources
  cfg pattern, single signal), run MARTeApp 3 s, confirm output rows equal input
  rows (proves the `TypeDescriptor.all` code is correct for that type). Repeat
  mentally/quickly for one int and one float type.

Run: (the orchestrator does this in Task 7; here just confirm bytes are sane)
`python3 -c "import struct;d=open('/tmp/chain_e2e/t.bin','rb').read();print(struct.unpack_from('<I',d,0))"`
Expected: prints the signal count.

- [ ] **Step 4: Commit**
```bash
git add Test/E2E/chain/gen_data.py
git commit -m "test(e2e-chain): deterministic typed/shaped data generator"
```

---

### Task 3: Config generator (`gen_cfg.py`)

**Files:**
- Create: `Test/E2E/chain/gen_cfg.py`
- Reference: `Test/E2E/datasources/E2ETest.cfg`, `Test/E2E/recorder/StreamHubRec.cfg`

**Interfaces:**
- Consumes: a scenario dict.
- Produces: `write_marte_cfg(scenario, path, input_bin, tap_bin|None)` and
  `write_hub_cfg(scenario, path)`. The MARTe cfg wires
  `LinuxTimer + FileReader(input_bin) -> IOGAM -> UDPStreamer(s)`; when
  `oracle in {fed,both}` it adds a second IOGAM branch `-> FileWriter(tap_bin)`
  of the same fed signals. The hub cfg declares one `Source` per UDPStreamer
  (with `MulticastGroup`/`DataPort` for multicast) on the scenario's `ws_port`.

- [ ] **Step 1: Write `gen_cfg.py`** generating both cfgs as strings from the
  scenario, honouring publishing mode keys, per-signal type/shape/timemode/quant,
  unicast vs multicast, payload, multi-source. Keep one GAM thread.

- [ ] **Step 2: Self-test cfg generation**

Run: `python3 -c "import sys;sys.path.insert(0,'Test/E2E/chain');import scenarios as S,gen_cfg as C;C.write_marte_cfg(S.SCENARIOS[0],'/tmp/chain_e2e/m.cfg','/tmp/chain_e2e/t.bin',None);C.write_hub_cfg(S.SCENARIOS[0],'/tmp/chain_e2e/h.cfg');print(open('/tmp/chain_e2e/h.cfg').read())"`
Expected: prints a valid-looking StreamHub cfg with `Sources` and the right port.

- [ ] **Step 3: Validate cfgs actually load** — run the full stack for scenario 1
  (StreamHub.ex + MARTeApp) for 5 s; confirm StreamHub log shows
  `session '...' started` and `Connected`. (Manual via Task 7 harness; here just
  eyeball the generated cfg text.)

- [ ] **Step 4: Commit**
```bash
git add Test/E2E/chain/gen_cfg.py
git commit -m "test(e2e-chain): MARTe + StreamHub config generator"
```

---

### Task 4: Go mock client — record + waveform + behavioural checks

**Files:**
- Create: `Test/E2E/chain/client/main.go`, `Test/E2E/chain/client/go.mod`
- Reference: `Test/E2E/streamhub/main.go` (parsers + driver to reuse)

**Interfaces:**
- CLI: `chain-client -hub host:port -scenario <id> -trigsig <src:sig>
  -trigthr <f> -checks live,zoom,window,trigger -out <dir> -dur <sec>`.
- Produces: `received_<id>.bin` (MARTe-binary-compatible: same header + rows
  re-assembled from v1 pushes per signal, time-ordered) and
  `checks_<id>.json` = `{ "live":{ok,frames,signals}, "zoom":[{range,n,inrange,
  pts}], "window":{...}, "trigger":[{edge,mode,fired,trigTime,preSec,postSec,
  capturePts,edgeOk}] }`.

- [ ] **Step 1: Copy the wire parsers** (`parsePush`, `parseCapture`, event
  types, `pump`/`waitFor`) from `Test/E2E/streamhub/main.go` into the new client;
  reuse `go.mod` deps (gorilla/websocket).

- [ ] **Step 2: Implement recording + live check** — accumulate v1 pushes for
  `dur` seconds; per `src:sig`, sort samples by time, dedupe; write
  `received_<id>.bin`; assert ≥10 frames and monotonic wall-clock time.

- [ ] **Step 3: Implement zoom + window checks** — issue zoom at a narrow and a
  wide range over the observed window; assert points within range and count ≤ n;
  record results. If historyInfo enabled, also historyZoom.

- [ ] **Step 4: Implement trigger matrix** — for each `edge ∈
  {rising,falling,both}` × `mode ∈ {normal,single}`: `setTrigger` on `-trigsig`
  at `-trigthr`, `arm`, wait for triggerState + v2 capture; record
  `trigTime/preSec/postSec`, verify capture window bounds and that the trigger
  signal crosses threshold in the correct direction near `trigTime`. For
  `normal` confirm re-arm (a second capture); for `single` confirm no second
  capture until `rearm`. Write `checks_<id>.json`.

- [ ] **Step 5: Build the client**

Run: `cd Test/E2E/chain/client && go build -o chain-client .`
Expected: builds, produces `chain-client`.

- [ ] **Step 6: Commit**
```bash
git add Test/E2E/chain/client
git commit -m "test(e2e-chain): Go mock client (record + zoom/window/trigger)"
```

---

### Task 5: Waveform validator (`validate_waveform.py`)

**Files:**
- Create: `Test/E2E/chain/validate_waveform.py`

**Interfaces:**
- Consumes: scenario dict, ground-truth dict (from `gen_data.write_input`),
  `received_<id>.bin`, optional `tap_<id>.bin`, `checks_<id>.json`.
- Produces: CLI `python3 validate_waveform.py --scenario <id> --truth <bin>
  --received <bin> [--tap <bin>] --checks <json> --out <metrics.json>`; returns
  exit 0 pass / 1 fail. `metrics_<id>.json` = per-signal
  `{max_abs_err, quant_step, rmse, corr, n_truth, n_recv, oracle, pass}` plus
  the client checks rolled in and an overall `pass`.

- [ ] **Step 1: Write the validator** — read received rows (reuse
  `validate_binary.read_binary`), reconstruct truth on the received timestamps
  via the ground-truth formula; apply the per-effect tolerance from the design
  (bit-exact no-quant; `quant_step/2` quantised; decimation stride; LTTB shape
  metric corr≥0.99 & nRMSE≤0.05). For `fed`/`both`, also compare received vs tap.

- [ ] **Step 2: Self-test on a synthetic pair** — craft a truth array and a
  copy with added quant noise ≤ step/2 → expect pass; noise > step → expect fail.

Run: `python3 Test/E2E/chain/validate_waveform.py --selftest`
Expected: prints `selftest OK` and exits 0.

- [ ] **Step 3: Commit**
```bash
git add Test/E2E/chain/validate_waveform.py
git commit -m "test(e2e-chain): per-effect waveform validator"
```

---

### Task 6: Plots (`plots.py`)

**Files:**
- Create: `Test/E2E/chain/plots.py`
- Reference: plotting block in `Test/E2E/datasources/run_e2e_report.sh`

**Interfaces:**
- Consumes: `input_<id>.bin`, `received_<id>.bin`, `metrics_<id>.json`,
  `checks_<id>.json`.
- Produces: `python3 plots.py --scenario <id> --dir <artifactdir>` writing
  `wave_<id>.png` (truth/received/diff), `trig_<id>.png` (captures),
  `zoom_<id>.png`.

- [ ] **Step 1: Write `plots.py`** (matplotlib Agg) producing the three PNGs;
  skip gracefully if an input is missing (write a placeholder note).

- [ ] **Step 2: Self-test**

Run: `python3 Test/E2E/chain/plots.py --scenario s01_scalar_uint32 --dir /tmp/chain_e2e 2>&1 | tail -3`
Expected: prints the PNG paths (or graceful "missing data" notes).

- [ ] **Step 3: Commit**
```bash
git add Test/E2E/chain/plots.py
git commit -m "test(e2e-chain): report plot generation"
```

---

### Task 7: Orchestrator (`run_chain_e2e.sh`) + results.json

**Files:**
- Create: `Test/E2E/chain/run_chain_e2e.sh`
- Reference: `Test/E2E/recorder/run_recorder_e2e.sh` (stack + LD_LIBRARY_PATH)

**Interfaces:**
- Consumes: all of the above.
- Produces: `Build/x86-linux/E2E/chain/results.json` (aggregate) and per-scenario
  artifacts; orchestrates build, the two-process stack, client, validation, plots.

- [ ] **Step 1: Write `run_chain_e2e.sh`** — flags `--skip-build`,
  `--only <id>`, `--pdf-only`. Source `env.sh`; set `LD_LIBRARY_PATH` as in
  `run_recorder_e2e.sh` (+ FileDataSource, IOGAM, LinuxTimer). Build UDPStream,
  UDPStreamer, StreamHub, and `chain-client`. For each scenario (a Python helper
  emits the list): gen data + cfgs; start `StreamHub.ex`; start `MARTeApp`; run
  `chain-client`; stop stack via trap; run `validate_waveform.py` and `plots.py`;
  append to `results.json`. Multicast scenarios: probe a loopback multicast
  route, else mark SKIP.

- [ ] **Step 2: Run the starter 3 scenarios end-to-end**

Run: `source env.sh && ./Test/E2E/chain/run_chain_e2e.sh --skip-build --only s01_scalar_uint32`
Expected: stack starts, client connects, `metrics_s01*.json` written with
overall `pass=true`; `results.json` contains the scenario.

- [ ] **Step 3: Run all 3 starter scenarios**

Run: `source env.sh && ./Test/E2E/chain/run_chain_e2e.sh --skip-build`
Expected: all 3 pass; `results.json` overall status PASS.

- [ ] **Step 4: Commit**
```bash
git add Test/E2E/chain/run_chain_e2e.sh
git commit -m "test(e2e-chain): orchestrator + results.json aggregation"
```

---

### Task 8: Expand to the full ~50-scenario covering matrix

**Files:**
- Modify: `Test/E2E/chain/scenarios.py`

- [ ] **Step 1: Add scenarios** filling the buckets in the design (per-type
  scalars; time-mode sweep; quant sweep; publishing modes; multicast variants;
  shapes/fragmentation; multi-source; edge cases) until every option value is
  covered ≥1×. Keep ports unique per scenario.

- [ ] **Step 2: Validate the whole matrix model**

Run: `python3 -c "import sys;sys.path.insert(0,'Test/E2E/chain');import scenarios as S;bad=[(s['id'],e) for s in S.SCENARIOS for e in [S.validate_scenario(s)] if e];print('INVALID',bad) if bad else print('OK',len(S.SCENARIOS),'scenarios')"`
Expected: `OK <N> scenarios` with N in 40–70.

- [ ] **Step 3: Run a representative subset live** (one per bucket via
  `--only`), fixing generator/cfg/validator gaps surfaced by real runs.

- [ ] **Step 4: Full run**

Run: `source env.sh && ./Test/E2E/chain/run_chain_e2e.sh`
Expected: suite completes; `results.json` shows the matrix; failures (if any)
are real findings, recorded per-scenario.

- [ ] **Step 5: Commit**
```bash
git add Test/E2E/chain/scenarios.py
git commit -m "test(e2e-chain): full covering scenario matrix"
```

---

### Task 9: Typst report (`E2E_Report.typ`)

**Files:**
- Create: `Test/E2E/chain/E2E_Report.typ`
- Reference: `Test/E2E/datasources/E2E_Report.typ`
- Modify: `run_chain_e2e.sh` (compile step)

- [ ] **Step 1: Write `E2E_Report.typ`** — title/env/summary table driven from
  `results.json` (read via a small Python pre-step that emits a `.typ` data
  include or a CSV the template loads), embed `wave_*/trig_*/zoom_*` PNGs for a
  representative subset, list per-scenario pass/fail.

- [ ] **Step 2: Wire PDF compile** into `run_chain_e2e.sh` (copy `.typ` to
  `OUT_DIR`, `typst compile`), guarded by `command -v typst`.

- [ ] **Step 3: Build the PDF**

Run: `source env.sh && ./Test/E2E/chain/run_chain_e2e.sh --pdf-only`
Expected: `Build/x86-linux/E2E/chain/E2E_Report.pdf` exists.

- [ ] **Step 4: Commit**
```bash
git add Test/E2E/chain/E2E_Report.typ Test/E2E/chain/run_chain_e2e.sh
git commit -m "test(e2e-chain): Typst PDF report"
```

---

### Task 10: Qt GUI smoke test

**Files:**
- Create: `Client/streamhub-qt/test/smoke_test.cpp`
- Modify: `Client/streamhub-qt/CMakeLists.txt` (add `enable_testing()` + a
  `QTest` executable + `add_test`)

**Interfaces:**
- Consumes: a live StreamHub on a port from env var `SHQ_TEST_HUB`
  (default `127.0.0.1:8090`).

- [ ] **Step 1: Write `smoke_test.cpp`** (`QTest`, `QT_QPA_PLATFORM=offscreen`):
  construct `MainWindow`, connect, spin the event loop a few hundred ms, assert
  `hub.sources()` non-empty and ≥1 repaint/tick occurred, exercise a layout
  change + pause toggle; pass if no crash and a source was seen (skip-pass if no
  hub, to keep it CI-safe).

- [ ] **Step 2: Add the test target** to `CMakeLists.txt` linking
  `Qt${QT_VERSION_MAJOR}::Test` + Widgets + WebSockets, reusing the client
  objects; register with `add_test(NAME shq_smoke ...)`.

- [ ] **Step 3: Build + run**

Run: `cd Client/streamhub-qt && cmake -B build && cmake --build build -j4 && QT_QPA_PLATFORM=offscreen ctest --test-dir build --output-on-failure`
Expected: `shq_smoke` passes (or skip-passes with no hub).

- [ ] **Step 4: Commit**
```bash
git add Client/streamhub-qt/test Client/streamhub-qt/CMakeLists.txt
git commit -m "test(streamhub-qt): offscreen QTest GUI smoke"
```

---

### Task 11: Docs + wire-up

**Files:**
- Modify: `CLAUDE.md` (E2E demo scripts line), `ARCHITECTURE.md` (test section)

- [ ] **Step 1: Document** the new suite under the build/test notes: how to run
  `run_chain_e2e.sh`, where artifacts land, what the matrix covers, and the Qt
  ctest smoke.

- [ ] **Step 2: Commit**
```bash
git add CLAUDE.md ARCHITECTURE.md
git commit -m "docs: streaming-chain E2E suite + Qt smoke test"
```

## Self-Review

- **Spec coverage:** matrix (T1,T8), waveform oracle A+B (T2,T5), client
  live/zoom/window/trigger (T4), disk artifact (T4 received_*.bin), Qt smoke
  (T10), PDF + JSON (T7,T9) — all mapped.
- **Type consistency:** scenario dict schema (T1) is consumed unchanged by
  gen_data (T2), gen_cfg (T3), client args (T4), validator (T5); artifact names
  `input_/received_/tap_/checks_/metrics_/wave_/trig_/zoom_<id>` consistent
  across T2–T9.
- **Placeholders:** none; commands and schemas are concrete.