# Stress Suite Report Integration — 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:** Track the existing stress matrix, extend its signal-size axis into the multi-fragment regime, and surface stress results in the E2E PDF report via an opt-in `--stress` flag (table + per-axis scaling curves + regression vs the previous run).

**Architecture:** The stress harness (`stress.py` matrix → `stress_run.py` orchestrator → `stress_results.json`) already exists and is functional. This plan (1) extends `stress.py` with larger multi-fragment packet cases and lifts an outdated 64 KB validation cap, (2) adds a `--stress` phase to `run_chain_e2e.sh` that runs `stress_run.py` after the correctness phase, (3) folds `stress_results.json` into `report_data.json` in `report_build.py` (block + per-axis scaling PNGs + history/regression), and (4) renders a `= Stress Tests` section in `E2E_Report.typ`. When `--stress` is not used, the PDF omits the section.

**Tech Stack:** Python 3 (matplotlib for plots), Bash, Typst. No new dependencies.

## Global Constraints

- Stress data files live in `Test/E2E/chain/`; report artifacts in `Build/x86-linux/E2E/chain/` (+ `stress/` subdir for `stress_results.json`).
- `report_data.json` keys consumed by `E2E_Report.typ` are a contract — adding keys is safe; renaming/removing is not.
- A missing artifact must degrade gracefully (no stress run ⇒ no stress section, no crash) — mirrors the existing `_load(path, default)` pattern in `report_build.py`.
- Per-case producer geometry flows through `scenarios.geometry(case)`; both `gen_data.py` and `gen_cfg.py` already read it, so new `row_dt`/`num_rows`/`producer_hz` values need no generator changes.
- The First/LastSample window constraint `(elements-1)/sampling_rate ≤ row_dt` (enforced by `scenarios.validate_scenario`) must hold for every new case, or the hub ring's time axis is non-monotonic.
- UDPSClient deliverable-packet cap is `UDPS_CLIENT_MAX_PACKET_BYTES = 1048576` (1 MiB), defined in `Source/Components/Interfaces/UDPStream/UDPSClient.h`. Mirror this value (verbatim) as a Python constant in `stress.py`.
- Run from repo root with `source env.sh` already done by the wrapper scripts; standalone Python validators need `cd Test/E2E/chain` first.

---

### Task 1: Extend the stress size axis into the multi-fragment regime (`stress.py`)

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

**Interfaces:**
- Consumes: `scenarios.ROW_DT` (= 1e-3), `scenarios.MAX_UDP_PAYLOAD`, `scenarios.UDPS_HEADER_SIZE`, `scenarios.validate_scenario`, `scenarios.geometry`.
- Produces: extended `STRESS_CASES` list (now 33 cases: the 3 size-axis lists each gain 3 multi-fragment cases — `ds_size_50000/100000/250000`, `hub_size_50000/100000/250000`); `mk_stress(...)` gains `row_dt=None, num_rows=None, producer_hz=None` keyword params; `_source(..., row_dt=S.ROW_DT)` gains a `row_dt` param; module constant `UDPS_CLIENT_MAX_PACKET_BYTES = 1048576`.

- [ ] **Step 1: Write the failing test**

Append to the bottom of `Test/E2E/chain/stress.py`'s `if __name__ == "__main__":` block is NOT where this goes — instead create a standalone check script invocation. Use this exact command as the test (it must fail before the change because the new ids do not exist yet):

Run (from `Test/E2E/chain`):
```bash
python3 -c "import stress as ST; ids={c['id'] for c in ST.STRESS_CASES}; need={'ds_size_50000','ds_size_100000','ds_size_250000','hub_size_50000','hub_size_100000','hub_size_250000'}; missing=need-ids; assert not missing, f'missing {missing}'; print('all multi-fragment size cases present')"
```
Expected: FAIL with `AssertionError: missing {...}`.

- [ ] **Step 2: Add the `row_dt` param to `_source` and the geometry params to `mk_stress`**

In `Test/E2E/chain/stress.py`, change the `_source` signature and rate computation (lines ~73-78):

```python
def _source(sid, n_signals, elements, multicast=False, row_dt=S.ROW_DT):
    """One UDPStreamer source: a uint64 ns anchor + n_signals float32 arrays.

    sampling_rate = elements / row_dt keeps each array's per-cycle window equal to
    one producer cycle (the First/LastSample monotonic-ring constraint) regardless
    of `elements` or a slowed-down producer."""
    rate = elements / row_dt
    sigs = [S._sig("Tns", "uint64", 1, unit="ns", formula="time_ns",
                   is_time=True)]
    sigs += [_f32_arr(f"S{i}", elements, rate) for i in range(n_signals)]
    return {
        "id": sid, "udp_port": next(_udp),
        "data_port": next(_data) if multicast else None,
        "multicast_group": S.MCAST_GROUP if multicast else None,
        "signals": sigs,
    }
```

Then change `mk_stress` to accept and store the geometry overrides (replace the signature line ~95-97 and the three `"row_dt": None, ...` keys ~106):

