From d2dbfc9c85a8375205626914232d7355ce24f30b Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Mon, 7 Sep 2026 12:34:48 +0300 Subject: [PATCH] fix(k1): translate onboard operation identities at plugin boundary --- .../test/sensorEnrollment.test.mjs | 21 ++++ apps/node-agent/packaging/build_deb.py | 2 +- ...026-09-07-k1-node-operation-identity-r4.md | 64 +++++++++++ packages/sensor-ui/src/enrollment.ts | 2 +- .../frontend/src/sensors/enrollment.ts | 5 + plugins/xgrids-k1/packaging/build_deb.py | 2 +- .../device_plugins/xgrids_k1/node_bridge.py | 82 ++++++++++---- .../device_plugins/xgrids_k1/node_sensor.py | 20 ++-- tests/test_node_k1_bridge.py | 100 +++++++++++++++++- 9 files changed, 264 insertions(+), 34 deletions(-) create mode 100644 docs/audits/2026-09-07-k1-node-operation-identity-r4.md diff --git a/apps/control-station/test/sensorEnrollment.test.mjs b/apps/control-station/test/sensorEnrollment.test.mjs index 61fc2a2..6ca74f5 100644 --- a/apps/control-station/test/sensorEnrollment.test.mjs +++ b/apps/control-station/test/sensorEnrollment.test.mjs @@ -109,3 +109,24 @@ test('delayed operation result cannot clear an observed board outage',()=>{ assert.equal(result.fresh,false); assert.equal(api.mergeEnrollmentState(result,{...initial,fresh:true,snapshot_revision:3}).fresh,true); }); + +test('failed or unknown Bluetooth search is never presented as an empty successful scan',async()=>{ + for(const state of ['failed','rejected','unknown']){ + let posts=0; + const transport={submit:async command=>{posts++;return { + operation_id:command.operation_id,state:state==='unknown'?'unknown':'complete', + error:'unconfirmed device command', + result:{...initial,candidates:[],command_result:{operation_id:command.operation_id,action:'scan',status:state}}, + };}}; + await assert.rejects(api.enroll(transport,initial,'scan'),/Не удалось выполнить поиск Bluetooth на БК/); + assert.equal(posts,1); + } + assert.equal(api.enrollmentNotice({...initial,command_result:{action:'scan',status:'failed'}}),''); +}); + +test('a confirmed scan with zero candidates remains a valid empty result',async()=>{ + const transport={submit:async command=>({operation_id:command.operation_id,state:'complete', + result:{...initial,candidates:[],command_result:{operation_id:command.operation_id,action:'scan',status:'succeeded'}}, + })}; + assert.deepEqual((await api.enroll(transport,initial,'scan')).candidates,[]); +}); diff --git a/apps/node-agent/packaging/build_deb.py b/apps/node-agent/packaging/build_deb.py index b5ab174..08cc9bf 100644 --- a/apps/node-agent/packaging/build_deb.py +++ b/apps/node-agent/packaging/build_deb.py @@ -11,7 +11,7 @@ import sys ROOT = Path(__file__).resolve().parents[1] -VERSION = "0.8.1" +VERSION = "0.8.2" sys.path.insert(0, str(ROOT.parents[1] / "scripts/packaging")) from debian import package diff --git a/docs/audits/2026-09-07-k1-node-operation-identity-r4.md b/docs/audits/2026-09-07-k1-node-operation-identity-r4.md new file mode 100644 index 0000000..7374246 --- /dev/null +++ b/docs/audits/2026-09-07-k1-node-operation-identity-r4.md @@ -0,0 +1,64 @@ +# Node K1 operation identity R4 + +The owner's next Core Fleet UI attempt at 12:19 MSK failed on installed Node +0.8.1 / K1 0.1.1+private.1. The newly added journal diagnostic recorded +`action=scan exception=ValueError`. The board still had discovery generation +zero. Its BlueZ controller was powered and not rfkill-blocked. This was an +execution failure before discovery, not evidence that K1 was absent. + +## Reproduced cause + +The Node protocol requires `op_` followed by 32 lowercase hexadecimal digits. +The K1 OperationJournal accepts a UUID, optionally prefixed by `op-`. Passing +the Node ID unchanged fails in device_lifecycle._identifier before any BLE +call. R3 corrected the missing runtime fence but its service stub hid this +second boundary mismatch. The new regression reproduced the exact ValueError +with the real NodeBridge, facade, compatibility service and operation journal. + +NodeBridge now translates the host ID into the plugin UUID format. The mapping +is deterministic and preserves all 128 bits. Host IDs remain unchanged in the +Go broker, durable Node journals and HTTP responses. The plugin's validators, +LAB callers and lifecycle authority are unchanged. Connection attempt and +recovery IDs are projected back to host IDs so continued UI observation can +still match the exact request after the Wi-Fi acknowledgement. + +The same conversion covers scan, provision, verify, START and STOP. The +acquisition prepare child uses a deterministic UUID derived from the host +intent, distinct from its START operation. It no longer appends an invalid +`_prepare` suffix to an operation ID. Repeated intent IDs reach the same journal +row; no random replacement IDs, automatic provisioning replay or migration of +old unknown operations were introduced. + +## Error evidence and UI + +An invocation failure reads only its exact existing plugin journal row. A +recorded scan failure reaches the caller as failed; an exception before any +row exists remains an unconfirmed execution. Neither is a successful empty +search. Both produce a Bluetooth search error in the shared enrollment UI, +while a confirmed zero-candidate result still displays K1 not found. Search +failures do not display the Wi-Fi connection notice. + +Unexpected worker failures log admitted action, exception class and source +function/line locations only. Exception text, source text, local variables, +request payloads and credentials are excluded. + +## Validation and limits + +The scan regression keeps the actual service, journal, BLE arbiter and scanner +logic, replacing only BleakScanner's OS radio endpoint. It verifies successful +discovery and a failed radio start, correct host result correlation, and one +radio invocation when the exact command is repeated. Acquisition IDs are +checked against the actual OperationJournal. Projection does not mutate the +plugin's native state. Frontend tests distinguish failed, rejected, unknown +and confirmed empty scans and retain network/bootstrap/recovery regressions. + +Focused Python bridge, installer, Fleet enrollment, BLE scanner and operation +journal suite: 73 passed. Core frontend suite: 789 passed, no failures or skips. +Build and installation acceptance is recorded below when completed. + +Release versions are Node 0.8.2 and optional K1 0.1.2+private.1. Existing private +release bytes and application material are retained. This is a Node adapter +correction; no Rerun profile, protocol frame, MQTT recovery state machine or +shared lifecycle validator was modified. Physical Bluetooth discovery, Bridge, +live camera/LiDAR and interruption recovery still require fresh-cache UI +acceptance. No agent CLI hardware command is part of this test. diff --git a/packages/sensor-ui/src/enrollment.ts b/packages/sensor-ui/src/enrollment.ts index c4c1df1..5157d13 100644 --- a/packages/sensor-ui/src/enrollment.ts +++ b/packages/sensor-ui/src/enrollment.ts @@ -5,7 +5,7 @@ export interface EnrollmentState { networks?:{ssid:string;signal:number;security:string}[]; connected?:boolean; ready_to_start?:boolean; selected_device_id?:string; ip?:string; snapshot_revision?:number; runtime_started_at?:string; allowed_actions?:string[]; - connection_attempt?:unknown; command_result?:{operation_id:string;status:string;error_code?:string}; + connection_attempt?:unknown; command_result?:{operation_id:string;action?:string;status:string;error_code?:string}; device_session?:{device_session_id:string;device_id:string}; observed_at?:string; } export interface EnrollmentCommand { diff --git a/plugins/xgrids-k1/frontend/src/sensors/enrollment.ts b/plugins/xgrids-k1/frontend/src/sensors/enrollment.ts index 6c61215..ba763cb 100644 --- a/plugins/xgrids-k1/frontend/src/sensors/enrollment.ts +++ b/plugins/xgrids-k1/frontend/src/sensors/enrollment.ts @@ -45,6 +45,7 @@ export interface EnrollmentObserver { pause?:()=>Promise; } const unknownResult='Результат подключения пока не подтверждён. Обновите состояние K1 перед новой попыткой.'; +const scanFailure='Не удалось выполнить поиск Bluetooth на БК. Проверьте Bluetooth на бортовом компьютере и повторите поиск.'; /** Submit once. HTTP delivery and the device's network/control result are separate. */ export async function enroll(transport:EnrollmentTransport,initial:EnrollmentState,action:EnrollmentCommand['action'],parameters:Record={},observer:EnrollmentObserver={}):Promise{ @@ -76,6 +77,9 @@ export async function enroll(transport:EnrollmentTransport,initial:EnrollmentSta try{operation=await transport.operation(command.operation_id);}catch{operation=null;} } if(action==='scan'||action==='networks'){ + if(action==='scan'&&(operation.state!=='complete'||!operation.result|| + (state.command_result?.operation_id===command.operation_id&&state.command_result.status!=='succeeded'))) + throw new Error(scanFailure); if(operation.state!=='complete'||!operation.result)throw new Error(operation.error||unknownResult); return state; } @@ -106,6 +110,7 @@ export async function enroll(transport:EnrollmentTransport,initial:EnrollmentSta export function enrollmentNotice(state:EnrollmentState):string { const attempt=connectionAttempt(state); if(!state.available||state.fresh===false)return 'Нет свежих сведений с БК. Восстанавливаем связь.'; + if(state.command_result?.action==='scan')return ''; // Scan failures belong to the search toast, not a Wi-Fi notice. if(attempt?.status==='accepted'||attempt?.status==='running'){ return attempt.phase==='network_applied'?'K1 подключился к Wi-Fi. Проверяем канал управления.':'Подключаем K1 к выбранной сети Wi-Fi.'; } diff --git a/plugins/xgrids-k1/packaging/build_deb.py b/plugins/xgrids-k1/packaging/build_deb.py index 0f931ea..f026347 100644 --- a/plugins/xgrids-k1/packaging/build_deb.py +++ b/plugins/xgrids-k1/packaging/build_deb.py @@ -22,7 +22,7 @@ from credential_install import PROFILE_ID, validate # noqa: E402 from debian import package # noqa: E402 from runtime_payload import files as runtime_files # noqa: E402 -VERSION = "0.1.1" +VERSION = "0.1.2" RESOURCES = ( "plugins/xgrids-k1/profile_loader.py", "plugins/xgrids-k1/plugin.manifest.json", diff --git a/src/k1link/device_plugins/xgrids_k1/node_bridge.py b/src/k1link/device_plugins/xgrids_k1/node_bridge.py index 895cc13..8c32790 100644 --- a/src/k1link/device_plugins/xgrids_k1/node_bridge.py +++ b/src/k1link/device_plugins/xgrids_k1/node_bridge.py @@ -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( { diff --git a/src/k1link/device_plugins/xgrids_k1/node_sensor.py b/src/k1link/device_plugins/xgrids_k1/node_sensor.py index 538e510..7523030 100644 --- a/src/k1link/device_plugins/xgrids_k1/node_sensor.py +++ b/src/k1link/device_plugins/xgrids_k1/node_sensor.py @@ -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, diff --git a/tests/test_node_k1_bridge.py b/tests/test_node_k1_bridge.py index 6e8c740..ecc996a 100644 --- a/tests/test_node_k1_bridge.py +++ b/tests/test_node_k1_bridge.py @@ -9,7 +9,7 @@ import pytest from missioncore_plugin_sdk.v0alpha2.session import DeviceSessionSnapshot from k1link.device_plugins.xgrids_k1.linux_host import nm_fields, route_fields -from k1link.device_plugins.xgrids_k1.node_bridge import NodeBridge +from k1link.device_plugins.xgrids_k1.node_bridge import NodeBridge, plugin_operation_id from k1link.device_plugins.xgrids_k1.node_sensor import NodeK1Sensor, project_sensor from k1link.viewer.node_rerun import NodeRerunHub @@ -107,16 +107,99 @@ def test_node_scan_reaches_service_through_real_facade_with_runtime_fence(): } output = await device.deliver(command) assert service.fences == ["runtime-one"] - assert service.scans == [command["operation_id"]] + assert service.scans == [plugin_operation_id(command["operation_id"])] assert len(output["candidates"]) == 1 command["runtime_id"] = "retired-runtime" rejected = await device.deliver(command) assert rejected["command_result"]["status"] == "rejected" - assert service.scans == [command["operation_id"]] + assert service.scans == [plugin_operation_id(command["operation_id"])] asyncio.run(run()) +@pytest.mark.parametrize("radio_failure", [False, True]) +def test_node_scan_real_service_preserves_operation_identity_and_replay( + tmp_path, monkeypatch, radio_failure, +): + from bleak.backends.device import BLEDevice + from bleak.backends.scanner import AdvertisementData + + from k1link.device_plugins.xgrids_k1 import facade + from k1link.device_plugins.xgrids_k1.ble import scanner + + calls = [] + + class Radio: + def __init__(self, detection_callback): + self.callback = detection_callback + + async def __aenter__(self): + calls.append("scan") + if radio_failure: + raise RuntimeError("synthetic radio unavailable") + self.callback( + BLEDevice("AA:BB:CC:DD:EE:FF", "XGR-TEST", None), + AdvertisementData( + local_name="XGR-TEST", manufacturer_data={}, service_data={}, + service_uuids=[], tx_power=None, rssi=-50, platform_data=(), + ), + ) + return self + + async def __aexit__(self, *_args): + pass + + # Keep the actual Node adapter, facade, service, journal and BLE arbiter. + # Replace only the OS radio endpoint and shorten its observation window. + monkeypatch.setattr(scanner, "BleakScanner", Radio) + monkeypatch.setattr(scanner, "BLE_SCAN_INITIAL_WINDOW_SECONDS", 0.01) + service = facade.XgridsK1CompatibilityService(tmp_path) + + async def run(): + device = NodeBridge(tmp_path, service=service) + current = await device.state() + command = { + "operation_id": "op_" + "b" * 32, + "runtime_id": current["runtime_id"], + "deadline_at": (datetime.now(UTC) + timedelta(seconds=60)).isoformat(), + "action": "scan", "parameters": {}, + } + result = await device.deliver(command) + repeated = await device.deliver(command) + assert len(calls) == 1 + assert result["discovery_generation"] == 1 + if radio_failure: + assert result["candidates"] == [] + else: + assert result["candidates"][0]["id"] == "AA:BB:CC:DD:EE:FF" + assert result["command_result"] == repeated["command_result"] == { + "operation_id": command["operation_id"], + "action": "scan", + "status": "failed" if radio_failure else "succeeded", + "error_code": "RuntimeError" if radio_failure else None, + } + + try: + asyncio.run(run()) + finally: + service.close() + + +def test_node_attempt_projection_keeps_host_correlation_without_mutating_plugin_state(): + current = state() + identifier = "op_" + "a" * 32 + current["connection_attempt"] = { + "schema_version": "missioncore.xgrids-k1-connection-attempt/v1", + "attempt_id": plugin_operation_id(identifier), + "recovery_operation_id": plugin_operation_id("op_" + "b" * 32), + "status": "failed", "public_error_code": "wifi-ssid-not-found", + } + result = NodeBridge.project(current) + assert result["connection_attempt"]["attempt_id"] == identifier + assert result["connection_attempt"]["recovery_operation_id"] == "op_" + "b" * 32 + assert current["connection_attempt"]["attempt_id"] == plugin_operation_id(identifier) + + def test_node_bridge_forces_reviewed_bridge_and_clears_input_secret(): async def run(): device = bridge() @@ -200,6 +283,17 @@ def test_one_start_intent_preserves_canonical_enter_prepare_start_sequence(): "acquisition.prepare", "acquisition.start", ] + from k1link.web.device_lifecycle import OperationJournal + + journal = OperationJournal() + for action, payload in device.facade.actions: + if action in {"acquisition.prepare", "acquisition.start"}: + row, created = journal.begin( + action, operation_id=payload["operation_id"], + idempotency_key=payload["idempotency_key"], + ) + assert created + assert journal.begin(action, operation_id=row.operation_id)[1] is False assert result["snapshot"]["acquisition"] == "streaming" command["parameters"]["control_generation"] = 2 with pytest.raises(ValueError):