From 78a2510a866da9beff5eb3aa1310bad8994a6bb5 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 15 Sep 2026 02:51:53 +0000 Subject: [PATCH] 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), ""); +});