```python
def mk_stress(sid, axis, level, sources, shape="hub", clients=1, hubs=1,
              reqrate=0.0, dur=6.0, network="unicast", publishing="Strict",
              ratio=None, min_refresh_hz=None, gate=None,
              row_dt=None, num_rows=None, producer_hz=None):
```

and within the returned dict replace:
```python
        "row_dt": None, "num_rows": None, "producer_hz": None,
```
with:
```python
        "row_dt": row_dt, "num_rows": num_rows, "producer_hz": producer_hz,
```

- [ ] **Step 3: Add the 1 MiB constant and the multi-fragment size cases**

In `Test/E2E/chain/stress.py`, after the `F32 = "float32"` line (~60) add:

```python
# Deliverable-packet cap mirrored verbatim from
# Source/Components/Interfaces/UDPStream/UDPSClient.h
# (UDPS_CLIENT_MAX_PACKET_BYTES). A single source's reassembled DATA payload must
# stay under this; the UDPStreamer still fragments it into MaxPayloadSize chunks.
UDPS_CLIENT_MAX_PACKET_BYTES = 1048576  # 1 MiB
```

After the existing `_DS_SIZE = [...]` list (ends ~133) append:

```python
# Multi-fragment size cases: one float32 array large enough that the DATA packet
# spans several UDP datagrams (>64 KB), exercising the UDPSClient reassembly path
# (cap 1 MiB). A slowed-down producer keeps bandwidth realistic:
#   50k≈195 KB @100 Hz≈20 MB/s, 100k≈390 KB @100 Hz≈39 MB/s, 250k≈954 KB @50 Hz≈48 MB/s.
# row_dt sets the producer cycle so the FirstSample window equals one cycle; a small
# num_rows keeps the FileReader input file bounded (50*250k*4 ≈ 50 MB).
_BIG_SIZE = [(50000, 0.01, 100), (100000, 0.01, 100), (250000, 0.02, 50)]

_DS_SIZE += [
    mk_stress(f"ds_size_{e}", "ds_signal_elements", e,
              [_source("src", 1, e, row_dt=rdt)],
              row_dt=rdt, num_rows=50, producer_hz=phz,
              gate=_gate(marte_rss=1024.0, hub_rss=2048.0))
    for (e, rdt, phz) in _BIG_SIZE
]
```

After the existing `_HUB_SIZE = [...]` list (ends ~161) append:

```python
_HUB_SIZE += [
    mk_stress(f"hub_size_{e}", "hub_signal_elements", e,
              [_source("src", 1, e, row_dt=rdt)],
              row_dt=rdt, num_rows=50, producer_hz=phz,
              gate=_gate(marte_rss=1024.0, hub_rss=2048.0))
    for (e, rdt, phz) in _BIG_SIZE
]
```

- [ ] **Step 4: Lift the 64 KB validation cap and update the docstring**

In `Test/E2E/chain/stress.py` `validate_case` (lines ~201-210), replace the single-datagram check:

```python
        if pb >= 65536:
            errs.append(f"{c['id']}: source {src['id']} packet {pb} B exceeds "
                        f"the 64 KB single-datagram cap")
```
with:
```python
        if pb >= UDPS_CLIENT_MAX_PACKET_BYTES:
            errs.append(f"{c['id']}: source {src['id']} packet {pb} B exceeds "
                        f"the {UDPS_CLIENT_MAX_PACKET_BYTES} B deliverable cap "
                        f"(UDPS_CLIENT_MAX_PACKET_BYTES)")
```

Then update the module docstring paragraph (lines ~44-47) from the sub-64 KB note to:

```python
The matrix keeps every datagram a single UDP fragment for the count/rate axes, but
the *size* axis deliberately crosses the 64 KB single-datagram boundary into the
multi-fragment regime: the UDPSClient reassembles up to UDPS_CLIENT_MAX_PACKET_BYTES
(1 MiB) per packet, so the size sweep runs to ~954 KB packets to exercise that path
under load.
```

- [ ] **Step 5: Run the test to verify it passes, plus the full matrix validator**

Run (from `Test/E2E/chain`):
```bash
python3 -c "import stress as ST; ids={c['id'] for c in ST.STRESS_CASES}; need={'ds_size_50000','ds_size_100000','ds_size_250000','hub_size_50000','hub_size_100000','hub_size_250000'}; missing=need-ids; assert not missing, f'missing {missing}'; print('all multi-fragment size cases present')"
python3 stress.py
```
Expected: first command prints `all multi-fragment size cases present`; `stress.py` prints `33 stress cases across 7 axes, ALL VALID` and exits 0 (the new `ds_size_250000`/`hub_size_250000` ≈ 1,000,008 B packets are under the 1 MiB cap and their windows fit `row_dt`).

- [ ] **Step 6: Integration-verify one multi-fragment case end-to-end**

