From 4b73a157653b5b67f71fcae4d4b3dff9169c108f Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 12 Aug 2026 22:57:09 +0300 Subject: [PATCH] feat(device-plane): add typed B2 service ping transport --- .../arusnavi-b2-adapter/src/index.mjs | 40 ++- .../arusnavi-b2-adapter/test/adapter.test.mjs | 30 +- .../src/index.mjs | 1 + .../services/device-control-core/src/app.mjs | 60 +++- .../src/device-gateway-core-runtime.mjs | 112 ++++++- .../src/discovery-repository.mjs | 5 +- .../src/edge-channel-supervisor.mjs | 21 +- .../src/gateway-ingest.mjs | 17 +- .../src/gateway-message-repository.mjs | 58 +++- .../src/postgres-repository.mjs | 36 ++- .../src/project-query-repository.mjs | 11 +- .../device-control-core/src/server.mjs | 8 +- .../src/typed-command-repository.mjs | 299 ++++++++++++++++++ .../src/typed-command-runtime.mjs | 149 +++++++++ .../device-control-core/test/app.test.mjs | 67 +++- .../test/gateway-ingest.test.mjs | 21 +- .../test/gateway-message-repository.test.mjs | 13 +- .../test/typed-command-runtime.test.mjs | 106 +++++++ .../device-edge-channel/src/runtime.mjs | 78 ++++- .../test/channel-integration.test.mjs | 61 +++- .../services/device-gateway/src/runtime.mjs | 121 ++++++- .../test/discovery-integration.test.mjs | 23 +- .../device-gateway/test/runtime.test.mjs | 63 +++- .../vps/edge-process/device-edge-runtime.mjs | 10 +- 24 files changed, 1322 insertions(+), 88 deletions(-) create mode 100644 device-plane/services/device-control-core/src/typed-command-repository.mjs create mode 100644 device-plane/services/device-control-core/src/typed-command-runtime.mjs create mode 100644 device-plane/services/device-control-core/test/typed-command-runtime.test.mjs diff --git a/device-plane/packages/arusnavi-b2-adapter/src/index.mjs b/device-plane/packages/arusnavi-b2-adapter/src/index.mjs index e74db69..8fc8c19 100644 --- a/device-plane/packages/arusnavi-b2-adapter/src/index.mjs +++ b/device-plane/packages/arusnavi-b2-adapter/src/index.mjs @@ -51,8 +51,8 @@ export const ARUSNAVI_B2_MODEL_PROFILE = deepFreeze({ package: "package-number-only", }, commandTransport: { - status: "disabled", - exportedCommandBuilders: 0, + status: "typed-service-ping-v1", + exportedCommandBuilders: 1, }, routeCompatibility: { gelios: "parallel-preserved", @@ -76,6 +76,8 @@ export const ARUSNAVI_B2_ADAPTER = defineDeviceAdapter({ buildMessageAcknowledgement(message) { return buildB2PackageAcknowledgement(message.packageNumber); }, + buildTypedCommand: buildB2TypedCommand, + parseTypedCommandResponse: tryParseB2TypedCommandResponse, }; }, }); @@ -238,6 +240,36 @@ export function buildB2PackageAcknowledgement(packageNumber) { ]); } +export function buildB2TypedCommand({ commandType, accessCode } = {}) { + if (commandType !== "service.ping") { + throw new TypeError("b2_typed_command_unsupported"); + } + if (typeof accessCode !== "string" || !/^\d{6}$/.test(accessCode)) { + throw new TypeError("b2_command_access_code_invalid"); + } + return Buffer.from(`${accessCode}*SERV*1.1`, "ascii"); +} + +export function tryParseB2TypedCommandResponse(input, { commandType } = {}) { + assertBuffer(input, "b2_command_response_buffer_required"); + if (commandType !== "service.ping") { + throw new TypeError("b2_typed_command_unsupported"); + } + const expected = Buffer.from("SERV OK", "ascii"); + const compared = Math.min(input.length, expected.length); + if (!input.subarray(0, compared).equals(expected.subarray(0, compared))) { + return Object.freeze({ status: "not-command" }); + } + if (input.length < expected.length) { + return Object.freeze({ status: "incomplete", minimumBytes: expected.length }); + } + return Object.freeze({ + status: "acknowledged", + bytesConsumed: expected.length, + resultCode: "serv_ok", + }); +} + export function assertB2ProfileInvariant(profile = ARUSNAVI_B2_MODEL_PROFILE) { if (profile.monitoringServerSlots !== 4) { throw new TypeError("b2_server_slot_count_invalid"); @@ -257,8 +289,8 @@ export function assertB2ProfileInvariant(profile = ARUSNAVI_B2_MODEL_PROFILE) { ) { throw new TypeError("b2_framing_specification_invalid"); } - if (profile.commandTransport.status !== "disabled") { - throw new TypeError("b2_command_transport_must_be_disabled"); + if (profile.commandTransport.status !== "typed-service-ping-v1") { + throw new TypeError("b2_command_transport_profile_invalid"); } if (profile.routeCompatibility.gelios !== "parallel-preserved") { throw new TypeError("b2_gelios_route_must_be_preserved"); diff --git a/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs b/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs index 5eada75..97c988d 100644 --- a/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs +++ b/device-plane/packages/arusnavi-b2-adapter/test/adapter.test.mjs @@ -7,8 +7,10 @@ import { assertB2ProfileInvariant, buildB2HeaderAcknowledgement, buildB2PackageAcknowledgement, + buildB2TypedCommand, tryParseB2Header2, tryParseB2Package, + tryParseB2TypedCommandResponse, } from "../src/index.mjs"; const specificationHeader = Buffer.from( @@ -131,10 +133,32 @@ test("fails closed on unsupported headers and malformed packages", () => { ); }); -test("exports no command builder and keeps transport disabled", () => { - assert.equal(ARUSNAVI_B2_MODEL_PROFILE.commandTransport.status, "disabled"); +test("exports only the typed service-ping command", () => { + assert.equal( + ARUSNAVI_B2_MODEL_PROFILE.commandTransport.status, + "typed-service-ping-v1", + ); assert.equal( ARUSNAVI_B2_MODEL_PROFILE.commandTransport.exportedCommandBuilders, - 0, + 1, + ); + assert.equal( + buildB2TypedCommand({ commandType: "service.ping", accessCode: "123456" }) + .toString("ascii"), + "123456*SERV*1.1", + ); + assert.deepEqual( + tryParseB2TypedCommandResponse(Buffer.from("SERV OK", "ascii"), { + commandType: "service.ping", + }), + { status: "acknowledged", bytesConsumed: 7, resultCode: "serv_ok" }, + ); + assert.throws( + () => buildB2TypedCommand({ commandType: "service.ping", accessCode: "12345" }), + /b2_command_access_code_invalid/, + ); + assert.throws( + () => buildB2TypedCommand({ commandType: "firmware.update", accessCode: "123456" }), + /b2_typed_command_unsupported/, ); }); diff --git a/device-plane/packages/device-edge-channel-contract/src/index.mjs b/device-plane/packages/device-edge-channel-contract/src/index.mjs index 14fc3d2..5a855cf 100644 --- a/device-plane/packages/device-edge-channel-contract/src/index.mjs +++ b/device-plane/packages/device-edge-channel-contract/src/index.mjs @@ -21,6 +21,7 @@ export const EDGE_TO_CORE_MESSAGE_KINDS = Object.freeze([ "discovery.observed", "adapter.message", "delivery.acknowledged", + "command.status", "channel.counters", ]); diff --git a/device-plane/services/device-control-core/src/app.mjs b/device-plane/services/device-control-core/src/app.mjs index 6a493b7..613d17a 100644 --- a/device-plane/services/device-control-core/src/app.mjs +++ b/device-plane/services/device-control-core/src/app.mjs @@ -56,6 +56,7 @@ export function createControlCoreApp({ managementToken = "", gatewayIngest = null, edgeChannelStatusProvider = null, + typedCommandRuntime = null, } = {}) { if (!repository || typeof repository.health !== "function") { throw new TypeError("device_repository_required"); @@ -97,6 +98,15 @@ export function createControlCoreApp({ ) { throw new TypeError("device_gateway_ingest_invalid"); } + if ( + typedCommandRuntime != null + && ( + typeof typedCommandRuntime.planServicePing !== "function" + || typeof typedCommandRuntime.status !== "function" + ) + ) { + throw new TypeError("device_typed_command_runtime_invalid"); + } if ( edgeChannelStatusProvider != null && typeof edgeChannelStatusProvider !== "function" @@ -126,7 +136,47 @@ export function createControlCoreApp({ edgeChannels: edgeChannelStatusProvider ? edgeChannelStatusProvider() : { enabled: false, configured: 0, accepted: 0, degraded: 0 }, - commandTransport: "disabled", + commandTransport: typedCommandRuntime + ? "typed-service-ping-v1" + : "disabled", + }); + } + + if ( + request.method === "POST" + && requestUrl.pathname === "/internal/v1/commands:service-ping" + ) { + if (!managementApiEnabled || !typedCommandRuntime) { + return writeJson(response, 404, { + ok: false, + error: "device_command_transport_disabled", + }); + } + if (!matchesBearer(request.headers.authorization, managementToken)) { + return writeJson(response, 401, { + ok: false, + error: "device_management_auth_required", + }); + } + const idempotencyKey = normalizeIdempotencyKey( + request.headers["idempotency-key"], + ); + const actor = managementActorFromHeaders(request.headers); + const input = await readJsonBody(request, 8 * 1024); + const execution = await typedCommandRuntime.planServicePing({ + idempotencyKey, + actor, + input, + }); + response.setHeader("Idempotency-Key", idempotencyKey); + response.setHeader( + "Idempotency-Replayed", + execution.replayed ? "true" : "false", + ); + return writeJson(response, 200, { + ok: true, + replayed: execution.replayed, + result: execution.command, }); } @@ -232,6 +282,11 @@ export function createControlCoreApp({ const workspace = await repository.getProjectWorkspace( actor, workspaceProjectId, + { + commandTransport: typedCommandRuntime + ? "typed-service-ping-v1" + : "disabled", + }, ); return writeJson(response, 200, { ok: true, workspace }); } @@ -280,7 +335,8 @@ export function createControlCoreApp({ } const input = await readJsonBody(request, 1024 * 1024); - const acceptance = await ingest.acceptMessage(input); + const receipt = await ingest.acceptMessage(input); + const acceptance = receipt.value; return writeJson(response, acceptance.replayed ? 200 : 201, { ok: true, acceptance, diff --git a/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs b/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs index a8686e0..0c10bf6 100644 --- a/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs +++ b/device-plane/services/device-control-core/src/device-gateway-core-runtime.mjs @@ -14,6 +14,7 @@ import { import { DEVICE_DISCOVERY_VIEW_SCHEMA, assertSafeProjection, + normalizeAdapterAcceptance, normalizeAdapterMessage, normalizeDiscoverySignal, } from "../../../packages/device-protocol-contract/src/index.mjs"; @@ -88,6 +89,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { edgeTrustGeneration: state?.observedEdgeIdentity?.generationRef ?? null, edgeCertificateFingerprint: state?.observedEdgeIdentity?.fingerprint ?? null, + negotiatedCommandTransport: state?.negotiatedCommandTransport ?? null, activeTrackerSessionChains: state?.sessionChains.size ?? 0, connectionAttempts: totalConnectionAttempts, channelsAccepted: totalChannelsAccepted, @@ -97,7 +99,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { protocolFailures: totalProtocolFailures, lastErrorCode, trackerIngress: "remote-edge-only", - commandTransport: "disabled", + commandTransport: config.commandTransport, }); } @@ -127,7 +129,9 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { maxEnvelopeBytes: config.maxEnvelopeBytes, }), sessionChains: new Map(), + trackerDevices: new Map(), channelGeneration: null, + negotiatedCommandTransport: null, observedEdgeIdentity: null, edgeSequence: 0, coreSequence: 0, @@ -245,15 +249,19 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { || envelope.payload?.transport !== "http2-mtls" || envelope.payload?.trustGeneration !== connection.observedEdgeIdentity?.generationRef - || envelope.payload?.commandTransport !== "disabled" + || !isCompatibleCommandTransport( + config.commandTransport, + envelope.payload?.commandTransport, + ) ) { throw new Error("device_gateway_core_channel_hello_invalid"); } + connection.negotiatedCommandTransport = envelope.payload.commandTransport; connection.channelGeneration = connection.registration.channelGeneration; send(connection, "channel.accepted", { status: "accepted", coreIdentity: config.coreIdentity, - commandTransport: "disabled", + commandTransport: connection.negotiatedCommandTransport, }, { trackerSessionId: CHANNEL_TRACKER_SESSION_ID, adapterProfileRef: CHANNEL_PROFILE_REF, @@ -269,7 +277,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { return; } if (envelope.messageKind === "channel.heartbeat") return; - if (["discovery.observed", "adapter.message"].includes(envelope.messageKind)) { + if (["discovery.observed", "adapter.message", "command.status"].includes(envelope.messageKind)) { scheduleTrackerEvent(connection, envelope); return; } @@ -287,7 +295,9 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { const work = (previous ?? Promise.resolve()) .then(() => envelope.messageKind === "discovery.observed" ? acceptDiscovery(connection, envelope) - : acceptAdapterMessage(connection, envelope)) + : envelope.messageKind === "adapter.message" + ? acceptAdapterMessage(connection, envelope) + : acceptCommandStatus(connection, envelope)) .catch((error) => failConnection(connection, error)) .finally(() => { if (connection.sessionChains.get(envelope.trackerSessionId) === work) { @@ -300,10 +310,27 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { async function acceptDiscovery(connection, envelope) { try { const signal = normalizeDiscoverySignal(envelope.payload?.signal); - const discovery = normalizeDiscoveryReceipt( + const receipt = normalizeDiscoveryReceipt( await config.observeDiscovery(signal), ); - sendEventResult(connection, envelope, { discovery }); + if (receipt.claimedDeviceRef) { + connection.trackerDevices.set( + envelope.trackerSessionId, + receipt.claimedDeviceRef, + ); + } else { + connection.trackerDevices.delete(envelope.trackerSessionId); + } + const commandOffer = ( + connection.negotiatedCommandTransport === "typed-service-ping-v1" + && receipt.claimedDeviceRef + ) + ? await config.offerCommand(receipt.claimedDeviceRef) + : null; + sendEventResult(connection, envelope, { + discovery: receipt.discovery, + ...(commandOffer ? { commandOffer } : {}), + }); totalEventsAccepted += 1; } catch (error) { sendEventRejection(connection, envelope, error); @@ -316,8 +343,38 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { const message = normalizeAdapterMessage(envelope.payload?.message, { maxBytes: config.maxEnvelopeBytes, }); - const acceptance = await config.acceptMessage(message); - sendEventResult(connection, envelope, { acceptance }); + const receipt = normalizeAdapterReceipt( + await config.acceptMessage(message), + ); + if (receipt.claimedDeviceRef) { + connection.trackerDevices.set( + envelope.trackerSessionId, + receipt.claimedDeviceRef, + ); + } + const claimedDeviceRef = receipt.claimedDeviceRef + ?? connection.trackerDevices.get(envelope.trackerSessionId); + const commandOffer = ( + connection.negotiatedCommandTransport === "typed-service-ping-v1" + && claimedDeviceRef + ) + ? await config.offerCommand(claimedDeviceRef) + : null; + sendEventResult(connection, envelope, { + acceptance: receipt.acceptance, + ...(commandOffer ? { commandOffer } : {}), + }); + totalEventsAccepted += 1; + } catch (error) { + sendEventRejection(connection, envelope, error); + totalEventsRejected += 1; + } + } + + async function acceptCommandStatus(connection, envelope) { + try { + await config.recordCommandStatus(envelope.payload?.status); + sendEventResult(connection, envelope, { status: "recorded" }); totalEventsAccepted += 1; } catch (error) { sendEventRejection(connection, envelope, error); @@ -489,7 +546,25 @@ function normalizeDiscoveryReceipt(input) { ) { throw new TypeError("device_gateway_core_discovery_receipt_invalid"); } - return discovery; + return Object.freeze({ + discovery, + claimedDeviceRef: input.claimedDeviceRef ?? null, + }); +} + +function normalizeAdapterReceipt(input) { + if (!input || typeof input !== "object" || Array.isArray(input)) { + throw new TypeError("device_gateway_core_adapter_receipt_invalid"); + } + const acceptanceValue = input.value ?? ( + input.schemaVersion === "nodedc.device-adapter-acceptance.v1" + ? input + : null + ); + return Object.freeze({ + acceptance: normalizeAdapterAcceptance(acceptanceValue), + claimedDeviceRef: input.claimedDeviceRef ?? null, + }); } function normalizeConfig(options) { @@ -499,6 +574,15 @@ function normalizeConfig(options) { if (typeof options.acceptMessage !== "function") { throw new TypeError("device_gateway_core_accept_message_invalid"); } + const commandTransport = options.commandTransport ?? "disabled"; + if (!["disabled", "typed-service-ping-v1"].includes(commandTransport)) { + throw new TypeError("device_gateway_core_command_transport_invalid"); + } + const offerCommand = options.offerCommand ?? (async () => null); + const recordCommandStatus = options.recordCommandStatus ?? (async () => undefined); + if (typeof offerCommand !== "function" || typeof recordCommandStatus !== "function") { + throw new TypeError("device_gateway_core_command_runtime_invalid"); + } const registrationProvider = typeof options.registrationProvider === "function" ? options.registrationProvider : async () => options.registration; @@ -540,6 +624,9 @@ function normalizeConfig(options) { coreIdentity: normalizeRef(options.coreIdentity, "core_identity"), observeDiscovery: options.observeDiscovery, acceptMessage: options.acceptMessage, + commandTransport, + offerCommand, + recordCommandStatus, keepaliveMs, deadPeerMs, connectTimeoutMs: normalizeInteger( @@ -633,6 +720,11 @@ function safeErrorCode(error) { : "device_gateway_core_event_rejected"; } +function isCompatibleCommandTransport(configured, offered) { + if (offered === configured) return true; + return configured === "typed-service-ping-v1" && offered === "disabled"; +} + function normalizeRef(value, field) { if (typeof value !== "string" || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(value)) { throw new TypeError(`device_gateway_core_${field}_invalid`); diff --git a/device-plane/services/device-control-core/src/discovery-repository.mjs b/device-plane/services/device-control-core/src/discovery-repository.mjs index f8bd595..dc53b16 100644 --- a/device-plane/services/device-control-core/src/discovery-repository.mjs +++ b/device-plane/services/device-control-core/src/discovery-repository.mjs @@ -133,7 +133,7 @@ export async function observeQuarantineDiscovery({ returning id, lifecycle_state, model_profile_ref, protocol, identifier_kind, identifier_masked, first_observed_at, last_observed_at, evidence, project_id, route_id, - enrollment_intent_id, (xmax = 0) as created`, + enrollment_intent_id, claimed_device_id, (xmax = 0) as created`, [ randomUUID(), safeView.identifier.kind, @@ -179,6 +179,9 @@ export async function observeQuarantineDiscovery({ return { created: row.created === true, value: discoveryView(row), + claimedDeviceRef: row.claimed_device_id + ? `device:${row.claimed_device_id}` + : null, }; } catch (error) { await client.query("rollback").catch(() => undefined); diff --git a/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs b/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs index b528ab4..dfacaba 100644 --- a/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs +++ b/device-plane/services/device-control-core/src/edge-channel-supervisor.mjs @@ -110,6 +110,11 @@ export function createDeviceEdgeChannelSupervisor(options = {}) { coreIdentity: config.coreIdentity.identityRef, observeDiscovery: config.gatewayIngest.observeDiscovery, acceptMessage: config.gatewayIngest.acceptMessage, + commandTransport: config.typedCommandRuntime + ? "typed-service-ping-v1" + : "disabled", + offerCommand: config.typedCommandRuntime?.offerForDevice, + recordCommandStatus: config.typedCommandRuntime?.recordStatus, }); assertClient(client); clients.set(registration.edgeRegistrationId, { client, digest }); @@ -169,7 +174,9 @@ export function createDeviceEdgeChannelSupervisor(options = {}) { degraded, reconciliationFailures, lastErrorCode, - commandTransport: "disabled", + commandTransport: config.typedCommandRuntime + ? "typed-service-ping-v1" + : "disabled", edges: Object.freeze(edges), }); } @@ -245,6 +252,7 @@ function normalizeConfiguration(options) { return Object.freeze({ repository: options.repository, gatewayIngest: options.gatewayIngest, + typedCommandRuntime: normalizeTypedCommandRuntime(options.typedCommandRuntime), coreIdentity: Object.freeze({ ...coreIdentity }), maxEdges, reconcileIntervalMs, @@ -254,6 +262,17 @@ function normalizeConfiguration(options) { }); } +function normalizeTypedCommandRuntime(value) { + if (value == null) return null; + if ( + typeof value.offerForDevice !== "function" + || typeof value.recordStatus !== "function" + ) { + throw new TypeError("device_edge_channel_typed_command_runtime_invalid"); + } + return value; +} + function normalizeRegistration(value) { if (!value || typeof value !== "object" || Array.isArray(value)) { throw new TypeError("device_edge_channel_registration_invalid"); diff --git a/device-plane/services/device-control-core/src/gateway-ingest.mjs b/device-plane/services/device-control-core/src/gateway-ingest.mjs index 3592371..890dacd 100644 --- a/device-plane/services/device-control-core/src/gateway-ingest.mjs +++ b/device-plane/services/device-control-core/src/gateway-ingest.mjs @@ -38,6 +38,7 @@ export function createDeviceGatewayIngest({ repository, identifierPepper } = {}) return Object.freeze({ created: discovery.created === true, value: assertSafeProjection(discovery.value), + claimedDeviceRef: discovery.claimedDeviceRef ?? null, }); }, @@ -60,13 +61,15 @@ export function createDeviceGatewayIngest({ repository, identifierPepper } = {}) payloadSchemaRef: safeView.payloadSchemaRef, payload: safeView.payload, }); - return normalizeAdapterAcceptance( - await repository.acceptAdapterMessage({ - identifierDigest, - requestDigest, - safeView, - }), - ); + const receipt = await repository.acceptAdapterMessage({ + identifierDigest, + requestDigest, + safeView, + }); + return Object.freeze({ + value: normalizeAdapterAcceptance(receipt.acceptance), + claimedDeviceRef: receipt.claimedDeviceRef ?? null, + }); }, }); } diff --git a/device-plane/services/device-control-core/src/gateway-message-repository.mjs b/device-plane/services/device-control-core/src/gateway-message-repository.mjs index df3d57e..f268cef 100644 --- a/device-plane/services/device-control-core/src/gateway-message-repository.mjs +++ b/device-plane/services/device-control-core/src/gateway-message-repository.mjs @@ -15,6 +15,13 @@ export async function acceptGatewayMessage({ const route = routeId == null ? null : await findActiveRoute(client, routeId, safeView); + const claimedDeviceRef = route == null + ? null + : await findClaimedDeviceRef(client, { + identifierDigest, + route, + safeView, + }); const id = randomUUID(); const inserted = await client.query( `insert into device_gateway_message_receipts ( @@ -67,7 +74,7 @@ export async function acceptGatewayMessage({ ); if (inserted.rows[0]) { await client.query("commit"); - return acceptanceView(inserted.rows[0], false); + return receiptView(inserted.rows[0], false, claimedDeviceRef); } const existing = await client.query( @@ -83,7 +90,7 @@ export async function acceptGatewayMessage({ throw domainError("device_gateway_idempotency_conflict", 409); } await client.query("commit"); - return acceptanceView(row, true); + return receiptView(row, true, claimedDeviceRef); } catch (error) { await client.query("rollback").catch(() => undefined); throw error; @@ -92,6 +99,36 @@ export async function acceptGatewayMessage({ } } +async function findClaimedDeviceRef(client, { + identifierDigest, + route, + safeView, +}) { + const result = await client.query( + `select claimed_device_id + from device_discoveries + where identifier_kind = $1 + and identifier_digest = $2 + and model_profile_ref = $3 + and lifecycle_state = 'claimed' + and project_id = $4 + and route_id = $5 + and claimed_device_id is not null + for share`, + [ + safeView.identifier.kind, + identifierDigest, + safeView.protocolProfileRef, + route.project_id, + route.id, + ], + ); + const row = result.rows[0]; + return row?.claimed_device_id + ? `device:${row.claimed_device_id}` + : null; +} + async function findActiveRoute(client, routeId, safeView) { const edgeId = parseEntityRef(safeView.edgeRef, "edge"); const result = await client.query( @@ -133,14 +170,17 @@ async function findActiveRoute(client, routeId, safeView) { return route; } -function acceptanceView(row, replayed) { +function receiptView(row, replayed, claimedDeviceRef) { return { - schemaVersion: "nodedc.device-adapter-acceptance.v1", - acceptanceRef: `acceptance:${row.id}`, - idempotencyKey: row.idempotency_key, - status: "accepted", - replayed, - acceptedAt: new Date(row.accepted_at).toISOString(), + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: `acceptance:${row.id}`, + idempotencyKey: row.idempotency_key, + status: "accepted", + replayed, + acceptedAt: new Date(row.accepted_at).toISOString(), + }, + claimedDeviceRef, }; } diff --git a/device-plane/services/device-control-core/src/postgres-repository.mjs b/device-plane/services/device-control-core/src/postgres-repository.mjs index e1fbc29..a3efb76 100644 --- a/device-plane/services/device-control-core/src/postgres-repository.mjs +++ b/device-plane/services/device-control-core/src/postgres-repository.mjs @@ -41,6 +41,11 @@ import { assertProjectCapability, toProjectRef, } from "./project-management.mjs"; +import { + dispatchTypedCommand, + planTypedServicePing, + recordTypedCommandStatus, +} from "./typed-command-repository.mjs"; const { Pool } = pg; const serviceRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); @@ -163,9 +168,9 @@ export class PostgresDeviceRepository { ); } - async getProjectWorkspace(actor, projectId) { + async getProjectWorkspace(actor, projectId, options) { return this.#executeRead((client) => - getDeviceProjectWorkspace(client, actor, projectId) + getDeviceProjectWorkspace(client, actor, projectId, options) ); } @@ -199,6 +204,33 @@ export class PostgresDeviceRepository { }); } + async planTypedServicePing(input) { + return this.#executeWrite((client) => planTypedServicePing(client, input)); + } + + async dispatchTypedCommand(input) { + return this.#executeWrite((client) => dispatchTypedCommand(client, input)); + } + + async recordTypedCommandStatus(input) { + return this.#executeWrite((client) => recordTypedCommandStatus(client, input)); + } + + async #executeWrite(operation) { + const client = await this.pool.connect(); + try { + await client.query("begin"); + const result = await operation(client); + await client.query("commit"); + return result; + } catch (error) { + await client.query("rollback").catch(() => undefined); + throw mapPostgresError(error); + } finally { + client.release(); + } + } + async #executeRead(operation) { const client = await this.pool.connect(); try { diff --git a/device-plane/services/device-control-core/src/project-query-repository.mjs b/device-plane/services/device-control-core/src/project-query-repository.mjs index bb68c4f..4373529 100644 --- a/device-plane/services/device-control-core/src/project-query-repository.mjs +++ b/device-plane/services/device-control-core/src/project-query-repository.mjs @@ -46,7 +46,12 @@ export async function listAccessibleDeviceProjects(client, actor) { }); } -export async function getDeviceProjectWorkspace(client, actor, projectId) { +export async function getDeviceProjectWorkspace( + client, + actor, + projectId, + { commandTransport = "disabled" } = {}, +) { const project = await findProjectWithCapability( client, actor, @@ -339,8 +344,8 @@ export async function getDeviceProjectWorkspace(client, actor, projectId) { auditEvents: auditEvents.rows.map(auditEventView), grants: projectGrants.rows.map(storedGrantView), policies: { - commandTransport: "disabled", - commandPlanningApi: "disabled", + commandTransport, + commandPlanningApi: commandTransport === "disabled" ? "disabled" : "enabled", identifierProjection: "masked-only", auditPayloadProjection: "metadata-only", }, diff --git a/device-plane/services/device-control-core/src/server.mjs b/device-plane/services/device-control-core/src/server.mjs index c4feac2..8874250 100644 --- a/device-plane/services/device-control-core/src/server.mjs +++ b/device-plane/services/device-control-core/src/server.mjs @@ -13,6 +13,7 @@ import { } from "./edge-channel-supervisor.mjs"; import { createDeviceGatewayIngest } from "./gateway-ingest.mjs"; import { PostgresDeviceRepository } from "./postgres-repository.mjs"; +import { createTypedCommandRuntime } from "./typed-command-runtime.mjs"; const config = await readConfig(); const repository = new PostgresDeviceRepository({ @@ -21,6 +22,9 @@ const repository = new PostgresDeviceRepository({ }); await repository.migrate(); +const typedCommandRuntime = config.managementApiEnabled && config.edgeChannelEnabled + ? createTypedCommandRuntime({ repository }) + : null; const gatewayIngest = config.discoveryIngestEnabled || config.edgeChannelEnabled ? createDeviceGatewayIngest({ @@ -36,6 +40,7 @@ const edgeChannelSupervisor = config.edgeChannelEnabled trustRoot: config.edgeChannelTrustRoot, maxEdges: config.edgeChannelMaxEdges, reconcileIntervalMs: config.edgeChannelReconcileIntervalMs, + typedCommandRuntime, }) : null; await edgeChannelSupervisor?.start(); @@ -51,6 +56,7 @@ const server = createControlCoreApp({ edgeChannelStatusProvider: edgeChannelSupervisor ? () => edgeChannelSupervisor.status() : null, + typedCommandRuntime, }); server.listen(config.port, config.host, () => { @@ -61,7 +67,7 @@ server.listen(config.port, config.host, () => { discoveryIngest: config.discoveryIngestEnabled, managementApi: config.managementApiEnabled, edgeChannels: config.edgeChannelEnabled, - commandTransport: "disabled", + commandTransport: typedCommandRuntime ? "typed-service-ping-v1" : "disabled", })); }); diff --git a/device-plane/services/device-control-core/src/typed-command-repository.mjs b/device-plane/services/device-control-core/src/typed-command-repository.mjs new file mode 100644 index 0000000..f93c2b8 --- /dev/null +++ b/device-plane/services/device-control-core/src/typed-command-repository.mjs @@ -0,0 +1,299 @@ +import { createHash, randomUUID } from "node:crypto"; + +import { assertProjectCapability } from "./project-management.mjs"; + +const SERVICE_PING_CATALOG = "arusnavi.b2.internal.v1:service-ping"; + +export async function planTypedServicePing(client, { + idempotencyKey, + requestDigest, + actor, + projectId, + deviceId, + expiresAt, +}) { + const commandKey = `service-ping-${createHash("sha256") + .update(`${actor.userRef}\0${idempotencyKey}`, "utf8") + .digest("hex").slice(0, 32)}`; + const project = await findProject(client, projectId); + const grants = await projectGrants(client, projectId); + assertProjectCapability(actor, grants, "command.plan"); + assertProjectCapability(actor, grants, "command.dispatch"); + const device = await findCommandableDevice(client, projectId, deviceId); + + const existing = await client.query( + `select dc.*, di.display_name as device_name + from device_commands dc + join device_instances di on di.id = dc.device_id + where dc.project_id = $1 and dc.command_key = $2 + for update of dc`, + [projectId, commandKey], + ); + if (existing.rows[0]) { + const row = existing.rows[0]; + if ( + row.device_id !== deviceId + || row.command_catalog_ref !== SERVICE_PING_CATALOG + || row.command_type !== "service.ping" + || row.parameters_digest !== requestDigest + ) { + throw domainError("device_command_idempotency_conflict", 409); + } + return { replayed: true, commandId: row.id, command: commandView(row) }; + } + + const commandId = randomUUID(); + const plannedAt = new Date(); + await client.query( + `insert into device_commands ( + id, owner_scope_id, project_id, device_id, command_key, + command_catalog_ref, command_type, risk_class, + parameters_digest, parameters_projection, lifecycle_state, + planned_by_ref, planned_at, expires_at + ) values ( + $1, $2, $3, $4, $5, $6, 'service.ping', 'low', + $7, $8::jsonb, 'queued', $9, $10, $11 + )`, + [ + commandId, + project.owner_scope_id, + projectId, + deviceId, + commandKey, + SERVICE_PING_CATALOG, + requestDigest, + JSON.stringify({ operation: "service.ping", profileRef: device.model_profile_ref }), + actor.userRef, + plannedAt, + expiresAt, + ], + ); + await insertEvent(client, commandId, deviceId, projectId, 1, null, "draft", actor.userRef, "operator_requested"); + await insertEvent(client, commandId, deviceId, projectId, 2, "draft", "planned", actor.userRef, "typed_policy_approved"); + await insertEvent(client, commandId, deviceId, projectId, 3, "planned", "queued", actor.userRef, "awaiting_tracker_session"); + await audit(client, { + eventType: "command.queued", + actorRef: actor.userRef, + projectId, + deviceId, + payload: { commandRef: `command:${commandId}`, commandType: "service.ping", riskClass: "low" }, + }); + return { + replayed: false, + commandId, + command: commandView({ + id: commandId, + device_id: deviceId, + device_name: device.display_name, + command_key: commandKey, + command_catalog_ref: SERVICE_PING_CATALOG, + command_type: "service.ping", + risk_class: "low", + lifecycle_state: "queued", + planned_at: plannedAt, + expires_at: expiresAt, + created_at: plannedAt, + updated_at: plannedAt, + }), + }; +} + +export async function dispatchTypedCommand(client, { + commandId, + transportMessageRef, + now, +}) { + const result = await client.query( + `select dc.*, di.display_name as device_name + from device_commands dc + join device_instances di on di.id = dc.device_id + where dc.id = $1 + for update of dc`, + [commandId], + ); + const row = result.rows[0]; + if (!row) throw domainError("device_command_not_found", 404); + if (row.lifecycle_state !== "queued") return null; + if (new Date(row.expires_at).getTime() <= now.getTime()) { + await client.query( + `update device_commands set lifecycle_state = 'expired', + terminal_at = $2, terminal_reason_code = 'ttl_elapsed', updated_at = $2 + where id = $1`, + [commandId, now], + ); + await insertEvent(client, commandId, row.device_id, row.project_id, 4, "queued", "expired", "workload:device-control-core", "ttl_elapsed"); + return null; + } + await client.query( + `update device_commands set lifecycle_state = 'dispatched', + dispatched_at = $2, transport_message_ref = $3, updated_at = $2 + where id = $1`, + [commandId, now, transportMessageRef], + ); + await insertEvent(client, commandId, row.device_id, row.project_id, 4, "queued", "dispatched", "workload:device-control-core", "tracker_session_allocated", transportMessageRef); + await audit(client, { + eventType: "command.dispatched", + actorRef: "workload:device-control-core", + projectId: row.project_id, + deviceId: row.device_id, + payload: { commandRef: `command:${commandId}`, commandType: row.command_type }, + }); + return commandView({ ...row, lifecycle_state: "dispatched", dispatched_at: now, transport_message_ref: transportMessageRef, updated_at: now }); +} + +export async function recordTypedCommandStatus(client, { + commandId, + transportMessageRef, + lifecycleState, + resultCode, + observedAt, +}) { + const result = await client.query( + `select * from device_commands where id = $1 for update`, + [commandId], + ); + const row = result.rows[0]; + if (!row) throw domainError("device_command_not_found", 404); + if (row.transport_message_ref !== transportMessageRef) { + throw domainError("device_command_transport_mismatch", 409); + } + if (["verified", "failed", "expired", "unknown"].includes(row.lifecycle_state)) { + return; + } + if (lifecycleState === "acknowledged" && row.lifecycle_state === "dispatched") { + await client.query( + `update device_commands set lifecycle_state = 'acknowledged', + acknowledged_at = $2, updated_at = $2 where id = $1`, + [commandId, observedAt], + ); + await insertEvent(client, commandId, row.device_id, row.project_id, 5, "dispatched", "acknowledged", "workload:device-edge", "protocol_reply_received", `result:${resultCode}`); + await client.query( + `update device_commands set lifecycle_state = 'verified', + terminal_at = $2, terminal_reason_code = 'protocol_reply_serv_ok', + updated_at = $2 where id = $1`, + [commandId, observedAt], + ); + await insertEvent(client, commandId, row.device_id, row.project_id, 6, "acknowledged", "verified", "workload:device-control-core", "protocol_reply_serv_ok", `result:${resultCode}`); + await audit(client, { + eventType: "command.verified", + actorRef: "workload:device-control-core", + projectId: row.project_id, + deviceId: row.device_id, + payload: { commandRef: `command:${commandId}`, commandType: row.command_type, resultCode }, + }); + return; + } + if (lifecycleState === "unknown" && row.lifecycle_state === "dispatched") { + await client.query( + `update device_commands set lifecycle_state = 'unknown', terminal_at = $2, + terminal_reason_code = $3, updated_at = $2 where id = $1`, + [commandId, observedAt, resultCode], + ); + await insertEvent(client, commandId, row.device_id, row.project_id, 5, "dispatched", "unknown", "workload:device-edge", resultCode); + } +} + +async function findProject(client, projectId) { + const result = await client.query( + `select id, owner_scope_id, lifecycle_state from device_projects where id = $1 for share`, + [projectId], + ); + const row = result.rows[0]; + if (!row) throw domainError("device_project_not_found", 404); + if (row.lifecycle_state !== "active") throw domainError("device_project_inactive", 409); + return row; +} + +async function findCommandableDevice(client, projectId, deviceId) { + const result = await client.query( + `select di.id, di.display_name, di.model_profile_ref, di.lifecycle_state + from device_instances di + where di.project_id = $1 and di.id = $2 + and exists ( + select 1 from device_routes dr + where dr.project_id = di.project_id + and dr.model_profile_ref = di.model_profile_ref + and dr.direction = 'bidirectional' + and dr.lifecycle_state = 'active' + ) + for share`, + [projectId, deviceId], + ); + const row = result.rows[0]; + if (!row) throw domainError("device_command_route_unavailable", 409); + if (row.model_profile_ref !== "arusnavi.b2.internal.v1") { + throw domainError("device_command_profile_unsupported", 409); + } + if (["suspended", "retired"].includes(row.lifecycle_state)) { + throw domainError("device_command_device_inactive", 409); + } + return row; +} + +async function projectGrants(client, projectId) { + const result = await client.query( + `select id, principal_kind, principal_ref, project_role, + capability_allow, capability_deny, lifecycle_state + from device_project_grants where project_id = $1 for share`, + [projectId], + ); + return result.rows.map((row) => ({ + grantRef: `grant:${row.id}`, + principalKind: row.principal_kind, + principalRef: row.principal_ref, + projectRole: row.project_role, + capabilityAllow: row.capability_allow ?? [], + capabilityDeny: row.capability_deny ?? [], + lifecycleState: row.lifecycle_state, + })); +} + +async function insertEvent(client, commandId, deviceId, projectId, sequence, from, to, actor, reason, evidence = null) { + await client.query( + `insert into device_command_events ( + id, command_id, device_id, project_id, sequence_number, + from_state, to_state, actor_ref, reason_code, evidence_ref + ) values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, + [randomUUID(), commandId, deviceId, projectId, sequence, from, to, actor, reason, evidence], + ); +} + +async function audit(client, { eventType, actorRef, projectId, deviceId, payload }) { + await client.query( + `insert into device_audit_events ( + id, event_type, actor_ref, project_id, device_id, payload + ) values ($1,$2,$3,$4,$5,$6::jsonb)`, + [randomUUID(), eventType, actorRef, projectId, deviceId, JSON.stringify(payload)], + ); +} + +function commandView(row) { + return Object.freeze({ + commandRef: `command:${row.id}`, + deviceRef: `device:${row.device_id}`, + deviceName: row.device_name, + commandKey: row.command_key, + commandCatalogRef: row.command_catalog_ref, + commandType: row.command_type, + riskClass: row.risk_class, + lifecycleState: row.lifecycle_state, + plannedAt: iso(row.planned_at), + expiresAt: iso(row.expires_at), + dispatchedAt: iso(row.dispatched_at), + acknowledgedAt: iso(row.acknowledged_at), + terminalAt: iso(row.terminal_at), + terminalReasonCode: row.terminal_reason_code ?? null, + createdAt: iso(row.created_at), + updatedAt: iso(row.updated_at), + }); +} + +function iso(value) { + return value == null ? null : new Date(value).toISOString(); +} + +function domainError(code, statusCode) { + const error = new Error(code); + error.statusCode = statusCode; + return error; +} diff --git a/device-plane/services/device-control-core/src/typed-command-runtime.mjs b/device-plane/services/device-control-core/src/typed-command-runtime.mjs new file mode 100644 index 0000000..3e60bfd --- /dev/null +++ b/device-plane/services/device-control-core/src/typed-command-runtime.mjs @@ -0,0 +1,149 @@ +import { createHash, randomUUID } from "node:crypto"; + +export function createTypedCommandRuntime({ repository, now = () => new Date() } = {}) { + if ( + !repository + || typeof repository.planTypedServicePing !== "function" + || typeof repository.dispatchTypedCommand !== "function" + || typeof repository.recordTypedCommandStatus !== "function" + ) { + throw new TypeError("device_typed_command_repository_required"); + } + const credentials = new Map(); + + return Object.freeze({ + async planServicePing({ idempotencyKey, actor, input }) { + const normalized = normalizeInput(input); + const expiresAt = new Date(now().getTime() + normalized.expiresInSeconds * 1000); + const requestDigest = `sha256:${createHash("sha256").update(JSON.stringify({ + actorRef: actor.userRef, + projectId: normalized.projectId, + deviceId: normalized.deviceId, + operation: "service.ping", + expiresInSeconds: normalized.expiresInSeconds, + }), "utf8").digest("hex")}`; + const execution = await repository.planTypedServicePing({ + idempotencyKey, + requestDigest, + actor, + projectId: normalized.projectId, + deviceId: normalized.deviceId, + expiresAt, + }); + if (!execution.replayed && execution.command.lifecycleState === "queued") { + credentials.set(execution.commandId, Object.freeze({ + deviceId: normalized.deviceId, + accessCode: normalized.accessCode, + expiresAt: new Date(execution.command.expiresAt).getTime(), + })); + } + return { replayed: execution.replayed, command: execution.command }; + }, + + async offerForDevice(deviceRef) { + const deviceId = entityId(deviceRef, "device"); + const at = now(); + for (const [commandId, secret] of credentials) { + if (secret.expiresAt <= at.getTime()) { + await repository.dispatchTypedCommand({ + commandId, + transportMessageRef: `edge-command:${randomUUID()}`, + now: at, + }); + credentials.delete(commandId); + continue; + } + if (secret.deviceId !== deviceId) continue; + const transportMessageRef = `edge-command:${randomUUID()}`; + const dispatched = await repository.dispatchTypedCommand({ + commandId, + transportMessageRef, + now: at, + }); + if (!dispatched) { + credentials.delete(commandId); + continue; + } + return Object.freeze({ + commandRef: `command:${commandId}`, + commandType: "service.ping", + accessCode: secret.accessCode, + transportMessageRef, + }); + } + return null; + }, + + async recordStatus(status) { + const commandId = entityId(status?.commandRef, "command"); + const normalized = normalizeStatus(status); + await repository.recordTypedCommandStatus({ commandId, ...normalized }); + credentials.delete(commandId); + }, + + status() { + return Object.freeze({ + commandTransport: "typed-service-ping-v1", + transientAuthorizations: credentials.size, + }); + }, + }); +} + +function normalizeInput(input) { + if (!input || typeof input !== "object" || Array.isArray(input)) { + throw domainError("device_service_ping_input_invalid", 400); + } + const keys = Object.keys(input).sort().join(","); + if (keys !== "accessCode,deviceRef,expiresInSeconds,projectRef") { + throw domainError("device_service_ping_input_invalid", 400); + } + if (typeof input.accessCode !== "string" || !/^\d{6}$/.test(input.accessCode)) { + throw domainError("device_service_ping_access_code_invalid", 400); + } + const expiresInSeconds = Number(input.expiresInSeconds); + if (!Number.isSafeInteger(expiresInSeconds) || expiresInSeconds < 30 || expiresInSeconds > 1800) { + throw domainError("device_service_ping_ttl_invalid", 400); + } + return { + projectId: entityId(input.projectRef, "project"), + deviceId: entityId(input.deviceRef, "device"), + accessCode: input.accessCode, + expiresInSeconds, + }; +} + +function normalizeStatus(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError("device_command_status_invalid"); + } + if (!['acknowledged', 'unknown'].includes(value.lifecycleState)) { + throw new TypeError("device_command_status_invalid"); + } + if (typeof value.transportMessageRef !== "string" || !/^edge-command:[0-9a-f-]{36}$/i.test(value.transportMessageRef)) { + throw new TypeError("device_command_status_invalid"); + } + if (typeof value.resultCode !== "string" || !/^[a-z][a-z0-9._-]{1,63}$/.test(value.resultCode)) { + throw new TypeError("device_command_status_invalid"); + } + const observedAt = new Date(value.observedAt); + if (Number.isNaN(observedAt.getTime())) throw new TypeError("device_command_status_invalid"); + return { + transportMessageRef: value.transportMessageRef.toLowerCase(), + lifecycleState: value.lifecycleState, + resultCode: value.resultCode, + observedAt, + }; +} + +function entityId(value, prefix) { + const match = String(value || "").match(new RegExp(`^${prefix}:([0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})$`, "i")); + if (!match) throw domainError(`device_${prefix}_ref_invalid`, 400); + return match[1].toLowerCase(); +} + +function domainError(code, statusCode) { + const error = new Error(code); + error.statusCode = statusCode; + return error; +} diff --git a/device-plane/services/device-control-core/test/app.test.mjs b/device-plane/services/device-control-core/test/app.test.mjs index cf448ad..9f7f6d0 100644 --- a/device-plane/services/device-control-core/test/app.test.mjs +++ b/device-plane/services/device-control-core/test/app.test.mjs @@ -255,6 +255,58 @@ test("management API exposes a repository idempotency conflict without retrying" } }); +test("typed service ping accepts a transient access code and never echoes it", async () => { + const projectRef = "project:11111111-1111-4111-8111-111111111111"; + const deviceRef = "device:22222222-2222-4222-8222-222222222222"; + let planned; + const runtime = await startTestServer({ + managementApiEnabled: true, + managementToken, + repository: { + health: async () => "ready", + executeManagementCommand: async () => ({ replayed: false, result: {} }), + }, + typedCommandRuntime: { + status: () => ({ commandTransport: "typed-service-ping-v1" }), + planServicePing: async (value) => { + planned = value; + return { + replayed: false, + command: { + commandRef: "command:33333333-3333-4333-8333-333333333333", + deviceRef, + commandType: "service.ping", + lifecycleState: "queued", + }, + }; + }, + }, + }); + try { + const response = await fetch( + `${runtime.baseUrl}/internal/v1/commands:service-ping`, + { + method: "POST", + headers: managementHeaders(), + body: JSON.stringify({ + projectRef, + deviceRef, + accessCode: "654321", + expiresInSeconds: 300, + }), + }, + ); + assert.equal(response.status, 200); + const body = await response.json(); + assert.equal(body.result.lifecycleState, "queued"); + assert.equal(JSON.stringify(body).includes("654321"), false); + assert.equal(planned.input.accessCode, "654321"); + assert.equal(planned.idempotencyKey, "phase2-test-0001"); + } finally { + await runtime.close(); + } +}); + test("project query is service-authenticated and forwards only the trusted actor", async () => { let queriedActor; const runtime = await startTestServer({ @@ -424,12 +476,15 @@ test("gateway message endpoint returns acceptance only after repository commit", acceptAdapterMessage: async (value) => { stored = value; return { - schemaVersion: "nodedc.device-adapter-acceptance.v1", - acceptanceRef: "acceptance:test-001", - idempotencyKey: value.safeView.idempotencyKey, - status: "accepted", - replayed: false, - acceptedAt: "2026-08-11T12:00:00.000Z", + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test-001", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }, + claimedDeviceRef: "device:11111111-1111-4111-8111-111111111111", }; }, }, diff --git a/device-plane/services/device-control-core/test/gateway-ingest.test.mjs b/device-plane/services/device-control-core/test/gateway-ingest.test.mjs index 148689f..64bfb75 100644 --- a/device-plane/services/device-control-core/test/gateway-ingest.test.mjs +++ b/device-plane/services/device-control-core/test/gateway-ingest.test.mjs @@ -25,12 +25,15 @@ test("shared gateway ingest masks identifiers for HTTP and Edge callers", async async acceptAdapterMessage(value) { stored.push(value); return { - schemaVersion: "nodedc.device-adapter-acceptance.v1", - acceptanceRef: "acceptance:test", - idempotencyKey: value.safeView.idempotencyKey, - status: "accepted", - replayed: false, - acceptedAt: "2026-08-11T12:00:00.000Z", + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:test", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }, + claimedDeviceRef: "device:11111111-1111-4111-8111-111111111111", }; }, }, @@ -40,7 +43,11 @@ test("shared gateway ingest masks identifiers for HTTP and Edge callers", async const acceptance = await ingest.acceptMessage(adapterMessage()); assert.equal(discovery.value.identifier.masked, "***********0001"); - assert.equal(acceptance.status, "accepted"); + assert.equal(acceptance.value.status, "accepted"); + assert.equal( + acceptance.claimedDeviceRef, + "device:11111111-1111-4111-8111-111111111111", + ); assert.match(stored[0].identifierDigest, /^hmac-sha256:[a-f0-9]{64}$/); assert.match(stored[1].requestDigest, /^sha256:[a-f0-9]{64}$/); assert.equal(JSON.stringify(stored).includes(rawImei), false); diff --git a/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs b/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs index a1e5044..6f7ddf6 100644 --- a/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs +++ b/device-plane/services/device-control-core/test/gateway-message-repository.test.mjs @@ -22,10 +22,11 @@ test("commits a gateway receipt before returning Core acceptance", async () => { const result = await acceptGatewayMessage(messageInput(client)); - assert.equal(result.status, "accepted"); - assert.equal(result.replayed, false); - assert.equal(result.idempotencyKey, idempotencyKey); - assert.equal(result.acceptedAt, acceptedAt.toISOString()); + assert.equal(result.acceptance.status, "accepted"); + assert.equal(result.acceptance.replayed, false); + assert.equal(result.acceptance.idempotencyKey, idempotencyKey); + assert.equal(result.acceptance.acceptedAt, acceptedAt.toISOString()); + assert.equal(result.claimedDeviceRef, null); assert.equal(client.remaining(), 0); assert.equal(client.released, true); }); @@ -47,8 +48,8 @@ test("replays one durable receipt for the same normalized request", async () => const result = await acceptGatewayMessage(messageInput(client)); - assert.equal(result.status, "accepted"); - assert.equal(result.replayed, true); + assert.equal(result.acceptance.status, "accepted"); + assert.equal(result.acceptance.replayed, true); assert.equal(client.remaining(), 0); assert.equal(client.released, true); }); diff --git a/device-plane/services/device-control-core/test/typed-command-runtime.test.mjs b/device-plane/services/device-control-core/test/typed-command-runtime.test.mjs new file mode 100644 index 0000000..6bbdb3c --- /dev/null +++ b/device-plane/services/device-control-core/test/typed-command-runtime.test.mjs @@ -0,0 +1,106 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { createTypedCommandRuntime } from "../src/typed-command-runtime.mjs"; + +const projectRef = "project:11111111-1111-4111-8111-111111111111"; +const deviceRef = "device:22222222-2222-4222-8222-222222222222"; + +test("keeps the B2 access code transient and emits only a typed offer", async () => { + const planned = []; + const dispatched = []; + const runtime = createTypedCommandRuntime({ + now: () => new Date("2026-08-12T18:00:00.000Z"), + repository: { + async planTypedServicePing(value) { + planned.push(value); + return { + replayed: false, + commandId: "33333333-3333-4333-8333-333333333333", + command: { + lifecycleState: "queued", + expiresAt: "2026-08-12T18:05:00.000Z", + }, + }; + }, + async dispatchTypedCommand(value) { + dispatched.push(value); + return { lifecycleState: "dispatched" }; + }, + async recordTypedCommandStatus() {}, + }, + }); + await runtime.planServicePing({ + idempotencyKey: "idem-00000001", + actor: { userRef: "user:test" }, + input: { projectRef, deviceRef, accessCode: "123456", expiresInSeconds: 300 }, + }); + assert.equal(JSON.stringify(planned).includes("123456"), false); + const offer = await runtime.offerForDevice(deviceRef); + assert.equal(offer.commandType, "service.ping"); + assert.equal(offer.accessCode, "123456"); + assert.match(offer.transportMessageRef, /^edge-command:/); + assert.equal(dispatched.length, 1); +}); + +test("expires a transient authorization through the durable ledger", async () => { + let current = new Date("2026-08-12T18:00:00.000Z"); + const dispatches = []; + const runtime = createTypedCommandRuntime({ + now: () => current, + repository: { + async planTypedServicePing() { + return { + replayed: false, + commandId: "33333333-3333-4333-8333-333333333333", + command: { + lifecycleState: "queued", + expiresAt: "2026-08-12T18:00:30.000Z", + }, + }; + }, + async dispatchTypedCommand(value) { + dispatches.push(value); + return null; + }, + async recordTypedCommandStatus() {}, + }, + }); + await runtime.planServicePing({ + idempotencyKey: "idem-00000002", + actor: { userRef: "user:test" }, + input: { projectRef, deviceRef, accessCode: "123456", expiresInSeconds: 30 }, + }); + current = new Date("2026-08-12T18:00:31.000Z"); + assert.equal(await runtime.offerForDevice(deviceRef), null); + assert.equal(dispatches.length, 1); + assert.equal(runtime.status().transientAuthorizations, 0); +}); + +test("does not recreate a transient authorization on an idempotent replay", async () => { + const runtime = createTypedCommandRuntime({ + repository: { + async planTypedServicePing() { + return { + replayed: true, + commandId: "33333333-3333-4333-8333-333333333333", + command: { + lifecycleState: "queued", + expiresAt: "2026-08-12T18:05:00.000Z", + }, + }; + }, + async dispatchTypedCommand() { + throw new Error("must_not_dispatch_replayed_secret"); + }, + async recordTypedCommandStatus() {}, + }, + }); + await runtime.planServicePing({ + idempotencyKey: "idem-00000003", + actor: { userRef: "user:test" }, + input: { projectRef, deviceRef, accessCode: "654321", expiresInSeconds: 300 }, + }); + assert.equal(runtime.status().transientAuthorizations, 0); + assert.equal(await runtime.offerForDevice(deviceRef), null); +}); diff --git a/device-plane/services/device-edge-channel/src/runtime.mjs b/device-plane/services/device-edge-channel/src/runtime.mjs index d5859f4..e5cd861 100644 --- a/device-plane/services/device-edge-channel/src/runtime.mjs +++ b/device-plane/services/device-edge-channel/src/runtime.mjs @@ -90,7 +90,7 @@ export function createDeviceEdgeChannelServer(options = {}) { status: "ready", transport: "http2-mtls", trustGeneration: trust.generationRef, - commandTransport: "disabled", + commandTransport: config.commandTransport, }, { trackerSessionId: CHANNEL_TRACKER_SESSION_ID, adapterProfileRef: CHANNEL_PROFILE_REF, @@ -156,7 +156,10 @@ export function createDeviceEdgeChannelServer(options = {}) { if (!result?.discovery) { throw new Error("device_edge_channel_discovery_acceptance_invalid"); } - return result.discovery; + return Object.freeze({ + ...result.discovery, + ...(result.commandOffer ? { commandOffer: result.commandOffer } : {}), + }); }, async submitAdapterMessage(message) { const normalized = normalizeAdapterMessage(message, { @@ -173,7 +176,24 @@ export function createDeviceEdgeChannelServer(options = {}) { if (acceptance.idempotencyKey !== normalized.idempotencyKey) { throw new Error("device_edge_channel_acceptance_mismatch"); } - return acceptance; + return Object.freeze({ + ...acceptance, + ...(result.commandOffer ? { commandOffer: result.commandOffer } : {}), + }); + }, + async submitCommandStatus(status) { + const normalized = normalizeCommandStatus(status); + const result = await submitEvent("command.status", { + status: normalized, + }, { + trackerSessionId: normalized.sessionRef, + adapterProfileRef: normalized.adapterProfileRef, + eventAt: normalized.observedAt, + }); + if (result?.status !== "recorded") { + throw new Error("device_edge_channel_command_status_invalid"); + } + return Object.freeze({ status: "recorded" }); }, status() { return Object.freeze({ @@ -191,7 +211,7 @@ export function createDeviceEdgeChannelServer(options = {}) { eventsRejected: totalEventsRejected, protocolFailures: totalProtocolFailures, trackerIngress: "disabled", - commandTransport: "disabled", + commandTransport: config.commandTransport, }); }, rotateTrust(next) { @@ -271,7 +291,10 @@ export function createDeviceEdgeChannelServer(options = {}) { if (envelope.messageKind !== "channel.accepted") { throw new Error("device_edge_channel_acceptance_required"); } - if (envelope.payload?.status !== "accepted") { + if ( + envelope.payload?.status !== "accepted" + || envelope.payload?.commandTransport !== config.commandTransport + ) { throw new Error("device_edge_channel_acceptance_invalid"); } state.accepted = true; @@ -410,6 +433,7 @@ function normalizeConfig(options) { return Object.freeze({ edgeRegistrationId, channelGeneration, + commandTransport: normalizeCommandTransport(options.commandTransport), trust, host: normalizeHost(options.host ?? "127.0.0.1"), port: normalizePort(options.port ?? 443), @@ -442,6 +466,50 @@ function normalizeConfig(options) { }); } +function normalizeCommandTransport(value) { + const normalized = value ?? "disabled"; + if (!["disabled", "typed-service-ping-v1"].includes(normalized)) { + throw new TypeError("device_edge_channel_command_transport_invalid"); + } + return normalized; +} + +function normalizeCommandStatus(value) { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError("device_edge_channel_command_status_invalid"); + } + const commandRef = normalizeRef(value.commandRef, "command_ref"); + const transportMessageRef = normalizeRef( + value.transportMessageRef, + "transport_message_ref", + ); + const sessionRef = normalizeRef(value.sessionRef, "tracker_session_ref"); + const adapterProfileRef = normalizeRef( + value.adapterProfileRef, + "adapter_profile_ref", + ); + if (!["acknowledged", "unknown"].includes(value.lifecycleState)) { + throw new TypeError("device_edge_channel_command_lifecycle_invalid"); + } + const resultCode = String(value.resultCode || ""); + if (!/^[a-z][a-z0-9._-]{1,63}$/.test(resultCode)) { + throw new TypeError("device_edge_channel_command_result_invalid"); + } + const observedAt = new Date(value.observedAt); + if (Number.isNaN(observedAt.getTime())) { + throw new TypeError("device_edge_channel_command_observed_at_invalid"); + } + return Object.freeze({ + commandRef, + transportMessageRef, + sessionRef, + adapterProfileRef, + lifecycleState: value.lifecycleState, + resultCode, + observedAt: observedAt.toISOString(), + }); +} + function normalizeTrust(value) { if (!value || typeof value !== "object") { throw new TypeError("device_edge_channel_tls_invalid"); diff --git a/device-plane/services/device-edge-channel/test/channel-integration.test.mjs b/device-plane/services/device-edge-channel/test/channel-integration.test.mjs index 1143ff4..1f90709 100644 --- a/device-plane/services/device-edge-channel/test/channel-integration.test.mjs +++ b/device-plane/services/device-edge-channel/test/channel-integration.test.mjs @@ -68,6 +68,59 @@ test("accepts synthetic discovery and durable message results over Core-initiate } }); +test("recovers the claimed device from telemetry after a Core restart and completes a typed command", async () => { + const commandRef = "command:11111111-1111-4111-8111-111111111111"; + const deviceRef = "device:22222222-2222-4222-8222-222222222222"; + const transportMessageRef = "edge-command:33333333-3333-4333-8333-333333333333"; + const recorded = []; + let offers = 0; + const pair = await startPair({ + commandTransport: "typed-service-ping-v1", + acceptMessage: async (message) => ({ + value: acceptanceFor(message, false), + claimedDeviceRef: deviceRef, + }), + offerCommand: async (offeredDeviceRef) => { + assert.equal(offeredDeviceRef, deviceRef); + offers += 1; + return { + commandRef, + commandType: "service.ping", + accessCode: "123456", + transportMessageRef, + }; + }, + recordCommandStatus: async (status) => recorded.push(status), + }); + + try { + const receipt = await pair.edge.submitAdapterMessage(adapterMessage()); + assert.equal(receipt.status, "accepted"); + assert.deepEqual(receipt.commandOffer, { + commandRef, + commandType: "service.ping", + accessCode: "123456", + transportMessageRef, + }); + assert.equal(offers, 1); + + await pair.edge.submitCommandStatus({ + commandRef, + transportMessageRef, + lifecycleState: "acknowledged", + resultCode: "serv_ok", + observedAt: "2026-08-12T17:30:00.000Z", + sessionRef: "session:pilot-1", + adapterProfileRef: "arusnavi.internal.b2.v1", + }); + assert.equal(recorded.length, 1); + assert.equal(recorded[0].resultCode, "serv_ok"); + assert.equal(pair.core.status().negotiatedCommandTransport, "typed-service-ping-v1"); + } finally { + await stopPair(pair); + } +}); + test("keeps the channel alive and reconnects without losing idempotency", async () => { const acceptedMessages = new Map(); const pair = await startPair({ @@ -458,6 +511,7 @@ function createEdgeServer(options = {}) { deadPeerMs: options.deadPeerMs ?? 150, acceptanceTimeoutMs: 500, maxPendingAcceptances: options.maxPendingAcceptances, + commandTransport: options.commandTransport, }); } @@ -490,6 +544,9 @@ function createCoreClient(options) { observeDiscovery: options.observeDiscovery ?? productionDiscoveryObserver(), acceptMessage: options.acceptMessage ?? (async (message) => acceptanceFor(message, false)), + commandTransport: options.commandTransport, + offerCommand: options.offerCommand, + recordCommandStatus: options.recordCommandStatus, }); } @@ -536,7 +593,7 @@ function discoverySignal() { modelProfileRef: "arusnavi.internal.b2.v1", protocol: "INTERNAL", observedAt: new Date().toISOString(), - identifier: { kind: "imei", value: "863151070211088" }, + identifier: { kind: "imei", value: "860000000000001" }, evidence: { transport: "tcp", bytesObserved: 10, @@ -559,7 +616,7 @@ function adapterMessage(overrides = {}) { sequence: 1, observedAt: new Date().toISOString(), idempotencyKey: `sha256:${"a".repeat(64)}`, - identifier: { kind: "imei", value: "863151070211088" }, + identifier: { kind: "imei", value: "860000000000001" }, payloadSchemaRef: "arusnavi.internal.package-metadata.v1", payload: { packageNumber: 1, diff --git a/device-plane/services/device-gateway/src/runtime.mjs b/device-plane/services/device-gateway/src/runtime.mjs index cd9bfcf..6900ede 100644 --- a/device-plane/services/device-gateway/src/runtime.mjs +++ b/device-plane/services/device-gateway/src/runtime.mjs @@ -52,6 +52,7 @@ export function createDeviceGatewayRuntime(options = {}) { processing: false, rejected: false, closed: false, + pendingCommand: null, }; sessions.set(socket, session); incrementAddressSessions(remoteAddress); @@ -103,7 +104,7 @@ export function createDeviceGatewayRuntime(options = {}) { publicIngress: config.publicIngressEnabled ? "telemetry-ingest" : "disabled", - commandTransport: "disabled", + commandTransport: config.commandTransport, sessions: { active: sessions.size, accepted: totalAccepted, @@ -149,7 +150,7 @@ export function createDeviceGatewayRuntime(options = {}) { totalMessagesAccepted, totalPackagesAcknowledged, totalBufferedBytes, - commandTransport: "disabled", + commandTransport: config.commandTransport, publicIngress: config.publicIngressEnabled ? "telemetry-ingest" : "disabled", @@ -186,9 +187,39 @@ export function createDeviceGatewayRuntime(options = {}) { await writeWithBackpressure(socket, session.adapterSession.buildHeaderAcknowledgement( Math.floor(new Date(observedAt).getTime() / 1000), )); + if (acceptedDiscovery.commandOffer) { + await dispatchCommandOffer( + socket, + session, + acceptedDiscovery.commandOffer, + ); + } continue; } + if (session.pendingCommand) { + const result = session.adapterSession.parseTypedCommandResponse( + session.buffer, + session.pendingCommand, + ); + if (result.status === "incomplete") return; + if (result.status === "acknowledged") { + const command = session.pendingCommand; + consumeBuffer(session, result.bytesConsumed); + session.pendingCommand = null; + await config.onCommandStatus({ + commandRef: command.commandRef, + transportMessageRef: command.transportMessageRef, + lifecycleState: "acknowledged", + resultCode: result.resultCode, + observedAt: config.now().toISOString(), + sessionRef: session.sessionRef, + adapterProfileRef: config.profile.profileRef, + }); + continue; + } + } + const parsed = session.adapterSession.parseMessage(session.buffer); if (parsed.status === "incomplete") return; const observedAt = config.now().toISOString(); @@ -215,9 +246,12 @@ export function createDeviceGatewayRuntime(options = {}) { payloadSchemaRef: parsed.payloadSchemaRef, payload: parsed.payload, }); - const acceptance = normalizeAdapterAcceptance( - await config.onMessage(message), - ); + const acceptedMessage = await config.onMessage(message); + const { + commandOffer, + ...acceptanceValue + } = acceptedMessage; + const acceptance = normalizeAdapterAcceptance(acceptanceValue); if (acceptance.idempotencyKey !== message.idempotencyKey) { throw new TypeError("device_gateway_core_acceptance_mismatch"); } @@ -228,9 +262,26 @@ export function createDeviceGatewayRuntime(options = {}) { socket, session.adapterSession.buildMessageAcknowledgement(parsed), ); + if (commandOffer) { + await dispatchCommandOffer( + socket, + session, + commandOffer, + ); + } } } + async function dispatchCommandOffer(socket, session, value) { + if (session.pendingCommand) { + throw new Error("device_gateway_command_already_pending"); + } + const command = normalizeCommandOffer(value); + const bytes = session.adapterSession.buildTypedCommand(command); + session.pendingCommand = command; + await writeWithBackpressure(socket, bytes); + } + function currentAddressSessions(remoteAddress) { return sessionsByAddress.get(remoteAddress) || 0; } @@ -287,8 +338,21 @@ export function createDeviceGatewayRuntime(options = {}) { function closeSession(socket, session) { if (session.closed) return; session.closed = true; - totalBufferedBytes -= session.buffer.length; + totalBufferedBytes = Math.max(0, totalBufferedBytes - session.buffer.length); session.buffer = Buffer.alloc(0); + if (session.pendingCommand && config.onCommandStatus) { + const command = session.pendingCommand; + session.pendingCommand = null; + void config.onCommandStatus({ + commandRef: command.commandRef, + transportMessageRef: command.transportMessageRef, + lifecycleState: "unknown", + resultCode: "tracker_session_closed", + observedAt: config.now().toISOString(), + sessionRef: session.sessionRef, + adapterProfileRef: config.profile.profileRef, + }).catch(() => undefined); + } sessions.delete(socket); decrementAddressSessions(session.remoteAddress); } @@ -302,8 +366,9 @@ export function createDeviceGatewayRuntime(options = {}) { } function consumeBuffer(session, bytesConsumed) { - session.buffer = session.buffer.subarray(bytesConsumed); - totalBufferedBytes -= bytesConsumed; + const consumed = Math.min(bytesConsumed, session.buffer.length); + session.buffer = session.buffer.subarray(consumed); + totalBufferedBytes = Math.max(0, totalBufferedBytes - consumed); } } @@ -326,11 +391,25 @@ function normalizeConfig(input) { const registration = listenEnabled ? resolveAdapterRegistration(input.adapterRegistry, input.protocolProfileRef) : null; + if (registration?.profile?.commandTransport?.status === "typed-service-ping-v1") { + const probe = assertDeviceAdapterSession(registration.adapter.createSession({ + profileRef: registration.profile.profileRef, + })); + if ( + typeof probe.buildTypedCommand !== "function" + || typeof probe.parseTypedCommandResponse !== "function" + || typeof input.onCommandStatus !== "function" + ) { + throw new TypeError("device_gateway_typed_command_runtime_required"); + } + } return { listenEnabled, publicIngressEnabled, adapter: registration?.adapter ?? null, profile: registration?.profile ?? null, + commandTransport: + registration?.profile?.commandTransport?.status ?? "disabled", edgeRef: listenEnabled ? normalizeOpaqueRef(input.edgeRef, "device_gateway_edge_ref_invalid") : "edge:disabled", @@ -412,10 +491,36 @@ function normalizeConfig(input) { onMessage: typeof input.onMessage === "function" ? input.onMessage : undefined, + onCommandStatus: typeof input.onCommandStatus === "function" + ? input.onCommandStatus + : undefined, now: typeof input.now === "function" ? input.now : () => new Date(), }; } +function normalizeCommandOffer(value) { + if ( + !value + || typeof value !== "object" + || Array.isArray(value) + || typeof value.commandRef !== "string" + || !/^command:[0-9a-f-]{36}$/i.test(value.commandRef) + || value.commandType !== "service.ping" + || typeof value.accessCode !== "string" + || !/^\d{6}$/.test(value.accessCode) + || typeof value.transportMessageRef !== "string" + || !/^edge-command:[0-9a-f-]{36}$/i.test(value.transportMessageRef) + ) { + throw new TypeError("device_gateway_command_offer_invalid"); + } + return Object.freeze({ + commandRef: value.commandRef.toLowerCase(), + commandType: value.commandType, + accessCode: value.accessCode, + transportMessageRef: value.transportMessageRef.toLowerCase(), + }); +} + function resolveAdapterRegistration(registry, profileRef) { if (!registry || typeof registry.resolveProfile !== "function") { throw new TypeError("device_gateway_adapter_registry_required"); diff --git a/device-plane/services/device-gateway/test/discovery-integration.test.mjs b/device-plane/services/device-gateway/test/discovery-integration.test.mjs index 4af836c..6404ed2 100644 --- a/device-plane/services/device-gateway/test/discovery-integration.test.mjs +++ b/device-plane/services/device-gateway/test/discovery-integration.test.mjs @@ -44,12 +44,14 @@ test("B2 HEADER2 becomes a persisted masked quarantine discovery before ACK", as acceptAdapterMessage: async (value) => { storedMessage = value; return { - schemaVersion: "nodedc.device-adapter-acceptance.v1", - acceptanceRef: "acceptance:integration-001", - idempotencyKey: value.safeView.idempotencyKey, - status: "accepted", - replayed: false, - acceptedAt: "2026-08-11T12:00:00.000Z", + acceptance: { + schemaVersion: "nodedc.device-adapter-acceptance.v1", + acceptanceRef: "acceptance:integration-001", + idempotencyKey: value.safeView.idempotencyKey, + status: "accepted", + replayed: false, + acceptedAt: "2026-08-11T12:00:00.000Z", + }, }; }, }, @@ -73,6 +75,7 @@ test("B2 HEADER2 becomes a persisted masked quarantine discovery before ACK", as now: () => new Date(0x52db95de * 1000), onDiscovery: observe.observeDiscovery, onMessage: observe.acceptMessage, + onCommandStatus: async () => undefined, }); const addresses = await gateway.start(); try { @@ -115,6 +118,7 @@ function listen(server) { function close(server) { return new Promise((resolve, reject) => { server.close((error) => (error ? reject(error) : resolve())); + server.closeAllConnections?.(); }); } @@ -134,5 +138,12 @@ function exchange(port, payload, expectedBytes) { } }); socket.on("error", reject); + socket.on("close", () => { + if (byteLength < expectedBytes) { + reject(new Error( + `device_gateway_test_socket_closed_early:${byteLength}/${expectedBytes}`, + )); + } + }); }); } diff --git a/device-plane/services/device-gateway/test/runtime.test.mjs b/device-plane/services/device-gateway/test/runtime.test.mjs index 7581d2e..7ef2b2e 100644 --- a/device-plane/services/device-gateway/test/runtime.test.mjs +++ b/device-plane/services/device-gateway/test/runtime.test.mjs @@ -81,7 +81,7 @@ test("telemetry ingress persists HEADER2 before acknowledging packages", async ( assert.equal(runtime.status().totalDiscoveries, 1); assert.equal(runtime.status().totalMessagesAccepted, 1); assert.equal(runtime.status().totalPackagesAcknowledged, 1); - assert.equal(runtime.status().commandTransport, "disabled"); + assert.equal(runtime.status().commandTransport, "typed-service-ping-v1"); assert.equal(runtime.status().publicIngress, "telemetry-ingest"); const response = await fetch( @@ -91,7 +91,52 @@ test("telemetry ingress persists HEADER2 before acknowledging packages", async ( assert.equal(body.framing, "verified-read-only"); assert.equal(body.tcpListener, "telemetry-ingest"); assert.equal(body.publicIngress, "telemetry-ingest"); - assert.equal(body.commandTransport, "disabled"); + assert.equal(body.commandTransport, "typed-service-ping-v1"); + } finally { + client.socket.destroy(); + await runtime.stop(); + } +}); + +test("dispatches a typed service ping on the next telemetry package and records SERV OK", async () => { + const statuses = []; + const runtime = createDeviceGatewayRuntime({ + healthPort: 0, + tcpPort: 0, + listenEnabled: true, + ...gatewayAdapterOptions(), + onDiscovery: async () => ({ lifecycleState: "claimed" }), + onMessage: async (message) => ({ + ...acceptanceFor(message), + commandOffer: { + commandRef: "command:11111111-1111-4111-8111-111111111111", + commandType: "service.ping", + accessCode: "123456", + transportMessageRef: "edge-command:22222222-2222-4222-8222-222222222222", + }, + }), + onCommandStatus: async (status) => statuses.push(status), + }); + const addresses = await runtime.start(); + const client = await connectAndCollect(addresses.tcpAddress.port); + try { + client.socket.write(Buffer.concat([specificationHeader, specificationPackage])); + await client.waitForBytes(28); + assert.equal( + client.bytes().subarray(13).toString("ascii"), + "123456*SERV*1.1", + ); + client.socket.write(Buffer.from("SERV OK", "ascii")); + await waitFor(() => statuses.length === 1); + assert.deepEqual(statuses[0], { + commandRef: "command:11111111-1111-4111-8111-111111111111", + transportMessageRef: "edge-command:22222222-2222-4222-8222-222222222222", + lifecycleState: "acknowledged", + resultCode: "serv_ok", + observedAt: statuses[0].observedAt, + sessionRef: statuses[0].sessionRef, + adapterProfileRef: "arusnavi.b2.internal.v1", + }); } finally { client.socket.destroy(); await runtime.stop(); @@ -319,6 +364,11 @@ function connectAndCollect(port) { for (const waiter of waiters.splice(0)) waiter.waitReject(error); reject(error); }); + socket.on("close", () => { + for (const waiter of waiters.splice(0)) { + waiter.waitReject(new Error("device_gateway_test_socket_closed")); + } + }); }); } @@ -339,6 +389,7 @@ function gatewayAdapterOptions() { adapterRegistry: DEVICE_ADAPTER_CATALOG.registry, protocolProfileRef: DEVICE_ADAPTER_CATALOG.defaultProfileRef, edgeRef: "edge:test-001", + onCommandStatus: async () => undefined, }; } @@ -352,3 +403,11 @@ function acceptanceFor(message, replayed = false) { acceptedAt: "2026-08-11T12:00:00.000Z", }; } + +async function waitFor(predicate, timeoutMs = 1_000) { + const deadline = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error("test_wait_timeout"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} diff --git a/device-plane/vps/edge-process/device-edge-runtime.mjs b/device-plane/vps/edge-process/device-edge-runtime.mjs index fb1e4c5..4376e3b 100644 --- a/device-plane/vps/edge-process/device-edge-runtime.mjs +++ b/device-plane/vps/edge-process/device-edge-runtime.mjs @@ -24,7 +24,10 @@ export async function main(environment = process.env) { environment, base.channel.edgeRegistrationId, ); - const channel = createDeviceEdgeChannelServer(base.channel); + const channel = createDeviceEdgeChannelServer({ + ...base.channel, + commandTransport: "typed-service-ping-v1", + }); const gateway = createDeviceGatewayRuntime({ listenEnabled: true, publicIngressEnabled: true, @@ -46,6 +49,7 @@ export async function main(environment = process.env) { sessionTimeoutMs: tracker.sessionTimeoutMs, onDiscovery: (signal) => channel.submitDiscovery(signal), onMessage: (message) => channel.submitAdapterMessage(message), + onCommandStatus: (status) => channel.submitCommandStatus(status), }); const health = createCombinedHealthServer(channel, gateway, base.health); let stopping = false; @@ -72,7 +76,7 @@ export async function main(environment = process.env) { edgeRegistrationId: base.channel.edgeRegistrationId, channelGeneration: base.channel.channelGeneration, trustGeneration: base.channel.trustGeneration, - commandTransport: "disabled", + commandTransport: "typed-service-ping-v1", })); process.on("SIGTERM", shutdown); @@ -175,7 +179,7 @@ function createCombinedHealthServer(channel, gateway, healthConfig) { ...channel.status(), trackerIngress: "telemetry-ingest", tracker: gateway.status(), - commandTransport: "disabled", + commandTransport: "typed-service-ping-v1", })); }); }