test(interop): daemon end-to-end vs secsgem-py reference host

daemon_interop.py drives a running secs_gemd through BOTH faces at once: a
gRPC tool client and a secsgem-py active host. Proves the gRPC<->HSMS bridge
against a reference GEM implementation, not just in-process:

  - gRPC GetControlState agrees with the HSMS-driven control state
  - gRPC SetVariables(ChamberPressure=2.5) + FireEvent(ProcessStarted) makes
    the host receive S6F11 CEID 300 carrying 2.5 (value flowed gRPC -> engine
    -> HSMS -> host)
  - unknown variable/event names rejected at the gRPC edge

Mirrors the existing host_vs_cpp_server.py pattern. New 'gemd' compose service
(HSMS :5000 + gRPC :50051); interop image gains grpcio/grpcio-tools (proto
stubs generated at runtime, flat to avoid the secsgem package-name clash).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

docker-compose.yml

@@ -55,6 +55,24 @@ services:
     command: ["/app/build/secs_client", "--host", "server", "--port", "5000", "--device", "0"]
     networks: [secs]
 
+  # The gRPC daemon: passive HSMS equipment on :5000 plus the gRPC tool API
+  # on :50051. Used by interop/daemon_interop.py (a gRPC tool + a secsgem-py
+  # host both drive it) to prove the gRPC<->HSMS bridge against a real host.
+  gemd:
+    <<: *base
+    depends_on:
+      builder:
+        condition: service_completed_successfully
+    command:
+      - /app/build/secs_gemd
+      - --port
+      - "5000"
+      - --grpc
+      - "0.0.0.0:50051"
+      - --config-dir
+      - /app/data
+    networks: [secs]
+
   # Python container preloaded with secsgem-py 0.3.0 for cross-validation
   # of our C++ HSMS/SECS-II/GEM implementation against the reference library.
   interop:

interop/Dockerfile

@@ -2,6 +2,6 @@
 # Kept separate from the C++ build image so we don't bloat the toolchain layer.
 FROM python:3.12-slim
 
-RUN pip install --no-cache-dir secsgem==0.3.0
+RUN pip install --no-cache-dir secsgem==0.3.0 grpcio==1.60.0 grpcio-tools==1.60.0 "setuptools<80"
 
 WORKDIR /interop

interop/daemon_interop.py