Run (from `Test/E2E/chain`, the chain must already be built — drop `--skip-build` if not):
```bash
./run_stress.sh --skip-build --only ds_size_100000 2>&1 | tail -8
```
Expected: `PASS` for `ds_size_100000` with `frames>=`(some n)≥5 and `survival` true (the 390 KB packet reassembles across ~6 fragments). If it FAILs on RSS, note the actual RSS and raise that case's `gate` ceiling; if it FAILs on frames, the producer is saturating — lower `producer_hz` for that case.

- [ ] **Step 7: Commit**

```bash
git add Test/E2E/chain/stress.py
git commit -m "$(cat <<'EOF'
test(e2e-stress): extend size axis into multi-fragment regime

Add 50k/100k/250k-element size cases (~195 KB–954 KB packets) to the DS and
hub size axes so the stress suite exercises UDPSClient multi-fragment
reassembly under load, and lift the now-outdated 64 KB validation cap to the
1 MiB deliverable cap (UDPS_CLIENT_MAX_PACKET_BYTES). Slowed producers keep
bandwidth realistic.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
EOF
)"
```

---

### Task 2: Add the `--stress` phase to `run_chain_e2e.sh`

**Files:**
- Modify: `Test/E2E/chain/run_chain_e2e.sh`

**Interfaces:**
- Consumes: `stress_run.py` CLI (`--marte --hub --client --work --out`), the same `MARTE_APP`/`STREAMHUB_EX`/`CLIENT` paths and `LD_LIBRARY_PATH` already set up in the script.
- Produces: a `--stress` flag (default off); when set, writes `${OUT_DIR}/stress/stress_results.json` before `report_build.py` runs.

- [ ] **Step 1: Add the `--stress` flag to the arg parser**

In `Test/E2E/chain/run_chain_e2e.sh`, add `STRESS=0` next to the other flag defaults (after line 27 `CPP_COV=0`):

```bash
STRESS=0
```

Add a case to the `while` arg loop (after the `--cpp-coverage)` case, ~line 33):

```bash
        --stress) STRESS=1 ;;
```

Update the usage line (line 34) to:

```bash
        --help|-h) echo "Usage: $0 [--skip-build] [--only <id>] [--pdf-only] [--cpp-coverage] [--stress]"; exit 0 ;;
```

- [ ] **Step 2: Add the stress phase before `report_build.py`**

In `Test/E2E/chain/run_chain_e2e.sh`, immediately before the `# ── Consolidated report data ...` block (the `${PY} "${SCRIPT_DIR}/report_build.py" ...` invocation, ~line 274), insert:

```bash
# ── Stress matrix (opt-in) ───────────────────────────────────────────────────
# Capacity sibling of the correctness phase: sweep one load axis at a time and
# gate survival/liveness (hard) + RSS/zoom-p95 (soft). Writes stress_results.json
# into OUT_DIR/stress, which report_build.py folds into the PDF when present.
STRESS_OUT="${OUT_DIR}/stress"
if [ "${STRESS}" -eq 1 ]; then
    echo ""
    echo "── Stress matrix ──"
    mkdir -p "${STRESS_OUT}"
    ${PY} "${SCRIPT_DIR}/stress.py" >/dev/null || { echo "stress matrix invalid"; exit 1; }
    ${PY} "${SCRIPT_DIR}/stress_run.py" \
        --marte "${MARTE_APP}" --hub "${STREAMHUB_EX}" --client "${CLIENT}" \
        --work "${WORK}" --out "${STRESS_OUT}" || true
fi
```

- [ ] **Step 3: Pass the stress results path to `report_build.py`**

In `Test/E2E/chain/run_chain_e2e.sh`, change the `report_build.py` invocation (~line 275-276) to add the `--stress-results` flag (the flag is added to `report_build.py` in Task 3; passing an absent file is handled gracefully there):

```bash
${PY} "${SCRIPT_DIR}/report_build.py" --repo "${REPO_ROOT}" \
    --results "${OUT_DIR}/results.json" --work "${WORK}" --out "${OUT_DIR}" \
    --stress-results "${OUT_DIR}/stress/stress_results.json" || true
```

- [ ] **Step 4: Update the script header comment**

In `Test/E2E/chain/run_chain_e2e.sh`, update the usage comment (line 13) to:

```bash
# Usage: ./run_chain_e2e.sh [--skip-build] [--only <id>] [--pdf-only] [--cpp-coverage] [--stress]
```

and add one line after it (line 13) documenting the flag:

```bash
# --stress also runs the capacity matrix (stress.py / stress_run.py) and embeds a
# Stress Tests section (table + per-axis scaling curves) in the PDF.
```

- [ ] **Step 5: Verify the script parses and the help/flag wiring is correct**

Run (from `Test/E2E/chain`):
```bash
bash -n run_chain_e2e.sh && echo "syntax OK"
./run_chain_e2e.sh --help
grep -n "STRESS\|stress_run.py\|--stress-results" run_chain_e2e.sh
```
Expected: `syntax OK`; help text shows `[--stress]`; grep shows the flag default, the `--stress) STRESS=1` case, the `stress_run.py` invocation guarded by `STRESS`, and the `--stress-results` arg on `report_build.py`.

