SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix(k1): translate onboard operation identities at plugin boundary

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-09-07 12:34:48 +03:00
parent 3dffe39c89
commit d2dbfc9c85
9 changed files with 264 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/device_plugins/xgrids_k1/node_bridge.py
+62 -20
View File
@@ -8,8 +8,11 @@ from __future__ import annotations
import asyncio
import logging
import re
import traceback
from datetime import UTC, datetime
from pathlib import Path
from uuid import UUID, uuid5
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
@@ -42,6 +45,29 @@ class NodeEnrollmentRejected(ValueError):
super().__init__(code)
def plugin_operation_id(node_id: str, *, stage: str | None = None) -> str:
"""Translate the host's opaque ID without changing either journal contract.
The same host intent always reaches the same plugin row. A composite action
uses deterministic child UUIDs so prepare and start cannot alias each other.
"""
if not isinstance(node_id, str) or re.fullmatch(r"op_[0-9a-f]{32}", node_id) is None:
raise NodeEnrollmentRejected("invalid-operation-id")
identifier = UUID(hex=node_id[3:])
if stage is not None:
identifier = uuid5(identifier, stage)
return f"op-{identifier}"
def node_operation_id(plugin_id: str | None) -> str | None:
"""Project only canonical plugin UUIDs back to the host correlation ID."""
if isinstance(plugin_id, str) and re.fullmatch(
r"op-[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", plugin_id
):
return "op_" + UUID(plugin_id[3:]).hex
return plugin_id
class NodeBridge:
def __init__(self, repository_root: Path, *, service=None):
self.rerun = NodeRerunHub()
@@ -64,6 +90,11 @@ class NodeBridge:
@staticmethod
def project(snapshot: dict) -> dict:
lifecycle = snapshot.get("connection_lifecycle", {})
attempt = compact_connection_attempt(snapshot.get("connection_attempt"))
if attempt is not None:
for key in ("attempt_id", "recovery_operation_id"):
if key in attempt:
attempt[key] = node_operation_id(attempt[key])
return {
"available": True,
"model": "XGRIDS K1",
@@ -81,7 +112,7 @@ class NodeBridge:
"connected": lifecycle.get("connection_ready") is True
and snapshot.get("active_connection_mode") == "bridge",
"ready_to_start": lifecycle.get("ready_to_start") is True,
"connection_attempt": compact_connection_attempt(snapshot.get("connection_attempt")),
"connection_attempt": attempt,
"allowed_actions": [
action
for action in lifecycle.get("allowed_actions", [])
@@ -113,6 +144,20 @@ class NodeBridge:
)
)
async def invoke_journalled(self, action: str, payload: dict, identifier: str) -> dict:
try:
return await self.invoke(action, payload, identifier)
except Exception:
# Read the exact result after a failure; never repeat an invocation
# or interpret an exception as evidence that a command was unsent.
result = await self.invoke("state.read", {}, identifier + "-observe")
if not any(
item.get("operation_id") == payload["operation_id"]
for item in result.get("operations", [])
):
raise
return result
async def execute(self, command: dict) -> dict:
try:
return await self._execute(command)
@@ -128,6 +173,7 @@ class NodeBridge:
result = await self.state()
result["command_result"] = {
"operation_id": command["operation_id"],
"action": command["action"],
"status": "rejected",
"error_code": error.code,
}
@@ -135,6 +181,7 @@ class NodeBridge:
async def _execute(self, command: dict) -> dict:
action, identifier = command["action"], command["operation_id"]
plugin_identifier = plugin_operation_id(identifier)
parameters = command.get("parameters", {})
if action not in {"scan", "networks", "connect", "verify"}:
raise NodeEnrollmentRejected("unsupported-action")
@@ -147,11 +194,11 @@ class NodeBridge:
if action == "networks":
return {**state, "networks": await asyncio.to_thread(wifi_networks)}
if action == "scan":
result = await self.invoke(
result = await self.invoke_journalled(
"discovery.scan",
{
"duration_seconds": BLE_SCAN_DEFAULT_TIMEOUT_SECONDS,
"operation_id": identifier,
"operation_id": plugin_identifier,
"expected_snapshot_runtime_id": command["runtime_id"],
},
identifier,
@@ -168,7 +215,7 @@ class NodeBridge:
payload = {
"device_id": parameters["device_id"],
"compatibility_attestation": ATTESTATION,
"operation_id": identifier,
"operation_id": plugin_identifier,
"expected_mode_revision": parameters["mode_revision"],
"expected_discovery_generation": parameters["discovery_generation"],
"expected_snapshot_runtime_id": state["runtime_id"],
@@ -177,26 +224,16 @@ class NodeBridge:
payload.update(
connection_mode="bridge",
allow_host_wifi_switch=False,
idempotency_key=identifier,
idempotency_key=plugin_identifier,
ssid=parameters.get("ssid"),
password=parameters.get("password"),
)
try:
result = await self.invoke(
result = await self.invoke_journalled(
"network.provision" if action == "connect" else "connection.verify",
payload,
identifier,
)
except Exception:
# A rejected/failed invocation may already have a durable
# result. Read that exact journal row; never resend the
# network command and never serialize exception/payload text.
result = await self.invoke("state.read", {}, identifier + "-observe")
if not any(
item.get("operation_id") == identifier
for item in result.get("operations", [])
):
raise
finally:
payload.pop("password", None)
parameters.pop("password", None)
@@ -205,7 +242,7 @@ class NodeBridge:
(
item
for item in reversed(result.get("operations", []))
if item.get("operation_id") == identifier
if item.get("operation_id") == plugin_identifier
),
None,
)
@@ -213,6 +250,7 @@ class NodeBridge:
error = operation.get("error") or {}
projected["command_result"] = {
"operation_id": identifier,
"action": action,
"status": operation.get("status"),
"error_code": error.get("code"),
}
@@ -291,15 +329,19 @@ def create_app(repository_root: Path):
body = await request.json()
return await bridge.deliver(body)
except Exception as error:
# Class and admitted action are sufficient for transport diagnosis.
# Never log the exception text, incoming payload or credentials.
# Source locations identify the failing boundary without exception
# text, source lines, local variables, payloads or credentials.
action = body.get("action") if isinstance(body, dict) else None
logging.getLogger(__name__).warning(
"K1 operation failed: action=%s exception=%s",
"K1 operation failed: action=%s exception=%s locations=%s",
action
if isinstance(action, str) and action in {"scan", "networks", "connect", "verify"}
else "invalid",
type(error).__name__,
";".join(
f"{Path(frame.filename).name}:{frame.name}:{frame.lineno}"
for frame in traceback.extract_tb(error.__traceback__)[-8:]
),
)
return JSONResponse(
{
src/k1link/device_plugins/xgrids_k1/node_sensor.py
+12 -8
View File
@@ -11,7 +11,7 @@ from .facade import (
XGRIDS_K1_PLUGIN_VERSION,
ViewerSettingsRequest,
)
from .node_bridge import ATTESTATION
from .node_bridge import ATTESTATION, plugin_operation_id
PHYSICAL_ACCEPTANCE_KEYS = (
"operator_present",
@@ -137,6 +137,7 @@ class NodeK1Sensor:
command.get("parameters", {}),
command["operation_id"],
)
plugin_identifier = plugin_operation_id(identifier)
if action == "details":
return item
if action == "close-peer":
@@ -170,7 +171,9 @@ class NodeK1Sensor:
return project_sensor(result, node_id)
if action == "verify":
result = await self.bridge.invoke(
"connection.verify", {"expected_snapshot_runtime_id": runtime}, identifier
"connection.verify",
{"expected_snapshot_runtime_id": runtime, "operation_id": plugin_identifier},
identifier,
)
return project_sensor(result, node_id)
if action not in {"start", "stop"} or params.get("operator_confirmed") is not True:
@@ -187,8 +190,8 @@ class NodeK1Sensor:
result = await self.bridge.invoke(
"acquisition.stop",
{
"operation_id": identifier,
"idempotency_key": identifier,
"operation_id": plugin_identifier,
"idempotency_key": plugin_identifier,
"acquisition_id": params.get("acquisition_id"),
"mode": "graceful",
"physical_acceptance": acceptance,
@@ -223,10 +226,11 @@ class NodeK1Sensor:
payload.update(self.cas(state, acquisition=False), operator_confirmed=True)
elif phase == "workspace-ready" and acquisition.get("state") != "prepared":
next_action = "acquisition.prepare"
prepare_identifier = plugin_operation_id(identifier, stage=next_action)
payload.update(
self.cas(state),
operation_id=identifier + "_prepare",
idempotency_key=identifier + "_prepare",
operation_id=prepare_identifier,
idempotency_key=prepare_identifier,
project_name="node-" + identifier[3:15],
compatibility_attestation=ATTESTATION,
)
@@ -234,8 +238,8 @@ class NodeK1Sensor:
next_action = "acquisition.start"
payload.update(
self.cas(state),
operation_id=identifier,
idempotency_key=identifier,
operation_id=plugin_identifier,
idempotency_key=plugin_identifier,
acquisition_id=acquisition["acquisition_id"],
expected_state_revision=acquisition["state_revision"],
physical_acceptance=acceptance,