feat(M4-P0D3): per-MU atomic scenario, readback and shared-grid TX gates

docs/M4_P0D3_STREAM_SCENARIO_TRANSACTIONS.md

@@ -0,0 +1,46 @@
+# M4 P0D3 — Atomic per-MU scenarios and shared-grid stream gates
+
+**Status:** feature PR / requires ESP-IDF + C++ CI; not physical hardware validation.
+
+Firmware reuses the existing `SvScenarioTransaction` in a bounded array of three slots. No XML, heap, mutex, extra timer, extra publisher task or per-frame log is introduced. The old single-SV CFG/START protocol and M3.H.3 release remain untouched.
+
+## Protocol
+
+Requires `MULTISV` committed roster + explicit `MULTISV SELECT`. Slot is `1`, `2`, or `3`.
+
+```text
+SVSTREAM 2 CFG BEGIN 12345
+SVSTREAM 2 CFG FREQ 50000
+SVSTREAM 2 CFG SET IA 100000 0 0 1
+SVSTREAM 2 CFG SET IB 100000 -120000 0 1
+SVSTREAM 2 CFG SET IC 100000 120000 0 1
+SVSTREAM 2 CFG SET IN 0 0 0 1
+SVSTREAM 2 CFG SET UA 5774 0 0 1
+SVSTREAM 2 CFG SET UB 5774 -120000 0 1
+SVSTREAM 2 CFG SET UC 5774 120000 0 1
+SVSTREAM 2 CFG SET UN 0 0 0 1
+SVSTREAM 2 CFG SHAPE 0 0 0 2 1000
+SVSTREAM 2 CFG COMMIT 12345 <16-character-host-computed-sv_scenario_hash-hex>
+SVSTREAM 2 CFG READBACK 12345
+SVSTREAM 2 SHOW
+```
+
+`FREQ` argument is mHz; `SET` arguments are wire RMS counts, phase millidegrees, uint32 quality, and Boolean channel enabled. `SHAPE` arguments are enabled, DC permille, harmonic permille, order and clip permille. SHA/CRC of the SCD envelope is a **different** check from `sv_scenario_hash`; compute the latter with the existing canonical host codec in `live_signal_transaction.hpp`, not floating-point JSON serialization.
+
+All 18 fields (frequency + shape + 8 SET + 8 enable states) are mandatory. A SET stages value and enabled status together, satisfying two completeness bits. One missing channel, invalid field, changed bank generation, wrong token, stale timeout (>8s) or hash mismatch rejects the whole transaction, preserving old output. `READBACK` checks applied bank generation/hash. A running scenario does not provide a fresh global START authority.
+
+```text
+SVSTREAM 2 STOP
+SVSTREAM 2 SHOW
+SVSTREAM 2 START
+```
+
+These operations toggle only the stream's Ethernet TX bit, **not** its signal-bank contents or the shared grid. Each 250us tick reads one 3-bit enable snapshot; all three DDS engines advance regardless of TX gating. MU2 rejoining takes the **current group smpCnt**, not 0. The global START/STOP handshake and physical link/PTP fail-closed policy remain authoritative; a per-MU START while the group is stopped merely arms that MU for the next verified group START. A newly committed SCD roster reinitializes all three TX-enable bits to ON.
+
+## Explicit limitations / next
+
+- Atomic **per-MU scenario**, not one atomic transaction covering all three MUs.
+- TX state is RAM-resident and persists through Studio disconnect while the ESP32 remains powered, not across ESP32 reboot.
+- Output follows one wire-time grid but no guarantee is claimed about zero Ethernet inter-packet skew.
+- Complete serial Studio deploy, 3-MU page editing and reconnect adoption are P0E.
+- Firmware CI is **not** proof of real 3×4000fps, 250us deadline, on-wire MAC or PTP phase; P0F must capture physical Ethernet.

embedded/esp32p4_smv_injector/main/app_main.cpp

@@ -1673,8 +1673,15 @@ void publisher_task(void* argument) {
                     continue;
                 }
             }
-            patch_packet(
-                canonical, canonical_sample_count, sample_synchronization, row, signal_state);
+            // One mask snapshot per batch. A disabled MU still advances its
+            // DDS phase on the shared 250us grid; it simply skips Ethernet.
+            const std::uint8_t multi_enabled_mask = multi_active
+                ? ar::esp32p4::smv::multi_profile_enabled_mask() : 0x01U;
+            const bool send_primary = (multi_enabled_mask & 0x01U) != 0U;
+            if (send_primary) {
+                patch_packet(
+                    canonical, canonical_sample_count, sample_synchronization, row, signal_state);
+            }
             // One timestamp BEFORE all three canonical EMAC submissions.
             // Budget telemetry is folded into the existing 400-tick stats
             // flush, never logged or malloc'ed on the real-time path.