- [ ] **Step 6: Commit**

```bash
git add Test/E2E/chain/run_chain_e2e.sh
git commit -m "$(cat <<'EOF'
test(e2e-chain): add opt-in --stress phase to the orchestrator

When --stress is passed, run the capacity matrix (stress.py/stress_run.py)
after the correctness phase, writing stress_results.json into OUT_DIR/stress
and handing its path to report_build.py for the PDF's Stress Tests section.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
EOF
)"
```

---

### Task 3: Fold stress results into `report_data.json` + scaling plots + regression (`report_build.py`)

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

**Interfaces:**
- Consumes: `stress_results.json` shape `{"overall": str, "cases": [{id, shape, axis, level, status, survival, clients, min_frames, marte_cpu_s, marte_rss_mb, hub_cpu_s, hub_rss_mb, zoom_count, zoom_fail, zoom_p50_ms, zoom_p95_ms, fails}]}`; existing `regression`, `_DIRECTION`, `_LABELS`, `trend_plots`, `_load`.
- Produces: a `--stress-results PATH` CLI arg; `build_stress(sr)` → `{"overall", "cases":[...], "by_axis":{axis:[cases sorted by level]}}`; `stress_headline(stress)` → flat dict of aggregate stress metrics merged into `headline`; `stress_plots(by_axis, out)` → list of PNG basenames; `regression(curr, prev, labels, directions)` (signature extended with `labels`/`directions`, defaulting to `_LABELS`/`_DIRECTION`); `report_data.json` gains a `stress` key (or `null`) and a `stress_plots` list.

- [ ] **Step 1: Write the failing test**

Create `Test/E2E/chain/test_report_stress.py`:

```python
import os
import report_build as RB

_SR = {
    "overall": "PASS",
    "cases": [
        {"id": "ds_size_1000", "shape": "hub", "axis": "ds_signal_elements",
         "level": 1000, "status": "PASS", "survival": True, "clients": 1,
         "min_frames": 200, "marte_cpu_s": 12.8, "marte_rss_mb": 10.4,
         "hub_cpu_s": 2.33, "hub_rss_mb": 28.3, "zoom_count": 0, "zoom_fail": 0,
         "zoom_p50_ms": 0.0, "zoom_p95_ms": 0.0, "fails": []},
        {"id": "ds_size_4000", "shape": "hub", "axis": "ds_signal_elements",
         "level": 4000, "status": "PASS", "survival": True, "clients": 1,
         "min_frames": 180, "marte_cpu_s": 20.0, "marte_rss_mb": 14.0,
         "hub_cpu_s": 3.0, "hub_rss_mb": 40.0, "zoom_count": 0, "zoom_fail": 0,
         "zoom_p50_ms": 0.0, "zoom_p95_ms": 0.0, "fails": []},
        {"id": "hub_reqrate_50", "shape": "hub", "axis": "hub_zoom_reqrate_hz",
         "level": 50, "status": "PASS", "survival": True, "clients": 4,
         "min_frames": 100, "marte_cpu_s": 5.0, "marte_rss_mb": 12.0,
         "hub_cpu_s": 8.0, "hub_rss_mb": 60.0, "zoom_count": 400, "zoom_fail": 0,
         "zoom_p50_ms": 12.0, "zoom_p95_ms": 35.0, "fails": []},
    ],
}


def test_build_stress_groups_by_axis_sorted_by_level():
    st = RB.build_stress(_SR)
    assert st["overall"] == "PASS"
    assert len(st["cases"]) == 3
    assert set(st["by_axis"]) == {"ds_signal_elements", "hub_zoom_reqrate_hz"}
    levels = [c["level"] for c in st["by_axis"]["ds_signal_elements"]]
    assert levels == [1000, 4000]  # sorted ascending


def test_stress_headline_aggregates():
    st = RB.build_stress(_SR)
    hl = RB.stress_headline(st)
    assert hl["stress_pass"] == 3
    assert hl["stress_fail"] == 0
    assert hl["stress_max_hub_rss_mb"] == 60.0
    assert hl["stress_max_marte_rss_mb"] == 14.0
    assert hl["stress_max_zoom_p95_ms"] == 35.0


def test_stress_plots_one_png_per_axis(tmp_path):
    st = RB.build_stress(_SR)
    made = RB.stress_plots(st["by_axis"], str(tmp_path))
    names = {os.path.basename(p) for p in made}
    assert "stress_ds_signal_elements.png" in names
    assert "stress_hub_zoom_reqrate_hz.png" in names
    for p in made:
        assert os.path.exists(p)


def test_regression_includes_stress_when_present():
    curr = {"e2e_pass": 5, "stress_max_hub_rss_mb": 60.0}
    prev = {"e2e_pass": 5, "stress_max_hub_rss_mb": 50.0}
    labels = dict(RB._LABELS); labels["stress_max_hub_rss_mb"] = "Stress max hub RSS (MB)"
    directions = dict(RB._DIRECTION); directions["stress_max_hub_rss_mb"] = False
    rows = RB.regression(curr, prev, labels, directions)
    row = next(r for r in rows if r["key"] == "stress_max_hub_rss_mb")
    assert row["delta"] == 10.0
    assert row["better"] is False  # RSS went up → worse
```

