From c33f4613b50c68f1a1c89a580921e1610d36f247 Mon Sep 17 00:00:00 2001 From: Kamil Krzywda Date: Fri, 25 Sep 2026 15:32:34 +0000 Subject: [PATCH] =?UTF-8?q?mesh:=20POST=20/api/respond=20+=20MCP=20respond?= =?UTF-8?q?=20tool=20=E2=80=94=20reply=20to=20a=20message=20by=20packet=20?= =?UTF-8?q?id?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mesh/fake-bridge.ts | 28 ++++ src/mesh/mcp.controller.spec.ts | 62 +++++++- src/mesh/mcp.controller.ts | 13 ++ src/mesh/mesh.controller.ts | 29 ++++ src/mesh/mesh.service.spec.ts | 252 ++++++++++++++++++++++++++++---- src/mesh/mesh.service.ts | 82 +++++++++-- 6 files changed, 422 insertions(+), 44 deletions(-) diff --git a/src/mesh/fake-bridge.ts b/src/mesh/fake-bridge.ts index e8c88fc..39c3c3d 100644 --- a/src/mesh/fake-bridge.ts +++ b/src/mesh/fake-bridge.ts @@ -148,6 +148,34 @@ export function telemetryPacketFrame( ); } +/** Inbound text message: TEXT_MESSAGE_APP packet with UTF-8 in Data.payload (fixed packet id 1). */ +export function textPacketFrame( + from: number, + to: number, + text: string, + channel = 0, +): Uint8Array { + const data = create(P.Mesh.DataSchema, { + portnum: P.Portnums.PortNum.TEXT_MESSAGE_APP, + payload: new TextEncoder().encode(text), + }); + const pkt = create(P.Mesh.MeshPacketSchema, { + from, + to, + id: 1, + channel, + payloadVariant: { case: 'decoded', value: data }, + }); + return frame( + toBinary( + P.Mesh.FromRadioSchema, + create(P.Mesh.FromRadioSchema, { + payloadVariant: { case: 'packet', value: pkt }, + }), + ), + ); +} + export const KAMIL_NUM = 123456789; export function kamilNodeInfo(): NodeInfoFields { diff --git a/src/mesh/mcp.controller.spec.ts b/src/mesh/mcp.controller.spec.ts index 6e011b5..a64b16d 100644 --- a/src/mesh/mcp.controller.spec.ts +++ b/src/mesh/mcp.controller.spec.ts @@ -11,6 +11,7 @@ import { nodeInfoFrame, startBridge, telemetryPacketFrame, + textPacketFrame, waitFor, } from './fake-bridge.js'; import { createMeshMcpServer } from './mcp.controller.js'; @@ -67,7 +68,7 @@ async function connectMcp(svc: MeshService): Promise<{ } describe('MCP server tools', () => { - it('lists all nine tools with descriptions and input schemas', async () => { + it('lists all ten tools with descriptions and input schemas', async () => { const env = await connectedService(); try { const mcp = await connectMcp(env.svc); @@ -82,6 +83,7 @@ describe('MCP server tools', () => { 'list_channels', 'list_nodes', 'mesh_status', + 'respond', 'send_message', ]); for (const t of tools) expect(t.description?.length).toBeGreaterThan(0); @@ -100,6 +102,13 @@ describe('MCP server tools', () => { 'text', 'from', ]); + + const respond = tools.find((t) => t.name === 'respond')!; + expect(respond.inputSchema.required).toEqual(['messageId', 'content']); + expect(Object.keys(respond.inputSchema.properties ?? {})).toEqual([ + 'messageId', + 'content', + ]); } finally { await mcp.close(); } @@ -196,6 +205,57 @@ describe('MCP server tools', () => { } }); + it('respond routes the reply automatically and surfaces failures as tool errors', async () => { + const env = await connectedService(); + try { + // rx DM from KAMIL on channel 3 (packet id 1) + env.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3)); + await waitFor(() => env.svc.messages().length === 1); + + const mcp = await connectMcp(env.svc); + try { + const r1 = await mcp.client.callTool({ + name: 'respond', + arguments: { messageId: 1, content: 'pong' }, + }); + expect(r1.isError).not.toBe(true); + const reply = JSON.parse(toolText(r1)) as { id: number }; + expect(reply.id).toBeGreaterThan(0); + // DM back to the original sender on the original channel + expect(env.svc.messages()[0]).toMatchObject({ + dir: 'tx', + to: KAMIL_NUM, + text: 'pong', + channel: 3, + }); + + const r2 = await mcp.client.callTool({ + name: 'respond', + arguments: { messageId: 999, content: 'x' }, + }); + expect(r2.isError).toBe(true); + expect(toolText(r2)).toContain('no message with id 999'); + + const r3 = await mcp.client.callTool({ + name: 'respond', + arguments: { messageId: 0, content: 'x' }, + }); + expect(r3.isError).toBe(true); // out-of-range rejected before the service sees it + + const r4 = await mcp.client.callTool({ + name: 'respond', + arguments: { messageId: 1, content: '' }, + }); + expect(r4.isError).toBe(true); + expect(toolText(r4)).toContain('content must be a non-empty string'); + } finally { + await mcp.close(); + } + } finally { + env.close(); + } + }); + it('inject_message records an rx entry without sending over RF', async () => { const env = await connectedService(); try { diff --git a/src/mesh/mcp.controller.ts b/src/mesh/mcp.controller.ts index f940fb3..4e612f8 100644 --- a/src/mesh/mcp.controller.ts +++ b/src/mesh/mcp.controller.ts @@ -46,6 +46,19 @@ export function createMeshMcpServer(mesh: MeshService): McpServer { }), ); + server.registerTool( + 'respond', + { + description: `Reply to an earlier message by its packet id (from get_messages); returns { id } — the Meshtastic packet id of the reply. Routing is automatic: rx records are answered as a DM back to the original sender, tx records go to their destination; broadcast originals and injected messages without a sender get a broadcast. The reply always rides the record's channel and carries replyId = messageId (threaded quote in the recipient app). Invalid messageId/content or an unknown id are tool errors.`, + inputSchema: { + messageId: z.number().int().min(1).max(0xffffffff), + content: z.string(), + }, + }, + async ({ messageId, content }) => + jsonResult(await mesh.respond(messageId, content)), + ); + server.registerTool( 'list_channels', { diff --git a/src/mesh/mesh.controller.ts b/src/mesh/mesh.controller.ts index 41b0b65..8b8808e 100644 --- a/src/mesh/mesh.controller.ts +++ b/src/mesh/mesh.controller.ts @@ -29,6 +29,17 @@ export class InjectMessageDto { from?: number; } +export class RespondDto { + @ApiProperty({ + description: + 'Packet id of the earlier message (from GET /api/messages) that this one replies to. Integer in 1..0xffffffff.', + example: 456789, + }) + messageId!: number; + @ApiProperty({ description: 'Reply text', example: 'on my way' }) + content!: string; +} + export class SendTextDto { @ApiProperty({ description: 'Message text', example: 'hello' }) text!: string; @@ -111,6 +122,24 @@ export class MeshController { return { id: await this.mesh.sendText(to, body.text, replyId) }; } + @Post('respond') + @ApiOperation({ + summary: 'Reply to an earlier message by packet id', + description: + "Routing is automatic — the caller never thinks about where it goes: rx records are answered as a DM back to the original sender, tx records go to their destination; broadcast originals (and injected messages without a sender) get a broadcast. The reply always rides the record's channel and carries replyId = messageId, so the recipient app renders it as a threaded quote of that packet. Returns { id } with the new packet id.", + }) + @ApiResponse({ + status: 200, + description: 'Packet queued', + example: { id: 123456789 }, + }) + @ApiResponse({ status: 400, description: 'Invalid messageId or content' }) + @ApiResponse({ status: 404, description: 'No message with the given id' }) + @ApiResponse({ status: 503, description: 'Bridge not connected' }) + async respond(@Body() body: RespondDto) { + return this.mesh.respond(body?.messageId, body?.content); + } + @Get('channels') @ApiOperation({ summary: "The device's Meshtastic channels (keys redacted)", diff --git a/src/mesh/mesh.service.spec.ts b/src/mesh/mesh.service.spec.ts index 4fa4703..c415a0a 100644 --- a/src/mesh/mesh.service.spec.ts +++ b/src/mesh/mesh.service.spec.ts @@ -1,5 +1,6 @@ import { create, fromBinary, toBinary } from '@bufbuild/protobuf'; import * as P from '@meshtastic/protobufs'; +import { BadRequestException, NotFoundException } from '@nestjs/common'; import { DatabaseSync } from 'node:sqlite'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { BROADCAST_ADDR, extractFrames, frame } from './codec.js'; @@ -12,6 +13,7 @@ import { nodeInfoFrame, startBridge, telemetryPacketFrame, + textPacketFrame, waitFor, } from './fake-bridge.js'; import { OpencodeService } from '../opencode/opencode.service.js'; @@ -74,34 +76,6 @@ function nodeInfoPacketFrame( ); } -// inbound text message: TEXT_MESSAGE_APP packet with UTF-8 in Data.payload -function textPacketFrame( - from: number, - to: number, - text: string, - channel = 0, -): Uint8Array { - const data = create(P.Mesh.DataSchema, { - portnum: P.Portnums.PortNum.TEXT_MESSAGE_APP, - payload: new TextEncoder().encode(text), - }); - const pkt = create(P.Mesh.MeshPacketSchema, { - from, - to, - id: 1, - channel, - payloadVariant: { case: 'decoded', value: data }, - }); - return frame( - toBinary( - P.Mesh.FromRadioSchema, - create(P.Mesh.FromRadioSchema, { - payloadVariant: { case: 'packet', value: pkt }, - }), - ), - ); -} - // ACK/NACK for our sends: ROUTING_APP packet with Data.requestId = original id and a serialized Routing in the payload function routingFrame( requestId: number, @@ -693,6 +667,228 @@ describe('MeshService message history', () => { }); }); +// outbound text packets written to the bridge, in order (want_config frames skipped) +function outboundPackets(out: Uint8Array[]): P.Mesh.MeshPacket[] { + const merged = new Uint8Array(out.reduce((n, c) => n + c.length, 0)); + let off = 0; + for (const c of out) { + merged.set(c, off); + off += c.length; + } + return extractFrames(merged).packets.flatMap((p) => { + const tr = fromBinary(P.Mesh.ToRadioSchema, p); + if (tr.payloadVariant?.case === 'packet') return [tr.payloadVariant.value]; + return []; + }); +} + +describe('MeshService respond', () => { + it('replies to an rx DM as a directed send back to the sender on the original channel with replyId = messageId', async () => { + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + // rx DM from KAMIL to our node on channel 2 (packet id 1) + bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2)); + await waitFor(() => svc.messages().length === 1); + + const out: Uint8Array[] = []; + bridge.sockets[0].on('data', (c) => out.push(c)); + const { id } = await svc.respond(1, 'pong'); + + expect(id).toBeGreaterThan(0); + // DM back to the original sender on the original channel + expect(svc.messages()[0]).toMatchObject({ + id, + dir: 'tx', + to: KAMIL_NUM, + text: 'pong', + channel: 2, + status: 'pending', + }); + + await waitFor(() => outboundPackets(out).length === 1); + const pkt = outboundPackets(out)[0]!; + expect(pkt.to).toBe(KAMIL_NUM); + expect(pkt.channel).toBe(2); + const data = + pkt.payloadVariant?.case === 'decoded' + ? pkt.payloadVariant.value + : undefined; + expect(data?.replyId).toBe(1); // threaded quote of the original packet + } finally { + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('replies to a broadcast original as a broadcast on the original channel with replyId = messageId', async () => { + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + // rx broadcast from KAMIL on channel 2 (packet id 1) + bridge.sockets[0].write( + textPacketFrame(KAMIL_NUM, BROADCAST_ADDR, 'hello all', 2), + ); + await waitFor(() => svc.messages().length === 1); + + const out: Uint8Array[] = []; + bridge.sockets[0].on('data', (c) => out.push(c)); + const { id } = await svc.respond(1, 'pong'); + + expect(id).toBeGreaterThan(0); + expect(svc.messages()[0]).toMatchObject({ + dir: 'tx', + to: BROADCAST_ADDR, // broadcast original → broadcast reply + text: 'pong', + channel: 2, + }); + + await waitFor(() => outboundPackets(out).length === 1); + const pkt = outboundPackets(out)[0]!; + expect(pkt.to).toBe(BROADCAST_ADDR); + expect(pkt.channel).toBe(2); + const data = + pkt.payloadVariant?.case === 'decoded' + ? pkt.payloadVariant.value + : undefined; + expect(data?.replyId).toBe(1); + } finally { + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('replies to a tx record targeting its destination on its channel with replyId = messageId', async () => { + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + const svc = new MeshService(); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + const origId = await svc.sendText(42, 'orig'); // tx record: to=42, default channel 3 + const out: Uint8Array[] = []; + bridge.sockets[0].on('data', (c) => out.push(c)); + const { id } = await svc.respond(origId, 'reply'); + + expect(id).toBeGreaterThan(0); + expect(svc.messages()[0]).toMatchObject({ + dir: 'tx', + to: 42, // the tx record's own destination + text: 'reply', + channel: 3, + }); + + // the orig send may share a chunk with the reply — take the newest packet + await waitFor(() => outboundPackets(out).length >= 1); + const pkts = outboundPackets(out); + const pkt = pkts[pkts.length - 1]!; + expect(pkt.to).toBe(42); + const data = + pkt.payloadVariant?.case === 'decoded' + ? pkt.payloadVariant.value + : undefined; + expect(data?.replyId).toBe(origId); + } finally { + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('throws NotFoundException when no message has the given id (memory or SQLite)', async () => { + const svc = new MeshService(); + await expect(svc.respond(999, 'hi')).rejects.toBeInstanceOf( + NotFoundException, + ); + await expect(svc.respond(999, 'hi')).rejects.toThrow( + 'no message with id 999', + ); + + const db = new DatabaseSync(':memory:'); // empty table — the DB lookup path also misses + initMessagesTable(db); + const svc2 = new MeshService(db); + await expect(svc2.respond(999, 'hi')).rejects.toBeInstanceOf( + NotFoundException, + ); + }); + + it('falls back to SQLite when the record is not in the in-memory buffer', async () => { + const db = new DatabaseSync(':memory:'); + initMessagesTable(db); + // 150 rows — onModuleInit seeds only the newest 100 into the buffer (ids 2050..2149) + const ins = db.prepare( + 'INSERT INTO messages(id, dir, to_node, from_node, text, channel, at, status, ack_error) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)', + ); + for (let i = 0; i < 150; i++) + ins.run( + 2000 + i, + 'rx', + 42, + KAMIL_NUM, + `old-${i}`, + 2, + 1_700_000_000_000 + i * 1000, + null, + null, + ); + + const bridge = await startBridge(); + process.env.MESH_HOST = '127.0.0.1'; + process.env.MESH_PORT = String(bridge.port); + const svc = new MeshService(db); + try { + svc.onModuleInit(); + await waitFor( + () => bridge.sockets.length === 1 && svc.status().connected, + ); + + // id 2000 is in SQLite but outside the seeded buffer + const { id } = await svc.respond(2000, 'pong'); + expect(id).toBeGreaterThan(0); + expect(svc.messages()[0]).toMatchObject({ + dir: 'tx', + to: KAMIL_NUM, // DM back to the sender of the old rx record + text: 'pong', + channel: 2, // the record's channel + }); + } finally { + svc.onModuleDestroy(); + bridge.close(); + } + }); + + it('rejects invalid messageId and empty content with BadRequestException', async () => { + const svc = new MeshService(); + for (const bad of [0, BROADCAST_ADDR + 1, 42.5, 'x' as unknown as number]) + await expect(svc.respond(bad, 'hi')).rejects.toBeInstanceOf( + BadRequestException, + ); + await expect(svc.respond(1, '')).rejects.toThrow( + 'content must be a non-empty string', + ); + await expect(svc.respond(1, undefined)).rejects.toBeInstanceOf( + BadRequestException, + ); + }); +}); + describe('MeshService opencode notifier', () => { // our own node number for the self-echo filter specs (distinct from KAMIL_NUM, an external user) const MY_NODE = 987654321; diff --git a/src/mesh/mesh.service.ts b/src/mesh/mesh.service.ts index eae173d..a5bd8fe 100644 --- a/src/mesh/mesh.service.ts +++ b/src/mesh/mesh.service.ts @@ -1,6 +1,8 @@ import { + BadRequestException, Injectable, Logger, + NotFoundException, OnModuleDestroy, OnModuleInit, ServiceUnavailableException, @@ -84,6 +86,22 @@ export function initMessagesTable(db: DatabaseSync): void { db.exec(MESSAGES_DDL); } +/** Map a messages-table row to a MessageRecord (NULLs → omitted optionals). */ +function recordFromRow(row: Record): MessageRecord { + const rec: MessageRecord = { + id: Number(row.id), + dir: row.dir === 'tx' ? 'tx' : 'rx', + text: String(row.text), + channel: Number(row.channel), + at: Number(row.at), + }; + if (row.to_node != null) rec.to = Number(row.to_node); + if (row.from_node != null) rec.from = Number(row.from_node); + if (row.status != null) rec.status = row.status as MessageRecord['status']; + if (row.ack_error != null) rec.ackError = String(row.ack_error); + return rec; +} + /** What we know about a node so far: full NodeInfo after sync, possibly just {num}+position for live-only nodes. */ type NodeEntry = Pick & Partial>; @@ -393,21 +411,55 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { 'SELECT id, dir, to_node, from_node, text, channel, at, status, ack_error FROM messages ORDER BY at DESC LIMIT ?', ).all(HISTORY_LIMIT) as Array>; // rows come back newest first — the buffer is oldest-first - for (const row of [...rows].reverse()) { - const rec: MessageRecord = { - id: Number(row.id), - dir: row.dir === 'tx' ? 'tx' : 'rx', - text: String(row.text), - channel: Number(row.channel), - at: Number(row.at), - }; - if (row.to_node != null) rec.to = Number(row.to_node); - if (row.from_node != null) rec.from = Number(row.from_node); - if (row.status != null) - rec.status = row.status as MessageRecord['status']; - if (row.ack_error != null) rec.ackError = String(row.ack_error); - this.history.push(rec); - } + for (const row of [...rows].reverse()) + this.history.push(recordFromRow(row)); + } + + /** Look up a history record by packet id: in-memory buffer first, then SQLite (newest 1000). */ + private findRecord(id: number): MessageRecord | undefined { + const hit = this.history.find((r) => r.id === id); + if (hit) return hit; + const db = this.db; + if (!db) return undefined; + const row = db + .prepare( + 'SELECT id, dir, to_node, from_node, text, channel, at, status, ack_error FROM messages WHERE id = ?', + ) + .get(id) as Record | undefined; + return row ? recordFromRow(row) : undefined; + } + + /** + * Reply to an earlier message by packet id — the caller never thinks about where it goes: + * rx records are answered as a DM back to the original sender, tx records go to their destination; + * broadcast originals (and injected messages without a sender) get a broadcast. The reply always + * rides the record's channel and carries replyId = messageId, so the recipient app renders it as a + * threaded quote of that packet. Returns { id } with the new packet id. Throws BadRequest on invalid + * arguments, NotFound when no message has the given id (in memory or in SQLite), 503 when the bridge is down. + */ + async respond(messageId: number, content: string): Promise<{ id: number }> { + if ( + !Number.isInteger(messageId) || + messageId < 1 || + messageId > BROADCAST_ADDR + ) + throw new BadRequestException( + 'messageId must be an integer in 1..0xffffffff', + ); + if (typeof content !== 'string' || content.length === 0) + throw new BadRequestException('content must be a non-empty string'); + const rec = this.findRecord(messageId); + if (!rec) throw new NotFoundException(`no message with id ${messageId}`); + // rx → DM back to the sender unless the original was a broadcast (then reply as a broadcast — that's what the stored destination is for); + // tx → its own destination; injected records without a sender → broadcast + const target = + rec.dir === 'rx' + ? rec.to === BROADCAST_ADDR + ? BROADCAST_ADDR + : rec.from + : rec.to; + const to = target ?? BROADCAST_ADDR; + return { id: await this.sendText(to, content, messageId, rec.channel) }; } /**