raphael/secs-gem
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(daemon): alarms by name + RequestControlState + WatchHealth (Phase A complete)

Browse Source 
A2 — alarms: optional 'name:' on alarm config (a LOCAL key — SEMI only
defines numeric ALID + freetext ALTX; field appended last so existing
{id, text, category} brace-inits compile unchanged), parsed by the loader,
checked by the validator, shipped in equipment.yaml. SetAlarm/ClearAlarm
RPCs resolve config name OR stringified ALID via a constructor snapshot.

A3 — control state + health: RequestControlState fires operator events on
the io thread (read_sync) and reports what the E30 table actually did —
ACCEPT iff the equipment landed in the requested state, CANNOT_DO_NOW naming
the actual state otherwise (the shipped table has no operator path to
EquipmentOffline; the test pins that honesty). ATTEMPT_ONLINE is rejected as
transient. WatchHealth streams an immediate snapshot then pushes on link/
control-state changes via service observers (add_link_observer +
add_control_state_observer — the HandlerSlot work paying off), spool depth
sampled at the 500ms poll; ends on cancel or engine stop.

Tests: daemon suite 61 -> 101 assertions (alarm lifecycle by name/id/unknown,
WatchHealth initial + change push, all four RequestControlState semantics);
loader test for the alarm name (present + absent fallback); core 467/3055.
Interop now 15 checks incl. gRPC SetAlarm -> host receives S5F1 ALCD=0x84
ALID=1, and RequestControlState(HOST_OFFLINE) -> GetControlState confirms.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-10 19:47:31 +02:00
parent 83593bb508
commit 1da56f973f
9 changed files with 326 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/equipment_service.hpp
+157 -2
View File
@@ -12,8 +12,13 @@
#include <grpcpp/grpcpp.h>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <map>
#include <mutex>
#include <optional>
#include <string>
#include <utility>
#include <vector>
@@ -155,8 +160,9 @@ inline pb::ControlState::State to_proto_state(gem::ControlState s) {
class EquipmentService final : public pb::Equipment::Service {
public:
// Snapshots the (immutable) name->id/format dictionaries. Construct before
// run_async() so the model is read while the io thread isn't running yet.
// Snapshots the (immutable) name->id/format dictionaries and registers the
// health observers. Construct before run_async() so the model is read (and
// observers land) while the io thread isn't running yet.
explicit EquipmentService(gem::EquipmentRuntime& rt) : rt_(rt) {
for (const auto& sv : rt.model().svids.all())
vars_.insert({sv.name, {sv.id, sv.value.format()}});
@@ -164,6 +170,21 @@ class EquipmentService final : public pb::Equipment::Service {
vars_.insert({dv.name, {dv.id, dv.value.format()}});
for (const auto& ev : rt.model().events.all_events())
events_.insert({ev.name, ev.id});
for (const auto& al : rt.model().alarms.all()) {
if (!al.name.empty()) alarms_.insert({al.name, al.id});
alarms_.insert({std::to_string(al.id), al.id}); // always addressable by id
}
// Health signals: bump a version + wake WatchHealth streams whenever the
// HSMS link or the control state changes (observers fire on the io thread;
// add_ observers survive register_default_handlers' primary set_).
rt.add_link_observer([this](bool selected) {
link_selected_.store(selected, std::memory_order_relaxed);
bump_health();
});
rt.add_control_state_observer(
[this](gem::ControlState, gem::ControlState, gem::ControlEvent) {
bump_health();
});
}
grpc::Status SetVariables(grpc::ServerContext*, const pb::VariableUpdate* req,
@@ -214,6 +235,16 @@ class EquipmentService final : public pb::Equipment::Service {
return grpc::Status::OK;
}
grpc::Status SetAlarm(grpc::ServerContext*, const pb::Alarm* req,
pb::Ack* resp) override {
return alarm_action(req->name(), /*set=*/true, resp);
}
grpc::Status ClearAlarm(grpc::ServerContext*, const pb::Alarm* req,
pb::Ack* resp) override {
return alarm_action(req->name(), /*set=*/false, resp);
}
grpc::Status GetVariables(grpc::ServerContext*, const pb::VariableQuery* req,
pb::VariableSnapshot* resp) override {
// Resolve names against the snapshot maps (empty query = everything).
@@ -258,7 +289,123 @@ class EquipmentService final : public pb::Equipment::Service {
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,
// CANNOT_DO_NOW (naming the actual state) otherwise. Note the shipped table
// has no operator path to EQUIPMENT_OFFLINE — operator_offline lands
// HOST_OFFLINE — so honesty matters here.
grpc::Status RequestControlState(grpc::ServerContext*,
const pb::ControlStateRequest* req,
pb::Ack* resp) override {
using PS = pb::ControlState;
const auto desired = req->desired();
if (desired == PS::ATTEMPT_ONLINE) {
resp->set_code(pb::Ack::PARAMETER_INVALID);
resp->set_message("ATTEMPT_ONLINE is transient; request a settled state");
return grpc::Status::OK;
}
auto final_state = rt_.read_sync([this, desired]() {
auto& sm = rt_.control();
switch (desired) {
case PS::ONLINE_LOCAL:
if (sm.state() == gem::ControlState::OnlineRemote) sm.operator_local();
else if (!sm.online()) sm.operator_online(); // table chains to OnlineLocal
break;
case PS::ONLINE_REMOTE:
if (sm.state() == gem::ControlState::OnlineLocal) sm.operator_remote();
else if (!sm.online()) {
sm.operator_online();
sm.operator_remote();
}
break;
case PS::HOST_OFFLINE:
case PS::EQUIPMENT_OFFLINE:
sm.operator_offline();
break;
default:
break;
}
return sm.state();
});
if (!final_state) {
resp->set_code(pb::Ack::CANNOT_DO_NOW);
resp->set_message("engine io thread did not answer (not running?)");
return grpc::Status::OK;
}
if (to_proto_state(*final_state) == desired) {
resp->set_code(pb::Ack::ACCEPT);
} else {
resp->set_code(pb::Ack::CANNOT_DO_NOW);
resp->set_message(std::string("equipment is in ") +
gem::control_state_name(*final_state));
}
return grpc::Status::OK;
}
// Streams a Health snapshot immediately, then again whenever the link or
// control state changes (and on spool-depth changes, sampled at the poll
// interval). Ends when the client cancels or the engine stops.
grpc::Status WatchHealth(grpc::ServerContext* ctx, const pb::Empty*,
grpc::ServerWriter<pb::Health>* writer) override {
bool first = true;
pb::Health last;
while (!ctx->IsCancelled()) {
const uint64_t seen = health_version_.load(std::memory_order_relaxed);
auto snap = make_health();
if (!snap) break; // engine stopped — end the stream
if (first || snap->link() != last.link() ||
snap->control_state() != last.control_state() ||
snap->spool_depth() != last.spool_depth()) {
if (!writer->Write(*snap)) break;
last = *snap;
first = false;
}
std::unique_lock<std::mutex> lk(health_mu_);
health_cv_.wait_for(lk, std::chrono::milliseconds(500), [&] {
return health_version_.load(std::memory_order_relaxed) != seen;
});
}
return grpc::Status::OK;
}
private:
void bump_health() {
health_version_.fetch_add(1, std::memory_order_relaxed);
health_cv_.notify_all();
}
// Build a Health snapshot. Control state + link come from atomics; spool
// depth is mutable engine state, read via read_sync. nullopt = engine down.
std::optional<pb::Health> make_health() {
auto depth = rt_.read_sync(
[this]() { return static_cast<uint32_t>(rt_.model().spool.size()); });
if (!depth) return std::nullopt;
pb::Health h;
// CONNECTED (TCP up, not yet SELECTED) is reserved: the runtime's link
// observer fires on SELECTED/closed only. TODO(daemon): surface the
// intermediate state if a tool ever needs it.
h.set_link(link_selected_.load(std::memory_order_relaxed)
? pb::Health::SELECTED
: pb::Health::DISCONNECTED);
h.set_spool_depth(*depth);
h.set_control_state(to_proto_state(rt_.control_state()));
return h;
}
grpc::Status alarm_action(const std::string& name, bool set, pb::Ack* resp) {
auto it = alarms_.find(name);
if (it == alarms_.end()) {
resp->set_code(pb::Ack::PARAMETER_INVALID);
resp->set_message("no alarm named '" + name + "'");
return grpc::Status::OK;
}
if (set) rt_.set_alarm(it->second);
else rt_.clear_alarm(it->second);
resp->set_code(pb::Ack::ACCEPT);
return grpc::Status::OK;
}
struct VarRef {
uint32_t vid;
s2::Format format; // declared wire format from equipment.yaml
@@ -267,6 +414,14 @@ class EquipmentService final : public pb::Equipment::Service {
gem::EquipmentRuntime& rt_;
std::map<std::string, VarRef> vars_; // SVIDs + DVIDs (SVIDs win on clash)
std::map<std::string, uint32_t> events_; // CEID by name
std::map<std::string, uint32_t> alarms_; // ALID by name AND stringified id
// WatchHealth plumbing: observers (io thread) bump the version and wake the
// per-stream wait loops (gRPC threads).
std::atomic<bool> link_selected_{false};
std::atomic<uint64_t> health_version_{0};
std::mutex health_mu_;
std::condition_variable health_cv_;
};
} // namespace secsgem::daemon