- [ ] **Step 2: Run the test to verify it fails**

Run (from `Test/E2E/chain`):
```bash
python3 -m pytest test_report_stress.py -v
```
Expected: FAIL — `AttributeError: module 'report_build' has no attribute 'build_stress'` (and `regression` takes 2 args).

- [ ] **Step 3: Add `build_stress`, `stress_headline`, stress labels/directions, and `stress_plots`**

In `Test/E2E/chain/report_build.py`, after the `regression(...)` function (ends ~line 193) add:

```python
# Stress-axis aggregate metrics tracked across runs (mirrors the headline scalars).
_STRESS_LABELS = {
    "stress_pass": "Stress cases passed",
    "stress_fail": "Stress cases failed",
    "stress_max_hub_rss_mb": "Stress max hub RSS (MB)",
    "stress_max_marte_rss_mb": "Stress max MARTe RSS (MB)",
    "stress_max_zoom_p95_ms": "Stress max zoom p95 (ms)",
}
_STRESS_DIRECTION = {
    "stress_pass": True, "stress_fail": False,
    "stress_max_hub_rss_mb": False, "stress_max_marte_rss_mb": False,
    "stress_max_zoom_p95_ms": False,
}


def build_stress(sr):
    """Shape stress_results.json into the report's stress block (+ by_axis)."""
    cases = sr.get("cases", []) or []
    by_axis = {}
    for c in cases:
        by_axis.setdefault(c.get("axis", "?"), []).append(c)
    for axis in by_axis:
        by_axis[axis].sort(key=lambda c: c.get("level", 0))
    return {"overall": sr.get("overall", "FAIL"), "cases": cases,
            "by_axis": by_axis}


def stress_headline(stress):
    cases = stress.get("cases", []) or []
    return {
        "stress_pass": sum(1 for c in cases if c.get("status") == "PASS"),
        "stress_fail": sum(1 for c in cases if c.get("status") == "FAIL"),
        "stress_max_hub_rss_mb": max((c.get("hub_rss_mb", 0) or 0
                                      for c in cases), default=0.0),
        "stress_max_marte_rss_mb": max((c.get("marte_rss_mb", 0) or 0
                                        for c in cases), default=0.0),
        "stress_max_zoom_p95_ms": max((c.get("zoom_p95_ms", 0) or 0
                                       for c in cases), default=0.0),
    }


# Which metrics to plot per axis (label, case-field). Mixed units share a "value"
# y-axis as trend_perf.png already does; all-zero series are dropped.
_STRESS_AXIS_METRICS = {
    "ds_signal_elements": [("MARTe RSS (MB)", "marte_rss_mb"),
                           ("hub RSS (MB)", "hub_rss_mb")],
    "hub_signal_elements": [("hub RSS (MB)", "hub_rss_mb"),
                            ("hub CPU (s)", "hub_cpu_s")],
    "ds_signal_count": [("MARTe RSS (MB)", "marte_rss_mb"),
                        ("MARTe CPU (s)", "marte_cpu_s")],
    "hub_source_count": [("hub RSS (MB)", "hub_rss_mb"),
                         ("MARTe RSS (MB)", "marte_rss_mb")],
    "hub_ws_clients": [("hub RSS (MB)", "hub_rss_mb"),
                       ("hub CPU (s)", "hub_cpu_s")],
    "ds_subscriber_hubs": [("hub RSS (MB)", "hub_rss_mb"),
                           ("MARTe CPU (s)", "marte_cpu_s")],
    "hub_zoom_reqrate_hz": [("zoom p95 (ms)", "zoom_p95_ms"),
                            ("zoom p50 (ms)", "zoom_p50_ms")],
}


def stress_plots(by_axis, out):
    """One scaling-curve PNG per axis: level (x) vs the axis's metrics (y)."""
    made = []
    for axis, cases in by_axis.items():
        series = _STRESS_AXIS_METRICS.get(
            axis, [("hub RSS (MB)", "hub_rss_mb"), ("MARTe RSS (MB)", "marte_rss_mb")])
        xs = [c.get("level") for c in cases]
        fig, ax = plt.subplots(figsize=(7, 3))
        plotted = False
        for lbl, field in series:
            ys = [c.get(field) for c in cases]
            if all((v is None or v == 0) for v in ys):
                continue
            ax.plot(xs, ys, "o-", label=lbl)
            plotted = True
        if not plotted:
            plt.close(fig)
            continue
        ax.set_title(f"Scaling: {axis}")
        ax.set_xlabel("load level")
        ax.set_ylabel("value")
        ax.grid(alpha=0.3)
        ax.legend(fontsize=8)
        fig.tight_layout()
        p = os.path.join(out, f"stress_{axis}.png")
        fig.savefig(p, dpi=110)
        plt.close(fig)
        made.append(p)
    return made
```

