From e69e2ddbd6385f98830ee1ad3a55d7e75964d09c Mon Sep 17 00:00:00 2001 From: mixelpixx Date: Mon, 27 Jul 2026 15:17:33 -0400 Subject: [PATCH] feat(serial): cursor-based reads and wait_for Serial reads were a lossy snapshot: read() returned the tail of a rolling buffer, so repeated reads overlapped, anything beyond the 512KB cap vanished silently, and capturing a long run meant polling and hoping the window never slid. There was also no way to wait for expected output - agents polled for that too. Reads are now cursor-based. Every read/wait_for response carries a global `cursor`; passing it back as `since` returns the FIRST max_lines complete lines at or after that offset plus a new cursor just past them - lossless paging. Buffer truncation and clear() advance a bufferStartOffset instead of destroying position, so a stale cursor reports `dropped: ` rather than silently mapping onto unrelated output. `has_more` says more complete lines are already buffered; a trailing partial line is held back until its newline arrives. Plain read() without since is unchanged (tail snapshot) apart from the new fields. New wait_for action: blocks until a line matching a substring (or regex with is_regex:true) arrives, with an optional `since` so already-buffered output is scanned first - output that arrived between calls cannot be missed. Resolves with {matched, line, cursor, elapsed_ms}; a timeout or a dropped connection resolves (not errors) with timed_out/disconnected flags. Timeout defaults to 30s, clamped 1-120 - deliberately lower than task waits since it holds an MCP request open. Waiters are flushed on disconnect(), ws close and ws error, so none can leak. The line scanner (processChunk/handleCompleteLine) assembles complete lines across chunk boundaries with a carried remainder; crash-signature detection will plug into the same funnel next. Verified on the ESP32-S3: paged reads returned exactly t:3..t:7 with dropped:0 and no overlap; wait_for matched "idle tempC" in 535ms; a never-matching pattern timed out cleanly with a cursor; a pre-clear cursor reported dropped:396 and clean lines; disconnecting mid-wait resolved disconnected:true. Co-Authored-By: Claude Opus 4.8 --- arduino-mcp-extension/README.md | 12 +- .../src/common/mcp-tool-router.ts | 3 +- arduino-mcp-extension/src/common/mcp-tools.ts | 23 +- .../src/node/mcp-serial-manager.ts | 269 +++++++++++++++++- arduino-mcp-extension/src/node/mcp-server.ts | 36 ++- 5 files changed, 328 insertions(+), 15 deletions(-) diff --git a/arduino-mcp-extension/README.md b/arduino-mcp-extension/README.md index 73f865b818b..a94b79bba37 100644 --- a/arduino-mcp-extension/README.md +++ b/arduino-mcp-extension/README.md @@ -363,12 +363,20 @@ error) and can keep waiting with `arduino_task_status {wait:true}`. Without | `list_ports` | - | List available serial ports | | `connect` | `port`, `baud_rate`, `fqbn` (optional) | Open connection | | `disconnect` | - | Close connection | -| `read` | `max_lines` | Read buffered board output | +| `read` | `max_lines`, `since` | Read buffered board output (cursor-based) | +| `wait_for` | `pattern`, `is_regex`, `timeout_seconds`, `since` | Block until a matching line arrives | | `write` | `data`, `line_ending` | Send data | -| `clear` | - | Clear read buffer | +| `clear` | - | Clear read buffer (cursors stay monotonic) | | `get_config` | - | Get current connection config | | `set_config` | `baud_rate` | Update connection settings | +Every `read`/`wait_for` response includes a `cursor`. Pass it back as `since` +to page through output **losslessly**: you get the next lines after that +offset, a `dropped` count if the 512 KB buffer overflowed in between, and +`has_more` when more complete lines are already buffered. Without `since`, +`read` returns the familiar tail snapshot. `wait_for` resolves with the +matching line, or `timed_out: true` / `disconnected: true` (never an error). + ### arduino_library | Action | Parameters | Description | diff --git a/arduino-mcp-extension/src/common/mcp-tool-router.ts b/arduino-mcp-extension/src/common/mcp-tool-router.ts index 7b71006ea98..202b63785fa 100644 --- a/arduino-mcp-extension/src/common/mcp-tool-router.ts +++ b/arduino-mcp-extension/src/common/mcp-tool-router.ts @@ -50,7 +50,8 @@ export const TOOL_CATEGORIES: ToolCategory[] = [ name: 'serial', description: 'Serial monitor operations - connect, read output, send data to devices.', toolNames: ['arduino_serial'], - useWhen: 'Debugging via serial output, sending commands to device, monitoring data', + useWhen: + 'Debugging via serial output, sending commands to device, waiting for specific output (wait_for), following logs losslessly with cursor-based reads', }, { name: 'library', diff --git a/arduino-mcp-extension/src/common/mcp-tools.ts b/arduino-mcp-extension/src/common/mcp-tools.ts index 7f8a7f7a809..4b906287051 100644 --- a/arduino-mcp-extension/src/common/mcp-tools.ts +++ b/arduino-mcp-extension/src/common/mcp-tools.ts @@ -277,7 +277,7 @@ export const ARDUINO_TOOLS: ToolDefinition[] = [ { name: 'arduino_serial', description: - 'Serial monitor operations - connect to a board, read its output, send data, change the baud rate. The connection is shared with the IDE serial monitor.', + 'Serial monitor operations - connect to a board, read its output, send data, change the baud rate. Reads are cursor-based: every read returns a `cursor`; pass it back as `since` to page through output losslessly. wait_for blocks until a line matching a pattern arrives. The connection is shared with the IDE serial monitor.', inputSchema: { type: 'object', properties: { @@ -287,6 +287,7 @@ export const ARDUINO_TOOLS: ToolDefinition[] = [ 'connect', 'disconnect', 'read', + 'wait_for', 'write', 'clear', 'get_config', @@ -322,6 +323,26 @@ export const ARDUINO_TOOLS: ToolDefinition[] = [ type: 'number', description: 'Maximum lines to return for read (default: 100)', }, + since: { + type: 'number', + description: + 'Cursor from a previous read/wait_for response (for read, wait_for). Returns only output at or after that offset, with `dropped` counting anything lost to buffer overflow - lossless paging instead of an overlapping tail.', + }, + pattern: { + type: 'string', + description: + 'Substring (or regex when is_regex=true) to wait for (for wait_for). Blocks until a matching line arrives.', + }, + is_regex: { + type: 'boolean', + description: + 'Treat pattern as a regular expression (for wait_for; default: false)', + }, + timeout_seconds: { + type: 'number', + description: + 'Max seconds wait_for blocks (default 30, clamp 1-120). A timeout returns matched:false with a cursor - not an error.', + }, }, required: ['action'], }, diff --git a/arduino-mcp-extension/src/node/mcp-serial-manager.ts b/arduino-mcp-extension/src/node/mcp-serial-manager.ts index d2ea5ad0015..66048a7c339 100644 --- a/arduino-mcp-extension/src/node/mcp-serial-manager.ts +++ b/arduino-mcp-extension/src/node/mcp-serial-manager.ts @@ -26,6 +26,24 @@ export const SUPPORTED_BAUD_RATES = [ const MAX_BUFFER_CHARS = 512 * 1024; // cap the capture buffer at 512 KB +/** Result shape for waitFor - see that method for the three resolutions. */ +export interface SerialWaitResult { + matched: boolean; + line?: string; + cursor: number; + elapsed_ms?: number; + timed_out?: boolean; + disconnected?: boolean; + hint?: string; +} + +interface SerialWaiter { + regex: RegExp; + startedAt: number; + timer: NodeJS.Timeout; + resolve: (result: SerialWaitResult) => void; +} + interface ActiveConnection { board: { name: string; fqbn: string }; port: Port; @@ -33,6 +51,22 @@ interface ActiveConnection { ws: WebSocket; buffer: string; connected: boolean; + /** + * Global char offset of buffer[0]. Cursors handed to callers are global + * offsets (bufferStartOffset + index into buffer), so they stay valid and + * monotonically increasing across buffer truncation and clear() - a stale + * cursor is reported as `dropped` chars instead of silently returning the + * wrong window. The total received so far is bufferStartOffset + buffer.length. + */ + bufferStartOffset: number; + /** Partial trailing line carried between chunks for the line scanner. */ + lineRemainder: string; + /** Pending wait_for calls, resolved by the line scanner. */ + waiters: SerialWaiter[]; +} + +function escapeRegExp(text: string): string { + return text.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); } export class MCPSerialManager { @@ -166,6 +200,9 @@ export class MCPSerialManager { ws, buffer: '', connected: false, + bufferStartOffset: 0, + lineRemainder: '', + waiters: [], }; ws.on('open', () => { @@ -178,19 +215,27 @@ export class MCPSerialManager { reject(err); } connection.connected = false; + this.flushWaiters(connection); }); ws.on('close', () => { connection.connected = false; + this.flushWaiters(connection); }); ws.on('message', (raw) => { try { const message = JSON.parse(raw.toString()); if (Array.isArray(message)) { // Board output: an array of string chunks. - connection.buffer += message.join(''); + const chunk = message.join(''); + connection.buffer += chunk; if (connection.buffer.length > MAX_BUFFER_CHARS) { + // Advance the global offset by exactly what we drop, so cursors + // handed out earlier stay meaningful (they report as `dropped`). + connection.bufferStartOffset += + connection.buffer.length - MAX_BUFFER_CHARS; connection.buffer = connection.buffer.slice(-MAX_BUFFER_CHARS); } + this.processChunk(connection, chunk); } else if ( message?.command === Monitor.MiddlewareCommand.ON_SETTINGS_DID_CHANGE ) { @@ -207,12 +252,78 @@ export class MCPSerialManager { }); } + /** + * Line scanner: assembles complete lines across chunk boundaries (chunks + * arrive mid-line) and feeds them to pending wait_for waiters. Each line + * carries the global cursor just past its newline. + */ + private processChunk(connection: ActiveConnection, chunk: string): void { + let data = connection.lineRemainder + chunk; + // Global offset of the start of `data`. Total received so far is + // bufferStartOffset + buffer.length (the chunk is already appended); back + // up over the chunk and the carried remainder to find where `data` begins. + let dataStart = + connection.bufferStartOffset + + connection.buffer.length - + chunk.length - + connection.lineRemainder.length; + let index: number; + while ((index = data.indexOf('\n')) !== -1) { + const line = data.slice(0, index).replace(/\r$/, ''); + const lineEndCursor = dataStart + index + 1; + this.handleCompleteLine(connection, line, lineEndCursor); + data = data.slice(index + 1); + dataStart += index + 1; + } + connection.lineRemainder = data; + } + + private handleCompleteLine( + connection: ActiveConnection, + line: string, + lineEndCursor: number + ): void { + if (!connection.waiters.length) { + return; + } + const matched = connection.waiters.filter((w) => w.regex.test(line)); + for (const waiter of matched) { + clearTimeout(waiter.timer); + waiter.resolve({ + matched: true, + line, + cursor: lineEndCursor, + elapsed_ms: Date.now() - waiter.startedAt, + }); + } + if (matched.length) { + connection.waiters = connection.waiters.filter( + (w) => !matched.includes(w) + ); + } + } + + /** Resolves every pending waiter as disconnected (close, error, disconnect). */ + private flushWaiters(connection: ActiveConnection): void { + const waiters = connection.waiters; + connection.waiters = []; + for (const waiter of waiters) { + clearTimeout(waiter.timer); + waiter.resolve({ + matched: false, + disconnected: true, + cursor: connection.bufferStartOffset + connection.buffer.length, + }); + } + } + async disconnect(): Promise { const connection = this.connection; if (!connection) { return; } this.connection = null; + this.flushWaiters(connection); try { connection.ws.close(); } catch { @@ -224,15 +335,143 @@ export class MCPSerialManager { .catch(() => undefined); } - read(maxLines: number): { lines: string[]; count: number } { + /** + * Reads captured output. + * + * Without `since`: the last `maxLines` lines (a tail snapshot), plus the + * current global `cursor` so the next call can page losslessly. + * + * With `since` (a cursor from a previous response): the FIRST `maxLines` + * complete lines at or after that offset, `cursor` set just past the last + * returned line, `dropped` counting chars lost to buffer truncation before + * `since`, and `has_more` when further complete lines are already buffered. + * A trailing partial line is held back until its newline arrives. + */ + read( + maxLines: number, + since?: number + ): { + lines: string[]; + count: number; + cursor: number; + dropped: number; + has_more: boolean; + } { const connection = this.requireConnection(); - const lines = connection.buffer.split(/\r?\n/); - // Drop a trailing empty segment caused by a terminating newline. - if (lines.length && lines[lines.length - 1] === '') { - lines.pop(); + const end = connection.bufferStartOffset + connection.buffer.length; + + if (since === undefined) { + const lines = connection.buffer.split(/\r?\n/); + // Drop a trailing empty segment caused by a terminating newline. + if (lines.length && lines[lines.length - 1] === '') { + lines.pop(); + } + const slice = lines.slice(-maxLines); + return { + lines: slice, + count: slice.length, + cursor: end, + dropped: 0, + has_more: false, + }; + } + + const clamped = Math.max( + connection.bufferStartOffset, + Math.min(since, end) + ); + const dropped = Math.max(0, connection.bufferStartOffset - since); + const region = connection.buffer.slice( + clamped - connection.bufferStartOffset + ); + + const lines: string[] = []; + let pos = 0; + while (lines.length < maxLines) { + const nl = region.indexOf('\n', pos); + if (nl === -1) { + break; + } + lines.push(region.slice(pos, nl).replace(/\r$/, '')); + pos = nl + 1; + } + return { + lines, + count: lines.length, + cursor: clamped + pos, + dropped, + has_more: region.indexOf('\n', pos) !== -1, + }; + } + + /** + * Blocks until a line matching `pattern` arrives, the timeout elapses, or + * the connection drops. When `since` is given, already-buffered complete + * lines from that cursor are scanned first, so output that arrived between + * calls cannot be missed. + */ + async waitFor( + pattern: string, + isRegex: boolean, + timeoutSeconds: number, + since?: number + ): Promise { + const connection = this.requireConnection(); + let regex: RegExp; + try { + regex = isRegex ? new RegExp(pattern) : new RegExp(escapeRegExp(pattern)); + } catch (e) { + throw new Error( + `Invalid regular expression "${pattern}": ${ + e instanceof Error ? e.message : e + }` + ); } - const slice = lines.slice(-maxLines); - return { lines: slice, count: slice.length }; + const startedAt = Date.now(); + const end = connection.bufferStartOffset + connection.buffer.length; + + // Scan what is already buffered (complete lines only) from `since`. + if (since !== undefined) { + const clamped = Math.max( + connection.bufferStartOffset, + Math.min(since, end) + ); + const region = connection.buffer.slice( + clamped - connection.bufferStartOffset + ); + let pos = 0; + let nl: number; + while ((nl = region.indexOf('\n', pos)) !== -1) { + const line = region.slice(pos, nl).replace(/\r$/, ''); + if (regex.test(line)) { + return { + matched: true, + line, + cursor: clamped + nl + 1, + elapsed_ms: Date.now() - startedAt, + }; + } + pos = nl + 1; + } + } + + return new Promise((resolve) => { + const waiter: SerialWaiter = { + regex, + startedAt, + resolve, + timer: setTimeout(() => { + connection.waiters = connection.waiters.filter((w) => w !== waiter); + resolve({ + matched: false, + timed_out: true, + cursor: connection.bufferStartOffset + connection.buffer.length, + hint: `No line matched within ${timeoutSeconds}s. Use read with since= to inspect what the board actually printed.`, + }); + }, timeoutSeconds * 1000), + }; + connection.waiters.push(waiter); + }); } write(data: string): { bytesSent: number } { @@ -267,8 +506,14 @@ export class MCPSerialManager { } clear(): void { - if (this.connection) { - this.connection.buffer = ''; + const c = this.connection; + if (c) { + // Cursors stay monotonic across clear(): advance the start offset so a + // stale cursor from before the clear reports `dropped` chars instead of + // silently mapping onto unrelated new output. + c.bufferStartOffset += c.buffer.length; + c.buffer = ''; + c.lineRemainder = ''; } } @@ -277,6 +522,8 @@ export class MCPSerialManager { port: string | null; baudRate: number | null; board: string | null; + cursor: number | null; + buffered_chars: number | null; } { const c = this.connection; return { @@ -284,6 +531,8 @@ export class MCPSerialManager { port: c?.port.address ?? null, baudRate: c?.baudRate ?? null, board: c?.board.name ?? null, + cursor: c ? c.bufferStartOffset + c.buffer.length : null, + buffered_chars: c ? c.buffer.length : null, }; } diff --git a/arduino-mcp-extension/src/node/mcp-server.ts b/arduino-mcp-extension/src/node/mcp-server.ts index 836b76d150c..43f1b46a3ea 100644 --- a/arduino-mcp-extension/src/node/mcp-server.ts +++ b/arduino-mcp-extension/src/node/mcp-server.ts @@ -91,6 +91,22 @@ function clampTaskTimeout(value: unknown): number { ); } +// Serial wait_for holds an MCP request open, so its ceiling is deliberately +// lower than task waits - long serial vigils should be re-issued. +const SERIAL_WAIT_DEFAULT_SECONDS = 30; +const SERIAL_WAIT_MIN_SECONDS = 1; +const SERIAL_WAIT_MAX_SECONDS = 120; + +function clampSerialWaitTimeout(value: unknown): number { + if (typeof value !== 'number' || !isFinite(value)) { + return SERIAL_WAIT_DEFAULT_SECONDS; + } + return Math.min( + SERIAL_WAIT_MAX_SECONDS, + Math.max(SERIAL_WAIT_MIN_SECONDS, value) + ); +} + interface StructuredBuildError { message: string; file?: string; @@ -1317,7 +1333,25 @@ export class ArduinoMCPServer { case 'read': { const maxLines = (args.max_lines as number) || 100; - return this.serial.read(maxLines); + const since = + typeof args.since === 'number' ? args.since : undefined; + return this.serial.read(maxLines, since); + } + + case 'wait_for': { + const pattern = args.pattern as string; + if (!pattern) { + throw new Error('pattern is required for wait_for'); + } + const timeout = clampSerialWaitTimeout(args.timeout_seconds); + const since = + typeof args.since === 'number' ? args.since : undefined; + return this.serial.waitFor( + pattern, + (args.is_regex as boolean) || false, + timeout, + since + ); } case 'write': {