From b1772cfefd11824a7b3b1c8a5244e49b7e53cdfb Mon Sep 17 00:00:00 2001 From: Raphael Maenle Date: Wed, 10 Jun 2026 23:53:45 +0200 Subject: [PATCH] =?UTF-8?q?feat(daemon):=20Phase=20D=20=E2=80=94=20GEM300?= =?UTF-8?q?=20in-the-loop=20(jobs,=20recipes,=20EC=20changes)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Semantics settled and documented: v1 is observe-and-report. The engine keeps acking S16/S3/S7/S2F15 from its FSM tables — exactly the behaviour both reference implementations validated — while the tool observes lifecycle events on the Subscribe stream and reports physical progress back. Gating stays the documented v2 deferred-reply item. Engine: two new store observers (HandlerSlot pattern) — RecipeStore fires (ppid, body) after an add (S7F3 downloads), EquipmentConstantStore fires (id, value) on ACCEPTED S2F15 writes only. Unit-tested. Daemon: the service registers PJ/recipe/EC observers (io thread; add_ observers coexist with register_default_handlers' primaries) and fans the new HostRequest variants out via push_request (fire-and-forget, no- buffering contract). ProcessJob carries action (Start->START, Resume-> RESUME, Paused->PAUSE, Stopping->STOP, Aborting->ABORT) + recipe + material bindings read store-side on the io thread. ReportProcessJob maps SETTING_UP ->SetupComplete, COMPLETE->ProcessComplete, ABORTED->AbortComplete via read_sync; PROCESSING is informational; unknown job => INVALID_OBJECT, table-rejected transition => CANNOT_DO_NOW. Carriers deferred (CarrierStore has no observer machinery; ReportCarrier stays UNIMPLEMENTED) — roadmap. Python client: on_process_job / on_recipe / on_constant_change decorators + report_job(job_id, state); ProcessJob dataclass exported. Tests: daemon suite 141 -> 175 assertions — the full in-process loop (S16F11 create -> tool setup -> S16F5 PJSTART -> stream ProcessJob with recipe+carriers -> ReportProcessJob(COMPLETE) -> FSM at ProcessComplete), rejection paths, S7F3 -> ProcessProgram, S2F15 -> ConstantChange with the configured name. Core 475/3097 (observer units). Live regression: daemon interop 20 checks + pyclient 13 checks still green against the running daemon. Co-Authored-By: Claude Fable 5 --- clients/python/secsgem_client/__init__.py | 4 +- .../__pycache__/__init__.cpython-312.pyc | Bin 663 -> 690 bytes .../__pycache__/_client.cpython-312.pyc | Bin 14574 -> 17719 bytes clients/python/secsgem_client/_client.py | 54 ++++++++- docs/DAEMON_ROADMAP.md | 29 ++++- include/secsgem/daemon/equipment_service.hpp | 95 ++++++++++++++++ .../secsgem/gem/store/equipment_constants.hpp | 10 ++ include/secsgem/gem/store/recipes.hpp | 12 +- tests/test_daemon_service.cpp | 107 ++++++++++++++++++ tests/test_data_model.cpp | 32 ++++++ 10 files changed, 333 insertions(+), 10 deletions(-) diff --git a/clients/python/secsgem_client/__init__.py b/clients/python/secsgem_client/__init__.py index f3b7279..276914a 100644 --- a/clients/python/secsgem_client/__init__.py +++ b/clients/python/secsgem_client/__init__.py @@ -6,7 +6,7 @@ stream/function, a format code, or a numeric id — just the names from your equipment.yaml. """ -from ._client import Command, Equipment, Health, SecsGemError +from ._client import Command, Equipment, Health, ProcessJob, SecsGemError -__all__ = ["Equipment", "Command", "Health", "SecsGemError"] +__all__ = ["Equipment", "Command", "Health", "ProcessJob", "SecsGemError"] __version__ = "0.1.0" diff --git a/clients/python/secsgem_client/__pycache__/__init__.cpython-312.pyc b/clients/python/secsgem_client/__pycache__/__init__.cpython-312.pyc index b48fc4eedad61f36ceddd8fd2455cdec71c8f6bb..fc850a211de058a54450ba02267ef5ae941f72af 100644 GIT binary patch delta 177 zcmbQvx`~zdG%qg~0}xc*(aicdkynz@V4}LbP%c{(8zVzHLkjaE#whj_mMD&mAwGB5>>1Q zdWL!iewv(^JyC8@Y>kfhqu<$tQ3C delta 175 zcmdnQI-QmGG%qg~0}$|>*UY*;kynyYXQH}1A0tCLLkjaE#wfNFmMHd(5k8FGEVtO5 z^K)|(^HOecx)zpZ7UZVpmE2OYM^JRXW*yFF^NfSvNltM9v@JY5r~UrfW!x8Mn=YuOl*uS-vt;L?JhIeKH{=z PV86j9+Q?nR15^M2otr8& diff --git a/clients/python/secsgem_client/__pycache__/_client.cpython-312.pyc b/clients/python/secsgem_client/__pycache__/_client.cpython-312.pyc index bbfeca5417c1fdcdb3858c951156e0b8da34b8d3..d03b2d7a2aafeee1e37a7a4d6735bc83f24685b1 100644 GIT binary patch delta 4983 zcma)Adu$xV8QFd;Ln{#7=@C!EyYElNjegXdKeZXD)1NRcxZr zEzsQK0?j)wqLd({s934(Mp#1az-1se z?vS0jtDnWgimLX;4<^3X9+NuPtdvH~A!#UnP|^k!X)vy8(pV^_sS+EFNwsk%sakww zM4>A}2jfgrsN5uNj-#wgf;KWZu13P4s5BO5M;dTAmXw$>0-~tUJt-9U5ca+(7)k54b21UrM1)p)JJ7nN4?-oKCP!d7%2dn544ar z&;p=Euy-1D=fb-g#nAbqS+|ir{Ijh8wo+k22pU@zeKLB>Qr?-p%ydsHNpnM`{F9vG za#-k24~+}5*e4S`LyHV6@lj3A;f?OYt1@v91P4Pg8daF?4w~l9DHe-5- zbN7h-Hqfdjxijb1g09{Z)=Q3}xh>m~mdE}bf*zBRQ6kPRYWbh6!C6aikhsed2cX)=;KfX-EsP5cabWY7q=4 zom~;8blw4HBV|mtDcECop@@RefN+Q>i)Au7{e1BXA}3%3kHn%H2gKSQ4Y6V7g*n}3 z@?)Ho){qCdT>9yC*0?&NN{B74V)vk-F}|j(hCDQVd)Wp859o>Vz2zM+o+wz@C{6Hc zG!~Br;r(dJ>FK{4ucV0|0zuP1*ERRh@VfK1cRL1dN?6 zXk|ab?lK5{|=`}TX60(E8SldTl;$PKPlP7tl+-QFaWuE4H*VTT)sn)uc>pcfdjK>(7K_G1RH0G`ZVAltQtik{gf>VA<8)kZ zf_Qd;8TM0vs{(O?yXJyK(sNZ1=IE=q0Q1c;;o}OwnGLS%{kL-d=Ic2hhD3E3JWqyV zl9Im6q;!gsYGa{cr9rv_H@*?b?&}7U2Z5y51WlER(yIcxX7*1@BJ)>5WL1vxxy8_6 zBQ7*MaL_07ZwUNfSyiGY@co6Q{MHN~&uj1$PKf7;Df@Ght{x^Wnv*|~QO^Gv@bMdp z3X?CScU}5o$I2jmliwf(c6ILFy``@^cx!(^nveqhJ8$Xe+_eivpu4bh10A<)@88zB zyVHja*B6ldNaW=3HG17=M97=`1rCp+d7@m4)l0&=_}&&r-vDv72jG}+#UZ$JXS1c5Y-!F@bT)6+ zzk0^Mde*;w#=m~nzj4OD@sGBP{;n$);_e|QEPr!kpYUj>5)US3-E}kWx^t^9x>x+g zv2yyfe3+E(0hPLa_aIY3RL=}zZKTA4(Rh4>-O4@n9ppWJTm7dSz5`djRg*IKU{m0WPT-2=m-9QV%m>PgX^GIgCh9Ie2(~b%ysmK6Pm69ZQ6hukizo1vS?62;h61M4Qe-#6cqKoR z;jxgRD()}$km{>?wP++OmC%TAnw%^&G}{(5?*I*TOLdw&>xA+8l&%ItbCVIms`JBeCCpfEG1 z8?p4grqPVFiAk&0mY$rj4BePgg$g$q%G)`~Elr-XowA255g{z@hfCmgcsP@`N&AFN zTMTW%Yc8JjScR}<(mrXM5YjJdGJ`rM9Zq4wdXGZ{oLF#y2wLZY3mMc!o##_Yi;1Vj zlYI~f={Eej5ua_)vE-47C!|=10T^gomk~k*eR>QItA_fJ)hF z0M!YAxzlJy0%b43P$GXZ@hr_`ALE)q*nLoih|8V%bNR(nLuWW$)s*!3rrOT5JvMpC z(|Dn3wz=c|=8n0-;-|77&pzEg=PjJ|E}!u(KRYy6SoDV$GT>g6M03(#KaeXPSDiV&x^zsta_s zwfFtj-lSopcE($KE^k(Dn~~dQ<(p^Zo9E>ErCmtnzD?v4Z)r-&V~Iiq7{} zbQ&t^X1sOhnt$2$eA{e&+f04ihxIKN4!(TkS4U=ByJuRvFV^> z{iT9N@&SKx)lbM@d3&?Z*+GPYz2rah4g9;!qdT@%39sjNuLkm_HNUqa=gs0}$XC01 zmy2(%%tC%+1>jp1cEsgbz0J0_nk~I8;#(~ZTm81T{T3L1yQ*oM&+?9!Z1YW5t?tCK ze`w91=pg*knppW>croUe`T%gc?GRLP8iQ0Jz^!?JA82VL5&lTaS~AK1+~Q9f#eM~@ zX+>Cza-zv&sh{E#Ia{FC#~OKGj}nS%gZp8^*gLoscr0fVsY#7NK{>7|>euBe7S*XJmQu68z!#Wt{~O*LMZh(2kS+-M*WejzLHJsL z5Nr}Z8sEcNV=bP9YlVxMgrk0h074~#2f?_!jl(#K-8&KPMz|N@egNG8sbxfA+PL{u z^aGrD9N`IsXAE6uk$o279KwqTzd^v8pS{FCYAtN&LKXq@gL(?!xbT5>>&Mk==jxg+ zTU@r9xj@Z4(krFhzjo`Yd9T%$H=k(%a2a5?t-6xqu$9hxMO*oNX0lO73kFDFcA%l1>3jp&LL0ooO#I6^(FGV;lmj!^-1aMzw-XW6h#@N}jEym_!LZ&11BRdkC@#jH_)PGcwbwZ_ z#)jBHkQJeXD6xL2E)OB5Y1KBCsz#frQXeW+1S<6sDH23v)K*a*D)l8dP^I>%=O06Y zqGqJ|{{Q^fbI!~;zkP*%{(}F{KA)F}=fQ{S071AqYm0ryftZK=*X3s** z!^e`WT6p{{#e726u&y-o3njoZtVYOM*2p%nfM{07n#3ZU!$BeIgbcBbEGT4HEM*hy zsyQufrm)($0X~mc!QGm_N@f_gcO@Vbbf*@`X!XOEW6%jv=tGs;p;?3EgrU)7p$hb&Hh-W zBD@(tLnpygzl)YCgY{u4a|r8TG0Vv`Q{z9xu@BLL*bf&H$$)|CNyKKv2&^SkTCUtp zJV8}1EQeJ2VYnDrZn3D{C5=z<5Dw*ay$oAYSvn6#Q=k2x9B;0qYy8_RehwRzVJw}c z3zaMBomA|^A2TOtA2c>5=;g}J#(&c2OBi(7ZgNa3&FYWl+}o5yYbA5So*%!2fxTG7bFesl<{48nbsk0I4-nt<;C(pU2}ha_($_23 zn-5EUSdm{v;7Y%(#9zbV_t!x0>(Hup(Ur=WI#1~V_@wm^{TZ~hHPau%;kGTx3dZ~d zF1Kx|Y;k5vpEldaOJzfU58mCHbIm*OM0()q>>vzuMjdp-2d6quIt_ZY z187S`S(FyYWB}f657HnsdV>yWG6Z9Rdha|n8Yp?pS^HEFzUcA<79A!o-F_CN$=Ekp zA{&ZbVQDf6r?w|SkHlyiE<{p=1+wN`k6%=nyy)_f1;3k$J}nGDn5~d%79P@Ais6)@$zZ3e7OKEdihQ3{p3=?q~>d{HPXfeYU_mgrp0%qFe5Pr|+k!yA8efVu`o>*Zix-FV5aL`%C_DtmCUJiJqYUxgUT( z_01kBxXB-C_r!#_sYddSz|CD%LjR!3oA*c`gng(tr3G&}6r{(OkI1(oj(l9Y72jNt z<=e7D*l)Yr3q6k8+i9Vv^2W|Q-UT1;nvm`g=<;_3Jqj1zm(ilwmR>0DP6zCb!CERU2>&W{?rBWF|0sjwmhmox zozsnh5y<$$7lUZDkDUFnrf@ieIEy%sSQN0_GrT;bbMxH#b@B>2mJxushPZ)v8*v@+ z6ru-_MA+Y`A2r0Fcnwj#fh>_voCiK_?!MRB{-whuXYXyu-beW$0^PX-J067Ga^ikS zl2iAqV{+s{rz8*2^@>hhu5A&CyruZb<}2LA)YH1GHT diff --git a/clients/python/secsgem_client/_client.py b/clients/python/secsgem_client/_client.py index b9e3cba..002ce00 100644 --- a/clients/python/secsgem_client/_client.py +++ b/clients/python/secsgem_client/_client.py @@ -63,6 +63,17 @@ class Command: self._done = True +@dataclass +class ProcessJob: + """An E40 process job the host wants run (or stopped/aborted). Do the + physical work, then report progress with Equipment.report_job().""" + + id: str + action: str # "START" | "STOP" | "PAUSE" | "RESUME" | "ABORT" + recipe: str + carriers: list + + @dataclass class Health: link: str # "DISCONNECTED" | "CONNECTED" | "SELECTED" @@ -79,6 +90,9 @@ class Equipment: grpc.channel_ready_future(self._channel).result(timeout=connect_timeout) self._stub = rpc.EquipmentStub(self._channel) self._handlers: Dict[str, Callable[[Command], object]] = {} + self._job_handler: Optional[Callable[[ProcessJob], object]] = None + self._recipe_handler: Optional[Callable[[str, bytes], object]] = None + self._constant_handler: Optional[Callable[[str, object], object]] = None self._listen_thread: Optional[threading.Thread] = None self._stop = threading.Event() @@ -167,6 +181,30 @@ class Equipment: return register + def on_process_job(self, fn: Callable[[ProcessJob], object]): + """Decorator: the host wants a process job run/stopped (E40). + Report progress with report_job().""" + self._job_handler = fn + return fn + + def on_recipe(self, fn: Callable[[str, bytes], object]): + """Decorator: the host downloaded a recipe (ppid, body).""" + self._recipe_handler = fn + return fn + + def on_constant_change(self, fn: Callable[[str, object], object]): + """Decorator: the host changed an equipment constant (name, value).""" + self._constant_handler = fn + return fn + + def report_job(self, job_id: str, state: str) -> None: + """Report physical job progress: "SETTING_UP" | "PROCESSING" | + "COMPLETE" | "ABORTED". The daemon drives the E40 state machine and + notifies the host (S16F9).""" + req = pb.ProcessJobState( + job_id=job_id, state=pb.ProcessJobState.State.Value(state)) + _check(self._stub.ReportProcessJob(req)) + def listen(self, background: bool = False) -> None: """Consume host requests and dispatch them to @eq.on handlers. Blocks; pass background=True to run on a daemon thread instead.""" @@ -188,8 +226,22 @@ class Equipment: for hr in self._stub.Subscribe(pb.SubscribeRequest(client="secsgem_client")): if self._stop.is_set(): return + if hr.HasField("process_job") and self._job_handler: + j = hr.process_job + self._job_handler(ProcessJob( + j.job_id, pb.ProcessJob.Action.Name(j.action), + j.recipe, list(j.carriers))) + continue + if hr.HasField("process_program") and self._recipe_handler: + self._recipe_handler(hr.process_program.ppid, + hr.process_program.body) + continue + if hr.HasField("constant") and self._constant_handler: + self._constant_handler(hr.constant.name, + from_value(hr.constant.value)) + continue if not hr.HasField("command"): - continue # future HostRequest variants (jobs, carriers, ...) + continue # carriers etc.: future variants cmd = hr.command handler = self._handlers.get(cmd.name) or self._handlers.get("*") command = Command(cmd.id, cmd.name, diff --git a/docs/DAEMON_ROADMAP.md b/docs/DAEMON_ROADMAP.md index 40273fb..373e5ea 100644 --- a/docs/DAEMON_ROADMAP.md +++ b/docs/DAEMON_ROADMAP.md @@ -202,12 +202,29 @@ debts tax every later phase, and the most valuable tests aren't automated. `cpp_mini_tool` is the worked example. ### Phase D — GEM300 in-the-loop (process/carrier tools) -10. ⬜ Settle job/carrier semantics (who acks S16F5/S3F17, gate vs observe — - see proto comments), then wire `ProcessJob`/`CarrierAction` onto the - stream + `ReportProcessJob`/`ReportCarrier` into the PJ/CJ/carrier stores. -11. ⬜ Recipe download (`ProcessProgram` on the stream when S7F3 lands) and - EC-change notification (`ConstantChange` when S2F15 lands). -12. ⬜ Interop scenarios for jobs/carriers vs secsgem-py + secs4j. +10. 🚧 Semantics SETTLED: v1 is **observe-and-report** — the engine keeps + acking S16/S3/S7/S2F15 from its FSM tables (what both references + validated); the tool observes lifecycle events on the stream and reports + physical progress back. Gating (tool decides the ack) = the documented + v2 deferred-reply item. ✅ `ProcessJob` on the stream (PJ store observer: + →Processing/Start=START, /Resume=RESUME, →Paused=PAUSE, →Stopping=STOP, + →Aborting=ABORT; carries recipe + material bindings) + + `ReportProcessJob` (SETTING_UP→SetupComplete, COMPLETE→ProcessComplete, + ABORTED→AbortComplete; PROCESSING informational; INVALID_OBJECT / + CANNOT_DO_NOW on unknown job / illegal transition). ⬜ Carriers deferred: + `CarrierStore` has no observer machinery yet — apply the HandlerSlot + pattern, then mirror the job wiring; `ReportCarrier` stays gRPC + UNIMPLEMENTED (the stub default) until then. +11. ✅ `ProcessProgram` on the stream when S7F3 lands (new RecipeStore + added-observer) and `ConstantChange` when S2F15 is ACCEPTED (new + EquipmentConstantStore changed-observer; rejected writes never fire). + Python client: `on_process_job`/`on_recipe`/`on_constant_change` + + `report_job`. +12. 🚧 In-process E2E covers the full job loop (S16F11→S16F5→stream→report→ + FSM, 175-assertion daemon suite); secs4j's 55 checks run against the + daemon (declarative path). ⬜ A live host-driven job loop with a + subscribed tool needs raw S16 frames host-side (secsgem-py 0.3.0 lacks + S16 builders — see interop/raw_gem300_harness.py for the frame source). ### Phase E — hardening & operations 13. ⬜ gRPC exposure: default to localhost + document UDS; optional TLS creds. diff --git a/include/secsgem/daemon/equipment_service.hpp b/include/secsgem/daemon/equipment_service.hpp index 2fe31a2..c2f4990 100644 --- a/include/secsgem/daemon/equipment_service.hpp +++ b/include/secsgem/daemon/equipment_service.hpp @@ -189,6 +189,54 @@ class EquipmentService final : public pb::Equipment::Service { bump_health(); }); + for (const auto& ec : rt.model().ecids.all()) + ecnames_.insert({ec.id, ec.name}); + + // GEM300 in-the-loop (Phase D, observe-and-report): forward job + // lifecycle, recipe downloads, and accepted EC writes onto the + // Subscribe stream. Observers fire on the io thread; push_request only + // takes the subscriber mutex briefly. add_ observers coexist with the + // primary handlers register_default_handlers installs (HandlerSlot). + rt.model().process_jobs.add_state_change_handler( + [this](const std::string& prjobid, gem::ProcessJobState, + gem::ProcessJobState to, gem::ProcessJobEvent trig) { + pb::ProcessJob::Action action; + using PS = gem::ProcessJobState; + using PE = gem::ProcessJobEvent; + if (to == PS::Processing && trig == PE::Start) action = pb::ProcessJob::START; + else if (to == PS::Processing && trig == PE::Resume) action = pb::ProcessJob::RESUME; + else if (to == PS::Paused) action = pb::ProcessJob::PAUSE; + else if (to == PS::Stopping) action = pb::ProcessJob::STOP; + else if (to == PS::Aborting) action = pb::ProcessJob::ABORT; + else return; // SettingUp/complete/etc.: not tool work + pb::HostRequest hr; + auto* job = hr.mutable_process_job(); + job->set_job_id(prjobid); + job->set_action(action); + // Observer runs on the io thread — reading the store is safe. + if (const auto* pj = rt_.model().process_jobs.get(prjobid)) { + job->set_recipe(pj->ppid); + for (const auto& m : pj->mtrloutspec) job->add_carriers(m); + } + push_request(std::move(hr)); + }); + rt.model().recipes.add_added_handler( + [this](const std::string& ppid, const std::string& body) { + pb::HostRequest hr; + hr.mutable_process_program()->set_ppid(ppid); + hr.mutable_process_program()->set_body(body); + push_request(std::move(hr)); + }); + rt.model().ecids.add_changed_handler( + [this](uint32_t ecid, const s2::Item& value) { + pb::HostRequest hr; + auto it = ecnames_.find(ecid); + hr.mutable_constant()->set_name( + it != ecnames_.end() ? it->second : std::to_string(ecid)); + *hr.mutable_constant()->mutable_value() = from_item(value); + push_request(std::move(hr)); + }); + // Host commands -> the Subscribe stream (the HCACK-4 contract). For every // YAML-declared command, install a behaviour handler that pushes the // command to subscribed tool clients and answers the host immediately @@ -367,6 +415,43 @@ class EquipmentService final : public pb::Equipment::Service { return grpc::Status::OK; } + // E40 progress reports (Phase D, observe-and-report): the tool advances + // the physical work and tells the engine; the engine drives the FSM and + // emits the matching S16F9 alert / CEIDs to the host. PROCESSING is + // informational (the engine entered it when the host commanded Start). + grpc::Status ReportProcessJob(grpc::ServerContext*, const pb::ProcessJobState* req, + pb::Ack* resp) override { + const std::string job_id = req->job_id(); + const auto state = req->state(); + auto outcome = rt_.read_sync([this, job_id, state]() -> std::optional { + auto& pjs = rt_.model().process_jobs; + if (!pjs.has(job_id)) return std::nullopt; + switch (state) { + case pb::ProcessJobState::SETTING_UP: + return pjs.fire_internal(job_id, gem::ProcessJobEvent::SetupComplete); + case pb::ProcessJobState::COMPLETE: + return pjs.fire_internal(job_id, gem::ProcessJobEvent::ProcessComplete); + case pb::ProcessJobState::ABORTED: + return pjs.fire_internal(job_id, gem::ProcessJobEvent::AbortComplete); + default: + return true; // PROCESSING: informational + } + }); + if (!outcome) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("engine io thread did not answer (not running?)"); + } else if (!*outcome) { + resp->set_code(pb::Ack::INVALID_OBJECT); + resp->set_message("no process job '" + job_id + "'"); + } else if (!**outcome) { + resp->set_code(pb::Ack::CANNOT_DO_NOW); + resp->set_message("E40 table rejects that transition from the current state"); + } else { + resp->set_code(pb::Ack::ACCEPT); + } + return grpc::Status::OK; + } + // Operator-panel control-state transitions (e.g. "offline for maintenance"). // Fires the operator events on the io thread and reports what the E30 table // actually did: ACCEPT iff the equipment landed in the requested state, @@ -448,6 +533,15 @@ class EquipmentService final : public pb::Equipment::Service { } private: + // Fan a HostRequest out to every subscriber (fire-and-forget + // notifications: jobs, recipes, EC changes). No subscriber = dropped, + // matching the no-buffering contract. + void push_request(pb::HostRequest hr) { + std::lock_guard lk(subs_mu_); + for (auto& sub : subs_) sub->queue.push_back(hr); + if (!subs_.empty()) subs_cv_.notify_all(); + } + // The command-forwarding handler (runs on the io thread during S2F41/F21/ // F49 dispatch). Subscriber present -> queue the command + HCACK 4; absent // -> the command's declarative YAML ack. Never blocks. @@ -514,6 +608,7 @@ class EquipmentService final : public pb::Equipment::Service { std::map vars_; // SVIDs + DVIDs (SVIDs win on clash) std::map events_; // CEID by name std::map alarms_; // ALID by name AND stringified id + std::map ecnames_; // ECID -> name (ConstantChange) // WatchHealth plumbing: observers (io thread) bump the version and wake the // per-stream wait loops (gRPC threads). diff --git a/include/secsgem/gem/store/equipment_constants.hpp b/include/secsgem/gem/store/equipment_constants.hpp index d04b7c4..dd2572c 100644 --- a/include/secsgem/gem/store/equipment_constants.hpp +++ b/include/secsgem/gem/store/equipment_constants.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include #include @@ -9,6 +10,7 @@ #include #include +#include "secsgem/gem/handler_slot.hpp" #include "secsgem/secs2/item.hpp" namespace secsgem::gem { @@ -59,10 +61,18 @@ class EquipmentConstantStore { if (it == by_id_.end()) return EquipmentAck::Denied_UnknownEcid; if (!in_range(it->second, value)) return EquipmentAck::Denied_OutOfRange; it->second.value = std::move(value); + if (on_changed_) on_changed_(id, it->second.value); return EquipmentAck::Accept; } + // Observe ACCEPTED host writes (S2F15): fires after the value is stored. + // The tool reacts to process-parameter tuning; rejected writes don't fire. + using ChangedHandler = std::function; + void add_changed_handler(ChangedHandler h) { on_changed_.add(std::move(h)); } + private: + HandlerSlot on_changed_; + // For numeric formats, parse min_str / max_str as integers / doubles and // compare against the first value of the array (we don't support setting // multi-element ECs in this implementation). diff --git a/include/secsgem/gem/store/recipes.hpp b/include/secsgem/gem/store/recipes.hpp index 61e8a5a..d92f016 100644 --- a/include/secsgem/gem/store/recipes.hpp +++ b/include/secsgem/gem/store/recipes.hpp @@ -1,12 +1,14 @@ #pragma once #include +#include #include #include #include #include #include +#include "secsgem/gem/handler_slot.hpp" #include "secsgem/secs2/item.hpp" namespace secsgem::gem { @@ -45,8 +47,15 @@ struct FormattedRecipe { class RecipeStore { public: + // Observe recipe arrivals (e.g. a host S7F3 download): fires after the + // store is updated, with (ppid, body). Multi-observer via HandlerSlot — + // the gRPC daemon forwards these onto its Subscribe stream. + using AddedHandler = std::function; + void add_added_handler(AddedHandler h) { on_added_.add(std::move(h)); } + void add(std::string ppid, std::string body) { - by_ppid_.insert_or_assign(std::move(ppid), std::move(body)); + auto it = by_ppid_.insert_or_assign(std::move(ppid), std::move(body)).first; + if (on_added_) on_added_(it->first, it->second); } std::optional get(const std::string& ppid) const { auto it = by_ppid_.find(ppid); @@ -95,6 +104,7 @@ class RecipeStore { private: std::map by_ppid_; + HandlerSlot on_added_; std::map formatted_; }; diff --git a/tests/test_daemon_service.cpp b/tests/test_daemon_service.cpp index b092ec4..e137c31 100644 --- a/tests/test_daemon_service.cpp +++ b/tests/test_daemon_service.cpp @@ -380,3 +380,110 @@ TEST_CASE("Subscribe: S2F41 -> stream -> HCACK 4; declarative fallback without s server->Shutdown(); rt.stop(); } + +// Phase D — GEM300 in-the-loop (observe-and-report): job lifecycle, recipe +// downloads, and EC writes flow host -> engine -> Subscribe stream; the tool +// reports physical progress back and the engine drives the E40 FSM. +TEST_CASE("Phase D: jobs, recipes, and EC changes on the stream; ReportProcessJob") { + gem::EquipmentRuntime rt(test_config()); + gem::register_default_handlers(rt); + dmn::EquipmentService svc(rt); + rt.run_async(); + + grpc::ServerBuilder builder; + builder.RegisterService(&svc); + std::unique_ptr server(builder.BuildAndStart()); + REQUIRE(server); + auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{})); + + grpc::ClientContext sub_ctx; + pb::SubscribeRequest sreq; + auto reader = stub->Subscribe(&sub_ctx, sreq); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); // registration + + auto dispatch = [&](s2::Message msg) { + auto reply = rt.read_sync([&rt, &msg]() { return rt.router().dispatch(msg); }); + REQUIRE(reply.has_value()); + REQUIRE(reply->has_value()); + }; + + // Host creates the PJ (S16F11), the tool sets up, the host starts it. + gem::PRJobCreateRequest pj{"PJ-D-1", gem::MaterialFlag::Substrate, + gem::ProcessRecipeMethod::RecipeOnly, + gem::RecipeSpec{"RECIPE-A", {}}, {"WFR-D-1"}, {}}; + dispatch(gem::s16f11_pr_job_create(pj)); + auto setup = rt.read_sync([&rt]() { + rt.model().process_jobs.fire_internal("PJ-D-1", gem::ProcessJobEvent::Select); + return rt.model().process_jobs.fire_internal("PJ-D-1", + gem::ProcessJobEvent::SetupComplete); + }); + REQUIRE(setup.has_value()); + REQUIRE(*setup); + dispatch(gem::s16f5_pr_job_command("PJ-D-1", "PJSTART")); + + // The tool's stream receives the job with recipe + material bindings. + pb::HostRequest hr; + REQUIRE(reader->Read(&hr)); + REQUIRE(hr.has_process_job()); + CHECK(hr.process_job().job_id() == "PJ-D-1"); + CHECK(hr.process_job().action() == pb::ProcessJob::START); + CHECK(hr.process_job().recipe() == "RECIPE-A"); + REQUIRE(hr.process_job().carriers_size() == 1); + CHECK(hr.process_job().carriers(0) == "WFR-D-1"); + + // The tool finishes the physical work and reports; the engine's FSM follows. + { + grpc::ClientContext ctx; + pb::ProcessJobState rep; + pb::Ack ack; + rep.set_job_id("PJ-D-1"); + rep.set_state(pb::ProcessJobState::COMPLETE); + REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok()); + CHECK(ack.code() == pb::Ack::ACCEPT); + } + auto state = rt.read_sync( + [&rt]() { return rt.model().process_jobs.state("PJ-D-1"); }); + REQUIRE(state.has_value()); + CHECK(*state == gem::ProcessJobState::ProcessComplete); + + // Reporting against an unknown job / an illegal transition is rejected. + { + grpc::ClientContext ctx; + pb::ProcessJobState rep; + pb::Ack ack; + rep.set_job_id("PJ-GHOST"); + rep.set_state(pb::ProcessJobState::COMPLETE); + REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok()); + CHECK(ack.code() == pb::Ack::INVALID_OBJECT); + } + { + grpc::ClientContext ctx; + pb::ProcessJobState rep; + pb::Ack ack; + rep.set_job_id("PJ-D-1"); // already ProcessComplete + rep.set_state(pb::ProcessJobState::COMPLETE); + REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok()); + CHECK(ack.code() == pb::Ack::CANNOT_DO_NOW); + } + + // Host downloads a recipe (S7F3) -> ProcessProgram on the stream. + dispatch(gem::s7f3_process_program_send("RECIPE-NEW", "step1\nstep2\n")); + REQUIRE(reader->Read(&hr)); + REQUIRE(hr.has_process_program()); + CHECK(hr.process_program().ppid() == "RECIPE-NEW"); + CHECK(hr.process_program().body() == "step1\nstep2\n"); + + // Host writes an EC (S2F15) -> ConstantChange with the configured name. + dispatch(gem::s2f15_ec_send({{10, s2::Item::u4(uint32_t{1})}})); + REQUIRE(reader->Read(&hr)); + REQUIRE(hr.has_constant()); + CHECK(hr.constant().name() == "TimeFormat"); + CHECK(hr.constant().value().integer() == 1); + + sub_ctx.TryCancel(); + pb::HostRequest drain; + while (reader->Read(&drain)) {} + (void)reader->Finish(); + server->Shutdown(); + rt.stop(); +} diff --git a/tests/test_data_model.cpp b/tests/test_data_model.cpp index 17833f6..a72bc48 100644 --- a/tests/test_data_model.cpp +++ b/tests/test_data_model.cpp @@ -444,3 +444,35 @@ TEST_CASE("spool persistence: clear deletes files") { fs::remove_all(dir); } + +TEST_CASE("RecipeStore added-observer fires after the store is updated") { + RecipeStore r; + std::string seen_ppid, seen_body; + r.add_added_handler([&](const std::string& p, const std::string& b) { + seen_ppid = p; + seen_body = b; + }); + r.add("R-1", "body"); + CHECK(seen_ppid == "R-1"); + CHECK(seen_body == "body"); + CHECK(r.get("R-1").has_value()); // observer saw the post-update store +} + +TEST_CASE("EquipmentConstantStore changed-observer fires on Accept only") { + EquipmentConstantStore ecids; + ecids.add({10, "Knob", "", s2::Item::u4(uint32_t{1}), + s2::Item::u4(uint32_t{1}), "0", "5"}); + int fired = 0; + ecids.add_changed_handler([&](uint32_t id, const s2::Item& v) { + ++fired; + CHECK(id == 10); + CHECK(v == s2::Item::u4(uint32_t{3})); + }); + CHECK(ecids.set_value(10, s2::Item::u4(uint32_t{3})) == EquipmentAck::Accept); + CHECK(fired == 1); + CHECK(ecids.set_value(10, s2::Item::u4(uint32_t{99})) == + EquipmentAck::Denied_OutOfRange); + CHECK(ecids.set_value(999, s2::Item::u4(uint32_t{1})) == + EquipmentAck::Denied_UnknownEcid); + CHECK(fired == 1); // rejected writes never fire +}