@@ -0,0 +1,195 @@
+"""End-to-end interop for secs_gemd: drive the daemon through BOTH faces.
+
+A gRPC tool client and a secsgem-py active host both connect to one running
+secs_gemd. This proves the daemon bridges gRPC <-> HSMS against a reference
+GEM host, not just that it compiles:
+
+  - gRPC GetControlState agrees with the HSMS-driven control state
+  - gRPC SetVariables(ChamberPressure=2.5) lands in the model, and
+  - gRPC FireEvent(ProcessStarted) makes the host receive S6F11 CEID 300
+    carrying that 2.5 (so the value flowed gRPC -> engine -> HSMS -> host)
+  - unknown variable / event names are rejected at the gRPC edge
+
+Exits 0 on success, non-zero on any failure.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import sys
+import tempfile
+import threading
+import time
+
+import grpc
+
+import secsgem.common
+import secsgem.gem
+import secsgem.hsms
+import secsgem.secs
+
+LOG = logging.getLogger("daemon-interop")
+F = secsgem.secs.functions
+
+
+def gen_stubs(proto_dir: str):
+    """Generate flat Python stubs from equipment.proto (proto package is
+    'secsgem.v1', which would collide with the secsgem-py package if we kept
+    the nested path — so generate flat and import directly)."""
+    from grpc_tools import protoc
+
+    out = tempfile.mkdtemp(prefix="gemd_pb_")
+    rc = protoc.main([
+        "protoc",
+        f"-I{proto_dir}",
+        f"--python_out={out}",
+        f"--grpc_python_out={out}",
+        os.path.join(proto_dir, "equipment.proto"),
+    ])
+    if rc != 0:
+        raise SystemExit("protoc stub generation failed")
+    sys.path.insert(0, out)
+    import equipment_pb2 as pb            # noqa: E402
+    import equipment_pb2_grpc as pb_grpc  # noqa: E402
+    return pb, pb_grpc
+
+
+def _scalars(body):
+    """Flatten a decoded secsgem-py body (nested dict/list) to scalar leaves."""
+    if isinstance(body, dict):
+        for v in body.values():
+            yield from _scalars(v)
+    elif isinstance(body, (list, tuple)):
+        for v in body:
+            yield from _scalars(v)
+    else:
+        yield body
+
+
+def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int,
+        proto_dir: str) -> int:
+    pb, pb_grpc = gen_stubs(proto_dir)
+    failures: list[str] = []
+
+    def check(label: str, ok: bool, detail: str = "") -> None:
+        LOG.info("[%s] %s%s", "OK " if ok else "FAIL", label,
+                 f" — {detail}" if detail else "")
+        if not ok:
+            failures.append(f"{label}: {detail or 'failed'}")
+
+    # ---- gRPC tool side ----
+    LOG.info("connecting gRPC tool to %s", grpc_addr)
+    channel = grpc.insecure_channel(grpc_addr)
+    try:
+        grpc.channel_ready_future(channel).result(timeout=15)
+    except grpc.FutureTimeoutError:
+        check("gRPC channel ready", False, f"no daemon at {grpc_addr}")
+        return 1
+    stub = pb_grpc.EquipmentStub(channel)
+
+    # ---- secsgem-py host side ----
+    settings = secsgem.hsms.HsmsSettings(
+        address=hsms_host, port=hsms_port, session_id=session_id,
+        connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE,
+        device_type=secsgem.common.DeviceType.HOST,
+    )
+    client = secsgem.gem.GemHostHandler(settings)
+
+    ceid300 = threading.Event()
+    last_s6f11 = {}
+
+    def on_s6f11(_handler, message):
+        decoded = client.settings.streams_functions.decode(message)
+        body = decoded.get()
+        ceid = body.get("CEID") if isinstance(body, dict) else None
+        LOG.info("[evt] S6F11 CEID=%s body=%r", ceid, body)
+        if ceid == 300:
+            last_s6f11["body"] = body
+            ceid300.set()
+        client.send_response(F.SecsS06F12(0), message.header.system)
+
+    client.register_stream_function(6, 11, on_s6f11)
+
+    client.enable()
+    try:
+        if not client.waitfor_communicating(timeout=15):
+            check("HSMS establish-communications", False)
+            return 1
+        check("HSMS establish-communications", True)
+
+        # Go online so control state -> OnlineRemote.
+        client.send_and_waitfor_response(F.SecsS01F17())
+        time.sleep(0.3)
+
+        # gRPC GetControlState should agree with the HSMS-driven state.
+        cs = stub.GetControlState(pb.Empty())
+        check("gRPC GetControlState == ONLINE_REMOTE",
+              cs.state == pb.ControlState.ONLINE_REMOTE,
+              pb.ControlState.State.Name(cs.state))
+
+        # Configure an event report: RPTID 1 = [VID 101], CEID 300 -> RPTID 1.
+        client.send_and_waitfor_response(
+            F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [101]}]}))
+        client.send_and_waitfor_response(
+            F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]}))
+        client.send_and_waitfor_response(
+            F.SecsS02F37({"CEED": True, "CEID": [300]}))
+        check("host configured report CEID 300 -> RPTID 1 (VID 101)", True)
+
+        # --- gRPC tool: negative paths (edge validation) ---
+        ack = stub.SetVariables(pb.VariableUpdate(
+            values={"definitely_not_a_var": pb.Value(real=1.0)}))
+        check("gRPC SetVariables(unknown) -> PARAMETER_INVALID",
+              ack.code == pb.Ack.PARAMETER_INVALID, pb.Ack.Code.Name(ack.code))
+        ack = stub.FireEvent(pb.Event(name="NoSuchEvent"))
+        check("gRPC FireEvent(unknown) -> PARAMETER_INVALID",
+              ack.code == pb.Ack.PARAMETER_INVALID, pb.Ack.Code.Name(ack.code))
+
+        # --- gRPC tool: set a value, then fire the event that reports it ---
+        ack = stub.SetVariables(pb.VariableUpdate(
+            values={"ChamberPressure": pb.Value(real=2.5)}))
+        check("gRPC SetVariables(ChamberPressure=2.5) -> ACCEPT",
+              ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code))
+
+        ack = stub.FireEvent(pb.Event(name="ProcessStarted"))
+        check("gRPC FireEvent(ProcessStarted) -> ACCEPT",
+              ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code))
+
+        # The host should now receive S6F11 CEID 300 carrying the 2.5.
+        got = ceid300.wait(timeout=10)
+        check("host received S6F11 CEID 300 (gRPC FireEvent bridged to HSMS)", got)
+        if got:
+            vals = list(_scalars(last_s6f11.get("body")))
+            near = any(isinstance(v, (int, float)) and abs(float(v) - 2.5) < 0.01
+                       for v in vals)
+            check("S6F11 report carries ChamberPressure=2.5 (value flowed end-to-end)",
+                  near, f"scalars={vals}")
+    finally:
+        client.disable()
+
+    if failures:
+        LOG.error("FAILURES (%d):", len(failures))
+        for f in failures:
+            LOG.error("  - %s", f)
+        return 1
+    LOG.info("all daemon interop checks passed")
+    return 0
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--grpc", default="gemd:50051")
+    ap.add_argument("--hsms-host", default="gemd")
+    ap.add_argument("--hsms-port", type=int, default=5000)
+    ap.add_argument("--session-id", type=int, default=0)
+    ap.add_argument("--proto-dir", default="/app/proto/secsgem/v1")
+    args = ap.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    return run(args.grpc, args.hsms_host, args.hsms_port, args.session_id,
+               args.proto_dir)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())