fix: 补齐 Python 节点运行时观测

2026-07-16 19:31:25 +02:00
parent 166f45e485
commit af4962e138
2 changed files with 122 additions and 1 deletions
@@ -615,7 +615,110 @@ class NodeOpsExecutor:
        argv = args.get("argv") if isinstance(args.get("argv"), list) else []
        if not command:
            return {"ok": False, "blockerCode": "invalid_command", "stderr": "command is required", "exitCode": None}
-        return self._spawn_output([str(command), *[str(item) for item in argv]], self._workspace_root(args), safe_int(args.get("timeoutMs"), default_timeout_ms))
+        command_binding = args.get("commandBinding") if isinstance(args.get("commandBinding"), dict) else {}
+        if command_binding.get("nativeHelper") == "io-probe-observation":
+            return self._io_probe_observation(args, command_binding, default_timeout_ms)
+        output = self._spawn_output([str(command), *[str(item) for item in argv]], self._workspace_root(args), safe_int(args.get("timeoutMs"), default_timeout_ms))
+        observation = self._command_observation(command_binding, output)
+        return {**output, **({"observation": observation} if observation else {})}
+
+    def _command_observation(self, command_binding: dict, output: dict) -> dict:
+        if command_binding.get("kind") not in ("io-probe", "board-comm"):
+            return {}
+        return self._parse_json(output.get("stdout"))
+
+    def _io_probe_observation(self, args: dict, command_binding: dict, default_timeout_ms: int) -> dict:
+        argv = args.get("argv") if isinstance(args.get("argv"), list) else []
+        input_data = self._parse_json(argv[1] if len(argv) > 1 else None)
+        command_base = input_data.get("commandBase") if isinstance(input_data.get("commandBase"), list) else []
+        if not command_base:
+            return {"ok": False, "blockerCode": "invalid_io_probe_binding", "stderr": "io-probe commandBase is required", "exitCode": None}
+        count = max(1, min(safe_int(input_data.get("count"), 1), 50))
+        interval_ms = self._nonnegative_int(input_data.get("intervalMs"))
+        settle_ms = self._nonnegative_int(input_data.get("settleMs"))
+        timeout_ms = safe_int(args.get("timeoutMs"), default_timeout_ms)
+        samples = []
+        response = None
+        exit_code = 0
+        commands = []
+        if settle_ms:
+            time.sleep(settle_ms / 1000)
+        for index in range(count):
+            command = self._io_probe_command(input_data, command_base)
+            output = self._spawn_output(command, self._workspace_root(args), timeout_ms)
+            commands.append(output)
+            response = self._parse_json(output.get("stdout"))
+            value = self._io_probe_value(response, str(input_data.get("valuePath") or ""))
+            if value is not None:
+                samples.append(value)
+            if not output.get("ok"):
+                exit_code = output.get("exitCode") or 1
+            if index < count - 1 and interval_ms:
+                time.sleep(interval_ms / 1000)
+        stats = {}
+        if samples:
+            stats = {"count": len(samples), "mean": sum(samples) / len(samples), "min": min(samples), "max": max(samples)}
+        observation = {
+            "observationId": f"obs_{input_data.get('probeId')}_{int(time.time() * 1000)}",
+            "probeId": input_data.get("probeId") or command_binding.get("probeId"),
+            "quantity": input_data.get("quantity") or command_binding.get("quantity"),
+            "unit": input_data.get("unit") or command_binding.get("unit"),
+            "samples": samples,
+            "stats": stats,
+            "response": response,
+            "rawArtifactRef": response.get("log_path") or response.get("logPath") or response.get("artifactRef") if isinstance(response, dict) else None,
+        }
+        final_exit_code = exit_code if exit_code or samples else 1
+        return {
+            "ok": final_exit_code == 0,
+            "command": [str(item) for item in command_base],
+            "cwd": str(self._workspace_root(args)),
+            "exitCode": final_exit_code,
+            "stdout": json.dumps(observation, ensure_ascii=False) + "\n",
+            "stderr": "".join(str(item.get("stderr") or "") for item in commands),
+            "observation": observation,
+        }
+
+    def _io_probe_command(self, input_data: dict, command_base: list) -> list[str]:
+        return [
+            *[str(item) for item in command_base],
+            "jrpctcp",
+            "--host",
+            str(input_data.get("host") or ""),
+            "--port",
+            str(input_data.get("port") or 8000),
+            *(["--transport", str(input_data["transport"])] if input_data.get("transport") else []),
+            *(["--target-node-id", str(input_data["targetNodeId"])] if input_data.get("targetNodeId") is not None else []),
+            str(input_data.get("method") or "get"),
+            str(input_data.get("path") or ""),
+            *[str(item) for item in input_data.get("params", [])],
+        ]
+
+    def _io_probe_value(self, response: dict, value_path: str) -> float | None:
+        candidates = []
+        nested_response = response.get("response") if isinstance(response, dict) else None
+        if isinstance(nested_response, dict) and isinstance(nested_response.get("response"), dict):
+            candidates.append(nested_response["response"])
+        if isinstance(nested_response, dict):
+            candidates.append(nested_response)
+        if isinstance(response, dict):
+            candidates.append(response)
+        keys = [item for item in value_path.removeprefix("$").removeprefix(".").split(".") if item]
+        for candidate in candidates:
+            value = candidate
+            for key in keys:
+                value = value.get(key) if isinstance(value, dict) else None
+            try:
+                return float(value)
+            except (TypeError, ValueError):
+                continue
+        return None
+
+    def _nonnegative_int(self, value: object) -> int:
+        try:
+            return max(0, int(value or 0))
+        except (TypeError, ValueError):
+            return 0

    def _debug_command(self, op: str, args: dict) -> dict:
        runs = args.get("commandRuns") if isinstance(args.get("commandRuns"), list) else None