diff --git a/Test/E2E/suite/scenarios.py b/Test/E2E/suite/scenarios.py
index c6ae614..9020a69 100644
--- a/Test/E2E/suite/scenarios.py
+++ b/Test/E2E/suite/scenarios.py
@@ -135,6 +135,18 @@ def _sig(name, type, elements=1, time_mode="PacketTime", time_signal=None,
 def validate_scenario(s):
     """Return a list of validity error strings (empty == valid)."""
     errs = []
+    kind = s.get("kind", "chain")
+    if kind == "direct":
+        if "cfg" not in s:
+            errs.append(f"direct scenario {s.get('id')} missing cfg")
+        return errs
+    if kind == "recorder":
+        if "hub_cfg" not in s or "marte_cfg" not in s:
+            errs.append(f"recorder scenario {s.get('id')} missing hub_cfg/marte_cfg")
+        return errs
+    if kind != "chain":
+        errs.append(f"unknown scenario kind: {kind}")
+        return errs
     row_dt, num_rows, _producer_hz, loop_hz = geometry(s)
     if s.get("network") not in ("unicast", "multicast"):
         errs.append("network must be unicast|multicast")
@@ -565,6 +577,47 @@ SCENARIOS = (_STARTERS + _TYPES + _ARRAYS + _TIMEMODES + _QUANT + _PUBLISH +
              _PAYLOAD + _MCAST + _MULTISRC + _INTERACT + _TRIGGER + _MISC +
              _HIGHRATE)
 
+for _s in SCENARIOS:
+    _s.setdefault("kind", "chain")
+
+# ── Non-chain scenario kinds ──────────────────────────────────────────────────
+# "direct" and "recorder" scenarios exercise the other two data paths described
+# in ARCHITECTURE.md (direct UDPStreamer<->UDPStreamerClient, and StreamHub's
+# BinaryRecorder disk sink) and intentionally do not carry the chain-only keys
+# (sources/network/oracle/...): they have their own orchestration in run_e2e.sh.
+_DIRECT = [
+    {
+        "id": "s52_direct_unicast",
+        "desc": "Direct UDPStreamer->UDPStreamerClient round-trip, unicast",
+        "kind": "direct",
+        "cfg": "Test/E2E/datasources/E2ETest.cfg",
+        "client_checks": [],
+        "known_issue": None,
+    },
+    {
+        "id": "s53_direct_multicast",
+        "desc": "Direct UDPStreamer->UDPStreamerClient round-trip, multicast",
+        "kind": "direct",
+        "cfg": "Test/E2E/datasources/E2EMulticastTest.cfg",
+        "client_checks": [],
+        "known_issue": None,
+    },
+]
+
+_RECORDER = [
+    {
+        "id": "s54_recorder",
+        "desc": "StreamHub BinaryRecorder disk-output round-trip",
+        "kind": "recorder",
+        "hub_cfg": "Test/E2E/recorder/StreamHubRec.cfg",
+        "marte_cfg": "Test/E2E/recorder/RecorderStreamer.cfg",
+        "client_checks": [],
+        "known_issue": None,
+    },
+]
+
+SCENARIOS = SCENARIOS + _DIRECT + _RECORDER
+
 
 if __name__ == "__main__":
     import sys
@@ -576,13 +629,14 @@ if __name__ == "__main__":
         if s["id"] in seen_ids:
             errs.append("duplicate id")
         seen_ids.add(s["id"])
-        if s["ws_port"] in seen_ws:
-            errs.append(f"duplicate ws_port {s['ws_port']}")
-        seen_ws.add(s["ws_port"])
-        for src in s["sources"]:
-            if src["udp_port"] in seen_udp:
-                errs.append(f"duplicate udp_port {src['udp_port']}")
-            seen_udp.add(src["udp_port"])
+        if s["kind"] == "chain":
+            if s["ws_port"] in seen_ws:
+                errs.append(f"duplicate ws_port {s['ws_port']}")
+            seen_ws.add(s["ws_port"])
+            for src in s["sources"]:
+                if src["udp_port"] in seen_udp:
+                    errs.append(f"duplicate udp_port {src['udp_port']}")
+                seen_udp.add(src["udp_port"])
         print(f"{s['id']:32s} {'OK' if not errs else errs}")
         ok = ok and not errs
     print(f"\n{len(SCENARIOS)} scenarios, {'ALL VALID' if ok else 'INVALID PRESENT'}")
diff --git a/Test/E2E/suite/tests_py.py b/Test/E2E/suite/tests_py.py
index 0e3df34..c16ec66 100644
--- a/Test/E2E/suite/tests_py.py
+++ b/Test/E2E/suite/tests_py.py
@@ -58,6 +58,24 @@ class TestScenarios(unittest.TestCase):
         errs = S.validate_scenario(s)
         self.assertTrue(any("quant only on float" in e for e in errs), errs)
 
+    def test_all_scenarios_have_kind(self):
+        for s in S.SCENARIOS:
+            self.assertIn("kind", s, f"{s['id']} missing kind")
+            self.assertIn(s["kind"], ("chain", "direct", "recorder", "debug", "tcplogger"))
+
+    def test_direct_and_recorder_scenarios_present(self):
+        kinds = {s["kind"] for s in S.SCENARIOS}
+        self.assertIn("direct", kinds)
+        self.assertIn("recorder", kinds)
+        direct_ids = {s["id"] for s in S.SCENARIOS if s["kind"] == "direct"}
+        self.assertEqual(direct_ids, {"s52_direct_unicast", "s53_direct_multicast"})
+        recorder_ids = {s["id"] for s in S.SCENARIOS if s["kind"] == "recorder"}
+        self.assertEqual(recorder_ids, {"s54_recorder"})
+
+    def test_validate_scenario_accepts_all_kinds(self):
+        for s in S.SCENARIOS:
+            S.validate_scenario(s)  # must not raise
+
 
 class TestGenData(unittest.TestCase):
     def test_ground_truth_shapes(self):