@@ -1692,7 +1699,7 @@ void publisher_task(void* argument) {
             SvPtpClockLease evidence_lease{};
             std::uint32_t evidence_session = 0U;
             std::uint32_t fault_generation_before = 0U;
-            if (epoch_armed && g_sample_clock_context.phase_enabled &&
+            if (send_primary && epoch_armed && g_sample_clock_context.phase_enabled &&
                 canonical_sample_count == 0U && missed_slots == 0U &&
                 active_profile.sample_counter_modulus ==
                     active_profile.publisher_rate_hz) {
@@ -1730,7 +1737,8 @@ void publisher_task(void* argument) {
             eth_mac_time_t egress_hw_time{};
             const std::int64_t submit_before_us =
                 take_egress_timestamp ? esp_timer_get_time() : 0;
-            const esp_err_t canonical_result = take_egress_timestamp
+            const esp_err_t canonical_result = !send_primary ? ESP_OK :
+                take_egress_timestamp
                 ? esp_eth_transmit_ctrl_vargs(
                     eth_handle, &egress_hw_time, 2U, canonical.bytes.data(),
                     static_cast<std::uint32_t>(canonical.bytes.size()))
@@ -1811,9 +1819,11 @@ void publisher_task(void* argument) {
                 }
             }
 #else
-            const esp_err_t canonical_result =
-                esp_eth_transmit(eth_handle, canonical.bytes.data(), canonical.bytes.size());
+            const esp_err_t canonical_result = send_primary
+                ? esp_eth_transmit(eth_handle, canonical.bytes.data(), canonical.bytes.size())
+                : ESP_OK;
 #endif
+            if (send_primary) {
             if (canonical_result == ESP_OK) {
                 ++local.canonical_ok;
                 if (consecutive_canonical_tx_failures != 0U)
@@ -1840,6 +1850,8 @@ void publisher_task(void* argument) {
 #endif
             }
 
+            } // primary-MU enable gate
+
             // M4: exactly two additional canonical SV frames from the same
             // hardware-timed tick. Each has independent prebuilt packet bytes
             // and DDS state. The three frames share smpCnt/smpSynch; egress
@@ -1849,6 +1861,9 @@ void publisher_task(void* argument) {
                 for (std::size_t slot = 0U; slot < multi_extra_packets.size(); ++slot) {
                     const auto extra_row = multi_extra_engines[slot].next(
                         multi_extra_signals[slot], 4000U);
+                    // Advance phase even when muted, so re-enabled frames
+                    // rejoin this same shared PTP sample epoch.
+                    if ((multi_enabled_mask & (1U << (slot + 1U))) == 0U) continue;
                     patch_packet(multi_extra_packets[slot], canonical_sample_count,
                                  sample_synchronization, extra_row,
                                  multi_extra_signals[slot]);
@@ -1902,7 +1917,7 @@ void publisher_task(void* argument) {
                     eth_handle, probe_multicast.data(), probe_multicast.size()));
             }
 #endif
-            if (multi_active) {
+            if (multi_active && multi_enabled_mask != 0U) {
                 const std::int64_t elapsed = esp_timer_get_time() - multi_batch_start_us;
                 const auto bounded = static_cast<std::uint32_t>(
                     std::clamp<std::int64_t>(elapsed, 0LL, 0xFFFFFFFFLL));

embedded/esp32p4_smv_injector/main/multi_profile_control.cpp

@@ -3,8 +3,10 @@
 #include "multi_profile_staging.hpp"
 #include "live_control.hpp"
 #include "ariec61850/sampled_values/m4_three_sv_signals.hpp"
+#include "ariec61850/sampled_values/live_signal_transaction.hpp"
 
 #include "esp_log.h"
+#include "esp_timer.h"
 
 #include <array>
 #include <atomic>
@@ -27,6 +29,11 @@ constexpr char kDelimiters[] = " \t\r\n";
 MultiSvConfigStore g_multi{};
 // Exactly three independent atomic waveform banks; only console task writes.
 ar::iec61850::sampled_values::M4ThreeSvSignalBanks g_mu_signals{};
+// Each slot independently stages a complete scenario. One UART writer; no
+// transactional parsing or allocation in the 4-kHz data-plane.
+std::array<ar::iec61850::sampled_values::SvScenarioTransaction, 3> g_mu_transactions{};
+// Toggled only by UART control; publisher takes one atomic mask per grid tick.
+ar::iec61850::sampled_values::M4ThreeSvStreamGate g_mu_stream_gate{};
 // Selection changes only in STOPPED control-plane commands; the active
 // publisher reads the immutable committed roster once at START.
 std::atomic<bool> g_multi_tx_selected{false};
@@ -161,6 +168,162 @@ multi_profile_signal_snapshot(const std::size_t slot) noexcept {
     return value;
 }
 
+
+[[nodiscard]] bool parse_u64_token(const char* text, std::uint64_t& output,
+                                   const int base) noexcept {
+    if (!text || !*text || *text == '-') return false;
+    errno = 0;
+    char* end = nullptr;
+    const unsigned long long value = std::strtoull(text, &end, base);
+    if (errno || end == text || *end) return false;
+    output = static_cast<std::uint64_t>(value);
+    return true;
+}
+
+[[nodiscard]] std::uint32_t serial_now_ms() noexcept {
+    return static_cast<std::uint32_t>(
+        static_cast<std::uint64_t>(esp_timer_get_time()) / 1000ULL);
+}
+
+void reset_mu_transactions() noexcept {
+    for (auto& tx : g_mu_transactions) tx.reset();
+}
+
+// The existing V1 SvScenarioTransaction is deliberately reused per MU:
+// complete 8-channel staging, generation guard, 8s timeout, canonical
+// wire-state hash and explicit readback. Only this UART task writes a bank.
+void handle_svstream_cfg(const std::size_t slot, char* save) noexcept {
+    using Tx = ar::iec61850::sampled_values::SvScenarioTransaction;
+    auto& tx = g_mu_transactions[slot];
+    auto* bank = g_mu_signals.control_bank_for_serial_writer(slot);
+    char* verb = strtok_r(nullptr, kDelimiters, &save);
+    if (!verb || bank == nullptr) {
+        tx.reject();
+        ESP_LOGE(kTag, "SVSTREAM CFG requires BEGIN|FREQ|SET|SHAPE|COMMIT|READBACK|ABORT");
+        return;
+    }
+    upper(verb);
+    const auto now = serial_now_ms();
+    const auto log_slot = static_cast<unsigned>(slot + 1U);
+    if (std::strcmp(verb, "ABORT") == 0) {
+        if (!no_extra(&save)) { tx.reject(); return; }
+        tx.reset();
+        ESP_LOGI(kTag, "SVSTREAM CFG ABORT slot=%u", log_slot);
+        return;
+    }
+    if (std::strcmp(verb, "BEGIN") == 0) {
+        std::uint64_t token{};
+        if (!parse_u64_token(strtok_r(nullptr, kDelimiters, &save), token, 10) ||
+            !no_extra(&save) ||
+            !tx.begin(token, bank->snapshot(), now, live_tx_running())) {
+            tx.reset();
+            ESP_LOGE(kTag, "SVSTREAM CFG BEGIN rejected slot=%u", log_slot);
+            return;
+        }
+        ESP_LOGI(kTag, "SVSTREAM CFG BEGIN slot=%u token=%llu",
+                 log_slot, static_cast<unsigned long long>(token));
+        return;
+    }
+    if (std::strcmp(verb, "READBACK") == 0) {
+        std::uint64_t token{};
+        if (!parse_u64_token(strtok_r(nullptr, kDelimiters, &save), token, 10) ||
+            !no_extra(&save) || !tx.confirm_readback(*bank, token, now)) {
+            ESP_LOGE(kTag, "SVSTREAM CFG READBACK rejected slot=%u", log_slot);
+            return;
+        }
+        ESP_LOGI(kTag,
+            "SVSTREAM CFG READBACK slot=%u token=%llu hash=%016llX generation=%llu",
+            log_slot, static_cast<unsigned long long>(token),
+            static_cast<unsigned long long>(tx.checksum()),
+            static_cast<unsigned long long>(tx.applied_generation()));
+        return;
+    }
+    if (std::strcmp(verb, "COMMIT") == 0) {
+        std::uint64_t token{}, expected_hash{};
+        if (!parse_u64_token(strtok_r(nullptr, kDelimiters, &save), token, 10) ||
+            !parse_u64_token(strtok_r(nullptr, kDelimiters, &save), expected_hash, 16) ||
+            !no_extra(&save) || tx.began_running() != live_tx_running()) {
+            tx.reject();
+            ESP_LOGE(kTag, "SVSTREAM CFG COMMIT rejected slot=%u malformed or RUN state changed",
+                     log_slot);
+            return;
+        }
+        const auto result = tx.commit(*bank, token, expected_hash, now);
+        if (result != Tx::Result::accepted) {
+            ESP_LOGE(kTag, "SVSTREAM CFG REJECT slot=%u code=%u", log_slot,
+                     static_cast<unsigned>(result));
+            return;
+        }
+        ESP_LOGI(kTag,
+            "SVSTREAM CFG COMMITTED slot=%u token=%llu hash=%016llX generation=%llu",
+            log_slot, static_cast<unsigned long long>(token),
+            static_cast<unsigned long long>(tx.checksum()),
+            static_cast<unsigned long long>(tx.applied_generation()));
+        return;
+    }
+
+    if (!tx.staging() || tx.began_running() != live_tx_running()) {
+        tx.reject();
+        ESP_LOGE(kTag, "SVSTREAM CFG rejected slot=%u: BEGIN required or RUN state changed",
+                 log_slot);
+        return;
+    }
+    auto& draft = tx.draft();
+    if (std::strcmp(verb, "FREQ") == 0) {
+        std::uint32_t frequency{};
+        if (!parse_uint(strtok_r(nullptr, kDelimiters, &save), frequency) ||
+            !no_extra(&save)) { tx.reject(); return; }
+        draft.frequency_millihz = frequency;
+        tx.mark_frequency();
+    } else if (std::strcmp(verb, "SET") == 0) {
+        const int index = stream_channel(strtok_r(nullptr, kDelimiters, &save));
+        std::int32_t rms{}, phase{};
+        std::uint32_t quality{}, enabled{};
+        if (index < 0 ||
+            !parse_signed(strtok_r(nullptr, kDelimiters, &save), rms) ||
+            !parse_signed(strtok_r(nullptr, kDelimiters, &save), phase) ||
+            !parse_uint(strtok_r(nullptr, kDelimiters, &save), quality) ||
+            !parse_uint(strtok_r(nullptr, kDelimiters, &save), enabled) ||
+            enabled > 1U || !no_extra(&save)) { tx.reject(); return; }
+        auto& ch = draft.channels[static_cast<std::size_t>(index)];
+        ch.rms_counts = rms;
+        ch.phase_millidegrees = phase;
+        ch.quality = quality;
+        ch.enabled = enabled != 0U;
+        tx.mark_set(static_cast<std::size_t>(index));
+        tx.mark_enabled(static_cast<std::size_t>(index));
+    } else if (std::strcmp(verb, "SHAPE") == 0) {
+        std::uint32_t enabled{}, harmonic{}, order{}, clip{};
+        std::int32_t dc{};
+        if (!parse_uint(strtok_r(nullptr, kDelimiters, &save), enabled) ||
+            enabled > 1U ||
+            !parse_signed(strtok_r(nullptr, kDelimiters, &save), dc) ||
+            !parse_uint(strtok_r(nullptr, kDelimiters, &save), harmonic) ||
+            !parse_uint(strtok_r(nullptr, kDelimiters, &save), order) ||
+            !parse_uint(strtok_r(nullptr, kDelimiters, &save), clip) ||
+            order > 255U || !no_extra(&save)) { tx.reject(); return; }
+        draft.current_shape = {enabled != 0U, dc, harmonic,
+                               static_cast<std::uint8_t>(order), clip};
+        tx.mark_shape();
+    } else {
+        tx.reject();
+        ESP_LOGE(kTag, "SVSTREAM CFG invalid command slot=%u", log_slot);
+        return;
+    }
+    // Even before COMMIT, reject an invalid individual field without exposing
+    // this draft to the publisher; full validation repeats on COMMIT.
+    if (!ar::iec61850::sampled_values::SvLiveSignalBank::validate(draft)) {
+        tx.reject();
+        ESP_LOGE(kTag, "SVSTREAM CFG invalid staged value slot=%u", log_slot);
+        return;
+    }
+    ESP_LOGI(kTag, "SVSTREAM CFG STAGED slot=%u verb=%s", log_slot, verb);
+}
+
+std::uint8_t multi_profile_enabled_mask() noexcept {
+    return g_mu_stream_gate.snapshot_mask();
+}
+
 void handle_svstream_command(char* arguments) noexcept {
     char* save = nullptr;
     const char* slot_text = strtok_r(arguments, kDelimiters, &save);
@@ -175,13 +338,30 @@ void handle_svstream_command(char* arguments) noexcept {
     char* verb = strtok_r(nullptr, kDelimiters, &save);
     if (!verb) { ESP_LOGE(kTag, "SVSTREAM usage: <1..3> SHOW|SET|FREQ"); return; }
     upper(verb);
+    if (std::strcmp(verb, "CFG") == 0) {
+        handle_svstream_cfg(slot, save);
+        return;
+    }
+    if (std::strcmp(verb, "START") == 0 || std::strcmp(verb, "STOP") == 0) {
+        if (!no_extra(&save)) {
+            ESP_LOGE(kTag, "SVSTREAM START|STOP takes no extra tokens");
+            return;
+        }
+        const bool enabled = std::strcmp(verb, "START") == 0;
+        static_cast<void>(g_mu_stream_gate.set_enabled(slot, enabled));
+        ESP_LOGI(kTag, "SVSTREAM slot=%lu enabled=%u mask=0x%02X effect=NEXT_COMMON_TICK",
+                 static_cast<unsigned long>(slot_number), enabled ? 1U : 0U,
+                 static_cast<unsigned>(multi_profile_enabled_mask()));
+        return;
+    }
     auto state = multi_profile_signal_snapshot(slot);
     if (std::strcmp(verb, "SHOW") == 0) {
         if (!no_extra(&save)) { ESP_LOGE(kTag, "SVSTREAM SHOW takes no extras"); return; }
-        ESP_LOGI(kTag, "SVSTREAM slot=%lu generation=%llu frequency_millihz=%lu",
+        ESP_LOGI(kTag, "SVSTREAM slot=%lu generation=%llu frequency_millihz=%lu tx_enabled=%u",
                  static_cast<unsigned long>(slot_number),
                  static_cast<unsigned long long>(state.generation),
-                 static_cast<unsigned long>(state.frequency_millihz));
+                 static_cast<unsigned long>(state.frequency_millihz),
+                 (multi_profile_enabled_mask() & (1U << slot)) ? 1U : 0U);
         static constexpr const char* names[] = {"IA", "IB", "IC", "IN", "UA", "UB", "UC", "UN"};
         for (std::size_t i = 0; i < state.channels.size(); ++i) {
             const auto& ch = state.channels[i];
@@ -259,6 +439,7 @@ void handle_multi_profile_command(char* arguments) noexcept {
             return;
         }
         const bool selected = !std::strcmp(command, "SELECT");
+        reset_mu_transactions();
         g_multi_tx_selected.store(selected, std::memory_order_release);
         ESP_LOGI(kTag, "MULTISV TXMODE selected=%u; next verified START uses %s",
                  selected ? 1U : 0U, selected ? "THREE_SV" : "LEGACY_ONE_SV");
@@ -330,11 +511,15 @@ void handle_multi_profile_command(char* arguments) noexcept {
             ESP_LOGE(kTag, "MULTISV COMMIT rejected: VALIDATE required or transaction invalid");
             return;
         }
+        // Roster identity changed: pending MU tickets and TX bits belong
+        // to the old set; all newly deployed MUs default to enabled.
+        reset_mu_transactions();
+        g_mu_stream_gate.reset_all();
         // A new committed profile set is never auto-activated by replacing
         // the old roster. Require an explicit SELECT after each COMMIT.
         g_multi_tx_selected.store(false, std::memory_order_release);
         ESP_LOGI(kTag,
-            "MULTISV COMMITTED transaction=%lu generation=%llu hash=%016llX bytes=%lu output=INACTIVE_UNTIL_M4_P0C2",
+            "MULTISV COMMITTED transaction=%lu generation=%llu hash=%016llX bytes=%lu output=CONFIG_COMMITTED_TX_REQUIRES_SELECT_AND_VERIFIED_START",
             static_cast<unsigned long>(txn),
             static_cast<unsigned long long>(g_multi.generation()),
             static_cast<unsigned long long>(g_multi.committed_hash()),

embedded/esp32p4_smv_injector/main/multi_profile_control.hpp

@@ -3,6 +3,7 @@
 #include "ariec61850/sampled_values/compiled_device_profile.hpp"
 #include "ariec61850/sampled_values/live_signal_state.hpp"
 #include <cstddef>
+#include <cstdint>
 
 // The console/control task is the ONLY writer to the committed roster.
 // A TX START serially follows SELECT, and COMMIT is forbidden while live SV
@@ -14,6 +15,9 @@ void multi_profile_abort_staging() noexcept;
 void handle_svstream_command(char* arguments) noexcept;
 [[nodiscard]] ar::iec61850::sampled_values::SvLiveSignalState
 multi_profile_signal_snapshot(std::size_t slot) noexcept;
+// One acquire snapshot at the start of each 250us batch. Bits 0/1/2 gate
+// canonical MU1/MU2/MU3 Ethernet submissions without resetting the grid.
+[[nodiscard]] std::uint8_t multi_profile_enabled_mask() noexcept;
 
 // Null means legacy V1 START. Non-null is the selected, validated, committed
 // 3-profile V2 roster. Never retain across STOP and a future control commit.