From 78a2510a866da9beff5eb3aa1310bad8994a6bb5 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 15 Sep 2026 02:51:53 +0000 Subject: [PATCH 1/4] P1 bridge follow-ups: send receipts, bounded WS transport, bounded thread reads Closes #34 (delivery states: rejected/accepted/unknown, IDs preserved), #35 (idempotent shutdown, handshake/fragmentation/budgets, no leaks), #36 (truncation metadata, bounded full reads, safe normalize, 1-200 limits). Co-authored-by: Zack Jackson --- .changeset/potato-p1-bridge.md | 4 + plugin/src/gbot.ts | 29 ++- plugin/src/mcp/grok-bot/tools/gbot_send.tsx | 17 +- plugin/src/mcp/grok-bot/tools/gbot_thread.tsx | 16 +- plugin/tests/route-unit/tools.test.ts | 46 +++- src/cli.js | 24 ++- src/codex-bridge.js | 202 +++++++++++++++--- src/gateway.js | 33 ++- src/transcript.js | 24 ++- test/codex-bridge.test.js | 157 +++++++++++++- test/transcript.test.js | 8 + 11 files changed, 497 insertions(+), 63 deletions(-) create mode 100644 .changeset/potato-p1-bridge.md diff --git a/.changeset/potato-p1-bridge.md b/.changeset/potato-p1-bridge.md new file mode 100644 index 0000000..18816bc --- /dev/null +++ b/.changeset/potato-p1-bridge.md @@ -0,0 +1,4 @@ +"grok-bot-cli": patch +--- + +P1 bridge follow-ups: preserve send receipts with rejected/accepted/unknown delivery states (Codex `CodexSendError` keeps thread/turn IDs, gateway `sendPrompt` returns `delivery` + `messageId` and marks post-write loss unknown); bound the Codex WebSocket transport (idempotent close settling pending requests, socket destroy on every failure, exact handshake validation, fragmentation/UTF-8/opcode handling, header/frame/message/buffer budgets); make `gbot_thread` bounded without losing replies (truncation metadata, `full` bounded full-read in tool and `--full` in CLI, safe string normalization, 1–200 limit consistency, gateway deadline and response-byte cap). diff --git a/plugin/src/gbot.ts b/plugin/src/gbot.ts index 9a66195..a0f6659 100644 --- a/plugin/src/gbot.ts +++ b/plugin/src/gbot.ts @@ -12,7 +12,15 @@ export const withRedactedErrors = async (run: () => Promise): Promise = try { return await run(); } catch (error) { - throw new Error(redactSecrets(error instanceof Error ? error.message : String(error))); + const wrapped: Error & { delivery?: string; targetId?: string } = new Error( + redactSecrets(error instanceof Error ? error.message : String(error)), + ); + if (error instanceof Error) { + const src = error as Error & { delivery?: unknown; targetId?: unknown }; + if (typeof src.delivery === 'string') wrapped.delivery = src.delivery; + if (typeof src.targetId === 'string') wrapped.targetId = src.targetId; + } + throw wrapped; } }; @@ -33,12 +41,16 @@ export const entrySchema = z.object({ kind: z.string(), role: z.string().optional(), text: z.string(), + truncated: z.boolean(), + fullLength: z.number().int().min(0), timestampMs: z.number().optional(), }); type Entry = z.infer; /** Match `gbot thread` CLI preview width so MCP hosts are not flooded. */ export const ENTRY_TEXT_MAX = 400; +// ponytail: fixed preview/full budgets; upgrade path is a paged thread resource instead of wider caps. +export const ENTRY_FULL_MAX = 20000; export const truncateEntryText = (text: string, max = ENTRY_TEXT_MAX): string => { if (text.length <= max) return text; @@ -52,19 +64,26 @@ const entryFields = z.object({ timestampMs: z.number().optional(), }); -const threadEntry = (raw: unknown): Entry => { +const threadEntry = (raw: unknown, opts: { full?: boolean } = {}): Entry => { const fields = entryFields.safeParse(raw); + const full = entryText(raw); + const max = opts.full ? ENTRY_FULL_MAX : ENTRY_TEXT_MAX; + const truncated = full.length > max; + const text = !truncated ? full : opts.full ? full.slice(0, max) : truncateEntryText(full); if (!fields.success) { - return { id: '', kind: 'unknown', text: truncateEntryText(JSON.stringify(raw)) }; + return { id: '', kind: 'unknown', text, truncated, fullLength: full.length }; } const { id, kind, role, timestampMs } = fields.data; return { id, kind, ...(role === undefined ? {} : { role }), - text: truncateEntryText(entryText(raw)), + text, + truncated, + fullLength: full.length, ...(timestampMs === undefined ? {} : { timestampMs }), }; }; -export const transcriptEntries = (transcript: unknown): Entry[] => unwrapEntries(transcript).map(threadEntry); +export const transcriptEntries = (transcript: unknown, opts: { full?: boolean } = {}): Entry[] => + unwrapEntries(transcript).map((raw) => threadEntry(raw, opts)); diff --git a/plugin/src/mcp/grok-bot/tools/gbot_send.tsx b/plugin/src/mcp/grok-bot/tools/gbot_send.tsx index b4ea313..e421a0d 100644 --- a/plugin/src/mcp/grok-bot/tools/gbot_send.tsx +++ b/plugin/src/mcp/grok-bot/tools/gbot_send.tsx @@ -18,16 +18,27 @@ export default defineTool( type: 'object', }, inputSchema: z.object({ message: z.string().min(1), target: z.string().min(1) }), - resultSchema: z.object({ result: z.record(z.string(), z.json()), target: targetSchema }), + resultSchema: z.object({ + result: z.record(z.string(), z.json()), + target: targetSchema, + delivery: z.enum(['accepted']), + messageId: z.string().optional(), + }), title: 'Send a message to Grok Bot', }, async ({ message, target }) => { const sent = await withRedactedErrors(async () => sendPrompt(await connectGateway(), target, message)); - const value = { result: sent.result, target: summarizeTarget(sent.target) }; + const messageId = typeof sent.messageId === 'string' ? (sent.messageId as string) : undefined; + const value = { + result: sent.result, + target: summarizeTarget(sent.target), + delivery: 'accepted' as const, + ...(messageId === undefined ? {} : { messageId }), + }; return ( - {`Sent to ${value.target.kind} ${value.target.name} (${value.target.id}). Read the reply with gbot_thread.`} + {`Sent to ${value.target.kind} ${value.target.name} (${value.target.id})${messageId === undefined ? ' (no receipt; check the thread before resending)' : ` as ${messageId}`}. Read the reply with gbot_thread.`} ); diff --git a/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx b/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx index 3df7f43..6422547 100644 --- a/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx +++ b/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx @@ -25,18 +25,28 @@ export default defineTool( description: 'How many trailing entries to return (1-200). Each entry text is capped at 400 characters.', type: 'number', }, + full: { + default: false, + description: + 'Return complete entry text up to a bounded full-read budget instead of the 400-character preview. Every entry still reports truncation metadata.', + type: 'boolean', + }, target: { description: 'Bot or group name or id, for example "General".', type: 'string' }, }, required: ['target'], type: 'object', }, - inputSchema: z.object({ limit: z.number().int().min(1).max(200).default(40), target: z.string().min(1) }), + inputSchema: z.object({ + limit: z.number().int().min(1).max(200).default(40), + full: z.boolean().default(false), + target: z.string().min(1), + }), resultSchema: z.object({ entries: z.array(entrySchema), target: targetSchema }), title: 'Read a Grok Bot thread', }, - async ({ limit, target }) => { + async ({ limit, target, full }) => { const tail = await withRedactedErrors(async () => getTranscriptTail(await connectGateway(), target, limit)); - const value = { entries: transcriptEntries(tail.transcript), target: summarizeTarget(tail.target) }; + const value = { entries: transcriptEntries(tail.transcript, { full }), target: summarizeTarget(tail.target) }; return ( {`${value.target.kind} ${value.target.name}: ${value.entries.length} entries.`} diff --git a/plugin/tests/route-unit/tools.test.ts b/plugin/tests/route-unit/tools.test.ts index 8d4ccaa..a633222 100644 --- a/plugin/tests/route-unit/tools.test.ts +++ b/plugin/tests/route-unit/tools.test.ts @@ -22,6 +22,7 @@ const roster = { { id: 'grp-1', memberAgentIds: ['bot-1'], name: 'Launch' }, { id: 'bot-2', isGroup: false, name: 'Legacy' }, { id: 'bot-3', isGroup: false, name: 'Proxy' }, + { id: 'bot-4', isGroup: false, name: 'Odd' }, ], }; const transcripts: Record = { @@ -42,6 +43,13 @@ const transcripts: Record = { ], nextBeforeSeq: 9, }, + 'bot-4': { + entries: [ + { content: { text: 5 }, id: 'o1', kind: 'note' }, + { id: 'o2', kind: 'mystery' }, + { id: 'o3', kind: 'note', text: `a${String.fromCharCode(0xd800)}b` }, + ], + }, }; const responses: Record) => [number, unknown]> = { getAgentTranscriptTail: (body) => [200, transcripts[String(body.id)]], @@ -111,6 +119,8 @@ describe('grok-bot MCP server', () => { }); expect(result.isError).toBe(false); expect(result.structuredContent).toEqual({ + delivery: 'accepted', + messageId: 'm-1', result: { messageId: 'm-1' }, target: { id: 'bot-1', kind: 'bot', name: 'General' }, }); @@ -130,9 +140,9 @@ describe('grok-bot MCP server', () => { expect(result.isError).toBe(false); expect(result.structuredContent).toEqual({ entries: [ - { id: 't1', kind: 'message', role: 'user', text: 'hello from the test', timestampMs: 1 }, - { id: 't2', kind: 'send-message', text: 'reply' }, - { id: 't3', kind: 'tool-call', text: '' }, + { id: 't1', kind: 'message', role: 'user', text: 'hello from the test', truncated: false, fullLength: 19, timestampMs: 1 }, + { id: 't2', kind: 'send-message', text: 'reply', truncated: false, fullLength: 5 }, + { id: 't3', kind: 'tool-call', text: '', truncated: false, fullLength: 0 }, ], target: { id: 'grp-1', kind: 'group', name: 'Launch' }, }); @@ -149,16 +159,38 @@ describe('grok-bot MCP server', () => { const legacy = await invokeMcpTool('gbot_thread', { input: { target: 'Legacy' }, server: 'grok-bot' }); expect(legacy.structuredContent).toMatchObject({ entries: [ - { id: 'l1', kind: 'message', text: 'direct text' }, - { id: 'l2', kind: 'note', text: 'part one\npart two\npart three' }, - { id: 'l3', kind: 'message', text: 'plain message' }, - { id: 'l4', kind: 'message', text: `${'x'.repeat(399)}…` }, + { id: 'l1', kind: 'message', text: 'direct text', truncated: false, fullLength: 11 }, + { id: 'l2', kind: 'note', text: 'part one\npart two\npart three', truncated: false, fullLength: 28 }, + { id: 'l3', kind: 'message', text: 'plain message', truncated: false, fullLength: 13 }, + { id: 'l4', kind: 'message', text: `${'x'.repeat(399)}…`, truncated: true, fullLength: 450 }, ], }); expect(contentText(legacy.content)).toContain('…'); expect(contentText(legacy.content)).not.toContain('x'.repeat(450)); }); + it('gbot_thread recovers a complete long reply with full:true and normalizes malformed entries', async () => { + const full = await invokeMcpTool('gbot_thread', { + input: { full: true, target: 'Legacy' }, + server: 'grok-bot', + }); + expect(full.isError).toBe(false); + const fullContent = full.structuredContent as { entries: { id: string; text: string; truncated: boolean; fullLength: number }[] }; + expect(fullContent.entries[3]).toEqual({ id: 'l4', kind: 'message', text: 'x'.repeat(450), truncated: false, fullLength: 450 }); + expect(contentText(full.content)).toContain('x'.repeat(450)); + + const odd = await invokeMcpTool('gbot_thread', { input: { target: 'Odd' }, server: 'grok-bot' }); + expect(odd.isError).toBe(false); + expect(odd.structuredContent).toEqual({ + entries: [ + { id: 'o1', kind: 'note', text: '5', truncated: false, fullLength: 1 }, + { id: 'o2', kind: 'mystery', text: '', truncated: false, fullLength: 0 }, + { id: 'o3', kind: 'note', text: 'a�b', truncated: false, fullLength: 3 }, + ], + target: { id: 'bot-4', kind: 'bot', name: 'Odd' }, + }); + }); + it('redacts a bearer token echoed by the gateway before the error reaches the host', async () => { const result = await invokeMcpTool('gbot_send', { input: { message: 'x', target: 'Proxy' }, diff --git a/src/cli.js b/src/cli.js index 4ca4a49..9cd7f18 100755 --- a/src/cli.js +++ b/src/cli.js @@ -48,7 +48,7 @@ function usage() { " groups set --member ID [--member ...]", " groups delete ", " send ", - " thread [--limit N] [--root MESSAGE_ID]", + " thread [--limit N] [--root MESSAGE_ID] [--full]", " chat alias for thread", " codex status", " codex list-threads [--limit N]", @@ -232,7 +232,7 @@ function formatRecord(rec, all) { return kind + " " + rec.name + title + "\n " + rec.id + desc + avatar + settingsLine + extra; } -function formatTranscript(out) { +function formatTranscript(out, { full = false } = {}) { const rec = out.target; const payload = out.transcript || out.thread || {}; const entries = transcriptEntries(payload); @@ -245,11 +245,17 @@ function formatTranscript(out) { const role = e.role || e.kind || e.sender || e.type || "msg"; const text = entryText(e); const id = e.id || e.messageId || ""; - lines.push("[" + role + (id ? " " + id : "") + "] " + String(text).slice(0, 400)); + lines.push("[" + role + (id ? " " + id : "") + "] " + (full ? text : truncateCliText(text))); } return lines.join("\n"); } +/** Bounded preview; --json/--full still retrieve the complete text. */ +function truncateCliText(text, max = 400) { + if (text.length <= max) return text; + return text.slice(0, max) + "… [+" + (text.length - max) + " chars; --full or --json for complete text]"; +} + function formatCodexStatus(s) { const lines = ["socket: " + s.socketPath]; if (!s.reachable) return lines.concat("reachable: no", s.message).join("\n"); @@ -459,20 +465,24 @@ async function main(argv) { const message = rest.join(" ").trim(); if (!ref || !message) throw new StoreError("gbot send "); const out = await backend.send(ref, message); - if (json) print({ id: out.target.id, name: out.target.name, kind: out.target.isGroup ? "group" : "bot", result: out.result }); - else print("Sent to " + (out.target.isGroup ? "group" : "bot") + " " + out.target.name + " (" + out.target.id + ")"); + const receipt = out.messageId ? " message " + out.messageId : ""; + if (json) print({ id: out.target.id, name: out.target.name, kind: out.target.isGroup ? "group" : "bot", result: out.result, delivery: out.delivery || "accepted", ...(out.messageId ? { messageId: out.messageId } : {}) }); + else print("Sent to " + (out.target.isGroup ? "group" : "bot") + " " + out.target.name + " (" + out.target.id + ")" + receipt); return; } if (cmd === "thread" || cmd === "chat") { const ref = sub; - if (!ref) throw new StoreError("gbot thread [--limit N] [--root MESSAGE_ID]"); + if (!ref) throw new StoreError("gbot thread [--limit N] [--root MESSAGE_ID] [--full]"); + const full = rest.includes("--full"); + if (full) rest.splice(rest.indexOf("--full"), 1); const limitRaw = takeFlag(rest, "--limit"); const rootId = takeFlag(rest, "--root"); const limit = limitRaw ? Number(limitRaw) : 40; + if (!Number.isInteger(limit) || limit < 1 || limit > 200) throw new StoreError("--limit must be an integer 1-200"); const out = rootId ? await backend.thread(ref, rootId) : await backend.transcript(ref, limit); if (json) print(out); - else print(formatTranscript(out)); + else print(formatTranscript(out, { full })); return; } diff --git a/src/codex-bridge.js b/src/codex-bridge.js index 85f3740..a46dd5f 100644 --- a/src/codex-bridge.js +++ b/src/codex-bridge.js @@ -16,6 +16,12 @@ export const UPSTREAM_DESKTOP_ISSUES = [ ]; const WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; +// Transport budgets: fail fast instead of buffering unbounded attacker-controlled bytes. +// ponytail: raise these only with streaming/pagination support; the app-server sends small JSON-RPC frames. +export const WS_MAX_HEADER_BYTES = 16 * 1024; +export const WS_MAX_MESSAGE_BYTES = 4 * 1024 * 1024; +export const WS_MAX_BUFFER_BYTES = 8 * 1024 * 1024; +const textDecoder = new TextDecoder("utf-8", { fatal: true }); const pkg = createRequire(import.meta.url)("../package.json"); export function codexSocketPath(env = process.env) { @@ -89,7 +95,35 @@ export function decodeFrame(buf) { if (buf.length < offset + len) return null; const payload = Buffer.from(buf.subarray(offset, offset + len)); if (mask) for (let i = 0; i < payload.length; i++) payload[i] ^= mask[i & 3]; - return { fin, opcode, payload, rest: buf.subarray(offset + len) }; + return { fin, opcode, masked, payload, rest: buf.subarray(offset + len) }; +} + +/** Claimed frame length without consuming; null when the length prefix is incomplete. */ +function peekFrameLength(buf) { + if (buf.length < 2) return null; + const marker = buf[1] & 0x7f; + if (marker < 126) return marker; + if (marker === 126) { + if (buf.length < 4) return null; + return buf.readUInt16BE(2); + } + if (buf.length < 10) return null; + const big = buf.readBigUInt64BE(2); + return big > BigInt(Number.MAX_SAFE_INTEGER) ? Infinity : Number(big); +} + +function validateUpgradeHead(head, key) { + const lines = head.split("\r\n"); + if (!/^HTTP\/1\.1 101/.test(lines[0])) return false; + const headers = new Map(); + for (const line of lines.slice(1)) { + const i = line.indexOf(":"); + if (i === -1) return false; + headers.set(line.slice(0, i).trim().toLowerCase(), line.slice(i + 1).trim()); + } + return headers.get("sec-websocket-accept") === websocketAccept(key) + && (headers.get("upgrade") || "").toLowerCase() === "websocket" + && (headers.get("connection") || "").toLowerCase().includes("upgrade"); } function upgradeRequest(key) { @@ -110,6 +144,17 @@ export class CodexRpcError extends Error { } } +export class CodexSendError extends Error { + constructor(message, { delivery, threadId, turnId, refused } = {}) { + super(message); + this.name = "CodexSendError"; + this.delivery = delivery; + if (threadId !== undefined) this.threadId = threadId; + if (turnId !== undefined) this.turnId = turnId; + if (refused !== undefined) this.refused = refused; + } +} + /** * Open a JSON-RPC session to the app-server over its Unix socket (WebSocket framing). * Server-initiated requests (approvals, user input) are refused with a JSON-RPC error @@ -125,14 +170,20 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { let buf = Buffer.alloc(0); let upgraded = false; let closed = false; + let fragOpcode = null; + let fragParts = []; + let fragBytes = 0; const failAll = (err) => { if (closed) return; closed = true; + if (err && err.delivery == null) err.delivery = pending.size ? "unknown" : "rejected"; for (const { reject: rej } of pending.values()) rej(err); pending.clear(); + try { socket.destroy(); } catch { /* already gone */ } reject(err); }; + const failProtocol = (detail) => failAll(new Error("Codex app-server violated the WebSocket protocol: " + detail)); const write = (opcode, payload) => { if (!socket.destroyed) socket.write(encodeFrame(opcode, payload, randomBytes(4))); }; @@ -159,10 +210,17 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { sendJson({ jsonrpc: "2.0", method, params }); }, close() { + if (closed) return; closed = true; - write(0x8, Buffer.from([0x03, 0xe8])); - socket.end(); - socket.unref(); + const err = new Error("Codex client closed"); + err.delivery = pending.size ? "unknown" : "rejected"; + for (const { reject: rej } of pending.values()) rej(err); + pending.clear(); + if (!socket.destroyed) { + if (upgraded) write(0x8, Buffer.from([0x03, 0xe8])); + socket.end(); + socket.unref(); + } }, }; @@ -183,6 +241,24 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { else entry.resolve(msg.result); }; + const onText = (payload) => { + let text; + try { + text = textDecoder.decode(payload); + } catch { + failAll(new Error("Codex app-server sent a non-UTF-8 text message")); + return; + } + let msg; + try { + msg = JSON.parse(text); + } catch (err) { + failAll(new Error("Codex app-server sent an unreadable message: " + err.message)); + return; + } + onMessage(msg); + }; + socket.setTimeout(timeoutMs, () => failAll(new Error("Timed out connecting to Codex app-server at " + path))); socket.once("error", (err) => failAll(new Error("Could not connect to Codex app-server at " + path + ": " + err.message))); socket.once("close", () => failAll(new Error("Codex app-server closed the connection"))); @@ -191,33 +267,68 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { buf = Buffer.concat([buf, chunk]); if (!upgraded) { const end = buf.indexOf("\r\n\r\n"); - if (end === -1) return; + if (end === -1) { + if (buf.length > WS_MAX_HEADER_BYTES) failAll(new Error("Codex app-server handshake headers exceed " + WS_MAX_HEADER_BYTES + " bytes")); + return; + } const head = buf.subarray(0, end).toString(); buf = buf.subarray(end + 4); - const ok = /^HTTP\/1\.1 101/.test(head) - && head.toLowerCase().includes("sec-websocket-accept: " + websocketAccept(key).toLowerCase()); - if (!ok) return failAll(new Error("Codex app-server refused the WebSocket upgrade: " + head.split("\r\n")[0])); + if (!validateUpgradeHead(head, key)) return failAll(new Error("Codex app-server refused the WebSocket upgrade: " + head.split("\r\n")[0])); upgraded = true; socket.setTimeout(0); resolve(client); } - // ponytail: unfragmented frames only; the app-server sends each JSON-RPC message as one text frame. for (;;) { const frame = decodeFrame(buf); - if (!frame) return; + if (!frame) { + const claimed = peekFrameLength(buf); + if (claimed != null && claimed > WS_MAX_MESSAGE_BYTES) { + failAll(new Error("Codex app-server frame exceeds " + WS_MAX_MESSAGE_BYTES + " bytes")); + } else if (buf.length > WS_MAX_BUFFER_BYTES) { + failAll(new Error("Codex app-server buffer exceeds " + WS_MAX_BUFFER_BYTES + " bytes")); + } + return; + } buf = frame.rest; - if (frame.opcode === 0x1) { - try { - onMessage(JSON.parse(frame.payload.toString())); - } catch (err) { - failAll(new Error("Codex app-server sent an unreadable message: " + err.message)); + if (frame.masked) return failProtocol("server frames must not be masked"); + if (frame.opcode >= 0x8) { + if (!frame.fin || frame.payload.length > 125) return failProtocol("bad control frame"); + if (frame.opcode === 0x9) write(0xa, frame.payload); + else if (frame.opcode === 0x8) { + write(0x8, frame.payload); + socket.end(); + failAll(new Error("Codex app-server closed the connection")); } - } else if (frame.opcode === 0x9) write(0xa, frame.payload); - else if (frame.opcode === 0x8) { - write(0x8, frame.payload); - socket.end(); - failAll(new Error("Codex app-server closed the connection")); + continue; // pong and other control frames carry nothing for us + } + if (frame.opcode === 0x0) { + if (fragOpcode == null) return failProtocol("continuation with nothing to continue"); + fragParts.push(frame.payload); + fragBytes += frame.payload.length; + if (fragBytes > WS_MAX_MESSAGE_BYTES) return failAll(new Error("Codex app-server message exceeds " + WS_MAX_MESSAGE_BYTES + " bytes")); + if (!frame.fin) continue; + const whole = Buffer.concat(fragParts, fragBytes); + const opcode = fragOpcode; + fragOpcode = null; + fragParts = []; + fragBytes = 0; + // ponytail: binary frames are unused by the app-server; only text is delivered. + if (opcode === 0x1) onText(whole); + continue; + } + if (frame.opcode === 0x1 || frame.opcode === 0x2) { + if (fragOpcode != null) return failProtocol("new message before finishing fragments"); + if (!frame.fin) { + fragOpcode = frame.opcode; + fragParts = [frame.payload]; + fragBytes = frame.payload.length; + continue; + } + // ponytail: binary frames are unused by the app-server; only text is delivered. + if (frame.opcode === 0x1) onText(frame.payload); + continue; } + return failProtocol("unknown opcode " + frame.opcode); } }); }); @@ -234,7 +345,13 @@ async function openSession(env = process.env) { const path = codexSocketPath(env); if (!socketPresent(path)) throw new Error(unreachableMessage(path)); const client = await connectCodexAppServer(path); - const init = await client.request("initialize", { clientInfo: { name: "gbot", version: pkg.version } }); + let init; + try { + init = await client.request("initialize", { clientInfo: { name: "gbot", version: pkg.version } }); + } catch (err) { + client.close(); + throw err; + } client.notify("initialized"); return { client, path, init }; } @@ -310,19 +427,48 @@ export async function sendToCodexThread(threadId, text, env = process.env) { try { resumed = await client.request("thread/resume", { threadId, excludeTurns: true }); } catch (err) { - throw explainSendError(err, threadId); + if (err instanceof CodexSendError) throw err; + throw new CodexSendError(explainSendError(err, threadId).message, { + delivery: err instanceof CodexRpcError ? "rejected" : (err && err.delivery) || "unknown", + threadId, + }); + } + // Scope refusals to this turn: server requests from earlier calls belong to another context. + const seenRefused = client.refused.length; + let turn; + try { + turn = await client.request("turn/start", { threadId, input: [{ type: "text", text }] }); + } catch (err) { + if (err instanceof CodexSendError) throw err; + const delivery = err instanceof CodexRpcError ? "rejected" : (err && err.delivery) || "unknown"; + const detail = err instanceof CodexRpcError + ? err.message + : "Lost the Codex turn/start response for thread " + threadId + ": " + ((err && err.message) || err) + + ". Delivery is unknown; check the thread before resending."; + // ponytail: no blind retry here; a stable receipt/correlation envelope is issue #37. + throw new CodexSendError(detail, { delivery, threadId }); + } + const turnId = turn && turn.turn && typeof turn.turn.id === "string" && turn.turn.id ? turn.turn.id : null; + if (!turnId) { + throw new CodexSendError( + "Codex app-server sent a malformed turn/start acknowledgment for thread " + threadId + ". Delivery is unknown; check the thread before resending.", + { delivery: "unknown", threadId }, + ); } - const turn = await client.request("turn/start", { threadId, input: [{ type: "text", text }] }); - if (client.refused.length) { - const methods = client.refused.map((r) => r.method).join(", "); - throw new Error( - "Turn " + turn.turn.id + " started on thread " + threadId + " but Codex asked for " + methods + ", which gbot refused. " + const freshRefused = client.refused.slice(seenRefused) + .filter((r) => !r.params || r.params.threadId == null || r.params.threadId === threadId); + if (freshRefused.length) { + const methods = freshRefused.map((r) => r.method).join(", "); + throw new CodexSendError( + "Turn " + turnId + " started on thread " + threadId + " but Codex asked for " + methods + ", which gbot refused. " + "Answer it in a Codex client, or set `approval_policy = \"never\"` in the daemon's config.toml for unattended sends.", + { delivery: "accepted", threadId, turnId, refused: freshRefused.map((r) => r.method) }, ); } return { + delivery: "accepted", threadId: resumed.thread.id, - turnId: turn.turn.id, + turnId, turnStatus: turn.turn.status, model: resumed.model, cwd: resumed.cwd, diff --git a/src/gateway.js b/src/gateway.js index daf9d7a..5e89869 100644 --- a/src/gateway.js +++ b/src/gateway.js @@ -13,6 +13,10 @@ export class GatewayError extends Error { } } +// ponytail: fixed 30 s deadline and buffered byte cap; upgrade path is per-method budgets plus streaming reads. +export const GATEWAY_TIMEOUT_MS = 30000; +export const GATEWAY_MAX_RESPONSE_BYTES = 2 * 1024 * 1024; + function backendBase() { return ( process.env.SAND_BACKEND_URL || @@ -80,6 +84,9 @@ export function hasGatewayAuth() { async function readJson(res) { const text = await res.text(); + if (text.length > GATEWAY_MAX_RESPONSE_BYTES) { + throw new GatewayError("Gateway response too large (" + text.length + " bytes, limit " + GATEWAY_MAX_RESPONSE_BYTES + ")"); + } if (!text) return {}; try { return JSON.parse(text); @@ -101,6 +108,7 @@ export async function ensureSandbox(accessToken) { const res = await fetch(url, { method: "POST", redirect: "error", + signal: AbortSignal.timeout(GATEWAY_TIMEOUT_MS), headers: ensureSandboxHeaders(accessToken), body: "{}", }); @@ -135,6 +143,7 @@ export async function gatewayCall(session, method, body = {}) { const res = await fetch(url, { method: "POST", redirect: "error", + signal: AbortSignal.timeout(GATEWAY_TIMEOUT_MS), headers: requestHeaders(session), body: JSON.stringify(body), }); @@ -319,13 +328,29 @@ export async function sendPrompt(session, ref, prompt, extra = {}) { clientNonce: extra.clientNonce || randomUUID(), }; if (extra.replyToId) body.replyToId = extra.replyToId; - const data = await gatewayCall(session, "sendPrompt", body); - return { target: rec, result: data }; + let data; + try { + data = await gatewayCall(session, "sendPrompt", body); + } catch (err) { + // Delivery states: the server answered no (rejected) vs the request may have landed (unknown). + // Never retry an unknown delivery blindly; read the thread first. + if (err && err.delivery == null) { + err.delivery = err instanceof GatewayError && err.status != null && err.status < 500 ? "rejected" : "unknown"; + } + if (err && err.targetId == null) err.targetId = rec.id; + if (err && err.delivery === "unknown" && err instanceof Error && !/delivery unknown/.test(err.message)) { + err.message += " (delivery unknown; check the thread before resending)"; + } + throw err; + } + const messageId = data && typeof data.messageId === "string" ? data.messageId : null; + return { target: rec, result: data, delivery: "accepted", ...(messageId ? { messageId } : {}) }; } -export async function getTranscriptTail(session, ref, limit = 50) { +export async function getTranscriptTail(session, ref, limit = 40) { const rec = await resolveRef(session, ref); - const data = await gatewayCall(session, "getAgentTranscriptTail", { id: rec.id, limit }); + const bounded = Math.min(Math.max(Math.trunc(limit) || 40, 1), 200); + const data = await gatewayCall(session, "getAgentTranscriptTail", { id: rec.id, limit: bounded }); return { target: rec, transcript: data }; } diff --git a/src/transcript.js b/src/transcript.js index 56dadea..060b4c3 100644 --- a/src/transcript.js +++ b/src/transcript.js @@ -1,6 +1,28 @@ // Shared by `gbot thread` and the grok-bot plugin's gbot_thread tool. +/** Coerce anything to a string without throwing (numbers, BigInt, unserializable objects). */ +export function toSafeText(value) { + if (typeof value === "string") return value; + if (value == null) return ""; + if (typeof value === "number" || typeof value === "boolean" || typeof value === "bigint") return String(value); + try { + const out = JSON.stringify(value); + return typeof out === "string" ? out : ""; + } catch { + return "[unserializable]"; + } +} + +/** Coerce to string and replace lone surrogates so downstream slicing/JSON never breaks. */ +export function normalizeText(value) { + return toSafeText(value).replace(/[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); socket.write( "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + "Sec-WebSocket-Accept: " + websocketAccept(req.headers["sec-websocket-key"]) + "\r\n\r\n", @@ -53,7 +57,14 @@ async function fakeAppServer(handlers) { socket.on("error", () => {}); }); await new Promise((resolve) => server.listen(socketPath, resolve)); - return { home, received, close: () => new Promise((resolve) => server.close(resolve)) }; + return { + home, + received, + close: () => new Promise((resolve) => { + for (const sock of sockets) sock.destroy(); + server.close(resolve); + }), + }; } const baseHandlers = { @@ -168,6 +179,7 @@ test("codex send resumes the thread, starts a turn, and prints the ids", async ( const { code, out } = await gbot(fake.home, "--json", "codex", "send", "t-1", "hello", "from", "gbot"); assert.equal(code, 0, out); assert.deepEqual(JSON.parse(out), { + delivery: "accepted", threadId: "t-1", turnId: "turn-9", turnStatus: "inProgress", @@ -249,15 +261,15 @@ test("codex send fails fast when the server sends a Close frame mid-request", as const started = Date.now(); const { code, err } = await gbot(fake.home, "codex", "send", "t-1", "go"); assert.equal(code, 1); - assert.equal(err, "Codex app-server closed the connection\n"); + assert.match(err, /^Lost the Codex turn\/start response for thread t-1: Codex app-server closed the connection\./); + assert.match(err, /Delivery is unknown; check the thread before resending\./); assert.ok(Date.now() - started < 5000, "did not wait for the request timeout"); } finally { await fake.close(); } }); -test("codex send keeps --json / --dir tokens that appear after the thread id", async () => { - const fake = await fakeAppServer(baseHandlers); +test("codex send keeps --json / --dir tokens that appear after the thread id", async () => { const fake = await fakeAppServer(baseHandlers); try { const { code, out } = await gbot( fake.home, @@ -314,3 +326,138 @@ test("codex list-threads strips terminal controls from names and previews", asyn await fake.close(); } }); + +test("codex send reassembles a fragmented turn/start reply split across TCP chunks", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + "turn/start": (params, ok, err, send, socket) => { + void ok; + void err; + const id = fake.received.at(-1).id; + const payload = Buffer.from(JSON.stringify({ jsonrpc: "2.0", id, result: { turn: { id: "turn-9", status: "inProgress", items: [] } } })); + const half = Math.floor(payload.length / 2); + const first = Buffer.concat([Buffer.from([0x01, half]), payload.subarray(0, half)]); + const second = Buffer.concat([Buffer.from([0x80, payload.length - half]), payload.subarray(half)]); + socket.write(first.subarray(0, 3)); + setTimeout(() => socket.write(Buffer.concat([first.subarray(3), second])), 20); + }, + }); + try { + const { code, out } = await gbot(fake.home, "--json", "codex", "send", "t-1", "go"); + assert.equal(code, 0, out); + assert.deepEqual(JSON.parse(out).turnId, "turn-9"); + } finally { + await fake.close(); + } +}); + +test("codex status fails on a wrong handshake and closes the socket instead of leaking it", async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-badhs-")); + mkdirSync(join(home, "app-server-control")); + const socketPath = join(home, "app-server-control", "app-server-control.sock"); + let serverSocket = null; + const server = createTcpServer((sock) => { + serverSocket = sock; + sock.on("data", () => sock.write("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")); + sock.on("error", () => {}); + }); + await new Promise((resolve) => server.listen(socketPath, resolve)); + const closed = new Promise((resolve) => server.on("connection", (sock) => sock.on("close", resolve))); + try { + const started = Date.now(); + const { code, err } = await gbot(home, "codex", "status"); + assert.equal(code, 1); + assert.match(err, /refused the WebSocket upgrade/); + assert.ok(Date.now() - started < 5000, "failed fast instead of hanging"); + await Promise.race([closed, new Promise((_, rej) => setTimeout(() => rej(new Error("client socket leaked")), 3000))]); + assert.ok(serverSocket.destroyed || serverSocket.closed, "server side sees the client go away"); + } finally { + await new Promise((resolve) => server.close(resolve)); + } +}); + +test("codex status fails fast on an oversized frame and settles the pending request", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + initialize: (params, ok, err, send, socket) => { + void params; void ok; void err; void send; + socket.write(Buffer.from([0x81, 0x7f, 0, 0, 0, 0, 0x10, 0, 0, 0])); + }, + }); + try { + const started = Date.now(); + const { code, err } = await gbot(fake.home, "codex", "status"); + assert.equal(code, 1); + assert.match(err, /exceeds/); + assert.ok(Date.now() - started < 5000, "did not wait for the request timeout"); + } finally { + await fake.close(); + } +}); + +test("client close is idempotent and settles pending requests", async () => { + const fake = await fakeAppServer({ ...baseHandlers, "turn/start": () => {} }); + const socketPath = join(fake.home, "app-server-control", "app-server-control.sock"); + try { + const client = await connectCodexAppServer(socketPath); + const pending = client.request("turn/start", { threadId: "t-1", input: [] }); + const settled = assert.rejects(pending, /closed/); + client.close(); + client.close(); + await settled; + } finally { + await fake.close(); + } +}); + +test("codex send keeps the thread id and reports unknown delivery on a malformed ack", async () => { + const fake = await fakeAppServer({ ...baseHandlers, "turn/start": (params, ok) => ok({ turn: { status: "inProgress" } }) }); + try { + const { code, err } = await gbot(fake.home, "codex", "send", "t-1", "go"); + assert.equal(code, 1); + assert.match(err, /malformed turn\/start acknowledgment for thread t-1/); + assert.match(err, /unknown/); + } finally { + await fake.close(); + } +}); + +test("codex send ignores server requests scoped to other threads", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + "thread/resume": (params, ok, err, send) => { + send({ jsonrpc: "2.0", id: "srv-other", method: "item/commandExecution/requestApproval", params: { threadId: "other" } }); + baseHandlers["thread/resume"](params, ok, err); + }, + }); + try { + const { code, out } = await gbot(fake.home, "--json", "codex", "send", "t-1", "go"); + assert.equal(code, 0, out); + assert.equal(JSON.parse(out).delivery, "accepted"); + } finally { + await fake.close(); + } +}); + +test("codex send preserves turn and thread ids with accepted delivery on refusal", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + "turn/start": (params, ok, err, send) => { + send({ jsonrpc: "2.0", id: "srv-1", method: "item/commandExecution/requestApproval", params: { threadId: params.threadId } }); + setTimeout(() => ok({ turn: { id: "turn-10", status: "inProgress", items: [] } }), 20); + }, + }); + const home = fake.home; + const env = { ...process.env, CODEX_HOME: home }; + try { + await assert.rejects(sendToCodexThread("t-1", "do it", env), (e) => { + assert.ok(e instanceof CodexSendError); + assert.equal(e.delivery, "accepted"); + assert.equal(e.threadId, "t-1"); + assert.equal(e.turnId, "turn-10"); + return true; + }); + } finally { + await fake.close(); + } +}); diff --git a/test/transcript.test.js b/test/transcript.test.js index 60f5c31..57fef6a 100644 --- a/test/transcript.test.js +++ b/test/transcript.test.js @@ -24,3 +24,11 @@ test("transcript containers unwrap to an entry list", () => { assert.deepEqual(transcriptEntries({ nextBeforeSeq: 2 }), []); assert.deepEqual(transcriptEntries(null), []); }); + +test("entryText never throws and normalizes malformed content to safe strings", () => { + assert.equal(entryText({ content: { text: 5 } }), "5"); + assert.equal(entryText({ content: { flag: true } }), '{"flag":true}'); + assert.equal(entryText({ content: { v: 1n } }), "[unserializable]"); + assert.equal(entryText({ text: "a\ud800b" }), "a\ufffdb"); + assert.equal(entryText(null), ""); +}); From 46821ee46d63fb28c13e45a9f31f685e7a54570f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 15 Sep 2026 03:16:01 +0000 Subject: [PATCH 2/4] Fix HOLD items: streaming byte cap, absolute deadline, null-RPC, JSON errors, ack validation, bounded full reads Co-authored-by: Zack Jackson --- plugin/src/gbot.ts | 19 ++- plugin/src/mcp/grok-bot/tools/gbot_send.tsx | 8 +- plugin/src/mcp/grok-bot/tools/gbot_thread.tsx | 4 +- plugin/tests/route-unit/tools.test.ts | 36 +++++- src/cli.js | 13 +- src/codex-bridge.js | 30 +++-- src/gateway.js | 35 +++++- test/codex-bridge.test.js | 116 +++++++++++++++++- test/gateway-send.test.js | 67 ++++++++++ 9 files changed, 302 insertions(+), 26 deletions(-) create mode 100644 test/gateway-send.test.js diff --git a/plugin/src/gbot.ts b/plugin/src/gbot.ts index a0f6659..3b27aa3 100644 --- a/plugin/src/gbot.ts +++ b/plugin/src/gbot.ts @@ -50,7 +50,10 @@ type Entry = z.infer; /** Match `gbot thread` CLI preview width so MCP hosts are not flooded. */ export const ENTRY_TEXT_MAX = 400; // ponytail: fixed preview/full budgets; upgrade path is a paged thread resource instead of wider caps. +// When an entry is cut by these budgets it still reports truncated/fullLength, and the +// remainder is retrievable with `gbot thread --full` / `--json` on the machine. export const ENTRY_FULL_MAX = 20000; +export const TRANSCRIPT_TOTAL_MAX = 200000; export const truncateEntryText = (text: string, max = ENTRY_TEXT_MAX): string => { if (text.length <= max) return text; @@ -64,12 +67,11 @@ const entryFields = z.object({ timestampMs: z.number().optional(), }); -const threadEntry = (raw: unknown, opts: { full?: boolean } = {}): Entry => { +const threadEntry = (raw: unknown, max: number, ellipsis: boolean): Entry => { const fields = entryFields.safeParse(raw); const full = entryText(raw); - const max = opts.full ? ENTRY_FULL_MAX : ENTRY_TEXT_MAX; const truncated = full.length > max; - const text = !truncated ? full : opts.full ? full.slice(0, max) : truncateEntryText(full); + const text = !truncated ? full : ellipsis ? truncateEntryText(full, max) : full.slice(0, max); if (!fields.success) { return { id: '', kind: 'unknown', text, truncated, fullLength: full.length }; } @@ -85,5 +87,12 @@ const threadEntry = (raw: unknown, opts: { full?: boolean } = {}): Entry => { }; }; -export const transcriptEntries = (transcript: unknown, opts: { full?: boolean } = {}): Entry[] => - unwrapEntries(transcript).map((raw) => threadEntry(raw, opts)); +export const transcriptEntries = (transcript: unknown, opts: { full?: boolean } = {}): Entry[] => { + const perEntry = opts.full ? ENTRY_FULL_MAX : ENTRY_TEXT_MAX; + let remaining = TRANSCRIPT_TOTAL_MAX; + return unwrapEntries(transcript).map((raw) => { + const entry = threadEntry(raw, Math.min(perEntry, remaining), !opts.full); + remaining = Math.max(0, remaining - entry.text.length); + return entry; + }); +}; diff --git a/plugin/src/mcp/grok-bot/tools/gbot_send.tsx b/plugin/src/mcp/grok-bot/tools/gbot_send.tsx index e421a0d..b035350 100644 --- a/plugin/src/mcp/grok-bot/tools/gbot_send.tsx +++ b/plugin/src/mcp/grok-bot/tools/gbot_send.tsx @@ -21,18 +21,20 @@ export default defineTool( resultSchema: z.object({ result: z.record(z.string(), z.json()), target: targetSchema, - delivery: z.enum(['accepted']), + delivery: z.enum(['accepted', 'unknown']), messageId: z.string().optional(), }), title: 'Send a message to Grok Bot', }, async ({ message, target }) => { const sent = await withRedactedErrors(async () => sendPrompt(await connectGateway(), target, message)); - const messageId = typeof sent.messageId === 'string' ? (sent.messageId as string) : undefined; + // Never claim acceptance the gateway didn't confirm; no-receipt sends stay unknown. + const delivery = sent.delivery === 'accepted' ? ('accepted' as const) : ('unknown' as const); + const messageId = delivery === 'accepted' && typeof sent.messageId === 'string' ? (sent.messageId as string) : undefined; const value = { result: sent.result, target: summarizeTarget(sent.target), - delivery: 'accepted' as const, + delivery, ...(messageId === undefined ? {} : { messageId }), }; return ( diff --git a/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx b/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx index 6422547..c10221c 100644 --- a/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx +++ b/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx @@ -28,7 +28,7 @@ export default defineTool( full: { default: false, description: - 'Return complete entry text up to a bounded full-read budget instead of the 400-character preview. Every entry still reports truncation metadata.', + 'Return complete entry text up to bounded budgets (20k chars per entry, 200k total) instead of the 400-character preview. Every entry still reports truncated/fullLength; read the remainder with `gbot thread --full` / `--json` on the machine.', type: 'boolean', }, target: { description: 'Bot or group name or id, for example "General".', type: 'string' }, @@ -37,6 +37,8 @@ export default defineTool( type: 'object', }, inputSchema: z.object({ + // ponytail: the route inputJsonSchema type cannot express minimum/maximum, so the + // 1-200 bound lives here in zod (and in the CLI/gateway); widen the route type to align them. limit: z.number().int().min(1).max(200).default(40), full: z.boolean().default(false), target: z.string().min(1), diff --git a/plugin/tests/route-unit/tools.test.ts b/plugin/tests/route-unit/tools.test.ts index a633222..15fa1d3 100644 --- a/plugin/tests/route-unit/tools.test.ts +++ b/plugin/tests/route-unit/tools.test.ts @@ -23,6 +23,8 @@ const roster = { { id: 'bot-2', isGroup: false, name: 'Legacy' }, { id: 'bot-3', isGroup: false, name: 'Proxy' }, { id: 'bot-4', isGroup: false, name: 'Odd' }, + { id: 'bot-5', isGroup: false, name: 'Noreceipt' }, + { id: 'bot-6', isGroup: false, name: 'Big' }, ], }; const transcripts: Record = { @@ -50,6 +52,9 @@ const transcripts: Record = { { id: 'o3', kind: 'note', text: `a${String.fromCharCode(0xd800)}b` }, ], }, + 'bot-6': { + entries: Array.from({ length: 11 }, (_, index) => ({ id: `b${index}`, kind: 'note', text: 'q'.repeat(20000) })), + }, }; const responses: Record) => [number, unknown]> = { getAgentTranscriptTail: (body) => [200, transcripts[String(body.id)]], @@ -57,7 +62,9 @@ const responses: Record) => [number, unkn sendPrompt: (body) => body.agentId === 'bot-3' ? [401, { message: 'upstream rejected authorization: Bearer test-token' }] - : [200, { messageId: 'm-1' }], + : body.agentId === 'bot-5' + ? [200, { ok: true }] + : [200, { messageId: 'm-1' }], }; const calls: GatewayCall[] = []; @@ -191,6 +198,33 @@ describe('grok-bot MCP server', () => { }); }); + it('gbot_send stays unknown when the gateway confirms no receipt', async () => { + const result = await invokeMcpTool('gbot_send', { + input: { message: 'ping', target: 'Noreceipt' }, + server: 'grok-bot', + }); + expect(result.isError).toBe(false); + expect(result.structuredContent).toEqual({ + delivery: 'unknown', + result: { ok: true }, + target: { id: 'bot-5', kind: 'bot', name: 'Noreceipt' }, + }); + expect(contentText(result.content)).toContain('no receipt'); + expect(contentText(result.content)).not.toContain(' as '); + }); + + it('gbot_thread caps aggregate output and keeps the remainder visible in metadata', async () => { + const full = await invokeMcpTool('gbot_thread', { + input: { full: true, target: 'Big' }, + server: 'grok-bot', + }); + expect(full.isError).toBe(false); + const entries = (full.structuredContent as { entries: { id: string; text: string; truncated: boolean; fullLength: number }[] }).entries; + expect(entries).toHaveLength(11); + expect(entries[9]).toEqual({ id: 'b9', kind: 'note', text: 'q'.repeat(20000), truncated: false, fullLength: 20000 }); + expect(entries[10]).toEqual({ id: 'b10', kind: 'note', text: '', truncated: true, fullLength: 20000 }); + }); + it('redacts a bearer token echoed by the gateway before the error reaches the host', async () => { const result = await invokeMcpTool('gbot_send', { input: { message: 'x', target: 'Proxy' }, diff --git a/src/cli.js b/src/cli.js index 9cd7f18..9fd4e47 100755 --- a/src/cli.js +++ b/src/cli.js @@ -15,7 +15,16 @@ function print(value) { function fail(err) { let message = err instanceof Error ? err.message : String(err); message = redactSecrets(message); - process.stderr.write(message + "\n"); + if (jsonErrors && err instanceof Error) { + const out = { error: message }; + if (err.delivery !== undefined) out.delivery = err.delivery; + if (err.threadId !== undefined) out.threadId = err.threadId; + if (err.turnId !== undefined) out.turnId = err.turnId; + if (err.targetId !== undefined) out.targetId = err.targetId; + process.stderr.write(JSON.stringify(out) + "\n"); + } else { + process.stderr.write(message + "\n"); + } process.exit(1); } @@ -106,6 +115,7 @@ function takeRepeating(args, name) { } /** Peel global CLI options only from the leading argv (before the command). */ +let jsonErrors = false; function takeLeadingGlobals(args) { let json = false; let gateway = false; @@ -119,6 +129,7 @@ function takeLeadingGlobals(args) { } if (a === "--json") { json = true; + jsonErrors = true; args.shift(); continue; } diff --git a/src/codex-bridge.js b/src/codex-bridge.js index a46dd5f..084f246 100644 --- a/src/codex-bridge.js +++ b/src/codex-bridge.js @@ -173,10 +173,15 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { let fragOpcode = null; let fragParts = []; let fragBytes = 0; + // Absolute handshake deadline: socket timeouts reset on any bytes, so trickled + // headers must not extend this. Per-request timers stay absolute after upgrade. + const handshakeTimer = setTimeout(() => failAll(new Error("Timed out connecting to Codex app-server at " + path)), timeoutMs); + if (typeof handshakeTimer.unref === "function") handshakeTimer.unref(); const failAll = (err) => { if (closed) return; closed = true; + clearTimeout(handshakeTimer); if (err && err.delivery == null) err.delivery = pending.size ? "unknown" : "rejected"; for (const { reject: rej } of pending.values()) rej(err); pending.clear(); @@ -212,15 +217,20 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { close() { if (closed) return; closed = true; + clearTimeout(handshakeTimer); const err = new Error("Codex client closed"); err.delivery = pending.size ? "unknown" : "rejected"; for (const { reject: rej } of pending.values()) rej(err); pending.clear(); - if (!socket.destroyed) { - if (upgraded) write(0x8, Buffer.from([0x03, 0xe8])); - socket.end(); - socket.unref(); - } + // Best-effort close frame, then guaranteed destruction so no path leaks the socket. + if (!socket.destroyed && upgraded) write(0x8, Buffer.from([0x03, 0xe8])); + if (socket.destroyed) return; + const forceDestroy = setTimeout(() => { try { socket.destroy(); } catch { /* already gone */ } }, 1000); + if (typeof forceDestroy.unref === "function") forceDestroy.unref(); + socket.end(() => { + clearTimeout(forceDestroy); + try { socket.destroy(); } catch { /* already gone */ } + }); }, }; @@ -256,10 +266,13 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { failAll(new Error("Codex app-server sent an unreadable message: " + err.message)); return; } + if (!msg || typeof msg !== "object" || Array.isArray(msg)) { + failAll(new Error("Codex app-server sent a malformed message")); + return; + } onMessage(msg); }; - socket.setTimeout(timeoutMs, () => failAll(new Error("Timed out connecting to Codex app-server at " + path))); socket.once("error", (err) => failAll(new Error("Could not connect to Codex app-server at " + path + ": " + err.message))); socket.once("close", () => failAll(new Error("Codex app-server closed the connection"))); socket.once("connect", () => socket.write(upgradeRequest(key))); @@ -271,11 +284,13 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { if (buf.length > WS_MAX_HEADER_BYTES) failAll(new Error("Codex app-server handshake headers exceed " + WS_MAX_HEADER_BYTES + " bytes")); return; } + // Terminated headers hit the cap too; size is checked before any decoding. + if (end > WS_MAX_HEADER_BYTES) return failAll(new Error("Codex app-server handshake headers exceed " + WS_MAX_HEADER_BYTES + " bytes")); const head = buf.subarray(0, end).toString(); buf = buf.subarray(end + 4); if (!validateUpgradeHead(head, key)) return failAll(new Error("Codex app-server refused the WebSocket upgrade: " + head.split("\r\n")[0])); upgraded = true; - socket.setTimeout(0); + clearTimeout(handshakeTimer); resolve(client); } for (;;) { @@ -290,6 +305,7 @@ export function connectCodexAppServer(path, { timeoutMs = 15000 } = {}) { return; } buf = frame.rest; + if (frame.payload.length > WS_MAX_MESSAGE_BYTES) return failAll(new Error("Codex app-server frame exceeds " + WS_MAX_MESSAGE_BYTES + " bytes")); if (frame.masked) return failProtocol("server frames must not be masked"); if (frame.opcode >= 0x8) { if (!frame.fin || frame.payload.length > 125) return failProtocol("bad control frame"); diff --git a/src/gateway.js b/src/gateway.js index 5e89869..d0da4a3 100644 --- a/src/gateway.js +++ b/src/gateway.js @@ -82,11 +82,32 @@ export function hasGatewayAuth() { return Boolean(gatewayOverride() || accessTokenFromEnv() || hasGrokBotGatewaySession()); } -async function readJson(res) { - const text = await res.text(); - if (text.length > GATEWAY_MAX_RESPONSE_BYTES) { - throw new GatewayError("Gateway response too large (" + text.length + " bytes, limit " + GATEWAY_MAX_RESPONSE_BYTES + ")"); +async function readTextCapped(res, maxBytes) { + if (!res.body || typeof res.body.getReader !== "function") { + const text = await res.text(); + if (Buffer.byteLength(text, "utf8") > maxBytes) { + throw new GatewayError("Gateway response too large (over " + maxBytes + " bytes)"); + } + return text; + } + const reader = res.body.getReader(); + const chunks = []; + let bytes = 0; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + bytes += value.byteLength ?? value.length; + if (bytes > maxBytes) { + try { await reader.cancel(); } catch { /* already closed */ } + throw new GatewayError("Gateway response too large (over " + maxBytes + " bytes)"); + } + chunks.push(value); } + return Buffer.concat(chunks.map((c) => Buffer.from(c.buffer ?? c, c.byteOffset ?? 0, c.byteLength ?? c.length))).toString("utf8"); +} + +async function readJson(res) { + const text = await readTextCapped(res, GATEWAY_MAX_RESPONSE_BYTES); if (!text) return {}; try { return JSON.parse(text); @@ -343,8 +364,10 @@ export async function sendPrompt(session, ref, prompt, extra = {}) { } throw err; } - const messageId = data && typeof data.messageId === "string" ? data.messageId : null; - return { target: rec, result: data, delivery: "accepted", ...(messageId ? { messageId } : {}) }; + const messageId = data && typeof data === "object" && typeof data.messageId === "string" ? data.messageId : null; + // Only a confirmed receipt counts as accepted; anything else is unknown, never a silent accept. + // ponytail: full send/execution correlation envelope stays in #37. + return { target: rec, result: data && typeof data === "object" ? data : {}, delivery: messageId ? "accepted" : "unknown", ...(messageId ? { messageId } : {}) }; } export async function getTranscriptTail(session, ref, limit = 40) { diff --git a/test/codex-bridge.test.js b/test/codex-bridge.test.js index 70c6a11..48d2821 100644 --- a/test/codex-bridge.test.js +++ b/test/codex-bridge.test.js @@ -351,8 +351,7 @@ test("codex send reassembles a fragmented turn/start reply split across TCP chun } }); -test("codex status fails on a wrong handshake and closes the socket instead of leaking it", async () => { - const home = mkdtempSync(join(tmpdir(), "gbot-codex-badhs-")); +test("codex status fails on a wrong handshake and closes the socket instead of leaking it", async () => { const home = mkdtempSync(join(tmpdir(), "gbot-codex-badhs-")); mkdirSync(join(home, "app-server-control")); const socketPath = join(home, "app-server-control", "app-server-control.sock"); let serverSocket = null; @@ -461,3 +460,116 @@ test("codex send preserves turn and thread ids with accepted delivery on refusal await fake.close(); } }); + +test("a JSON null message fails as malformed instead of crashing on msg.id", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + initialize: (params, ok, err, send) => { + void params; void ok; void err; + send(null); + }, + }); + try { + const started = Date.now(); + const { code, err } = await gbot(fake.home, "codex", "status"); + assert.equal(code, 1); + assert.match(err, /malformed message/); + assert.ok(Date.now() - started < 5000, "did not hang on the bad message"); + } finally { + await fake.close(); + } +}); + +test("handshake timeout is absolute; trickled bytes do not extend it", async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-trickle-")); + mkdirSync(join(home, "app-server-control")); + const socketPath = join(home, "app-server-control", "app-server-control.sock"); + const server = createTcpServer((sock) => { + sock.on("data", () => { + const t = setInterval(() => sock.write("X"), 50); + sock.on("close", () => clearInterval(t)); + }); + sock.on("error", () => {}); + }); + await new Promise((resolve) => server.listen(socketPath, resolve)); + try { + const started = Date.now(); + await assert.rejects(connectCodexAppServer(socketPath, { timeoutMs: 300 }), /Timed out connecting/); + assert.ok(Date.now() - started < 5000, "absolute deadline fired instead of waiting on the trickle"); + } finally { + await new Promise((resolve) => server.close(resolve)); + } +}); + +test("codex status rejects terminated oversized handshake headers", async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-bighdr-")); + mkdirSync(join(home, "app-server-control")); + const socketPath = join(home, "app-server-control", "app-server-control.sock"); + const server = createTcpServer((sock) => { + sock.on("data", () => sock.write("HTTP/1.1 101 Switching Protocols\r\nX-Pad: " + "y".repeat(20000) + "\r\n\r\n")); + sock.on("error", () => {}); + }); + await new Promise((resolve) => server.listen(socketPath, resolve)); + try { + const started = Date.now(); + const { code, err } = await gbot(home, "codex", "status"); + assert.equal(code, 1); + assert.match(err, /exceed/); + assert.ok(Date.now() - started < 5000, "failed fast instead of decoding the headers"); + } finally { + await new Promise((resolve) => server.close(resolve)); + } +}); + +test("codex status rejects a complete oversized frame without buffering it", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + initialize: (params, ok, err, send, socket) => { + void params; void ok; void err; void send; + socket.write(encodeFrame(0x1, Buffer.alloc(5 * 1024 * 1024))); + }, + }); + try { + const started = Date.now(); + const { code, err } = await gbot(fake.home, "codex", "status"); + assert.equal(code, 1); + assert.match(err, /exceed/); + assert.ok(Date.now() - started < 5000, "failed fast instead of buffering the frame"); + } finally { + await fake.close(); + } +}); + +test("codex send emits structured JSON errors with delivery and ids", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const unknown = await gbot(fake.home, "--json", "codex", "send", "nope", "hi"); + assert.equal(unknown.code, 1); + assert.deepEqual(JSON.parse(unknown.err), { + error: "Unknown Codex thread nope. Run `gbot codex list-threads` to see reachable threads.", + delivery: "rejected", + threadId: "nope", + }); + assert.equal(unknown.out, ""); + } finally { + await fake.close(); + } + const refusing = await fakeAppServer({ + ...baseHandlers, + "turn/start": (params, ok, err, send) => { + send({ jsonrpc: "2.0", id: "srv-1", method: "item/commandExecution/requestApproval", params: { threadId: params.threadId } }); + setTimeout(() => ok({ turn: { id: "turn-10", status: "inProgress", items: [] } }), 20); + }, + }); + try { + const { code, err } = await gbot(refusing.home, "--json", "codex", "send", "t-1", "do it"); + assert.equal(code, 1); + const parsed = JSON.parse(err); + assert.equal(parsed.delivery, "accepted"); + assert.equal(parsed.threadId, "t-1"); + assert.equal(parsed.turnId, "turn-10"); + assert.match(parsed.error, /^Turn turn-10 started on thread t-1/); + } finally { + await refusing.close(); + } +}); diff --git a/test/gateway-send.test.js b/test/gateway-send.test.js new file mode 100644 index 0000000..9005ce5 --- /dev/null +++ b/test/gateway-send.test.js @@ -0,0 +1,67 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { GATEWAY_MAX_RESPONSE_BYTES, getTranscriptTail, sendPrompt } from "../src/gateway.js"; + +const session = { gatewayUrl: "https://box.cursor.sh", gatewayToken: "t" }; +const roster = { agents: [{ id: "bot-1", name: "General" }] }; + +function mockGateway(t, send) { + t.mock.method(globalThis, "fetch", async (url) => { + if (String(url).endsWith("/api/listAgents")) return new Response(JSON.stringify(roster), { status: 200 }); + return send(); + }); +} + +test("sendPrompt accepts only a confirmed messageId receipt", async (t) => { + mockGateway(t, () => new Response(JSON.stringify({ messageId: "m-1" }), { status: 200 })); + const out = await sendPrompt(session, "General", "hi"); + assert.equal(out.delivery, "accepted"); + assert.equal(out.messageId, "m-1"); +}); + +test("sendPrompt without a receipt is unknown, not accepted", async (t) => { + mockGateway(t, () => new Response(JSON.stringify({ ok: true }), { status: 200 })); + const out = await sendPrompt(session, "General", "hi"); + assert.equal(out.delivery, "unknown"); + assert.equal(out.messageId, undefined); +}); + +test("sendPrompt rejects on a 4xx and preserves the target id", async (t) => { + mockGateway(t, () => new Response(JSON.stringify({ message: "nope" }), { status: 400 })); + await assert.rejects(sendPrompt(session, "General", "hi"), (e) => { + assert.equal(e.delivery, "rejected"); + assert.equal(e.targetId, "bot-1"); + return true; + }); +}); + +test("sendPrompt marks transport loss unknown with a no-retry hint", async (t) => { + t.mock.method(globalThis, "fetch", async (url) => { + if (String(url).endsWith("/api/listAgents")) return new Response(JSON.stringify(roster), { status: 200 }); + throw new TypeError("fetch failed"); + }); + await assert.rejects(sendPrompt(session, "General", "hi"), (e) => { + assert.equal(e.delivery, "unknown"); + assert.match(e.message, /delivery unknown; check the thread before resending/); + return true; + }); +}); + +test("oversized gateway responses abort mid-stream instead of buffering", async (t) => { + mockGateway(t, () => new Response("x".repeat(GATEWAY_MAX_RESPONSE_BYTES + 8), { status: 200 })); + await assert.rejects(sendPrompt(session, "General", "hi"), /too large/); +}); + +test("getTranscriptTail clamps the limit to 1-200 like the CLI and tool", async (t) => { + const seen = []; + t.mock.method(globalThis, "fetch", async (url, options) => { + if (String(url).endsWith("/api/listAgents")) return new Response(JSON.stringify(roster), { status: 200 }); + seen.push(JSON.parse(options.body)); + return new Response(JSON.stringify({ entries: [] }), { status: 200 }); + }); + await getTranscriptTail(session, "General", 99999); + assert.equal(seen[0].limit, 200); + await getTranscriptTail(session, "General", -3); + assert.equal(seen[1].limit, 1); +}); From c6bf539db5b6e18ba6dfd9372e4654bd55cee295 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 15 Sep 2026 03:19:05 +0000 Subject: [PATCH 3/4] Sync changeset with HOLD-fix revision Co-authored-by: Zack Jackson --- .changeset/potato-p1-bridge.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/potato-p1-bridge.md b/.changeset/potato-p1-bridge.md index 18816bc..92c3661 100644 --- a/.changeset/potato-p1-bridge.md +++ b/.changeset/potato-p1-bridge.md @@ -1,4 +1,4 @@ "grok-bot-cli": patch --- -P1 bridge follow-ups: preserve send receipts with rejected/accepted/unknown delivery states (Codex `CodexSendError` keeps thread/turn IDs, gateway `sendPrompt` returns `delivery` + `messageId` and marks post-write loss unknown); bound the Codex WebSocket transport (idempotent close settling pending requests, socket destroy on every failure, exact handshake validation, fragmentation/UTF-8/opcode handling, header/frame/message/buffer budgets); make `gbot_thread` bounded without losing replies (truncation metadata, `full` bounded full-read in tool and `--full` in CLI, safe string normalization, 1–200 limit consistency, gateway deadline and response-byte cap). +P1 bridge follow-ups: preserve send receipts with rejected/accepted/unknown delivery states (Codex `CodexSendError` keeps thread/turn IDs, gateway `sendPrompt` returns `delivery` + `messageId` only on a confirmed receipt); bound the Codex WebSocket transport (idempotent close with guaranteed destruction, absolute handshake deadline, exact handshake validation, fragmentation/UTF-8/opcode handling, header/frame/message/buffer budgets); make `gbot_thread` bounded without losing replies (truncation metadata, bounded `full` reads, `--full` in CLI, safe string normalization, 1–200 limit consistency, gateway deadline and streaming response-byte cap). From 3fda090af4ebcfba2045b900f6f7a46d9819b35b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 15 Sep 2026 03:49:17 +0000 Subject: [PATCH 4/4] Contract gaps: trailing --json before dispatch, MCP metadata budgets Co-authored-by: Zack Jackson --- plugin/src/gbot.ts | 29 ++++++--- plugin/src/mcp/grok-bot/tools/gbot_thread.tsx | 2 +- plugin/tests/route-unit/tools.test.ts | 37 ++++++++++- src/cli.js | 24 +++++++- test/codex-bridge.test.js | 61 ++++++++++++++++++- 5 files changed, 140 insertions(+), 13 deletions(-) diff --git a/plugin/src/gbot.ts b/plugin/src/gbot.ts index 3b27aa3..ac4be59 100644 --- a/plugin/src/gbot.ts +++ b/plugin/src/gbot.ts @@ -54,6 +54,8 @@ export const ENTRY_TEXT_MAX = 400; // remainder is retrievable with `gbot thread --full` / `--json` on the machine. export const ENTRY_FULL_MAX = 20000; export const TRANSCRIPT_TOTAL_MAX = 200000; +// Metadata fields are capped too: an uncapped id/kind/role would bypass the total budget. +export const ENTRY_META_MAX = 200; export const truncateEntryText = (text: string, max = ENTRY_TEXT_MAX): string => { if (text.length <= max) return text; @@ -67,6 +69,8 @@ const entryFields = z.object({ timestampMs: z.number().optional(), }); +const capMeta = (value: string): string => (value.length > ENTRY_META_MAX ? `${value.slice(0, ENTRY_META_MAX)}…` : value); + const threadEntry = (raw: unknown, max: number, ellipsis: boolean): Entry => { const fields = entryFields.safeParse(raw); const full = entryText(raw); @@ -77,9 +81,9 @@ const threadEntry = (raw: unknown, max: number, ellipsis: boolean): Entry => { } const { id, kind, role, timestampMs } = fields.data; return { - id, - kind, - ...(role === undefined ? {} : { role }), + id: capMeta(id), + kind: capMeta(kind), + ...(role === undefined ? {} : { role: capMeta(role) }), text, truncated, fullLength: full.length, @@ -87,12 +91,23 @@ const threadEntry = (raw: unknown, max: number, ellipsis: boolean): Entry => { }; }; -export const transcriptEntries = (transcript: unknown, opts: { full?: boolean } = {}): Entry[] => { +const metaLength = (entry: Entry): number => entry.id.length + entry.kind.length + (entry.role?.length ?? 0); + +export const transcriptEntries = (transcript: unknown, opts: { full?: boolean; limit?: number } = {}): Entry[] => { + const rows = unwrapEntries(transcript); + // Enforce the requested count locally: a gateway ignoring `limit` cannot inflate output. + const wanted = + typeof opts.limit === 'number' && Number.isInteger(opts.limit) && opts.limit > 0 ? Math.min(opts.limit, 200) : rows.length; const perEntry = opts.full ? ENTRY_FULL_MAX : ENTRY_TEXT_MAX; let remaining = TRANSCRIPT_TOTAL_MAX; - return unwrapEntries(transcript).map((raw) => { - const entry = threadEntry(raw, Math.min(perEntry, remaining), !opts.full); - remaining = Math.max(0, remaining - entry.text.length); + return rows.slice(0, wanted).map((raw) => { + const entry = threadEntry(raw, perEntry, !opts.full); + const allowText = Math.max(0, Math.min(entry.text.length, remaining - metaLength(entry))); + if (allowText < entry.text.length) { + entry.text = allowText <= 0 ? '' : !opts.full ? truncateEntryText(entry.text, allowText) : entry.text.slice(0, allowText); + entry.truncated = entry.fullLength > entry.text.length; + } + remaining = Math.max(0, remaining - metaLength(entry) - entry.text.length); return entry; }); }; diff --git a/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx b/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx index c10221c..a242ce2 100644 --- a/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx +++ b/plugin/src/mcp/grok-bot/tools/gbot_thread.tsx @@ -48,7 +48,7 @@ export default defineTool( }, async ({ limit, target, full }) => { const tail = await withRedactedErrors(async () => getTranscriptTail(await connectGateway(), target, limit)); - const value = { entries: transcriptEntries(tail.transcript, { full }), target: summarizeTarget(tail.target) }; + const value = { entries: transcriptEntries(tail.transcript, { full, limit }), target: summarizeTarget(tail.target) }; return ( {`${value.target.kind} ${value.target.name}: ${value.entries.length} entries.`} diff --git a/plugin/tests/route-unit/tools.test.ts b/plugin/tests/route-unit/tools.test.ts index 15fa1d3..fbba0c5 100644 --- a/plugin/tests/route-unit/tools.test.ts +++ b/plugin/tests/route-unit/tools.test.ts @@ -25,6 +25,7 @@ const roster = { { id: 'bot-4', isGroup: false, name: 'Odd' }, { id: 'bot-5', isGroup: false, name: 'Noreceipt' }, { id: 'bot-6', isGroup: false, name: 'Big' }, + { id: 'bot-7', isGroup: false, name: 'Meta' }, ], }; const transcripts: Record = { @@ -55,6 +56,9 @@ const transcripts: Record = { 'bot-6': { entries: Array.from({ length: 11 }, (_, index) => ({ id: `b${index}`, kind: 'note', text: 'q'.repeat(20000) })), }, + 'bot-7': { + entries: [{ id: 'i'.repeat(300), kind: 'k'.repeat(300), role: 'R'.repeat(5000), text: 'hi' }], + }, }; const responses: Record) => [number, unknown]> = { getAgentTranscriptTail: (body) => [200, transcripts[String(body.id)]], @@ -220,11 +224,40 @@ describe('grok-bot MCP server', () => { }); expect(full.isError).toBe(false); const entries = (full.structuredContent as { entries: { id: string; text: string; truncated: boolean; fullLength: number }[] }).entries; - expect(entries).toHaveLength(11); - expect(entries[9]).toEqual({ id: 'b9', kind: 'note', text: 'q'.repeat(20000), truncated: false, fullLength: 20000 }); + expect(entries.length).toBe(11); + expect(entries[9]?.text.length).toBe(19940); + expect(entries[9]).toMatchObject({ id: 'b9', truncated: true, fullLength: 20000 }); expect(entries[10]).toEqual({ id: 'b10', kind: 'note', text: '', truncated: true, fullLength: 20000 }); }); + it('gbot_thread enforces the requested count even when the gateway ignores the limit', async () => { + const capped = await invokeMcpTool('gbot_thread', { + input: { limit: 3, target: 'Big' }, + server: 'grok-bot', + }); + expect(capped.isError).toBe(false); + const entries = (capped.structuredContent as { entries: unknown[] }).entries; + expect(entries.length).toBe(3); + }); + + it('gbot_thread bounds id/kind/role metadata that would bypass the text budget', async () => { + const odd = await invokeMcpTool('gbot_thread', { input: { target: 'Meta' }, server: 'grok-bot' }); + expect(odd.isError).toBe(false); + expect(odd.structuredContent).toEqual({ + entries: [ + { + id: `${'i'.repeat(200)}…`, + kind: `${'k'.repeat(200)}…`, + role: `${'R'.repeat(200)}…`, + text: 'hi', + truncated: false, + fullLength: 2, + }, + ], + target: { id: 'bot-7', kind: 'bot', name: 'Meta' }, + }); + }); + it('redacts a bearer token echoed by the gateway before the error reaches the host', async () => { const result = await invokeMcpTool('gbot_send', { input: { message: 'x', target: 'Proxy' }, diff --git a/src/cli.js b/src/cli.js index 4ef7f73..d2eb07d 100755 --- a/src/cli.js +++ b/src/cli.js @@ -127,6 +127,17 @@ function hasFlag(args, name) { return true; } +/** Peel a trailing flag that `--` does not protect; free-text commands keep mid-text tokens. */ +function takeTrailingFlag(args, name) { + const stop = args.indexOf("--"); + const end = stop === -1 ? args.length : stop; + if (end > 0 && args[end - 1] === name) { + args.splice(end - 1, 1); + return true; + } + return false; +} + /** Peel global CLI options only from the leading argv (before the command). */ let jsonErrors = false; function takeLeadingGlobals(args) { @@ -312,6 +323,11 @@ function formatCodexThread(t) { } async function runCodex(sub, rest, json) { + // Structured subcommands take no free text, so --json peels anywhere. Send + // peels a trailing --json only; mid-message tokens stay message content. + if (sub === "status" || sub === "list-threads") { + if (hasFlag(rest, "--json")) { json = true; jsonErrors = true; } + } if (sub === "status") { const status = await codexStatus(); print(json ? status : formatCodexStatus(status)); @@ -330,6 +346,7 @@ async function runCodex(sub, rest, json) { } if (sub === "send") { const threadId = rest.shift(); + if (takeTrailingFlag(rest, "--json")) { json = true; jsonErrors = true; } if (rest[0] === "--") rest.shift(); const message = rest.join(" ").trim(); if (!threadId || threadId.startsWith("-") || !message) throw new StoreError("gbot codex send "); @@ -416,6 +433,12 @@ async function main(argv) { return; } + // Peel command-local --json before touching the backend so auth/gateway + // failures honor it. Send peels a trailing --json only (mid-text tokens stay + // message content); `--` protects everything after it. + if (cmd === "send" && takeTrailingFlag(rest, "--json")) { json = true; jsonErrors = true; } + if ((cmd === "thread" || cmd === "chat") && hasFlag(rest, "--json")) { json = true; jsonErrors = true; } + const backend = await openBackend({ root: rootFlag, gateway, files: filesMode }); if (cmd === "bots" && sub === "list") { @@ -539,7 +562,6 @@ async function main(argv) { if (cmd === "thread" || cmd === "chat") { const ref = sub; if (!ref) throw new StoreError("gbot thread [--limit N] [--root MESSAGE_ID] [--full]"); - if (hasFlag(rest, "--json")) { json = true; jsonErrors = true; } const full = hasFlag(rest, "--full"); const limitRaw = takeFlag(rest, "--limit"); const rootId = takeFlag(rest, "--root"); diff --git a/test/codex-bridge.test.js b/test/codex-bridge.test.js index 48d2821..980c361 100644 --- a/test/codex-bridge.test.js +++ b/test/codex-bridge.test.js @@ -540,8 +540,7 @@ test("codex status rejects a complete oversized frame without buffering it", asy } }); -test("codex send emits structured JSON errors with delivery and ids", async () => { - const fake = await fakeAppServer(baseHandlers); +test("codex send emits structured JSON errors with delivery and ids", async () => { const fake = await fakeAppServer(baseHandlers); try { const unknown = await gbot(fake.home, "--json", "codex", "send", "nope", "hi"); assert.equal(unknown.code, 1); @@ -573,3 +572,61 @@ test("codex send emits structured JSON errors with delivery and ids", async () = await refusing.close(); } }); + +test("codex send parses a trailing --json as a flag in both placements", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const trailing = await gbot(fake.home, "codex", "send", "t-1", "hi", "--json"); + assert.equal(trailing.code, 0, trailing.err); + assert.equal(JSON.parse(trailing.out).turnId, "turn-9"); + assert.equal(fake.received.at(-1).params.input[0].text, "hi"); + + const leading = await gbot(fake.home, "--json", "codex", "send", "t-1", "hi"); + assert.equal(leading.code, 0, leading.err); + assert.equal(JSON.parse(leading.out).turnId, "turn-9"); + assert.equal(fake.received.at(-1).params.input[0].text, "hi"); + } finally { + await fake.close(); + } +}); + +test("codex send keeps a --json protected by -- as message content", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const { code, out } = await gbot(fake.home, "codex", "send", "t-1", "--", "--json"); + assert.equal(code, 0, out); + assert.match(out, /^Started turn turn-9/); + assert.equal(fake.received.at(-1).params.input[0].text, "--json"); + } finally { + await fake.close(); + } +}); + +test("codex send failures honor a trailing --json with structured errors", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const { code, err, out } = await gbot(fake.home, "codex", "send", "nope", "hi", "--json"); + assert.equal(code, 1); + assert.equal(out, ""); + assert.deepEqual(JSON.parse(err), { + error: "Unknown Codex thread nope. Run `gbot codex list-threads` to see reachable threads.", + delivery: "rejected", + threadId: "nope", + }); + } finally { + await fake.close(); + } +}); + +test("send failures honor a trailing --json before backend auth runs", async () => { + const env = { ...process.env, CODEX_HOME: "/nonexistent", PATH: "/nonexistent" }; + const run = (...args) => new Promise((resolve) => { + execFile(process.execPath, [CLI, ...args], { encoding: "utf8", env }, (error, out, err) => { + resolve({ code: error ? error.code : 0, out, err }); + }); + }); + const { code, err } = await run("--dir", "/nonexistent-gbot-dir", "send", "Nobody", "hi", "--json"); + assert.equal(code, 1); + const parsed = JSON.parse(err); + assert.equal(typeof parsed.error, "string"); +});