DPL MCP: allow client to control signposts

Author: ktf

Framework/Core/scripts/dpl-mcp-server/dpl_mcp_server.py

@@ -237,6 +237,37 @@ async def get_logs(max_lines: int = 100) -> str:
     return "\n".join(lines)
 
 
+@mcp.tool()
+async def enable_signpost(device: str, streams: list[str]) -> str:
+    """Enable one or more signpost log streams for a DPL device.
+
+    Signpost streams produce detailed trace output visible in the device logs.
+    Use get_logs() after subscribing to see the output.
+
+    Known stream names (full form): ch.cern.aliceo2.device,
+    ch.cern.aliceo2.completion, ch.cern.aliceo2.monitoring_service,
+    ch.cern.aliceo2.data_processor_context, ch.cern.aliceo2.stream_context.
+
+    Args:
+        device: Device name as shown by list_devices, or "" for the driver.
+        streams: List of full signpost log names to enable.
+    """
+    await _send({"cmd": "enable_signpost", "device": device, "streams": streams})
+    return f"Enabled {len(streams)} signpost stream(s) for '{device or 'driver'}': {', '.join(streams)}"
+
+
+@mcp.tool()
+async def disable_signpost(device: str, streams: list[str]) -> str:
+    """Disable one or more signpost log streams for a DPL device.
+
+    Args:
+        device: Device name as shown by list_devices, or "" for the driver.
+        streams: List of full signpost log names to disable.
+    """
+    await _send({"cmd": "disable_signpost", "device": device, "streams": streams})
+    return f"Disabled {len(streams)} signpost stream(s) for '{device or 'driver'}': {', '.join(streams)}"
+
+
 @mcp.tool()
 async def get_updates(max_updates: int = 50) -> str:
     """Drain and return buffered metric update frames received since the last call.

Framework/Core/src/StatusWebSocketHandler.cxx

@@ -12,11 +12,15 @@
 #include "StatusWebSocketHandler.h"
 #include "DPLWebSocket.h"
 #include "DriverServerContext.h"
+#include "Framework/DeviceControl.h"
+#include "Framework/DeviceController.h"
 #include "Framework/DeviceInfo.h"
 #include "Framework/DeviceMetricsInfo.h"
 #include "Framework/DeviceSpec.h"
+#include "Framework/DeviceState.h"
 #include "Framework/DeviceStateEnums.h"
 #include "Framework/LogParsingHelpers.h"
+#include "Framework/Signpost.h"
 #include <algorithm>
 #include <cstdio>
 #include <string>
@@ -250,6 +254,10 @@ void StatusWebSocketHandler::frame(char const* data, size_t s)
     handleSubscribeLogs(deviceName);
   } else if (cmd == "unsubscribe_logs") {
     handleUnsubscribeLogs(deviceName);
+  } else if (cmd == "enable_signpost") {
+    handleEnableSignpost(deviceName, extractArrayField(msg, "streams"));
+  } else if (cmd == "disable_signpost") {
+    handleDisableSignpost(deviceName, extractArrayField(msg, "streams"));
   }
 }
 
@@ -433,6 +441,69 @@ size_t StatusWebSocketHandler::findDeviceIndex(std::string_view name) const
   return SIZE_MAX;
 }
 
+void StatusWebSocketHandler::handleEnableSignpost(std::string_view deviceName, std::string_view streamsArr)
+{
+  if (streamsArr.empty()) {
+    return;
+  }
+  if (deviceName.empty()) {
+    // Driver process — toggle in-process via o2_walk_logs.
+    forEachStringInArray(streamsArr, [](std::string_view streamName) {
+      std::string target(streamName);
+      o2_walk_logs([](char const* name, void* l, void* context) -> bool {
+        auto* log = static_cast<_o2_log_t*>(l);
+        if (static_cast<std::string*>(context)->compare(name) == 0) {
+          _o2_log_set_stacktrace(log, log->defaultStacktrace);
+          return false;
+        }
+        return true;
+      }, &target);
+    });
+  } else {
+    size_t di = findDeviceIndex(deviceName);
+    if (di == SIZE_MAX || di >= mContext.controls->size() || !(*mContext.controls)[di].controller) {
+      return;
+    }
+    auto* controller = (*mContext.controls)[di].controller;
+    forEachStringInArray(streamsArr, [controller](std::string_view name) {
+      std::string cmd = "/signpost:enable ";
+      cmd += name;
+      controller->write(cmd.c_str(), cmd.size());
+    });
+  }
+}
+
+void StatusWebSocketHandler::handleDisableSignpost(std::string_view deviceName, std::string_view streamsArr)
+{
+  if (streamsArr.empty()) {
+    return;
+  }
+  if (deviceName.empty()) {
+    forEachStringInArray(streamsArr, [](std::string_view streamName) {
+      std::string target(streamName);
+      o2_walk_logs([](char const* name, void* l, void* context) -> bool {
+        auto* log = static_cast<_o2_log_t*>(l);
+        if (static_cast<std::string*>(context)->compare(name) == 0) {
+          _o2_log_set_stacktrace(log, 0);
+          return false;
+        }
+        return true;
+      }, &target);
+    });
+  } else {
+    size_t di = findDeviceIndex(deviceName);
+    if (di == SIZE_MAX || di >= mContext.controls->size() || !(*mContext.controls)[di].controller) {
+      return;
+    }
+    auto* controller = (*mContext.controls)[di].controller;
+    forEachStringInArray(streamsArr, [controller](std::string_view name) {
+      std::string cmd = "/signpost:disable ";
+      cmd += name;
+      controller->write(cmd.c_str(), cmd.size());
+    });
+  }
+}
+
 void StatusWebSocketHandler::handleSubscribeLogs(std::string_view deviceName)
 {
   size_t di = findDeviceIndex(deviceName);

Framework/Core/src/StatusWebSocketHandler.h

@@ -41,6 +41,17 @@ struct WSDPLHandler;
 ///   {"cmd":"unsubscribe_logs","device":"<name>"}
 ///     → driver stops pushing log lines for the device
 ///
+///   {"cmd":"enable_signpost","device":"<name>","streams":["device","completion",...]}
+///     → enable the named signpost log streams for a device (or the driver if device=="")
+///     → known streams: "device","completion","monitoring_service","data_processor_context","stream_context"
+///
+///   {"cmd":"disable_signpost","device":"<name>","streams":["device","completion",...]}
+///     → disable the named signpost log streams for a device
+///
+///   {"cmd":"list_signposts"}
+///     → driver replies with {"type":"signposts_list","streams":["device","completion",...]}
+///     → lists the known stream names
+///
 /// Protocol (driver → client):
 ///   {"type":"snapshot","devices":[{"name","pid","active","streamingState","deviceState"},...]}
 ///     → sent once on connect; contains no metrics or logs
@@ -84,6 +95,8 @@ struct StatusWebSocketHandler : public WebSocketHandler {
   void handleUnsubscribe(std::string_view deviceName, std::string_view metricsJson);
   void handleSubscribeLogs(std::string_view deviceName);
   void handleUnsubscribeLogs(std::string_view deviceName);
+  void handleEnableSignpost(std::string_view deviceName, std::string_view streamsArr);
+  void handleDisableSignpost(std::string_view deviceName, std::string_view streamsArr);
   size_t findDeviceIndex(std::string_view name) const;
 
   DriverServerContext& mContext;