#pragma once #include #include #include #include #include #include #include #include #include #include #include #include #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 { // One Process Job record. The FSM is heap-allocated through unique_ptr so // the per-PJ state-change handler can capture a stable pointer to // `this`-style state without invalidation on map rehash. // // The MF / recipe-method / rcp-vars / process-params fields are the // optional E40-0705 §10.2 trailers on S16F11. Simple callers leave // them defaulted; tools that actually need recipe-variable tuning or // per-job process parameters populate them. struct RcpVar { std::string name; secs2::Item value; }; struct ProcessParam { std::string name; secs2::Item value; }; struct ProcessJob { std::string prjobid; std::string ppid; // recipe identifier std::vector mtrloutspec; // material identifiers bool alert_enabled = true; // S16F9 alerts on/off MaterialFlag mf = MaterialFlag::Substrate; ProcessRecipeMethod prrecipemethod = ProcessRecipeMethod::RecipeOnly; std::vector rcpvars; std::vector prprocessparams; std::unique_ptr fsm; std::filesystem::path journal_path; }; class ProcessJobStore { public: using TransitionTableFactory = std::function; using StateChangeHandler = std::function; ProcessJobStore() : factory_([] { return ProcessJobTransitionTable::default_table(); }) {} // The per-PJ FSM closes over `this`, so the store must keep a stable // address. unique_ptr makes copying impossible anyway; moves would // silently dangle the per-PJ lambdas — disallow them explicitly. ProcessJobStore(const ProcessJobStore&) = delete; ProcessJobStore& operator=(const ProcessJobStore&) = delete; ProcessJobStore(ProcessJobStore&&) = delete; ProcessJobStore& operator=(ProcessJobStore&&) = delete; void set_table_factory(TransitionTableFactory f) { factory_ = std::move(f); } void set_state_change_handler(StateChangeHandler h) { on_change_ = std::move(h); } enum class CreateResult { Created, Denied_AlreadyExists, Denied_InvalidPpid, }; // Validate `ppid` against the optional callback; if accepted, create // the PJ in Queued and fire the Created -> Queued change handler. CreateResult create(std::string prjobid, std::string ppid, std::vector materials, const std::function& ppid_exists = [](const std::string&) { return true; }) { if (jobs_.count(prjobid)) return CreateResult::Denied_AlreadyExists; if (!ppid_exists(ppid)) return CreateResult::Denied_InvalidPpid; install_(prjobid, std::move(ppid), std::move(materials), /*alert=*/true, MaterialFlag::Substrate, ProcessRecipeMethod::RecipeOnly, ProcessJobState::Queued); order_.push_back(prjobid); if (persistent_) { write_record_(prjobid); write_order_(); } // Synthetic NoState -> Queued so subscribers observe creation. if (on_change_) { on_change_(jobs_.find(prjobid)->first, ProcessJobState::NoState, ProcessJobState::Queued, ProcessJobEvent::Created); } return CreateResult::Created; } // After `create`, populate the optional E40-0705 fields on the new PJ. // Returns true if the PJ exists. These fields don't influence the FSM; // they're carried so the tool's recipe engine can read them later. bool set_e40_extras(const std::string& prjobid, MaterialFlag mf, ProcessRecipeMethod prrecipemethod, std::vector rcpvars, std::vector params) { auto it = jobs_.find(prjobid); if (it == jobs_.end()) return false; it->second.mf = mf; it->second.prrecipemethod = prrecipemethod; it->second.rcpvars = std::move(rcpvars); it->second.prprocessparams = std::move(params); if (persistent_) write_record_(prjobid); return true; } bool has(const std::string& prjobid) const { return jobs_.count(prjobid) > 0; } const ProcessJob* get(const std::string& prjobid) const { auto it = jobs_.find(prjobid); return it == jobs_.end() ? nullptr : &it->second; } ProcessJob* get(const std::string& prjobid) { auto it = jobs_.find(prjobid); return it == jobs_.end() ? nullptr : &it->second; } ProcessJobState state(const std::string& prjobid) const { auto it = jobs_.find(prjobid); return it == jobs_.end() ? ProcessJobState::NoState : it->second.fsm->state(); } // Host-initiated S16F5 PRJobCommand. Returns InvalidObject for unknown // PJs; CannotDoNow when the current PJ state has no row for the event. // For HOQ the FSM gates legality (Queued only) and the store performs // the actual reorder so the next CJ Select picks this PJ first. HostCmdAck on_host_command(const std::string& prjobid, ProcessJobEvent event) { auto* pj = get(prjobid); if (!pj) return HostCmdAck::InvalidObject; HostCmdAck ack = pj->fsm->on_host_command(event); if (ack == HostCmdAck::Accept) { if (event == ProcessJobEvent::HeadOfQueue) { move_to_head(prjobid); if (persistent_) write_order_(); } else if (persistent_) { write_record_(prjobid); } } return ack; } // Position of `prjobid` in the queue order (insertion order; HOQ rewrites // it). Returns -1 if unknown. Exposed for tests and CJ promotion logic. int position(const std::string& prjobid) const { auto it = std::find(order_.begin(), order_.end(), prjobid); return it == order_.end() ? -1 : static_cast(it - order_.begin()); } // Internal events from the application (recipe runner etc.). bool fire_internal(const std::string& prjobid, ProcessJobEvent event) { auto* pj = get(prjobid); if (!pj) return false; bool ok = pj->fsm->on_internal(event); if (ok && persistent_) write_record_(prjobid); return ok; } // Dequeue S16F13: only legal while QUEUED. HostCmdAck dequeue(const std::string& prjobid) { auto it = jobs_.find(prjobid); if (it == jobs_.end()) return HostCmdAck::InvalidObject; if (it->second.fsm->state() != ProcessJobState::Queued) return HostCmdAck::CannotDoNow; if (persistent_ && !it->second.journal_path.empty()) { std::error_code ec; std::filesystem::remove(it->second.journal_path, ec); } erase_from_order(prjobid); jobs_.erase(it); if (persistent_) write_order_(); return HostCmdAck::Accept; } // Remove a terminal job from the store (caller responsibility to ensure // ProcessComplete before deleting). Used after CJ Completed cleanup. bool remove(const std::string& prjobid) { auto it = jobs_.find(prjobid); if (it == jobs_.end()) return false; if (persistent_ && !it->second.journal_path.empty()) { std::error_code ec; std::filesystem::remove(it->second.journal_path, ec); } erase_from_order(prjobid); jobs_.erase(it); if (persistent_) write_order_(); return true; } // Per-PJ S16F9 alert gate. E40 §10.3 leaves PRALERT control to the host // (S16F1/F2 in the full multi-create form, which we don't model yet); // exposed here so application code can toggle alerts directly. bool set_alert(const std::string& prjobid, bool enabled) { auto* pj = get(prjobid); if (!pj) return false; pj->alert_enabled = enabled; if (persistent_) write_record_(prjobid); return true; } std::size_t size() const { return jobs_.size(); } // Insertion-order list of PRJOBIDs, mutated by HOQ. Used by CJ promotion // logic and by tests that need to assert ordering. const std::vector& ids() const { return order_; } // ---- Persistence ---------------------------------------------------- // One file per PJ (`.pj`); a single `order.idx` index records the // queue ordering (insertion-order modulo HOQ). Replay reads the index // first to preserve ordering across restarts. Version 2 of the record // (current) journals the E40 extras (rcpvars + prprocessparams) by // round-tripping each value through secs2::encode/decode. Version 1 // 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); persistent_ = true; std::error_code ec; fs::create_directories(journal_dir_, ec); std::map staging; std::map paths; std::vector> existing; for (auto& entry : fs::directory_iterator(journal_dir_, ec)) { if (!entry.is_regular_file()) continue; const auto& p = entry.path(); if (p.extension() != ".pj") continue; uint64_t seq = 0; try { seq = std::stoull(p.stem().string()); } catch (...) { continue; } existing.emplace_back(seq, p); } std::sort(existing.begin(), existing.end(), [](const auto& a, const auto& b) { return a.first < b.first; }); for (auto& [seq, path] : existing) { auto rec = load_record_(path); if (!rec) { fs::remove(path, ec); continue; } 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; pj->rcpvars = std::move(rec->rcpvars); pj->prprocessparams = std::move(rec->prprocessparams); } if (seq >= next_seq_) next_seq_ = seq + 1; } // Rebuild order_ from the index file when present; otherwise fall // back to creation order (already insertion-sorted by seq above). std::vector idx = read_order_(); order_.clear(); if (!idx.empty()) { for (auto& id : idx) if (jobs_.count(id)) order_.push_back(id); // Append any PJs not in the index (defensive: shouldn't happen). for (auto& kv : jobs_) { if (std::find(order_.begin(), order_.end(), kv.first) == order_.end()) order_.push_back(kv.first); } } else { for (auto& kv : jobs_) order_.push_back(kv.first); } } bool persistence_enabled() const { return persistent_; } std::filesystem::path journal_dir() const { return journal_dir_; } private: struct Record { std::string prjobid; std::string ppid; std::vector materials; bool alert; MaterialFlag mf; ProcessRecipeMethod prrecipemethod; ProcessJobState state; std::vector rcpvars; std::vector prprocessparams; }; // PJ record (version 2): // [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 = 0x02; void install_(const std::string& prjobid, std::string ppid, std::vector materials, bool alert, MaterialFlag mf, ProcessRecipeMethod prrecipemethod, ProcessJobState restored) { auto fsm = std::make_unique(factory_(), restored); fsm->set_state_change_handler( [this, id = prjobid](ProcessJobState from, ProcessJobState to, ProcessJobEvent trig) { if (on_change_) on_change_(id, from, to, trig); }); ProcessJob pj; pj.prjobid = prjobid; pj.ppid = std::move(ppid); pj.mtrloutspec = std::move(materials); pj.alert_enabled = alert; pj.mf = mf; pj.prrecipemethod = prrecipemethod; pj.fsm = std::move(fsm); jobs_.emplace(prjobid, std::move(pj)); } void write_record_(const std::string& prjobid) { namespace fs = std::filesystem; auto* pj = get(prjobid); if (!pj) return; if (pj->journal_path.empty()) { char namebuf[32]; std::snprintf(namebuf, sizeof(namebuf), "%010llu.pj", static_cast(next_seq_++)); pj->journal_path = journal_dir_ / namebuf; } fs::path tmp = pj->journal_path; tmp += ".tmp"; std::ofstream out(tmp, std::ios::binary | std::ios::trunc); if (!out) return; uint8_t header[6] = { kMagic, kVersion, static_cast(pj->fsm->state()), static_cast(pj->alert_enabled ? 1 : 0), static_cast(pj->mf), static_cast(pj->prrecipemethod)}; out.write(reinterpret_cast(header), sizeof(header)); auto write_str = [&](const std::string& str) { uint16_t n = static_cast(str.size()); uint8_t lb[2] = {static_cast(n >> 8), static_cast(n)}; out.write(reinterpret_cast(lb), 2); out.write(str.data(), static_cast(n)); }; write_str(pj->prjobid); write_str(pj->ppid); auto write_be16 = [&](uint16_t v) { uint8_t b[2] = {static_cast(v >> 8), static_cast(v)}; out.write(reinterpret_cast(b), 2); }; auto write_be32 = [&](uint32_t v) { uint8_t b[4] = {static_cast((v >> 24) & 0xFF), static_cast((v >> 16) & 0xFF), static_cast((v >> 8) & 0xFF), static_cast(v & 0xFF)}; out.write(reinterpret_cast(b), 4); }; write_be16(static_cast(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(enc.size())); out.write(reinterpret_cast(enc.data()), static_cast(enc.size())); }; write_be16(static_cast(pj->rcpvars.size())); for (const auto& v : pj->rcpvars) write_named(v.name, v.value); write_be16(static_cast(pj->prprocessparams.size())); for (const auto& v : pj->prprocessparams) write_named(v.name, v.value); out.flush(); out.close(); std::error_code ec; fs::rename(tmp, pj->journal_path, ec); if (ec) fs::remove(tmp, ec); } static std::optional load_record_(const std::filesystem::path& p) { std::ifstream in(p, std::ios::binary); if (!in) return std::nullopt; uint8_t header[6]; in.read(reinterpret_cast(header), sizeof(header)); 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(header[2]); r.alert = header[3] != 0; r.mf = static_cast(header[4]); r.prrecipemethod = static_cast(header[5]); auto read_str = [&](std::string& out) { uint8_t lb[2]; in.read(reinterpret_cast(lb), 2); if (!in) return false; uint16_t n = static_cast((uint16_t(lb[0]) << 8) | lb[1]); out.resize(n); in.read(out.data(), n); return static_cast(in) || n == 0; }; auto read_be16 = [&](uint16_t& v) { uint8_t b[2]; in.read(reinterpret_cast(b), 2); if (!in) return false; v = static_cast((uint16_t(b[0]) << 8) | b[1]); return true; }; auto read_be32 = [&](uint32_t& v) { uint8_t b[4]; in.read(reinterpret_cast(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; uint16_t mc = 0; if (!read_be16(mc)) return std::nullopt; 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& 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 bytes(enc_len); if (enc_len) { in.read(reinterpret_cast(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 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; } void write_order_() { namespace fs = std::filesystem; fs::path target = journal_dir_ / "order.idx"; fs::path tmp = target; tmp += ".tmp"; std::ofstream out(tmp, std::ios::binary | std::ios::trunc); if (!out) return; for (const auto& id : order_) { out << id << '\n'; } out.flush(); out.close(); std::error_code ec; fs::rename(tmp, target, ec); if (ec) fs::remove(tmp, ec); } std::vector read_order_() const { std::vector out; std::ifstream in(journal_dir_ / "order.idx"); if (!in) return out; std::string line; while (std::getline(in, line)) { if (!line.empty()) out.push_back(line); } return out; } void erase_from_order(const std::string& prjobid) { auto it = std::find(order_.begin(), order_.end(), prjobid); if (it != order_.end()) order_.erase(it); } void move_to_head(const std::string& prjobid) { auto it = std::find(order_.begin(), order_.end(), prjobid); if (it == order_.end() || it == order_.begin()) return; std::rotate(order_.begin(), it, it + 1); } std::map jobs_; std::vector order_; // queue position (E40 HOQ-aware) TransitionTableFactory factory_; StateChangeHandler on_change_; bool persistent_ = false; std::filesystem::path journal_dir_; uint64_t next_seq_ = 0; }; } // namespace secsgem::gem