persistence: PJ rcpvars + prprocessparams in v2 record format

Closes the v1 caveat: the optional E40-0705 trailers on S16F11 —
recipe variables (RcpVar) and process parameters (ProcessParam),
each carrying a secs2::Item value of arbitrary type — now survive
restart.

Record format bumps to v2:
  v2 header = v1 header
  + [u16 rcpvar_count][repeat: u16 name_len, name, u32 enc_len,
                       secs2::encode(value)]
  + [u16 ppparam_count][...same shape]

v1 records are still accepted by load_record_ (no extras come back).

Two new tests:
  - round-trip mixed F4 / ASCII / U4 / nested-list values through
    rcpvars + prprocessparams
  - hand-crafted v1 record on disk still loads cleanly, just with
    empty extras (proves backwards compat)

Closes the "PJ rcpvars / prprocessparams persistence" caveat from
the post-#1-13 status writeup.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

include/secsgem/gem/store/process_jobs.hpp

@@ -15,6 +15,7 @@
 
 #include "secsgem/gem/e40_constants.hpp"
 #include "secsgem/gem/process_job_state.hpp"
+#include "secsgem/secs2/codec.hpp"
 #include "secsgem/secs2/item.hpp"
 
 namespace secsgem::gem {
@@ -226,10 +227,10 @@ class ProcessJobStore {
   // ---- Persistence ----------------------------------------------------
   // One file per PJ (`<seq>.pj`); a single `order.idx` index records the
   // queue ordering (insertion-order modulo HOQ).  Replay reads the index
-  // first to preserve ordering across restarts.  Extras (rcpvars,
+  // first to preserve ordering across restarts.  Version 2 of the record
-  // prprocessparams) are skipped in v1 — they carry secs2::Item variants
+  // (current) journals the E40 extras (rcpvars + prprocessparams) by
-  // and aren't necessary for the FSM lifecycle.  Callers that need them
+  // round-tripping each value through secs2::encode/decode.  Version 1
-  // post-restart should call set_e40_extras() again.
+  // records are accepted on replay; their extras come back empty.
   void enable_persistence(std::filesystem::path dir) {
     namespace fs = std::filesystem;
     journal_dir_ = std::move(dir);
@@ -258,7 +259,11 @@ class ProcessJobStore {
       install_(rec->prjobid, rec->ppid, std::move(rec->materials),
                rec->alert, rec->mf, rec->prrecipemethod, rec->state);
       auto* pj = get(rec->prjobid);
-      if (pj) pj->journal_path = path;
+      if (pj) {
+        pj->journal_path = path;
+        pj->rcpvars = std::move(rec->rcpvars);
+        pj->prprocessparams = std::move(rec->prprocessparams);
+      }
       if (seq >= next_seq_) next_seq_ = seq + 1;
     }
 
@@ -290,16 +295,23 @@ class ProcessJobStore {
     MaterialFlag mf;
     ProcessRecipeMethod prrecipemethod;
     ProcessJobState state;
+    std::vector<RcpVar> rcpvars;
+    std::vector<ProcessParam> prprocessparams;
   };
 
-  // PJ record:
+  // PJ record (version 2):
-  //   [u8 magic = 0xC7][u8 version = 1]
+  //   [u8 magic = 0xC7][u8 version]
   //   [u8 state][u8 alert][u8 mf][u8 prrecipemethod]
   //   [u16 prjobid_len][prjobid_bytes]
   //   [u16 ppid_len][ppid_bytes]
   //   [u16 material_count][repeat: u16 len + bytes]
+  //   --- v2 additions ---
+  //   [u16 rcpvar_count][repeat: u16 name_len, name, u32 enc_len, secs2::encode(value)]
+  //   [u16 ppparam_count][repeat: u16 name_len, name, u32 enc_len, secs2::encode(value)]
+  // v1 records (no extras) are accepted on replay; we just skip the
+  // trailers and the in-memory rcpvars/prprocessparams stay empty.
   static constexpr uint8_t kMagic   = 0xC7;
-  static constexpr uint8_t kVersion = 0x01;
+  static constexpr uint8_t kVersion = 0x02;
 
   void install_(const std::string& prjobid, std::string ppid,
                 std::vector<std::string> materials, bool alert,
@@ -351,10 +363,30 @@ class ProcessJobStore {
     };
     write_str(pj->prjobid);
     write_str(pj->ppid);
-    uint16_t mc = static_cast<uint16_t>(pj->mtrloutspec.size());
+    auto write_be16 = [&](uint16_t v) {
-    uint8_t mb[2] = {static_cast<uint8_t>(mc >> 8), static_cast<uint8_t>(mc)};
+      uint8_t b[2] = {static_cast<uint8_t>(v >> 8), static_cast<uint8_t>(v)};
-    out.write(reinterpret_cast<const char*>(mb), 2);
+      out.write(reinterpret_cast<const char*>(b), 2);
+    };
+    auto write_be32 = [&](uint32_t v) {
+      uint8_t b[4] = {static_cast<uint8_t>((v >> 24) & 0xFF),
+                      static_cast<uint8_t>((v >> 16) & 0xFF),
+                      static_cast<uint8_t>((v >> 8) & 0xFF),
+                      static_cast<uint8_t>(v & 0xFF)};
+      out.write(reinterpret_cast<const char*>(b), 4);
+    };
+    write_be16(static_cast<uint16_t>(pj->mtrloutspec.size()));
     for (const auto& m : pj->mtrloutspec) write_str(m);
+    auto write_named = [&](const std::string& name, const secs2::Item& value) {
+      write_str(name);
+      auto enc = secs2::encode(value);
+      write_be32(static_cast<uint32_t>(enc.size()));
+      out.write(reinterpret_cast<const char*>(enc.data()),
+                static_cast<std::streamsize>(enc.size()));
+    };
+    write_be16(static_cast<uint16_t>(pj->rcpvars.size()));
+    for (const auto& v : pj->rcpvars) write_named(v.name, v.value);
+    write_be16(static_cast<uint16_t>(pj->prprocessparams.size()));
+    for (const auto& v : pj->prprocessparams) write_named(v.name, v.value);
     out.flush();
     out.close();
     std::error_code ec;
@@ -367,7 +399,10 @@ class ProcessJobStore {
     if (!in) return std::nullopt;
     uint8_t header[6];
     in.read(reinterpret_cast<char*>(header), sizeof(header));
-    if (!in || header[0] != kMagic || header[1] != kVersion) return std::nullopt;
+    if (!in || header[0] != kMagic) return std::nullopt;
+    // v1 and v2 share the same prefix; v2 adds rcpvars+prprocessparams.
+    if (header[1] != 0x01 && header[1] != 0x02) return std::nullopt;
+    const uint8_t version = header[1];
     Record r;
     r.state          = static_cast<ProcessJobState>(header[2]);
     r.alert          = header[3] != 0;
@@ -382,16 +417,57 @@ class ProcessJobStore {
       in.read(out.data(), n);
       return static_cast<bool>(in) || n == 0;
     };
+    auto read_be16 = [&](uint16_t& v) {
+      uint8_t b[2];
+      in.read(reinterpret_cast<char*>(b), 2);
+      if (!in) return false;
+      v = static_cast<uint16_t>((uint16_t(b[0]) << 8) | b[1]);
+      return true;
+    };
+    auto read_be32 = [&](uint32_t& v) {
+      uint8_t b[4];
+      in.read(reinterpret_cast<char*>(b), 4);
+      if (!in) return false;
+      v = (uint32_t(b[0]) << 24) | (uint32_t(b[1]) << 16) |
+          (uint32_t(b[2]) << 8) | uint32_t(b[3]);
+      return true;
+    };
     if (!read_str(r.prjobid)) return std::nullopt;
     if (!read_str(r.ppid))    return std::nullopt;
-    uint8_t mb[2];
+    uint16_t mc = 0;
-    in.read(reinterpret_cast<char*>(mb), 2);
+    if (!read_be16(mc)) return std::nullopt;
-    if (!in) return std::nullopt;
-    uint16_t mc = static_cast<uint16_t>((uint16_t(mb[0]) << 8) | mb[1]);
     r.materials.resize(mc);
     for (auto& m : r.materials) {
       if (!read_str(m)) return std::nullopt;
     }
+    if (version >= 2) {
+      auto read_named_list = [&](std::vector<RcpVar>& dst) -> bool {
+        uint16_t n = 0;
+        if (!read_be16(n)) return false;
+        dst.resize(n);
+        for (auto& kv : dst) {
+          if (!read_str(kv.name)) return false;
+          uint32_t enc_len = 0;
+          if (!read_be32(enc_len)) return false;
+          std::vector<uint8_t> bytes(enc_len);
+          if (enc_len) {
+            in.read(reinterpret_cast<char*>(bytes.data()), enc_len);
+            if (!in) return false;
+          }
+          try { kv.value = secs2::decode(bytes); }
+          catch (...) { return false; }
+        }
+        return true;
+      };
+      if (!read_named_list(r.rcpvars)) return std::nullopt;
+      // ProcessParam has same shape as RcpVar; reuse via reinterpret-style:
+      std::vector<RcpVar> pp_tmp;
+      if (!read_named_list(pp_tmp)) return std::nullopt;
+      r.prprocessparams.reserve(pp_tmp.size());
+      for (auto& kv : pp_tmp) {
+        r.prprocessparams.push_back({std::move(kv.name), std::move(kv.value)});
+      }
+    }
     return r;
   }
 

tests/test_job_persistence.cpp

@@ -157,6 +157,76 @@ TEST_CASE("ControlJobStore persistence: state + prjob list replay") {
   fs::remove_all(dir);
 }
 
+TEST_CASE("ProcessJobStore persistence: rcpvars + prprocessparams survive restart") {
+  auto dir = scratch_dir("pj-extras");
+  {
+    ProcessJobStore s;
+    s.enable_persistence(dir);
+    REQUIRE(s.create("PJ-X", "RCP-1", {"W1"}) ==
+            ProcessJobStore::CreateResult::Created);
+    std::vector<RcpVar> rcpvars = {
+        {"setpoint_c",   secsgem::secs2::Item::f4(350.5f)},
+        {"chamber_name", secsgem::secs2::Item::ascii("CHAMBER-A")},
+        {"flags",        secsgem::secs2::Item::u4(uint32_t{0xDEADBEEFu})},
+    };
+    std::vector<ProcessParam> params = {
+        {"recipe_var",   secsgem::secs2::Item::list({
+                            secsgem::secs2::Item::ascii("nested"),
+                            secsgem::secs2::Item::u2(uint16_t{42})})},
+    };
+    REQUIRE(s.set_e40_extras("PJ-X", MaterialFlag::Substrate,
+                              ProcessRecipeMethod::RecipeOnly,
+                              rcpvars, params));
+  }
+  {
+    ProcessJobStore s;
+    s.enable_persistence(dir);
+    const auto* pj = s.get("PJ-X");
+    REQUIRE(pj);
+    REQUIRE(pj->rcpvars.size() == 3);
+    CHECK(pj->rcpvars[0].name == "setpoint_c");
+    CHECK(pj->rcpvars[0].value.format() == secsgem::secs2::Format::F4);
+    CHECK(pj->rcpvars[1].value.as_ascii() == "CHAMBER-A");
+    CHECK(pj->rcpvars[2].name == "flags");
+
+    REQUIRE(pj->prprocessparams.size() == 1);
+    CHECK(pj->prprocessparams[0].name == "recipe_var");
+    CHECK(pj->prprocessparams[0].value.is_list());
+    CHECK(pj->prprocessparams[0].value.as_list().size() == 2);
+    CHECK(pj->prprocessparams[0].value.as_list()[0].as_ascii() == "nested");
+  }
+  fs::remove_all(dir);
+}
+
+TEST_CASE("ProcessJobStore persistence: v1 record (no extras) still loads") {
+  auto dir = scratch_dir("pj-v1");
+  // Hand-craft a v1 record on disk.
+  {
+    std::ofstream f(dir / "0000000001.pj", std::ios::binary);
+    // [magic=0xC7][version=1][state=Queued=0][alert=1][mf=0][prrecipemethod=0]
+    const uint8_t header[6] = {0xC7, 0x01, 0x00, 0x01, 0x00, 0x00};
+    f.write(reinterpret_cast<const char*>(header), 6);
+    auto write_str = [&](const std::string& s) {
+      uint16_t n = static_cast<uint16_t>(s.size());
+      uint8_t lb[2] = {static_cast<uint8_t>(n >> 8), static_cast<uint8_t>(n)};
+      f.write(reinterpret_cast<const char*>(lb), 2);
+      f.write(s.data(), n);
+    };
+    write_str("PJ-LEGACY");   // prjobid
+    write_str("RCP-OLD");     // ppid
+    const uint8_t mc[2] = {0x00, 0x00};
+    f.write(reinterpret_cast<const char*>(mc), 2);
+    // No v2 trailer.
+  }
+  ProcessJobStore s;
+  s.enable_persistence(dir);
+  REQUIRE(s.has("PJ-LEGACY"));
+  CHECK(s.get("PJ-LEGACY")->ppid == "RCP-OLD");
+  CHECK(s.get("PJ-LEGACY")->rcpvars.empty());
+  CHECK(s.get("PJ-LEGACY")->prprocessparams.empty());
+  fs::remove_all(dir);
+}
+
 TEST_CASE("ControlJobStore persistence: remove deletes journal file") {
   auto dir = scratch_dir("cj-remove");
   ControlJobStore s;