- [ ] **Step 4: Extend `regression(...)` to accept label/direction maps**

In `Test/E2E/chain/report_build.py`, change the `regression` signature and body (lines ~177-193) to:

```python
def regression(curr, prev, labels=None, directions=None):
    labels = labels if labels is not None else _LABELS
    directions = directions if directions is not None else _DIRECTION
    rows = []
    for k, label in labels.items():
        c = curr.get(k)
        p = prev.get(k) if prev else None
        better = None
        delta = None
        if isinstance(c, (int, float)) and isinstance(p, (int, float)):
            delta = round(c - p, 4)
            if delta == 0:
                better = None
            else:
                better = (delta > 0) == directions[k]
        rows.append({"name": label, "key": k, "current": c, "previous": p,
                     "delta": delta, "better": better,
                     "higher_better": directions[k]})
    return rows
```

- [ ] **Step 5: Run the test to verify it passes**

Run (from `Test/E2E/chain`):
```bash
python3 -m pytest test_report_stress.py -v
```
Expected: all 4 tests PASS.

- [ ] **Step 6: Wire it into `main()`**

In `Test/E2E/chain/report_build.py` `main()`:

Add the CLI arg after `--out` (~line 244):
```python
    ap.add_argument("--stress-results", default="",
                    help="path to stress_results.json (optional)")
```

After `cov = _load(...)` (~line 250) add:
```python
    sr = _load(args.stress_results) if args.stress_results else None
    stress = build_stress(sr) if sr else None
```

After `hl = headline(e2e, ut, cov)` (~line 258) add:
```python
    labels, directions = _LABELS, _DIRECTION
    if stress:
        hl.update(stress_headline(stress))
        labels = {**_LABELS, **_STRESS_LABELS}
        directions = {**_DIRECTION, **_STRESS_DIRECTION}
```

Change the `reg = regression(hl, prev)` line (~line 272) to:
```python
    reg = regression(hl, prev, labels, directions)
```

After the `entry["overall"] = e2e["overall"]` line (~line 278) add (so per-case stress detail is retained in history for future curves):
```python
    if stress:
        entry["stress"] = [
            {k: c.get(k) for k in ("id", "axis", "level", "status",
                                   "marte_cpu_s", "marte_rss_mb",
                                   "hub_cpu_s", "hub_rss_mb",
                                   "zoom_p95_ms", "min_frames")}
            for c in stress["cases"]
        ]
```

After `plots = [os.path.basename(p) for p in trend_plots(history, args.out)]` (~line 283) add:
```python
    splots = ([os.path.basename(p) for p in stress_plots(stress["by_axis"], args.out)]
              if stress else [])
```

In the `doc = {...}` dict (~line 285-290) add two keys:
```python
        "stress": stress, "stress_plots": splots,
```

- [ ] **Step 7: Verify end-to-end against the real stress_results.json**

Run (from `Test/E2E/chain`; uses the existing single-case result from the Task-1 smoke test):
```bash
python3 report_build.py --repo ../../.. \
    --results ../../../Build/x86-linux/E2E/chain/results.json \
    --work /tmp/chain_e2e \
    --out ../../../Build/x86-linux/E2E/chain \
    --stress-results ../../../Build/x86-linux/E2E/chain/stress/stress_results.json
python3 -c "import json; d=json.load(open('../../../Build/x86-linux/E2E/chain/report_data.json')); assert d.get('stress'), 'no stress block'; print('stress block present:', list(d['stress']['by_axis']), 'plots:', d['stress_plots'])"
```
Expected: prints the stress block's axes and a non-empty `stress_plots` list; `stress_*.png` files exist in the out dir. (If `results.json`/`/tmp/chain_e2e` are absent from a prior run, the e2e part degrades to nulls — only the stress assertion matters here.)

- [ ] **Step 8: Commit**

```bash
git add Test/E2E/chain/report_build.py Test/E2E/chain/test_report_stress.py
git commit -m "$(cat <<'EOF'
test(e2e-chain): fold stress results into report_data.json

report_build.py reads stress_results.json (when --stress-results given),
adds a stress block (cases + by_axis), per-axis scaling-curve PNGs, aggregate
stress headline metrics, and stress regression rows vs the previous run.
Degrades to no stress section when the file is absent.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
EOF
)"
```

---

### Task 4: Render the Stress Tests section in `E2E_Report.typ`

**Files:**
- Modify: `Test/E2E/chain/E2E_Report.typ`

**Interfaces:**
- Consumes: `data.stress` (`{overall, cases:[{id, axis, level, status, survival, clients, min_frames, marte_cpu_s, marte_rss_mb, hub_cpu_s, hub_rss_mb, zoom_p50_ms, zoom_p95_ms}], by_axis}`) and `data.stress_plots` (list of PNG basenames). Existing helpers `status_badge`, `fnum`.
- Produces: a `= Stress Tests` section inserted after Performance, omitted entirely when `data.stress == none`.

- [ ] **Step 1: Add the stress section after the Performance table**

In `Test/E2E/chain/E2E_Report.typ`, immediately after the Performance table's closing `)` (line 221) and before `// ── per-scenario waveform fidelity ──` (line 223), insert:

```typ
// ── stress / capacity ─────────────────────────────────────────────────────────
#let stress = data.at("stress", default: none)
#if stress != none [
  = Stress Tests #h(6pt) #status_badge(stress.overall)
  Capacity matrix: one load axis swept at a time. Hard gates — survival + client
  liveness; soft gates — peak RSS and zoom p95 latency. The size axis crosses into
  the multi-fragment (>64 KB packet) regime.
  #v(4pt)
  #table(
    columns: (1.5fr, 1.6fr, 0.7fr, 0.7fr, 1fr, 1fr, 1fr, 1fr),
    align: (left, left, right, center, right, right, right, right),
    stroke: 0.4pt + rgb("#d0d7de"),
    inset: 4pt,
    table.header([*Case*], [*Axis*], [*Level*], [*Status*],
                 [*MARTe RSS (MB)*], [*Hub RSS (MB)*],
                 [*Hub CPU (s)*], [*Zoom p95 (ms)*]),
    ..stress.cases.map(c => (
      raw(c.id),
      text(size: 8pt)[#c.axis],
      [#c.level],
      status_badge(c.status),
      fnum(c.at("marte_rss_mb", default: none), digits: 1),
      fnum(c.at("hub_rss_mb", default: none), digits: 1),
      fnum(c.at("hub_cpu_s", default: none), digits: 2),
      fnum(c.at("zoom_p95_ms", default: none), digits: 1),
    )).flatten()
  )
  #let splots = data.at("stress_plots", default: ())
  #if splots.len() > 0 [
    #v(6pt)
    == Scaling curves
    #grid(columns: 2, gutter: 8pt,
      ..splots.map(p => image(p, width: 100%))
    )
  ]
]
```

- [ ] **Step 2: Verify the template compiles against a stress-bearing report_data.json**

Run (from `Build/x86-linux/E2E/chain`, after Task 3 Step 7 produced `report_data.json` + `stress_*.png` there):
```bash
cp /home/martino/Projects/MARTe_Integrated_components/Test/E2E/chain/E2E_Report.typ .
typst compile E2E_Report.typ E2E_Report_test.pdf && echo "compiled OK"
```
Expected: `compiled OK`; the PDF contains a "Stress Tests" section with a per-case table and the scaling-curve images. Remove the test PDF afterward: `rm -f E2E_Report_test.pdf`.

- [ ] **Step 3: Verify the section is omitted when no stress data is present**

Run (from a temp dir):
```bash
cd /tmp && python3 -c "import json; json.dump({'meta':{'git_sha':'x','timestamp':'t','target':'x86-linux'},'e2e':{'overall':'PASS','scenarios':[],'agg':{'mean_corr':None,'mean_peak_rss_mb':None,'mean_cpu_s':None,'mean_throughput_sps':None},'n_pass':0,'n_fail':0,'n_skip':0,'n_xfail':0,'n_xpass':0},'unit_tests':{'suites':[],'totals':{}},'coverage':{'languages':[]},'regression':[],'headline':{'e2e_pass':0,'e2e_total':0,'unit_pass':0,'unit_total':0,'mean_corr':None,'mean_throughput_sps':None,'cov_python':None,'cov_go':None,'mean_peak_rss_mb':None,'mean_cpu_s':None,'e2e_xfail':0,'e2e_xpass':0},'trend_plots':[],'history_len':1,'is_first_run':True}, open('report_data.json','w'))"
cp /home/martino/Projects/MARTe_Integrated_components/Test/E2E/chain/E2E_Report.typ .
typst compile E2E_Report.typ no_stress.pdf && echo "no-stress compiled OK"
rm -f no_stress.pdf report_data.json E2E_Report.typ
```
Expected: `no-stress compiled OK` (the `#if stress != none` guard skips the section cleanly).

- [ ] **Step 4: Commit**

```bash
cd /home/martino/Projects/MARTe_Integrated_components
git add Test/E2E/chain/E2E_Report.typ
git commit -m "$(cat <<'EOF'
test(e2e-chain): render Stress Tests section in the PDF report

Add a Stress Tests section (per-case table + per-axis scaling-curve images)
after Performance, guarded so it is omitted when no stress data is present.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
EOF
)"
```

---

### Task 5: Track stress files in git + document `--stress`

**Files:**
- Modify: `CLAUDE.md`
- Track (git add): `Test/E2E/chain/stress.py`, `Test/E2E/chain/stress_run.py`, `Test/E2E/chain/run_stress.sh`, `Test/E2E/chain/client/main_test.go` (and confirm `Test/E2E/chain/client/main.go` modifications are staged).

**Interfaces:**
- Consumes: nothing.
- Produces: documented `--stress` workflow; the stress suite under version control.

- [ ] **Step 1: Confirm which stress files are untracked**

Run (from repo root):
```bash
git status --short Test/E2E/chain/stress.py Test/E2E/chain/stress_run.py Test/E2E/chain/run_stress.sh Test/E2E/chain/client/main_test.go Test/E2E/chain/test_report_stress.py
```
Expected: lists the untracked (`??`) / modified (` M`) stress files. (`stress.py`, `report_build.py`, `run_chain_e2e.sh`, `E2E_Report.typ`, `test_report_stress.py` were committed in Tasks 1–4; this step targets the remaining harness files.)

- [ ] **Step 2: Update the E2E paragraph in `CLAUDE.md`**

In `CLAUDE.md`, find the "Streaming-chain E2E suite" paragraph (the one describing `run_chain_e2e.sh`). Append this sentence to the end of that paragraph:

```markdown
 A `--stress` flag additionally runs the capacity matrix (`stress.py` declarative axes → `stress_run.py` orchestrator → `stress_results.json`): it sweeps signal size (into the multi-fragment >64 KB regime), signal count, source count, WS-client count, subscriber fan-out, and zoom request-rate one axis at a time, gating survival + liveness (hard) and peak RSS + zoom-p95 latency (soft), and embeds a Stress Tests section (per-case table + per-axis scaling curves, with regression vs the previous run) into the PDF. Standalone: `./run_stress.sh [--skip-build] [--only <id>] [--axis <axis>]`.
```

- [ ] **Step 3: Stage and commit the harness files + docs**

```bash
git add Test/E2E/chain/stress.py Test/E2E/chain/stress_run.py \
        Test/E2E/chain/run_stress.sh Test/E2E/chain/client/main.go \
        Test/E2E/chain/client/main_test.go CLAUDE.md
git commit -m "$(cat <<'EOF'
test(e2e-stress): track capacity harness + document --stress

Bring the stress matrix (stress.py), orchestrator (stress_run.py), standalone
wrapper (run_stress.sh) and the chain-client stress mode under version control,
and document the --stress workflow in CLAUDE.md.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
EOF
)"
```

- [ ] **Step 4: Full opt-in verification (slow — runs the whole matrix once)**

Run (from `Test/E2E/chain`; ~4–6 min for 33 stress cases plus the correctness phase):
```bash
./run_chain_e2e.sh --skip-build --stress 2>&1 | tail -25
```
Expected: the correctness phase runs as before, then a `── Stress matrix ──` phase prints per-case PASS/FAIL lines and `stress_results.json: N/33 pass`, then `report_build.py` and `typst compile` succeed with `PDF: .../E2E_Report.pdf`. Open the PDF and confirm the Stress Tests section shows the table + scaling curves. If any large size case FAILs on an RSS gate, record the real RSS and bump that case's ceiling in `stress.py` (commit as a fixup).

---

## Self-Review

**Spec coverage:**
- Spec §1 (matrix extension + lift cap) → Task 1. ✓
- Spec §2 (report data wiring: `--stress` + `report_build.py` block) → Tasks 2 & 3. ✓
- Spec §3 (PDF section + scaling plots) → Task 4 (Typst) + Task 3 (`stress_plots`). ✓
- Spec §4 (regression + history) → Task 3 (`stress_headline`, `_STRESS_DIRECTION`, `entry["stress"]`, extended `regression`). ✓
- Spec §5 (`--stress` flag plumbing + docs) → Task 2 (flag) + Task 5 (CLAUDE.md, header in Task 2 Step 4). ✓
- Spec "full flow" diagram → realized by Task 2's phase ordering (stress before `report_build.py`). ✓
- Spec verification bullets → Task 1 Step 6, Task 3 Step 7, Task 4 Steps 2-3, Task 5 Step 4. ✓

**Placeholder scan:** No TBD/TODO/"handle errors appropriately"; every code step shows full code; every command shows expected output.

**Type consistency:** `build_stress` returns `{overall, cases, by_axis}` — consumed by `stress_headline` (reads `cases`), `stress_plots` (reads `by_axis`), Typst (`stress.cases`, `stress.overall`, `data.stress_plots`). `regression(curr, prev, labels, directions)` — the new 4-arg form is used in `main()` (Task 3 Step 6) and the test (Task 3 Step 1) and remains back-compatible (defaults). Stress metric keys (`stress_pass`, `stress_max_hub_rss_mb`, …) match across `stress_headline`, `_STRESS_LABELS`, `_STRESS_DIRECTION`. Case field names (`marte_rss_mb`, `hub_rss_mb`, `hub_cpu_s`, `zoom_p95_ms`, `level`, `axis`) match `stress_run.py`'s `_evaluate` output verbatim.