diff --git a/.changeset/codex-contract-envelope.md b/.changeset/codex-contract-envelope.md new file mode 100644 index 0000000..43de0b0 --- /dev/null +++ b/.changeset/codex-contract-envelope.md @@ -0,0 +1,5 @@ +--- +"grok-bot-cli": patch +--- + +Make `gbot codex` dependable for automation and safe for agent relays: `codex status --json` reports `socketState`, a stable failure `mode` (`socket-absent`, `permission-denied`, `not-a-socket`, `connect-failed`, `handshake-failed`, `bad-response`, `windows-unsupported`), `schema.compatibility` separate from reachability, a bounded `codex --version` probe (`cliVersionProbe`), and `desktopAttached: "unknown"`; `codex list-threads` adds `--cursor`, bounds `--limit` to 1–200, rejects unknown arguments, validates the response shape, and strips terminal controls from every text field in JSON too. `codex send` and `send` accept `--correlation-id`, `--reply-to`, `--hop`, and `--envelope`, return receipts with `messageId` (sent as Codex's `clientUserMessageId`), `correlationId`, `replyTo`, `hop`, and `maxHops`, refuse relays at `GROK_BOT_MAX_HOPS` (default 4) with `reason: "hop-limit"`, honor the operator allowlist `GROK_BOT_CODEX_THREADS`, refuse `active` threads with `reason: "busy"` instead of steering a running turn (or, with `--when-busy queue` and `GROK_BOT_CODEX_EXPERIMENTAL=1`, hand them to Codex's experimental `thread/queue/add` and report `delivery: "queued"`; `codex queue ` lists that queue), and emit `reason`/`mode` in every `--json` send failure. Fixes #37, #38, #39. diff --git a/README.md b/README.md index cabe478..9058b6b 100644 --- a/README.md +++ b/README.md @@ -72,12 +72,31 @@ gbot codex send "Grok here: the build is green, please continue." **ChatGPT Desktop limitation.** Desktop runs its own private stdio app-server and does not publish the shared control socket, so external clients cannot reach live Desktop tasks. When the socket is absent, `gbot codex status` exits 1 and says so, naming the upstream issues: [openai/codex#41014](https://github.com/openai/codex/issues/41014) and [openai/codex#41112](https://github.com/openai/codex/issues/41112). `gbot` never reads Desktop's temporary `CODEX_APP_TOOLS_PIPE_PATH` sockets under `/tmp/codex-browser-use/`; that channel is private to Desktop. -**Failure modes.** +**Status contract (`gbot codex status --json`).** `reachable` is endpoint reachability only. `socketState` is `socket`, `absent`, `permission-denied`, or `not-a-socket`; `mode` is `daemon` for a usable daemon, otherwise the failure: `socket-absent`, `permission-denied` (the file or the connect refused this user), `not-a-socket`, `connect-failed` (socket present, nothing completed the WebSocket upgrade), `handshake-failed` (upgrade or `initialize` failed), `windows-unsupported`, or `bad-response` (reachable, but `initialize` returned something off-schema — `reachable` stays `true`). `schema.compatibility` is `exact` when the daemon reports the pinned version, `unverified` when it differs (methods usually survive upgrades, but the shapes are not re-checked), or `unknown`. `cliVersionProbe` reports whether `codex --version` answered (`ok`, `missing`, `timeout` after 3 s, `error`). Whether ChatGPT Desktop owns a thread is not observable from the socket, so `desktopAttached` is always `"unknown"`. Every failure exits 1; automation reads `mode`/`reason`, not the exit code. -- Socket absent: no daemon, or Desktop-private mode. Start the daemon or wait for the upstream fixes. -- Unknown thread: `send` fails with "Unknown Codex thread"; use `list-threads`. -- Thread open elsewhere: a thread with an active writer (VS Code, TUI) fails with "open in another client"; close it there first. -- Approvals: `gbot` never approves commands or file changes on your behalf. If Codex asks while `gbot` is still connected, `send` refuses the request, exits 1, and tells you the turn id. `send` disconnects as soon as the turn starts, so later approval requests stay with the daemon for a Codex client to answer; for unattended sends set `approval_policy = "never"` in the daemon's `config.toml`. +**Thread discovery.** `list-threads --limit N` (1–200) pages with the opaque `--cursor` from the previous `nextCursor`; JSON keeps the cursor verbatim, text output prints a sanitized `more: --cursor …` hint. Text fields are stripped of terminal control sequences in both outputs (single-line fields also lose line breaks; `preview` keeps its newlines; a structured `source` such as `{ "custom": … }` passes through unchanged), `status` is one of `notLoaded | idle | active | systemError | unknown`, and non-numeric `updatedAt` becomes `null`. Unknown arguments are rejected before the socket is touched; a response that does not match the pinned schema (including an entry without a string `id`) fails with `reason: "bad-response"`. + +**Routes, attribution, and loops.** `gbot codex send` runs on the machine that owns `CODEX_HOME`, as the user who owns the socket, with that user's Codex credentials; the socket path comes only from `CODEX_HOME`, never from the message or an agent-supplied argument. A cloud-hosted Grok Bot cannot reach a desktop socket directly — run `gbot` locally (for example from a Codex skill or an agent on that machine). `GROK_BOT_CODEX_THREADS=id,id` lets the operator pin `send` to approved threads (`reason: "route-not-allowed"` otherwise). Every send gets a delivery envelope: `messageId` (also sent as Codex's native `clientUserMessageId`), `correlationId` (defaults to the message id), optional `replyTo`, and `hop`. A reply passes the original correlation id and `hop` + 1: + +```sh +gbot codex send "Grok here: build is green" # receipt: messageId M, correlationId M, hop 0 +gbot codex send --correlation-id M --reply-to M --hop 1 "ack" # the answer, one hop later +``` + +Sends at `hop >= GROK_BOT_MAX_HOPS` (default 4) are refused with `reason: "hop-limit"` before anything reaches the daemon, so two agents cannot acknowledge each other forever; `gbot` never auto-acknowledges. `--envelope` (implied by any envelope flag) prepends a one-line `[gbot msg=… corr=… reply-to=… hop=… from=user@host]` header so the receiving agent can quote the ids back. That header is caller-authored provenance for the reader, not authentication: the daemon authenticates the local user through the socket, nothing else. The same flags work for `gbot send` to Grok bots and groups (the receipt carries `envelopeId`, `correlationId`, `hop`; Grok's own `messageId` remains the delivery receipt). Private ChatGPT Desktop pipes and arbitrary ChatGPT chats stay out of scope; only Codex threads on a reachable app-server daemon are routes. + +**Busy threads.** `send` reads the thread status on resume. Only `idle` and `notLoaded` threads start a turn. An `active` thread (a turn in progress, or waiting on approval / user input) is refused with `reason: "busy"`: in app-server 0.154.0 a `turn/start` on an active thread steers that turn rather than queueing behind it, and `gbot` never steers or interrupts work a human may be doing. Either wait for `list-threads` to show `idle` and resend, or pass `--when-busy queue` to hand the message to the daemon's own queue through Codex's experimental `thread/queue/add` — that needs `GROK_BOT_CODEX_EXPERIMENTAL=1`, returns `delivery: "queued"` with `queuedSubmissionId`, and `gbot codex queue ` shows what is still waiting. `systemError` threads are refused with `reason: "thread-error"`, statuses this version does not know with `reason: "unknown-status"`. Receipts distinguish `delivery: "accepted"` (turn started; `turnId`, `turnStatus`), `"queued"`, `"rejected"` (nothing was sent; see `reason`), and `"unknown"` (the request left but no acknowledgment came back — look for `messageId` in the thread or queue before resending). The decision record, with the schema evidence and a live probe of the queue API, is in [`docs/codex-busy-threads.md`](docs/codex-busy-threads.md). + +**Failure modes.** Every `send` / `codex send` failure under `--json` is `{ error, delivery, reason, messageId, correlationId, hop, … }` on stderr with exit 1; argument mistakes are `{ error, reason: "usage" }`; `codex status` never fails this way — it prints the status document on stdout and exits 1 when `mode` is anything but `daemon`. `reason` values are stable: + +- `socket-absent` / `permission-denied` / `not-a-socket` / `connect-failed` / `handshake-failed` / `windows-unsupported`: the route is unavailable. Start the daemon, fix the socket, or wait for the upstream Desktop fixes. +- `unknown-thread`: use `list-threads`. +- `external-owner`: a thread with an active writer (VS Code, TUI) is open in another client; close it there first. +- `busy` / `thread-error` / `unknown-status`: see above. +- `route-not-allowed` / `hop-limit` / `experimental-disabled`: refused by operator policy, the relay bound, or the experimental-API gate. +- `unsupported`: the daemon does not offer the (experimental) method `--when-busy queue` needs. +- `approval-refused` (`delivery: "accepted"`): `gbot` never approves commands or file changes on your behalf. If Codex asks while `gbot` is still connected, `send` refuses the request, exits 1, and tells you the turn id. `send` disconnects as soon as the turn starts, so later approval requests stay with the daemon for a Codex client to answer; for unattended sends set `approval_policy = "never"` in the daemon's `config.toml`. +- `transport` / `bad-response` (`delivery: "unknown"`): the connection dropped or the daemon answered off-schema after the request left. ## Talking to Grok Bot from Codex diff --git a/docs/codex-busy-threads.md b/docs/codex-busy-threads.md new file mode 100644 index 0000000..8c8a079 --- /dev/null +++ b/docs/codex-busy-threads.md @@ -0,0 +1,74 @@ +# Busy-thread delivery for `gbot codex send` + +Decision record for [#38](https://github.com/ScriptedAlchemy/grok-bot-cli/issues/38). Evidence is the +app-server schema emitted by `codex app-server generate-json-schema` for codex-cli 0.154.0 — the release +`src/codex-bridge.js` is pinned to — generated both without and with `--experimental`, plus a live probe +against a 0.154.0 daemon (below). + +## What the protocol offers + +| Surface | In 0.154.0 | Notes | +| --- | --- | --- | +| `turn/start` on an idle thread | stable client request | Starts a new turn; `TurnStartResponse.turn.id` is the new turn. | +| `turn/start` on an **active** thread | stable client request | Steers the running turn. `TurnStartParams.turnTrigger` is documented as "Ignored when this request steers an already-active turn", so the call does not queue behind the human's turn — it joins it. | +| `turn/steer` | stable client request | Explicit steer; requires `expectedTurnId` and fails when it no longer matches the active turn. | +| `turn/interrupt` | stable client request | Stops the active turn. | +| `thread/queue/add` / `list` / `update` / `delete` / `reorder` / `start` | **experimental** client requests | Present only in the `--experimental` schema and only for a client that initialized with `capabilities.experimentalApi: true`. `add` takes `{ threadId, clientUserMessageId, input }` and returns a `QueuedSubmission { id, clientUserMessageId, input }`; `list` pages with `cursor`/`limit`; `start` runs one queued submission as a turn. | +| `thread/queue/changed` | server notification | Emitted when a thread's queue changes. | +| `ThreadStatus` | type | `notLoaded`, `idle`, `active` (with `activeFlags`: `waitingOnApproval`, `waitingOnUserInput`), `systemError`. | + +### Live probe (0.154.0 daemon, 2026-09-15) + +With `experimentalApi: true`: `thread/queue/list` → `{ data: [], nextCursor: null }`; `thread/queue/add` +returned a `QueuedSubmission` with the supplied `clientUserMessageId`; `list` then showed it first in +insertion order; `thread/queue/delete` → `{ deleted: true }` and the list was empty again. The target +thread was `notLoaded` — the queue is server-side state, not a property of a loaded session. Not tested +on this machine: whether queued submissions survive a daemon restart, and whether the daemon starts them +itself when the active turn ends or only on `thread/queue/start` (the TUI drains its own queue; a +daemon-only thread may need an explicit start). Both stay documented as unknown until measured. + +## Decision + +1. **Default: refuse busy destinations.** `send` resumes the thread (`thread/resume`, `excludeTurns: true`) + and reads `thread.status`. `active` → `delivery: "rejected", reason: "busy"`; `systemError` → + `reason: "thread-error"`; any status this pin does not know → `reason: "unknown-status"`. Only `idle` + and `notLoaded` (no turn can be running) proceed to `turn/start`. `turn/steer` and `turn/interrupt` + are never called: they change or stop work a human may be doing in the same thread. +2. **Opt-in: hand busy sends to the daemon's queue.** `--when-busy queue` requires + `GROK_BOT_CODEX_EXPERIMENTAL=1` (the experimental API is gated, so its use is an operator decision, not + a default) and initializes the session with `experimentalApi: true`. On an `active` thread `send` calls + `thread/queue/add` with the envelope's `messageId` as `clientUserMessageId` and returns + `delivery: "queued"` with `queuedSubmissionId`. A daemon without the method answers `-32601`, reported + as `reason: "unsupported"`. `gbot codex queue ` lists the queue (same gate) so the caller can + see whether the submission is still waiting. Idle threads are never queued — they start directly. +3. **No gbot-side queue.** A local queue would live in one process on one machine, be lost on restart, + and race the human's own submissions. The daemon's queue is the right owner; when it stabilizes the + gate goes away. +4. **Receipts distinguish states.** `accepted` (turn started; `turnId`, `turnStatus`), `queued` + (`queuedSubmissionId`; not started), `rejected` (nothing left `gbot`; `reason` says why), `unknown` + (the request left but no acknowledgment came back; `messageId` is Codex's `clientUserMessageId`, so + the caller can look for it in the thread or the queue before resending). There is no `completed` + state: `send` returns when the turn starts, and waiting for a result is a separate bounded operation. +5. **Steer and interrupt stay out.** Adding either is a new flag plus a schema re-pin, never a default. + +## Known ceiling + +Status is read at resume time and `turn/start` follows on the same connection a few milliseconds later. +A human who starts a turn inside that window turns our `turn/start` into a steer. The daemon exposes no +compare-and-start request, so the race cannot be closed client-side; it is small, and the receipt's +`turnId` plus `clientUserMessageId` make it auditable. The upgrade path is `thread/queue/add` followed by +`thread/queue/start` once the queue API leaves experimental. + +## Test matrix (`test/codex-bridge.test.js`) + +- idle → `accepted`, `turn/start` carries `clientUserMessageId` and `turnTrigger: "gbot"`. +- active (`waitingOnUserInput`) → `busy`, no `turn/start` / `turn/steer` / `turn/interrupt` sent. +- active + `--when-busy queue` + gate on → `queued`, `initialize` carried `experimentalApi: true`, + `thread/queue/add` carried the envelope id; `gbot codex queue` lists and sanitizes; a daemon without the + method → `unsupported`. +- `--when-busy queue` with the gate off → `experimental-disabled`, nothing sent. +- systemError → `thread-error`; unknown status → `unknown-status`. +- active writer elsewhere → `external-owner`; unknown thread → `unknown-thread`. +- connection dropped after `turn/start` left → `delivery: "unknown"`, `reason: "transport"`, receipt keeps `messageId`. +- malformed acknowledgment → `delivery: "unknown"`, `reason: "bad-response"`. +- every rejection, including route and allowlist refusals, carries `messageId` / `correlationId` / `hop`. diff --git a/src/cli.js b/src/cli.js index a610822..57e6bd2 100755 --- a/src/cli.js +++ b/src/cli.js @@ -6,22 +6,28 @@ import { inspectGrokBotGatewaySession } from "./app-session.js"; import { entryText, transcriptDelta, transcriptEntries } from "./transcript.js"; import { historyPath, readHistory, saveHistory } from "./history.js"; import { redactSecrets } from "./url-policy.js"; -import { codexStatus, listCodexThreads, sendToCodexThread } from "./codex-bridge.js"; +import { buildEnvelope, codexStatus, listCodexQueue, listCodexThreads, sendToCodexThread, singleLine, stripTerminalControls, withEnvelopeHeader } from "./codex-bridge.js"; function print(value) { if (typeof value === "string") process.stdout.write(value + "\n"); else process.stdout.write(JSON.stringify(value, null, 2) + "\n"); } +/** Exit 1 for every failure; `--json` callers read `reason` / `mode` / `delivery` instead of the exit code. */ function fail(err) { let message = err instanceof Error ? err.message : String(err); message = redactSecrets(message); 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; + for (const key of ["delivery", "reason", "mode", "threadId", "turnId", "targetId", "messageId", "correlationId"]) { + if (err[key] !== undefined) out[key] = err[key]; + } + if (out.reason === undefined && err instanceof StoreError) out.reason = "usage"; + if (err.envelope && typeof err.envelope === "object") { + out.messageId = err.envelope.messageId; + out.correlationId = err.envelope.correlationId; + out.hop = err.envelope.hop; + } process.stderr.write(JSON.stringify(out) + "\n"); } else { process.stderr.write(message + "\n"); @@ -57,14 +63,19 @@ function usage() { " groups remove ", " groups set --member ID [--member ...]", " groups delete ", - " send ", + " send [envelope flags] ", " thread [--limit N] [--after ENTRY_ID] [--root MESSAGE_ID] [--full]", " chat alias for thread", " history [bot-or-group] [--search TEXT] [--limit N] (offline)", " history --path print the local JSONL file path", " codex status", - " codex list-threads [--limit N]", - " codex send ", + " codex list-threads [--limit N] [--cursor CURSOR]", + " codex send [envelope flags] [--when-busy reject|queue] ", + " codex queue (experimental: GROK_BOT_CODEX_EXPERIMENTAL=1)", + "", + "Envelope flags (before the target): --correlation-id ID --reply-to MESSAGE_ID --hop N --envelope", + " Receipts carry messageId/correlationId/hop; a reply passes the original correlation id and hop+1.", + " Sends at hop >= GROK_BOT_MAX_HOPS (default 4) are refused. --envelope prepends the [gbot ...] header.", "", "Max group members: " + MAX_GROUP_MEMBERS, "--description / --instructions is the UI Instructions field (same key).", @@ -74,17 +85,19 @@ function usage() { "Auth: GROK_BOT_GATEWAY_URL + GROK_BOT_GATEWAY_TOKEN, or the Grok Bot app session, or CURSOR_ACCESS_TOKEN", "File fallback: GROK_BOT_AGENTS_DIR", "Codex: talks to the local app-server daemon socket under CODEX_HOME (default ~/.codex)", + " GROK_BOT_CODEX_THREADS=id,id restricts `codex send` to operator-approved threads", "History: opt-in plaintext JSONL at ~/.grok-bot-cli/history.jsonl", " GROK_BOT_HISTORY=on to record; --history-dir / GROK_BOT_HISTORY_DIR to relocate", " --no-history to skip one command", ].join("\n"); } -function takeFlag(args, name) { +/** `opaque` values may start with `-` (pagination cursors); ordinary values may not. */ +function takeFlag(args, name, { opaque = false } = {}) { const i = args.indexOf(name); if (i === -1) return undefined; const value = args[i + 1]; - if (value == null || value.startsWith("-")) throw new StoreError(name + " needs a value"); + if (value == null || value === "" || (!opaque && value.startsWith("-"))) throw new StoreError(name + " needs a value"); args.splice(i, 2); return value; } @@ -193,15 +206,6 @@ function takeLeadingGlobals(args) { return { json, gateway, files, dir, noHistory, historyDir }; } -/** Strip CSI/OSC and other C0/C1 controls so thread fields cannot drive the terminal. */ -function stripTerminalControls(text) { - return String(text) - .replace(/\u001b\[[0-9;?]*[ -/]*[@-~]/g, "") - .replace(/\u001b\][^\u0007\u001b]*(?:\u0007|\u001b\\)/g, "") - .replace(/\u001b./g, "") - .replace(/[\u0000-\u001F\u007F-\u009F]/g, ""); -} - function parseOnOff(value, flag) { const v = String(value).trim().toLowerCase(); if (v === "on" || v === "true" || v === "1" || v === "yes") return true; @@ -306,10 +310,13 @@ function truncateCliText(text, max = 400) { } function formatCodexStatus(s) { - const lines = ["socket: " + s.socketPath]; - if (!s.reachable) return lines.concat("reachable: no", s.message).join("\n"); + const lines = ["socket: " + s.socketPath + " (" + s.socketState + ")"]; + if (!s.reachable) return lines.concat("reachable: no (" + s.mode + ")", s.message).join("\n"); + if (s.mode !== "daemon") return lines.concat("reachable: yes, but unusable (" + s.mode + ")", s.message).join("\n"); lines.push("reachable: yes (daemon)"); - lines.push("daemon version: " + (s.daemonVersion ?? "unknown") + " cli version: " + (s.cliVersion ?? "unknown") + " pinned schema: " + s.pinnedVersion); + const cli = s.cliVersion ?? (s.cliVersionProbe === "ok" ? "unknown" : "unknown, probe " + s.cliVersionProbe); + lines.push("daemon version: " + (s.daemonVersion ?? "unknown") + " cli version: " + cli + " pinned schema: " + s.pinnedVersion + " (" + s.schema.compatibility + ")"); + lines.push("desktop attached: unknown (not observable from the socket)"); if (s.versionMismatch) lines.push("warning: daemon and CLI versions differ; `codex app-server daemon restart` picks up the installed CLI"); return lines.join("\n"); } @@ -322,40 +329,106 @@ function formatCodexThread(t) { return stripTerminalControls(t.id) + " " + stripTerminalControls(t.status) + title + "\n " + stripTerminalControls(t.cwd ?? "") + preview; } -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; } +/** Structured subcommands accept only their documented flags; anything left over is an error. */ +function rejectUnknownArgs(rest, usageLine) { + if (rest.length) throw new StoreError("Unknown argument " + JSON.stringify(rest[0]) + ". Usage: " + usageLine); +} + +/** + * Envelope flags live before the free-text target so message bodies keep their own `--` tokens. + * Returns a built envelope (or a hop-limit refusal) plus whether any flag was given. + */ +function takeEnvelopeFlags(rest, { busyPolicy = false } = {}) { + let correlationId; + let replyTo; + let hop; + let envelope = false; + let whenBusy = "reject"; + for (;;) { + const a = rest[0]; + if (busyPolicy && a === "--when-busy") { + rest.shift(); + whenBusy = rest.shift(); + if (whenBusy !== "reject" && whenBusy !== "queue") throw new StoreError("--when-busy must be reject or queue"); + continue; + } + if (a === "--correlation-id") { rest.shift(); correlationId = rest.shift(); if (correlationId == null) throw new StoreError("--correlation-id needs a value"); continue; } + if (a === "--reply-to") { rest.shift(); replyTo = rest.shift(); if (replyTo == null) throw new StoreError("--reply-to needs a value"); continue; } + if (a === "--hop") { + rest.shift(); + const raw = rest.shift(); + if (raw == null || !/^\d+$/.test(raw)) throw new StoreError("--hop must be a non-negative integer"); + hop = Number(raw); + continue; + } + if (a === "--envelope") { rest.shift(); envelope = true; continue; } + break; } - if (sub === "status") { - const status = await codexStatus(); - print(json ? status : formatCodexStatus(status)); - if (!status.reachable) process.exitCode = 1; - return; + try { + return { envelope: buildEnvelope({ correlationId, replyTo, hop, envelope }), whenBusy }; + } catch (err) { + if (err instanceof RangeError) throw new StoreError(err.message); + throw err; } - if (sub === "list-threads") { - const limitRaw = takeFlag(rest, "--limit"); - const limit = limitRaw ? Number(limitRaw) : 20; - if (!Number.isInteger(limit) || limit < 1) throw new StoreError("--limit must be a positive integer"); - const out = await listCodexThreads({ limit }); - if (json) print(out); - else if (out.threads.length === 0) print("No Codex threads."); - else print(out.threads.map(formatCodexThread).join("\n\n")); - return; - } - 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 "); - const out = await sendToCodexThread(threadId, message); - if (json) print(out); - else print("Started turn " + out.turnId + " (" + out.turnStatus + ") on Codex thread " + out.threadId); - return; +} + +async function codexStatusCommand(rest, json) { + rejectUnknownArgs(rest, "gbot codex status [--json]"); + const status = await codexStatus(); + print(json ? status : formatCodexStatus(status)); + // Exit 0 only for a usable daemon; `mode` says why otherwise (reachable but off-schema included). + if (!status.reachable || status.mode !== "daemon") process.exitCode = 1; +} + +async function codexListCommand(rest, json) { + const limitRaw = takeFlag(rest, "--limit"); + const cursor = takeFlag(rest, "--cursor", { opaque: true }); + rejectUnknownArgs(rest, "gbot codex list-threads [--limit N] [--cursor CURSOR] [--json]"); + const limit = limitRaw ? Number(limitRaw) : 20; + let out; + try { + out = await listCodexThreads({ limit, cursor }); + } catch (err) { + if (err instanceof RangeError) throw new StoreError(err.message); + throw err; } - throw new StoreError("gbot codex status | list-threads [--limit N] | send "); + if (json) print(out); + else if (out.threads.length === 0) print("No Codex threads."); + else print(out.threads.map(formatCodexThread).join("\n\n") + (out.nextCursor ? "\n\nmore: --cursor " + JSON.stringify(singleLine(out.nextCursor)) : "")); +} + +async function codexQueueCommand(rest, json) { + const threadId = rest.shift(); + rejectUnknownArgs(rest, "gbot codex queue [--json]"); + if (!threadId || threadId.startsWith("-")) throw new StoreError("gbot codex queue [--json]"); + const out = await listCodexQueue(threadId); + if (json) print(out); + else if (out.queued.length === 0) print("No queued submissions on Codex thread " + threadId + "."); + else print(out.queued.map((q) => q.id + " " + (q.clientUserMessageId ?? "") + "\n " + singleLine(q.text).slice(0, 200)).join("\n\n")); +} + +async function codexSendCommand(rest, json) { + const { envelope, whenBusy } = takeEnvelopeFlags(rest, { busyPolicy: true }); + const threadId = rest.shift(); + if (rest[0] === "--") rest.shift(); + const message = rest.join(" ").trim(); + if (!threadId || threadId.startsWith("-") || !message) throw new StoreError("gbot codex send [envelope flags] [--when-busy reject|queue] "); + const out = await sendToCodexThread(threadId, message, { envelope, whenBusy }); + if (json) print(out); + else if (out.delivery === "queued") print("Queued " + out.queuedSubmissionId + " on busy Codex thread " + out.threadId + "; message " + out.messageId); + else print("Started turn " + out.turnId + " (" + out.turnStatus + ") on Codex thread " + out.threadId + "; message " + out.messageId); +} + +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. + const structured = sub === "status" || sub === "list-threads" || sub === "queue"; + if (structured ? hasFlag(rest, "--json") : sub === "send" && takeTrailingFlag(rest, "--json")) { json = true; jsonErrors = true; } + if (sub === "status") return codexStatusCommand(rest, json); + if (sub === "list-threads") return codexListCommand(rest, json); + if (sub === "queue") return codexQueueCommand(rest, json); + if (sub === "send") return codexSendCommand(rest, json); + throw new StoreError("gbot codex status | list-threads [--limit N] [--cursor CURSOR] | queue | send [envelope flags] [--when-busy reject|queue] "); } async function main(argv) { @@ -547,15 +620,35 @@ async function main(argv) { } if (cmd === "send") { - const ref = sub; - if (hasFlag(rest, "--json")) { json = true; jsonErrors = true; } - const message = rest.join(" ").trim(); - if (!ref || !message) throw new StoreError("gbot send "); - const out = await backend.send(ref, message); + // Envelope flags precede the target: `gbot send --reply-to M --hop 1 `. + const sendArgs = [sub, ...rest].filter((a) => a !== undefined); + // Trailing --json is a flag; `--` protects a message that ends with one. + if (takeTrailingFlag(sendArgs, "--json")) { json = true; jsonErrors = true; } + const { envelope } = takeEnvelopeFlags(sendArgs); + const ref = sendArgs.shift(); + if (sendArgs[0] === "--") sendArgs.shift(); + const message = sendArgs.join(" ").trim(); + if (!ref || ref.startsWith("-") || !message) throw new StoreError("gbot send [envelope flags] "); + const out = await backend.send(ref, withEnvelopeHeader(message, envelope)); saveHistory(out, { dir: historyDir, disabled: noHistory, event: "send", prompt: message }); + // The gateway's message id (when returned) is the delivery receipt; the local envelope id + // is the correlation handle a reply quotes back with --reply-to. 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); + 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 } : {}), + envelopeId: envelope.messageId, + correlationId: envelope.correlationId, + ...(envelope.replyTo ? { replyTo: envelope.replyTo } : {}), + hop: envelope.hop, + maxHops: envelope.maxHops, + }); + } else print("Sent to " + (out.target.isGroup ? "group" : "bot") + " " + out.target.name + " (" + out.target.id + ")" + receipt + "; envelope " + envelope.messageId); return; } diff --git a/src/codex-bridge.js b/src/codex-bridge.js index 084f246..1b241fb 100644 --- a/src/codex-bridge.js +++ b/src/codex-bridge.js @@ -1,7 +1,7 @@ -import { createHash, randomBytes } from "node:crypto"; +import { createHash, randomBytes, randomUUID } from "node:crypto"; import { statSync } from "node:fs"; import { createConnection } from "node:net"; -import { homedir } from "node:os"; +import { homedir, hostname } from "node:os"; import { join } from "node:path"; import { spawnSync } from "node:child_process"; import { createRequire } from "node:module"; @@ -29,14 +29,29 @@ export function codexSocketPath(env = process.env) { return join(home, "app-server-control", "app-server-control.sock"); } -export function socketPresent(path) { +/** `socket` | `absent` | `permission-denied` | `not-a-socket`; permission failures are not absence. */ +export function socketState(path) { try { - return statSync(path).isSocket(); - } catch { - return false; + return statSync(path).isSocket() ? "socket" : "not-a-socket"; + } catch (err) { + return err && (err.code === "EACCES" || err.code === "EPERM") ? "permission-denied" : "absent"; } } +/** Strip ANSI/OSC sequences and C0/C1 controls (tab and newline stay) from server-supplied text. */ +export function stripTerminalControls(text) { + return String(text) + .replace(/\u001b\[[0-9;?]*[ -/]*[@-~]/g, "") + .replace(/\u001b\][^\u0007\u001b]*(?:\u0007|\u001b\\)/g, "") + .replace(/\u001b./g, "") + .replace(/[\u0000-\u0008\u000B-\u001F\u007F-\u009F]/g, ""); +} + +/** Single-line fields (ids, names, paths, cursors, header tokens): no line breaks or tabs survive. */ +export function singleLine(text) { + return stripTerminalControls(text).replace(/[\t\n\r\u2028\u2029]+/g, " "); +} + export function unreachableMessage(path) { return [ "No Codex app-server control socket at " + path + ".", @@ -145,16 +160,73 @@ export class CodexRpcError extends Error { } export class CodexSendError extends Error { - constructor(message, { delivery, threadId, turnId, refused } = {}) { + constructor(message, { delivery, threadId, turnId, refused, reason, envelope } = {}) { super(message); this.name = "CodexSendError"; this.delivery = delivery; + if (reason !== undefined) this.reason = reason; if (threadId !== undefined) this.threadId = threadId; if (turnId !== undefined) this.turnId = turnId; if (refused !== undefined) this.refused = refused; + if (envelope !== undefined) this.envelope = envelope; } } +/** The route to the app-server is unavailable; `mode` is a stable machine-readable state. */ +export class CodexRouteError extends Error { + constructor(message, mode) { + super(message); + this.name = "CodexRouteError"; + this.mode = mode; + this.delivery = "rejected"; + this.reason = mode; + } +} + +/** The app-server answered with a shape this pinned schema does not describe. */ +export class CodexProtocolError extends Error { + constructor(method, detail) { + super("Codex app-server returned an unexpected " + method + " response: " + detail + + ". gbot is pinned to app-server schema " + PINNED_CODEX_VERSION + "; run `gbot codex status --json` to compare versions."); + this.name = "CodexProtocolError"; + this.method = method; + this.mode = "bad-response"; + this.reason = "bad-response"; + } +} + +function isObject(value) { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +/** Failures before the WebSocket upgrade completes: the socket is present but unusable. */ +function connectError(err, path) { + const code = err && err.cause && err.cause.code ? err.cause.code : err && err.code; + const message = err && err.message ? err.message : String(err); + if (code === "EACCES" || code === "EPERM" || /\bEACCES\b|\bEPERM\b/.test(message)) { + return new CodexRouteError("Codex app-server control socket at " + path + " refused the connection for this user (" + code + "). " + + "gbot runs as the user who owns CODEX_HOME; check the socket's owner and mode.", "permission-denied"); + } + if (/refused the WebSocket upgrade|violated the WebSocket protocol|handshake headers exceed/.test(message)) { + return new CodexRouteError(message, "handshake-failed"); + } + return new CodexRouteError(message + " (socket present at " + path + ", but no app-server completed the connection).", "connect-failed"); +} + +/** Failures while initializing an upgraded connection: reachable transport, unusable session. */ +function handshakeError(err) { + if (err instanceof CodexProtocolError) return err; + const message = err && err.message ? err.message : String(err); + return new CodexRouteError("Codex app-server accepted the connection but initialize failed: " + message, "handshake-failed"); +} + +/** Failures after a session exists: the request may or may not have been processed. */ +function transportError(err) { + if (err instanceof CodexRpcError || err instanceof CodexProtocolError || err instanceof CodexSendError || err instanceof CodexRouteError) return err; + if (err && typeof err === "object" && err.reason === undefined) err.reason = "transport"; + return err; +} + /** * 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 @@ -356,69 +428,154 @@ function appServerVersion(initResult) { return m ? m[1] : null; } -async function openSession(env = process.env) { - if (process.platform === "win32") throw new Error(windowsUnsupportedMessage()); +/** Throws CodexRouteError when the operator's socket cannot be used; the path comes only from CODEX_HOME. */ +export function assertRoute(path) { + if (process.platform === "win32") throw new CodexRouteError(windowsUnsupportedMessage(), "windows-unsupported"); + const state = socketState(path); + if (state === "socket") return; + if (state === "permission-denied") { + throw new CodexRouteError("Codex app-server control socket at " + path + " exists but this user may not access it. " + + "gbot runs as the user who owns CODEX_HOME; check the socket's owner and mode.", "permission-denied"); + } + if (state === "not-a-socket") { + throw new CodexRouteError(path + " exists but is not a Unix socket; remove the stale file and restart the daemon.", "not-a-socket"); + } + throw new CodexRouteError(unreachableMessage(path), "socket-absent"); +} + +async function openSession(env = process.env, { experimental = false } = {}) { const path = codexSocketPath(env); - if (!socketPresent(path)) throw new Error(unreachableMessage(path)); - const client = await connectCodexAppServer(path); + assertRoute(path); + let client; + try { + client = await connectCodexAppServer(path); + } catch (err) { + throw connectError(err, path); + } let init; try { - init = await client.request("initialize", { clientInfo: { name: "gbot", version: pkg.version } }); + init = await client.request("initialize", { + clientInfo: { name: "gbot", version: pkg.version }, + ...(experimental ? { capabilities: { experimentalApi: true } } : {}), + }); } catch (err) { client.close(); - throw err; + throw handshakeError(err); + } + if (!isObject(init)) { + client.close(); + throw new CodexProtocolError("initialize", "result is not an object"); } client.notify("initialized"); return { client, path, init }; } -export function localCodexVersion() { - const out = spawnSync("codex", ["--version"], { encoding: "utf8" }); +export const CODEX_VERSION_PROBE_TIMEOUT_MS = 3000; + +/** Bounded `codex --version` probe: `{ version, probe }` where probe is ok | missing | timeout | error. */ +export function probeLocalCodexVersion(timeoutMs = CODEX_VERSION_PROBE_TIMEOUT_MS) { + const out = spawnSync("codex", ["--version"], { encoding: "utf8", timeout: timeoutMs }); + if (out.error) { + if (out.error.code === "ENOENT") return { version: null, probe: "missing" }; + if (out.error.code === "ETIMEDOUT") return { version: null, probe: "timeout" }; + return { version: null, probe: "error" }; + } const m = out.status === 0 ? /(\d+\.\d+\.\d+\S*)/.exec(out.stdout) : null; - return m ? m[1] : null; + return m ? { version: m[1], probe: "ok" } : { version: null, probe: "error" }; } +/** + * Status contract: `reachable` is endpoint reachability only. `schema.compatibility` is + * `exact` when the daemon reports the pinned version, otherwise `unverified` (methods + * usually survive upgrades) or `unknown`. Whether ChatGPT Desktop owns any thread is not + * observable from the socket, so `desktopAttached` is always `"unknown"`. + */ export async function codexStatus(env = process.env) { const path = codexSocketPath(env); - const base = { socketPath: path, pinnedVersion: PINNED_CODEX_VERSION, cliVersion: localCodexVersion() }; - if (process.platform === "win32") { - return { ...base, reachable: false, mode: "windows-unsupported", message: windowsUnsupportedMessage() }; - } - if (!socketPresent(path)) { - return { ...base, reachable: false, mode: "socket-absent", message: unreachableMessage(path) }; + const cli = probeLocalCodexVersion(); + const base = { + socketPath: path, + socketState: socketState(path), + pinnedVersion: PINNED_CODEX_VERSION, + cliVersion: cli.version, + cliVersionProbe: cli.probe, + desktopAttached: "unknown", + }; + let session; + try { + session = await openSession(env); + } catch (err) { + if (err instanceof CodexRouteError) return { ...base, reachable: false, mode: err.mode, message: err.message }; + // The endpoint answered; what it said does not match the pinned schema. + if (err instanceof CodexProtocolError) return { ...base, reachable: true, mode: err.mode, message: err.message }; + throw err; } - const { client, init } = await openSession(env); - client.close(); - const daemonVersion = appServerVersion(init); + session.client.close(); + const daemonVersion = appServerVersion(session.init); return { ...base, reachable: true, mode: "daemon", daemonVersion, - codexHome: init.codexHome ?? null, + codexHome: typeof session.init.codexHome === "string" ? session.init.codexHome : null, + schema: { + pinned: PINNED_CODEX_VERSION, + daemon: daemonVersion, + compatibility: daemonVersion == null ? "unknown" : daemonVersion === PINNED_CODEX_VERSION ? "exact" : "unverified", + }, versionMismatch: Boolean(base.cliVersion && daemonVersion && base.cliVersion !== daemonVersion), }; } +const THREAD_STATUSES = new Set(["notLoaded", "idle", "active", "systemError"]); + +function lineField(value) { + return value == null ? null : singleLine(value); +} + +/** Server text fields are sanitized; structured values (Codex's `source: { custom }`) pass through untouched. */ +function sourceField(value) { + if (value == null) return null; + return typeof value === "string" ? singleLine(value) : value; +} + export function summarizeThread(t) { + if (!isObject(t) || typeof t.id !== "string" || !t.id) throw new CodexProtocolError("thread/list", "entry without a string `id`"); + const type = isObject(t.status) && typeof t.status.type === "string" ? t.status.type : "unknown"; return { - id: t.id, - status: t.status && t.status.type ? t.status.type : "unknown", - name: t.name ?? null, - preview: t.preview ?? "", - cwd: t.cwd ?? null, - source: t.source ?? null, - updatedAt: t.updatedAt ?? null, + id: singleLine(t.id), + status: THREAD_STATUSES.has(type) ? type : "unknown", + activeFlags: isObject(t.status) && Array.isArray(t.status.activeFlags) ? t.status.activeFlags.map((f) => singleLine(f)) : [], + name: lineField(t.name), + preview: typeof t.preview === "string" ? stripTerminalControls(t.preview) : "", + cwd: lineField(t.cwd), + source: sourceField(t.source), + updatedAt: typeof t.updatedAt === "number" ? t.updatedAt : null, }; } -export async function listCodexThreads({ limit = 20, env = process.env } = {}) { +export const THREAD_LIST_MAX_LIMIT = 200; + +export async function listCodexThreads({ limit = 20, cursor, env = process.env } = {}) { + if (!Number.isInteger(limit) || limit < 1 || limit > THREAD_LIST_MAX_LIMIT) { + throw new RangeError("--limit must be an integer 1-" + THREAD_LIST_MAX_LIMIT); + } + if (cursor !== undefined && (typeof cursor !== "string" || !cursor)) throw new RangeError("--cursor must be a non-empty string"); const { client } = await openSession(env); try { // The default listing rescans every rollout file to repair metadata (26 s on a busy machine); // the state DB already holds what we print. - const out = await client.request("thread/list", { limit, useStateDbOnly: true }); - return { threads: out.data.map(summarizeThread), nextCursor: out.nextCursor ?? null }; + const params = { limit, useStateDbOnly: true, ...(cursor !== undefined ? { cursor } : {}) }; + let out; + try { + out = await client.request("thread/list", params); + } catch (err) { + throw transportError(err); + } + if (!isObject(out) || !Array.isArray(out.data)) throw new CodexProtocolError("thread/list", "missing `data` array"); + if (out.nextCursor != null && typeof out.nextCursor !== "string") throw new CodexProtocolError("thread/list", "`nextCursor` is not a string"); + const threads = out.data.map(summarizeThread); + return { threads, nextCursor: out.nextCursor ?? null, limit }; } finally { client.close(); } @@ -436,8 +593,168 @@ function explainSendError(err, threadId) { return err; } -export async function sendToCodexThread(threadId, text, env = process.env) { - const { client } = await openSession(env); +export const DEFAULT_MAX_HOPS = 4; +const ID_PATTERN = /^[A-Za-z0-9_.:-]{1,128}$/; + +/** + * Delivery envelope: `messageId` names this send, `correlationId` names the conversation it + * belongs to, `replyTo` names the message it answers, `hop` counts agent-to-agent forwards. + * `maxHops` (GROK_BOT_MAX_HOPS) bounds relays: a reply must carry hop = incoming hop + 1, and a + * send at or past the bound is refused, so two agents cannot ack each other forever. + * The textual header is caller-authored provenance for the reader, not authentication. + */ +export function buildEnvelope({ correlationId, replyTo, hop, envelope = false, env = process.env } = {}) { + const maxHops = Number.parseInt(env.GROK_BOT_MAX_HOPS ?? "", 10); + const bound = Number.isInteger(maxHops) && maxHops >= 0 ? maxHops : DEFAULT_MAX_HOPS; + for (const [name, value] of [["--correlation-id", correlationId], ["--reply-to", replyTo]]) { + if (value !== undefined && !ID_PATTERN.test(value)) throw new RangeError(name + " must be 1-128 characters of [A-Za-z0-9_.:-]"); + } + const hopCount = hop === undefined ? 0 : hop; + if (!Number.isInteger(hopCount) || hopCount < 0) throw new RangeError("--hop must be a non-negative integer"); + if (replyTo !== undefined && correlationId === undefined) { + throw new RangeError("--reply-to needs the original --correlation-id so the reply stays in its conversation"); + } + const messageId = randomUUID(); + const out = { + messageId, + correlationId: correlationId ?? messageId, + hop: hopCount, + maxHops: bound, + header: envelope || replyTo !== undefined || hop !== undefined || correlationId !== undefined, + }; + if (replyTo !== undefined) out.replyTo = replyTo; + if (hopCount >= bound) { + throw new CodexSendError( + "Refusing to send: hop " + hopCount + " reaches the relay bound " + bound + " (GROK_BOT_MAX_HOPS). " + + "This message is an agent-to-agent relay that has already been forwarded too many times.", + { delivery: "rejected", reason: "hop-limit", envelope: out }, + ); + } + return out; +} + +/** One-line header a receiving agent can read to reply with `--reply-to` and `--hop N+1`. */ +const identityToken = (value) => String(value ?? "").replace(/[^A-Za-z0-9_.:-]/g, "").slice(0, 64) || "unknown"; + +export function envelopeHeader(envelope, env = process.env) { + const from = identityToken(env.USER || env.USERNAME) + "@" + hostnameSafe(); + const parts = ["msg=" + envelope.messageId, "corr=" + envelope.correlationId]; + if (envelope.replyTo) parts.push("reply-to=" + envelope.replyTo); + parts.push("hop=" + envelope.hop, "from=" + from); + return "[gbot " + parts.join(" ") + "]"; +} + +function hostnameSafe() { + try { + return identityToken(hostname()); + } catch { + return "unknown"; + } +} + +export function withEnvelopeHeader(text, envelope, env = process.env) { + return envelope.header ? envelopeHeader(envelope, env) + "\n" + text : text; +} + +/** Operator-controlled destinations: GROK_BOT_CODEX_THREADS="id,id" restricts `codex send`. */ +export function assertThreadAllowed(threadId, env = process.env) { + const raw = env.GROK_BOT_CODEX_THREADS; + if (raw == null || raw.trim() === "") return; + const allowed = raw.split(",").map((s) => s.trim()).filter(Boolean); + if (!allowed.includes(threadId)) { + throw new CodexSendError( + "Codex thread " + threadId + " is not in GROK_BOT_CODEX_THREADS; the operator allows only: " + allowed.join(", "), + { delivery: "rejected", reason: "route-not-allowed", threadId }, + ); + } +} + +/** + * Busy destinations. In app-server 0.154.0 `turn/start` on a thread with an active turn steers + * that turn instead of queueing (TurnStartParams.turnTrigger: "Ignored when this request steers + * an already-active turn"). gbot never steers or interrupts human work: an `active` thread is + * refused with a `busy` receipt, or, with `whenBusy: "queue"`, handed to the daemon's own queue + * through the experimental `thread/queue/add`. See docs/codex-busy-threads.md. + */ +function threadState(resumed, threadId) { + const status = resumed.thread.status; + const type = isObject(status) && typeof status.type === "string" ? status.type : "unknown"; + if (type === "idle" || type === "notLoaded") return { type, busy: false }; + if (type === "active") { + const flags = Array.isArray(status.activeFlags) ? status.activeFlags.map((f) => singleLine(f)) : []; + return { type, busy: true, flags }; + } + if (type === "systemError") { + throw new CodexSendError("Codex thread " + threadId + " is in systemError state; open it in a Codex client first.", + { delivery: "rejected", reason: "thread-error", threadId }); + } + throw new CodexSendError("Codex thread " + threadId + " reports status " + JSON.stringify(type) + ", which this gbot (pinned to app-server " + + PINNED_CODEX_VERSION + ") does not know; not sending.", { delivery: "rejected", reason: "unknown-status", threadId }); +} + +export function experimentalEnabled(env = process.env) { + return /^(1|true|on)$/i.test(env.GROK_BOT_CODEX_EXPERIMENTAL || ""); +} + +function requireExperimental(env, what) { + if (experimentalEnabled(env)) return; + throw new CodexSendError(what + " uses Codex's experimental app-server API (thread/queue/*), which is off by default. " + + "Set GROK_BOT_CODEX_EXPERIMENTAL=1 to opt in; method names are pinned to Codex " + PINNED_CODEX_VERSION + ".", + { delivery: "rejected", reason: "experimental-disabled" }); +} + +function unsupportedOrRpc(err, method, threadId, envelope) { + if (err instanceof CodexRpcError && err.rpc && err.rpc.code === -32601) { + return new CodexSendError("Codex app-server does not offer " + method + " (daemon predates it, or experimentalApi was not granted). " + + "Upgrade Codex or send without --when-busy queue.", { delivery: "rejected", reason: "unsupported", threadId, envelope }); + } + if (err instanceof CodexRpcError) return new CodexSendError(err.message, { delivery: "rejected", reason: "rejected", threadId, envelope }); + return new CodexSendError("Lost the Codex " + method + " response for thread " + threadId + ": " + ((err && err.message) || err) + + ". Delivery is unknown; list the queue before resending.", { delivery: (err && err.delivery) || "unknown", reason: "transport", threadId, envelope }); +} + +/** Read the daemon's queue for one thread (experimental `thread/queue/list`). */ +export async function listCodexQueue(threadId, { env = process.env, limit = 50, cursor } = {}) { + requireExperimental(env, "gbot codex queue"); + const { client } = await openSession(env, { experimental: true }); + try { + let out; + try { + out = await client.request("thread/queue/list", { threadId, limit, ...(cursor !== undefined ? { cursor } : {}) }); + } catch (err) { + throw unsupportedOrRpc(err, "thread/queue/list", threadId); + } + if (!isObject(out) || !Array.isArray(out.data)) throw new CodexProtocolError("thread/queue/list", "missing `data` array"); + return { + threadId, + queued: out.data.map((q) => ({ + id: isObject(q) && typeof q.id === "string" ? singleLine(q.id) : null, + clientUserMessageId: isObject(q) && typeof q.clientUserMessageId === "string" ? singleLine(q.clientUserMessageId) : null, + text: isObject(q) && Array.isArray(q.input) ? q.input.map((part) => (isObject(part) && typeof part.text === "string" ? part.text : "")).filter(Boolean).join("\n") : "", + })), + nextCursor: typeof out.nextCursor === "string" ? out.nextCursor : null, + }; + } finally { + client.close(); + } +} + +export async function sendToCodexThread(threadId, text, { env = process.env, envelope = buildEnvelope({ env }), whenBusy = "reject" } = {}) { + try { + return await sendToCodexThreadInner(threadId, text, { env, envelope, whenBusy }); + } catch (err) { + // Every receipt names the message, including refusals that never reached the daemon. + if ((err instanceof CodexSendError || err instanceof CodexRouteError || err instanceof CodexProtocolError) && err.envelope === undefined) err.envelope = envelope; + throw err; + } +} + +async function sendToCodexThreadInner(threadId, text, { env, envelope, whenBusy }) { + if (whenBusy !== "reject" && whenBusy !== "queue") throw new RangeError("--when-busy must be reject or queue"); + assertThreadAllowed(threadId, env); + if (whenBusy === "queue") requireExperimental(env, "--when-busy queue"); + const body = withEnvelopeHeader(text, envelope, env); + const { client } = await openSession(env, { experimental: whenBusy === "queue" }); try { let resumed; try { @@ -446,14 +763,62 @@ export async function sendToCodexThread(threadId, text, env = process.env) { if (err instanceof CodexSendError) throw err; throw new CodexSendError(explainSendError(err, threadId).message, { delivery: err instanceof CodexRpcError ? "rejected" : (err && err.delivery) || "unknown", + reason: err instanceof CodexRpcError ? (/no rollout found|thread not found/i.test(String(err.rpc && err.rpc.message)) ? "unknown-thread" + : /active writer/i.test(String(err.rpc && err.rpc.message)) ? "external-owner" : "rejected") : "transport", threadId, + envelope, }); } + if (!isObject(resumed) || !isObject(resumed.thread) || typeof resumed.thread.id !== "string") { + throw new CodexProtocolError("thread/resume", "missing `thread.id`"); + } + const state = threadState(resumed, threadId); + const receiptBase = { + threadId: resumed.thread.id, + threadStatus: state.type, + model: resumed.model, + cwd: resumed.cwd, + approvalPolicy: resumed.approvalPolicy, + messageId: envelope.messageId, + correlationId: envelope.correlationId, + ...(envelope.replyTo ? { replyTo: envelope.replyTo } : {}), + hop: envelope.hop, + maxHops: envelope.maxHops, + }; + if (state.busy && whenBusy === "reject") { + const flags = state.flags.length ? " (" + state.flags.join(", ") + ")" : ""; + throw new CodexSendError( + "Codex thread " + threadId + " has an active turn" + flags + "; sending now would steer that turn. " + + "Wait for it to go idle (`gbot codex list-threads`) and resend, or pass --when-busy queue.", + { delivery: "rejected", reason: "busy", threadId, envelope }, + ); + } + if (state.busy) { + let queued; + try { + queued = await client.request("thread/queue/add", { threadId, clientUserMessageId: envelope.messageId, input: [{ type: "text", text: body }] }); + } catch (err) { + throw unsupportedOrRpc(err, "thread/queue/add", threadId, envelope); + } + const submission = isObject(queued) && isObject(queued.queuedSubmission) && typeof queued.queuedSubmission.id === "string" ? queued.queuedSubmission : null; + if (!submission) { + throw new CodexSendError("Codex app-server sent a malformed thread/queue/add acknowledgment for thread " + threadId + + ". Delivery is unknown; list the queue before resending.", { delivery: "unknown", reason: "bad-response", threadId, envelope }); + } + return { delivery: "queued", ...receiptBase, queuedSubmissionId: submission.id, activeFlags: state.flags }; + } + // ponytail: idle-at-resume then turn/start is a small race with a human starting a turn first; + // the daemon exposes no compare-and-start request. Upgrade path: thread/queue/add + thread/queue/start once stable. // 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 }] }); + turn = await client.request("turn/start", { + threadId, + input: [{ type: "text", text: body }], + clientUserMessageId: envelope.messageId, + turnTrigger: "gbot", + }); } catch (err) { if (err instanceof CodexSendError) throw err; const delivery = err instanceof CodexRpcError ? "rejected" : (err && err.delivery) || "unknown"; @@ -461,14 +826,14 @@ export async function sendToCodexThread(threadId, text, env = process.env) { ? 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 }); + // No blind retry: the receipt carries messageId so the caller can look for it before resending. + throw new CodexSendError(detail, { delivery, reason: delivery === "rejected" ? "rejected" : "transport", threadId, envelope }); } 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 }, + { delivery: "unknown", reason: "bad-response", threadId, envelope }, ); } const freshRefused = client.refused.slice(seenRefused) @@ -478,18 +843,10 @@ export async function sendToCodexThread(threadId, text, env = process.env) { 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) }, + { delivery: "accepted", reason: "approval-refused", threadId, turnId, refused: freshRefused.map((r) => r.method), envelope }, ); } - return { - delivery: "accepted", - threadId: resumed.thread.id, - turnId, - turnStatus: turn.turn.status, - model: resumed.model, - cwd: resumed.cwd, - approvalPolicy: resumed.approvalPolicy, - }; + return { delivery: "accepted", ...receiptBase, turnId, turnStatus: turn.turn.status }; } finally { client.close(); } diff --git a/test/codex-bridge.test.js b/test/codex-bridge.test.js index 980c361..c13dbc5 100644 --- a/test/codex-bridge.test.js +++ b/test/codex-bridge.test.js @@ -1,6 +1,6 @@ import assert from "node:assert/strict"; import { execFile } from "node:child_process"; -import { mkdirSync, mkdtempSync } from "node:fs"; +import { chmodSync, mkdirSync, mkdtempSync, writeFileSync } from "node:fs"; import { createServer } from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -149,17 +149,18 @@ test("codex list-threads passes --limit and prints threads", async () => { try { const text = await gbot(fake.home, "codex", "list-threads", "--limit", "1"); assert.equal(text.code, 0, text.err); - assert.equal(text.out, "t-1 idle - Fix the build\n /repo/a\n please fix the build\n"); + assert.equal(text.out, "t-1 idle - Fix the build\n /repo/a\n please fix the build\n\nmore: --cursor \"cursor-2\"\n"); assert.deepEqual(fake.received.find((m) => m.method === "thread/list").params, { limit: 1, useStateDbOnly: true }); const json = await gbot(fake.home, "--json", "codex", "list-threads"); assert.equal(json.code, 0, json.err); assert.deepEqual(JSON.parse(json.out), { threads: [ - { id: "t-1", status: "idle", name: "Fix the build", preview: "please fix the build", cwd: "/repo/a", source: "vscode", updatedAt: 1700000001 }, - { id: "t-2", status: "notLoaded", name: null, preview: "second thread\npreview", cwd: "/repo/b", source: "cli", updatedAt: 1700000000 }, + { id: "t-1", status: "idle", activeFlags: [], name: "Fix the build", preview: "please fix the build", cwd: "/repo/a", source: "vscode", updatedAt: 1700000001 }, + { id: "t-2", status: "notLoaded", activeFlags: [], name: null, preview: "second thread\npreview", cwd: "/repo/b", source: "cli", updatedAt: 1700000000 }, ], nextCursor: null, + limit: 20, }); assert.deepEqual(fake.received.at(-1).params, { limit: 20, useStateDbOnly: true }); } finally { @@ -170,7 +171,9 @@ test("codex list-threads passes --limit and prints threads", async () => { test("codex list-threads rejects a bad --limit", async () => { const { code, err } = await gbot("/nonexistent", "codex", "list-threads", "--limit", "0"); assert.equal(code, 1); - assert.equal(err, "--limit must be a positive integer\n"); + assert.equal(err, "--limit must be an integer 1-200\n"); + const big = await gbot("/nonexistent", "codex", "list-threads", "--limit", "201"); + assert.equal(big.err, "--limit must be an integer 1-200\n"); }); test("codex send resumes the thread, starts a turn, and prints the ids", async () => { @@ -178,21 +181,35 @@ test("codex send resumes the thread, starts a turn, and prints the ids", async ( try { 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), { + const receipt = JSON.parse(out); + assert.match(receipt.messageId, /^[0-9a-f-]{36}$/); + assert.equal(receipt.correlationId, receipt.messageId); + assert.deepEqual(receipt, { delivery: "accepted", threadId: "t-1", + threadStatus: "idle", turnId: "turn-9", turnStatus: "inProgress", model: "gpt-6", cwd: "/repo/a", approvalPolicy: "never", + messageId: receipt.messageId, + correlationId: receipt.messageId, + hop: 0, + maxHops: 4, }); assert.deepEqual(fake.received.map((m) => m.method), ["initialize", "initialized", "thread/resume", "turn/start"]); assert.deepEqual(fake.received[2].params, { threadId: "t-1", excludeTurns: true }); - assert.deepEqual(fake.received[3].params, { threadId: "t-1", input: [{ type: "text", text: "hello from gbot" }] }); + // Plain sends keep the body verbatim; the id rides Codex's native clientUserMessageId. + assert.deepEqual(fake.received[3].params, { + threadId: "t-1", + input: [{ type: "text", text: "hello from gbot" }], + clientUserMessageId: receipt.messageId, + turnTrigger: "gbot", + }); const text = await gbot(fake.home, "codex", "send", "t-1", "again"); - assert.equal(text.out, "Started turn turn-9 (inProgress) on Codex thread t-1\n"); + assert.match(text.out, /^Started turn turn-9 \(inProgress\) on Codex thread t-1; message [0-9a-f-]{36}\n$/); } finally { await fake.close(); } @@ -246,7 +263,7 @@ test("codex send returns once the turn starts; later approval requests are the d try { const { code, out } = await gbot(fake.home, "codex", "send", "t-1", "go"); assert.equal(code, 0); - assert.equal(out, "Started turn turn-11 (inProgress) on Codex thread t-1\n"); + assert.match(out, /^Started turn turn-11 \(inProgress\) on Codex thread t-1; message /); } finally { await fake.close(); } @@ -364,9 +381,12 @@ test("codex status fails on a wrong handshake and closes the socket instead of l 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"); + const { code, out } = await gbot(home, "--json", "codex", "status"); assert.equal(code, 1); - assert.match(err, /refused the WebSocket upgrade/); + const status = JSON.parse(out); + assert.equal(status.reachable, false); + assert.equal(status.mode, "handshake-failed"); + assert.match(status.message, /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"); @@ -385,9 +405,11 @@ test("codex status fails fast on an oversized frame and settles the pending requ }); try { const started = Date.now(); - const { code, err } = await gbot(fake.home, "codex", "status"); + const { code, out } = await gbot(fake.home, "--json", "codex", "status"); assert.equal(code, 1); - assert.match(err, /exceeds/); + const status = JSON.parse(out); + assert.equal(status.mode, "handshake-failed"); + assert.match(status.message, /exceeds/); assert.ok(Date.now() - started < 5000, "did not wait for the request timeout"); } finally { await fake.close(); @@ -449,7 +471,7 @@ test("codex send preserves turn and thread ids with accepted delivery on refusal const home = fake.home; const env = { ...process.env, CODEX_HOME: home }; try { - await assert.rejects(sendToCodexThread("t-1", "do it", env), (e) => { + 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"); @@ -471,9 +493,11 @@ test("a JSON null message fails as malformed instead of crashing on msg.id", asy }); try { const started = Date.now(); - const { code, err } = await gbot(fake.home, "codex", "status"); + const { code, out } = await gbot(fake.home, "--json", "codex", "status"); assert.equal(code, 1); - assert.match(err, /malformed message/); + const status = JSON.parse(out); + assert.equal(status.mode, "handshake-failed"); + assert.match(status.message, /malformed message/); assert.ok(Date.now() - started < 5000, "did not hang on the bad message"); } finally { await fake.close(); @@ -512,9 +536,10 @@ test("codex status rejects terminated oversized handshake headers", async () => await new Promise((resolve) => server.listen(socketPath, resolve)); try { const started = Date.now(); - const { code, err } = await gbot(home, "codex", "status"); + const { code, out } = await gbot(home, "codex", "status"); assert.equal(code, 1); - assert.match(err, /exceed/); + assert.match(out, /reachable: no \(handshake-failed\)/); + assert.match(out, /exceed/); assert.ok(Date.now() - started < 5000, "failed fast instead of decoding the headers"); } finally { await new Promise((resolve) => server.close(resolve)); @@ -531,9 +556,10 @@ test("codex status rejects a complete oversized frame without buffering it", asy }); try { const started = Date.now(); - const { code, err } = await gbot(fake.home, "codex", "status"); + const { code, out } = await gbot(fake.home, "codex", "status"); assert.equal(code, 1); - assert.match(err, /exceed/); + assert.match(out, /reachable: no \(handshake-failed\)/); + assert.match(out, /exceed/); assert.ok(Date.now() - started < 5000, "failed fast instead of buffering the frame"); } finally { await fake.close(); @@ -544,10 +570,16 @@ test("codex send emits structured JSON errors with delivery and ids", async () = try { const unknown = await gbot(fake.home, "--json", "codex", "send", "nope", "hi"); assert.equal(unknown.code, 1); - assert.deepEqual(JSON.parse(unknown.err), { + const failure = JSON.parse(unknown.err); + assert.match(failure.messageId, /^[0-9a-f-]{36}$/); + assert.deepEqual(failure, { error: "Unknown Codex thread nope. Run `gbot codex list-threads` to see reachable threads.", delivery: "rejected", + reason: "unknown-thread", threadId: "nope", + messageId: failure.messageId, + correlationId: failure.messageId, + hop: 0, }); assert.equal(unknown.out, ""); } finally { @@ -608,11 +640,11 @@ test("codex send failures honor a trailing --json with structured errors", async 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", - }); + const failure = JSON.parse(err); + assert.equal(failure.reason, "unknown-thread"); + assert.equal(failure.delivery, "rejected"); + assert.equal(failure.threadId, "nope"); + assert.match(failure.messageId, /^[0-9a-f-]{36}$/); } finally { await fake.close(); } @@ -630,3 +662,485 @@ test("send failures honor a trailing --json before backend auth runs", async () const parsed = JSON.parse(err); assert.equal(typeof parsed.error, "string"); }); + +// ---- #39: machine-readable status and thread discovery ---- + +test("codex status distinguishes permission-denied and stale files from an absent socket", { skip: process.platform === "win32" || process.getuid?.() === 0 }, async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-perm-")); + mkdirSync(join(home, "app-server-control"), { mode: 0o000 }); + try { + const denied = await gbot(home, "--json", "codex", "status"); + assert.equal(denied.code, 1); + const status = JSON.parse(denied.out); + assert.equal(status.reachable, false); + assert.equal(status.mode, "permission-denied"); + assert.equal(status.socketState, "permission-denied"); + assert.match(status.message, /may not access it/); + } finally { + chmodSync(join(home, "app-server-control"), 0o700); + } + writeFileSync(join(home, "app-server-control", "app-server-control.sock"), "stale"); + const stale = await gbot(home, "--json", "codex", "status"); + assert.equal(JSON.parse(stale.out).mode, "not-a-socket"); + assert.equal(JSON.parse(stale.out).socketState, "not-a-socket"); +}); + +test("codex status reports connect-failed when the socket exists but nothing answers", async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-dead-")); + mkdirSync(join(home, "app-server-control")); + const socketPath = join(home, "app-server-control", "app-server-control.sock"); + const server = createTcpServer((sock) => sock.destroy()); + await new Promise((resolve) => server.listen(socketPath, resolve)); + try { + const { code, out } = await gbot(home, "--json", "codex", "status"); + assert.equal(code, 1); + const status = JSON.parse(out); + assert.equal(status.reachable, false); + assert.equal(status.socketState, "socket"); + assert.equal(status.mode, "connect-failed"); + assert.equal(status.desktopAttached, "unknown"); + } finally { + await new Promise((resolve) => server.close(resolve)); + } +}); + +test("codex status separates schema compatibility from reachability and bounds the CLI probe", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + initialize: (params, ok) => ok({ userAgent: "codex/0.160.0 (linux) client (" + params.clientInfo.name + ")", codexHome: 7 }), + }); + try { + const { code, out } = await gbot(fake.home, "--json", "codex", "status"); + assert.equal(code, 0, out); + const status = JSON.parse(out); + assert.equal(status.reachable, true); + assert.deepEqual(status.schema, { pinned: "0.154.0", daemon: "0.160.0", compatibility: "unverified" }); + assert.equal(status.codexHome, null, "non-string codexHome is not passed through"); + assert.equal(status.cliVersionProbe, "missing", "PATH has no codex binary"); + assert.equal(status.desktopAttached, "unknown"); + const text = await gbot(fake.home, "codex", "status"); + assert.match(text.out, /pinned schema: 0\.154\.0 \(unverified\)/); + assert.match(text.out, /desktop attached: unknown/); + } finally { + await fake.close(); + } +}); + +test("codex status times out a hung codex --version probe instead of hanging", async () => { + const bin = mkdtempSync(join(tmpdir(), "gbot-codex-bin-")); + writeFileSync(join(bin, "codex"), "#!/bin/sh\nexec /bin/sleep 30\n", { mode: 0o755 }); + const home = mkdtempSync(join(tmpdir(), "gbot-codex-nosock-")); + const started = Date.now(); + const result = await new Promise((resolve) => { + execFile(process.execPath, [CLI, "--json", "codex", "status"], { encoding: "utf8", env: { ...process.env, CODEX_HOME: home, PATH: bin } }, (error, out) => resolve({ code: error ? error.code : 0, out })); + }); + assert.ok(Date.now() - started < 10000, "probe was bounded"); + const status = JSON.parse(result.out); + assert.equal(status.cliVersion, null); + assert.equal(status.cliVersionProbe, "timeout"); +}); + +test("codex status rejects unknown arguments before touching the socket", async () => { + const { code, err } = await gbot("/nonexistent", "codex", "status", "--verbose"); + assert.equal(code, 1); + assert.match(err, /^Unknown argument "--verbose"\. Usage: gbot codex status/); +}); + +test("codex list-threads pages with --cursor, echoes nextCursor, and rejects unknown args", async () => { + const seen = []; + const fake = await fakeAppServer({ + ...baseHandlers, + "thread/list": (params, ok) => { + seen.push(params); + if (params.cursor === "page-2") return ok({ data: [THREADS[1]], nextCursor: null }); + ok({ data: [THREADS[0]], nextCursor: "page-2\u001b[31m" }); + }, + }); + try { + const first = await gbot(fake.home, "--json", "codex", "list-threads", "--limit", "1"); + assert.equal(first.code, 0, first.err); + const page = JSON.parse(first.out); + assert.equal(page.nextCursor, "page-2\u001b[31m", "JSON keeps the opaque cursor verbatim"); + assert.equal(page.limit, 1); + const text = await gbot(fake.home, "codex", "list-threads", "--limit", "1"); + assert.match(text.out, /more: --cursor "page-2"\n$/, "text output strips controls from the cursor and quotes it"); + assert.doesNotMatch(text.out, /\u001b/); + + const second = await gbot(fake.home, "--json", "codex", "list-threads", "--limit", "1", "--cursor", "page-2"); + assert.deepEqual(JSON.parse(second.out).threads.map((t) => t.id), ["t-2"]); + assert.equal(JSON.parse(second.out).nextCursor, null); + assert.deepEqual(seen.at(-1), { limit: 1, useStateDbOnly: true, cursor: "page-2" }); + + const unknown = await gbot(fake.home, "codex", "list-threads", "--all"); + assert.equal(unknown.code, 1); + assert.match(unknown.err, /^Unknown argument "--all"/); + const empty = await gbot(fake.home, "codex", "list-threads", "--cursor"); + assert.equal(empty.err, "--cursor needs a value\n"); + } finally { + await fake.close(); + } +}); + +test("codex list-threads fails with bad-response when the daemon returns an unknown shape", async () => { + const fake = await fakeAppServer({ ...baseHandlers, "thread/list": (params, ok) => ok({ threads: [] }) }); + try { + const { code, err } = await gbot(fake.home, "--json", "codex", "list-threads"); + assert.equal(code, 1); + const failure = JSON.parse(err); + assert.equal(failure.reason, "bad-response"); + assert.equal(failure.mode, "bad-response"); + assert.match(failure.error, /unexpected thread\/list response: missing `data` array/); + assert.match(failure.error, /pinned to app-server schema 0\.154\.0/); + } finally { + await fake.close(); + } +}); + +test("codex list-threads sanitizes text fields, keeps structured source, and rejects idless entries", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + "thread/list": (params, ok) => ok({ + data: [ + { id: "t-\u001b[2Jx\nline", status: { type: "active", activeFlags: ["waitingOnApproval\u0007"] }, name: "a\tb", preview: "keep\nnewlines\u001b[31m", cwd: "/r\u001b]0;x\u0007", source: { custom: "hauler\u001b[0m" }, updatedAt: "soon" }, + { id: "t-odd", status: { type: "somethingNew" }, source: "cli\r" }, + ], + nextCursor: null, + }), + }); + try { + const { code, out } = await gbot(fake.home, "--json", "codex", "list-threads"); + assert.equal(code, 0); + const { threads } = JSON.parse(out); + assert.deepEqual(threads, [ + { id: "t-x line", status: "active", activeFlags: ["waitingOnApproval"], name: "a b", preview: "keep\nnewlines", cwd: "/r", source: { custom: "hauler\u001b[0m" }, updatedAt: null }, + { id: "t-odd", status: "unknown", activeFlags: [], name: null, preview: "", cwd: null, source: "cli", updatedAt: null }, + ]); + } finally { + await fake.close(); + } + const broken = await fakeAppServer({ ...baseHandlers, "thread/list": (params, ok) => ok({ data: [{ status: { type: "idle" } }], nextCursor: null }) }); + try { + const { code, err } = await gbot(broken.home, "--json", "codex", "list-threads"); + assert.equal(code, 1); + assert.equal(JSON.parse(err).reason, "bad-response"); + assert.match(JSON.parse(err).error, /entry without a string `id`/); + } finally { + await broken.close(); + } +}); + +// ---- #37: attributable routes and loop-safe correlation ---- + +test("codex send carries correlation and reply metadata, prepends the header, and bounds hops", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const reply = await gbot(fake.home, "--json", "codex", "send", "--correlation-id", "corr-1", "--reply-to", "msg-0", "--hop", "1", "t-1", "on it", "--dir", "src"); + assert.equal(reply.code, 0, reply.err); + const receipt = JSON.parse(reply.out); + assert.equal(receipt.correlationId, "corr-1"); + assert.equal(receipt.replyTo, "msg-0"); + assert.equal(receipt.hop, 1); + assert.notEqual(receipt.messageId, "corr-1"); + const start = fake.received.findLast((m) => m.method === "turn/start"); + assert.equal(start.params.clientUserMessageId, receipt.messageId); + const [header, ...body] = start.params.input[0].text.split("\n"); + assert.match(header, new RegExp("^\\[gbot msg=" + receipt.messageId + " corr=corr-1 reply-to=msg-0 hop=1 from=[^\\s\\]]+\\]$")); + assert.equal(body.join("\n"), "on it --dir src", "mid-message tokens stay message content"); + + const looped = await gbot(fake.home, "--json", "codex", "send", "--correlation-id", "corr-1", "--reply-to", "msg-3", "--hop", "4", "t-1", "ack"); + assert.equal(looped.code, 1); + const refusal = JSON.parse(looped.err); + assert.equal(refusal.delivery, "rejected"); + assert.equal(refusal.reason, "hop-limit"); + assert.equal(refusal.correlationId, "corr-1"); + assert.match(refusal.error, /relay bound 4 \(GROK_BOT_MAX_HOPS\)/); + assert.equal(fake.received.filter((m) => m.method === "turn/start").length, 1, "hop-limit refusals never reach the daemon"); + + const orphan = await gbot(fake.home, "codex", "send", "--reply-to", "msg-0", "t-1", "hi"); + assert.equal(orphan.code, 1); + assert.match(orphan.err, /--reply-to needs the original --correlation-id/); + const badId = await gbot(fake.home, "codex", "send", "--correlation-id", "has space", "t-1", "hi"); + assert.match(badId.err, /--correlation-id must be 1-128 characters/); + const badHop = await gbot(fake.home, "codex", "send", "--hop", "-1", "t-1", "hi"); + assert.match(badHop.err, /--hop must be a non-negative integer/); + } finally { + await fake.close(); + } +}); + +test("GROK_BOT_MAX_HOPS and --envelope are honored; operator allowlist rejects other threads", async () => { + const fake = await fakeAppServer(baseHandlers); + const run = (env, ...args) => new Promise((resolve) => { + execFile(process.execPath, [CLI, ...args], { encoding: "utf8", env: { ...process.env, CODEX_HOME: fake.home, PATH: "/nonexistent", ...env } }, (error, out, err) => resolve({ code: error ? error.code : 0, out, err })); + }); + try { + const tight = await run({ GROK_BOT_MAX_HOPS: "1" }, "--json", "codex", "send", "--hop", "1", "t-1", "x"); + assert.equal(JSON.parse(tight.err).reason, "hop-limit"); + const loose = await run({ GROK_BOT_MAX_HOPS: "9" }, "--json", "codex", "send", "--hop", "8", "t-1", "x"); + assert.equal(loose.code, 0, loose.err); + assert.equal(JSON.parse(loose.out).maxHops, 9); + + const enveloped = await run({}, "--json", "codex", "send", "--envelope", "t-1", "plain"); + assert.equal(enveloped.code, 0, enveloped.err); + const start = fake.received.findLast((m) => m.method === "turn/start"); + assert.match(start.params.input[0].text, /^\[gbot msg=[0-9a-f-]{36} corr=[0-9a-f-]{36} hop=0 from=[^\]]+\]\nplain$/); + + const blocked = await run({ GROK_BOT_CODEX_THREADS: "t-9, t-8" }, "--json", "codex", "send", "t-1", "x"); + assert.equal(blocked.code, 1); + const refusal = JSON.parse(blocked.err); + assert.equal(refusal.reason, "route-not-allowed"); + assert.equal(refusal.delivery, "rejected"); + assert.match(refusal.error, /operator allows only: t-9, t-8/); + const before = fake.received.length; + const allowed = await run({ GROK_BOT_CODEX_THREADS: "t-1" }, "--json", "codex", "send", "t-1", "x"); + assert.equal(allowed.code, 0, allowed.err); + assert.ok(fake.received.length > before); + } finally { + await fake.close(); + } +}); + +test("codex send reports an unavailable route as a structured rejected receipt", async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-noroute-")); + const { code, err } = await gbot(home, "--json", "codex", "send", "t-1", "hello"); + assert.equal(code, 1); + const failure = JSON.parse(err); + assert.equal(failure.delivery, "rejected"); + assert.equal(failure.reason, "socket-absent"); + assert.equal(failure.mode, "socket-absent"); + assert.match(failure.error, /No Codex app-server control socket/); +}); + +// ---- #38: busy-thread delivery ---- + +test("codex send refuses active and systemError threads without steering them", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + "thread/resume": (params, ok) => ok({ + thread: { + id: params.threadId, + status: params.threadId === "t-busy" ? { type: "active", activeFlags: ["waitingOnUserInput"] } : { type: "systemError" }, + }, + model: "gpt-6", + cwd: "/repo", + approvalPolicy: "never", + }), + }); + try { + const busy = await gbot(fake.home, "--json", "codex", "send", "t-busy", "hi"); + assert.equal(busy.code, 1); + const refusal = JSON.parse(busy.err); + assert.equal(refusal.delivery, "rejected"); + assert.equal(refusal.reason, "busy"); + assert.equal(refusal.threadId, "t-busy"); + assert.match(refusal.error, /active turn \(waitingOnUserInput\); sending now would steer that turn/); + + const broken = await gbot(fake.home, "--json", "codex", "send", "t-err", "hi"); + assert.equal(JSON.parse(broken.err).reason, "thread-error"); + assert.equal(fake.received.filter((m) => m.method === "turn/start").length, 0, "no turn/start, turn/steer, or turn/interrupt was sent"); + assert.equal(fake.received.filter((m) => /steer|interrupt/.test(m.method)).length, 0); + } finally { + await fake.close(); + } +}); + +test("codex send distinguishes external-owner, unknown-thread, and transport reasons", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const owned = JSON.parse((await gbot(fake.home, "--json", "codex", "send", "t-2", "hi")).err); + assert.equal(owned.reason, "external-owner"); + assert.equal(owned.delivery, "rejected"); + const unknown = JSON.parse((await gbot(fake.home, "--json", "codex", "send", "nope", "hi")).err); + assert.equal(unknown.reason, "unknown-thread"); + } finally { + await fake.close(); + } + const dropping = await fakeAppServer({ + ...baseHandlers, + "turn/start": (params, ok, err, send, socket) => socket.destroy(), + }); + try { + const lost = JSON.parse((await gbot(dropping.home, "--json", "codex", "send", "t-1", "hi")).err); + assert.equal(lost.delivery, "unknown"); + assert.equal(lost.reason, "transport"); + assert.match(lost.messageId, /^[0-9a-f-]{36}$/, "the receipt names the message to look for before resending"); + } finally { + await dropping.close(); + } +}); + +// ---- review follow-ups: classification, envelopes, injection, queue ---- + +test("codex status classifies initialize failures as handshake-failed and off-schema init as bad-response", async () => { + const rejecting = await fakeAppServer({ initialize: (params, ok, err) => err({ code: -32000, message: "nope" }) }); + try { + const status = JSON.parse((await gbot(rejecting.home, "--json", "codex", "status")).out); + assert.equal(status.reachable, false); + assert.equal(status.mode, "handshake-failed"); + assert.match(status.message, /initialize failed: Codex app-server rejected initialize: nope/); + } finally { + await rejecting.close(); + } + const offSchema = await fakeAppServer({ initialize: (params, ok) => ok("not-an-object") }); + try { + const { code, out } = await gbot(offSchema.home, "--json", "codex", "status"); + assert.equal(code, 1); + const status = JSON.parse(out); + assert.equal(status.reachable, true, "the endpoint answered"); + assert.equal(status.mode, "bad-response"); + } finally { + await offSchema.close(); + } +}); + +test("codex status reports permission-denied when connect fails with EACCES", { skip: process.platform === "win32" || process.getuid?.() === 0 }, async () => { + const home = mkdtempSync(join(tmpdir(), "gbot-codex-eacces-")); + mkdirSync(join(home, "app-server-control")); + const socketPath = join(home, "app-server-control", "app-server-control.sock"); + const server = createTcpServer((sock) => sock.destroy()); + await new Promise((resolve) => server.listen(socketPath, resolve)); + chmodSync(socketPath, 0o000); + try { + const { code, out } = await gbot(home, "--json", "codex", "status"); + assert.equal(code, 1); + const status = JSON.parse(out); + assert.equal(status.socketState, "socket"); + assert.equal(status.mode, "permission-denied"); + } finally { + chmodSync(socketPath, 0o600); + await new Promise((resolve) => server.close(resolve)); + } +}); + +test("every codex send rejection carries the envelope, and unknown statuses are refused", async () => { + const fake = await fakeAppServer({ + ...baseHandlers, + "thread/resume": (params, ok) => ok({ thread: { id: params.threadId, status: { type: "hibernating" } }, model: "m", cwd: "/", approvalPolicy: "never" }), + }); + const run = (env, ...args) => new Promise((resolve) => { + execFile(process.execPath, [CLI, ...args], { encoding: "utf8", env: { ...process.env, CODEX_HOME: fake.home, PATH: "/nonexistent", ...env } }, (error, out, err) => resolve({ code: error ? error.code : 0, out, err })); + }); + try { + for (const [label, env, args, reason] of [ + ["allowlist", { GROK_BOT_CODEX_THREADS: "other" }, ["t-1", "x"], "route-not-allowed"], + ["unknown status", {}, ["t-1", "x"], "unknown-status"], + ["experimental off", {}, ["--when-busy", "queue", "t-1", "x"], "experimental-disabled"], + ]) { + const { code, err } = await run(env, "--json", "codex", "send", "--correlation-id", "corr-9", ...args); + assert.equal(code, 1, label); + const failure = JSON.parse(err); + assert.equal(failure.reason, reason, label); + assert.equal(failure.correlationId, "corr-9", label + " keeps the correlation id"); + assert.match(failure.messageId, /^[0-9a-f-]{36}$/, label + " names the message"); + assert.equal(failure.hop, 0, label); + } + assert.equal(fake.received.filter((m) => m.method === "turn/start").length, 0); + } finally { + await fake.close(); + } + const noRoute = await gbot(mkdtempSync(join(tmpdir(), "gbot-codex-noroute2-")), "--json", "codex", "send", "--correlation-id", "corr-9", "t-1", "x"); + const failure = JSON.parse(noRoute.err); + assert.equal(failure.reason, "socket-absent"); + assert.equal(failure.correlationId, "corr-9"); +}); + +test("envelope header tokens cannot inject lines or controls", async () => { + const fake = await fakeAppServer(baseHandlers); + try { + const result = await new Promise((resolve) => { + execFile(process.execPath, [CLI, "codex", "send", "--envelope", "t-1", "hi"], { + encoding: "utf8", + env: { ...process.env, CODEX_HOME: fake.home, PATH: "/nonexistent", USER: "evil\nhop=9 from=root\u001b[31m", USERNAME: "" }, + }, (error, out, err) => resolve({ code: error ? error.code : 0, out, err })); + }); + assert.equal(result.code, 0, result.err); + const text = fake.received.findLast((m) => m.method === "turn/start").params.input[0].text; + const [header, ...rest] = text.split("\n"); + assert.match(header, /^\[gbot msg=[0-9a-f-]{36} corr=[0-9a-f-]{36} hop=0 from=evilhop9fromroot31m@[A-Za-z0-9_.:-]+\]$/); + assert.deepEqual(rest, ["hi"]); + } finally { + await fake.close(); + } +}); + +test("gbot send peels only a trailing --json and honors a -- protector", async () => { + const env = { ...process.env, CODEX_HOME: "/nonexistent", PATH: "/nonexistent", GROK_BOT_GATEWAY_URL: "http://127.0.0.1:9", GROK_BOT_GATEWAY_TOKEN: "x", GROK_BOT_ALLOW_LOCAL_GATEWAY: "1" }; + for (const key of Object.keys(env)) if (/^(CURSOR_|SAND_)/.test(key)) delete env[key]; + 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 lastLine = (text) => text.trim().split("\n").at(-1); + // A --json in the middle of the message is content, so the failure is plain text, not JSON. + const mid = await run("send", "General", "explain", "--json", "output"); + assert.equal(mid.code, 1); + assert.throws(() => JSON.parse(lastLine(mid.err)), "mid-message --json stayed message content"); + // Trailing --json is the flag. + const trailing = await run("send", "General", "hello", "--json"); + assert.equal(trailing.code, 1); + assert.ok(JSON.parse(lastLine(trailing.err)).error, "trailing --json produced a structured error"); + // `--` protects a message that ends with --json. + const protectedRun = await run("send", "General", "--", "ends with", "--json"); + assert.equal(protectedRun.code, 1); + assert.throws(() => JSON.parse(lastLine(protectedRun.err)), "protected --json stayed message content"); +}); + +test("codex list-threads accepts an opaque cursor that starts with a dash", async () => { + const seen = []; + const fake = await fakeAppServer({ ...baseHandlers, "thread/list": (params, ok) => { seen.push(params); ok({ data: [], nextCursor: null }); } }); + try { + const { code } = await gbot(fake.home, "codex", "list-threads", "--cursor", "-abc=="); + assert.equal(code, 0); + assert.equal(seen.at(-1).cursor, "-abc=="); + } finally { + await fake.close(); + } +}); + +test("--when-busy queue hands a busy thread to the daemon queue under the experimental gate", async () => { + const inits = []; + const fake = await fakeAppServer({ + ...baseHandlers, + initialize: (params, ok) => { inits.push(params); ok({ userAgent: "codex/0.154.0 (x) y (gbot)", codexHome: "/fake" }); }, + "thread/resume": (params, ok) => ok({ thread: { id: params.threadId, status: { type: "active", activeFlags: [] } }, model: "m", cwd: "/", approvalPolicy: "never" }), + "thread/queue/add": (params, ok) => ok({ queuedSubmission: { id: "q-1", clientUserMessageId: params.clientUserMessageId, input: params.input } }), + "thread/queue/list": (params, ok) => ok({ data: [{ id: "q-1", clientUserMessageId: "m-1", input: [{ type: "text", text: "queued\u001b[31m body" }] }], nextCursor: null }), + }); + const run = (...args) => new Promise((resolve) => { + execFile(process.execPath, [CLI, ...args], { encoding: "utf8", env: { ...process.env, CODEX_HOME: fake.home, PATH: "/nonexistent", GROK_BOT_CODEX_EXPERIMENTAL: "1" } }, (error, out, err) => resolve({ code: error ? error.code : 0, out, err })); + }); + try { + const { code, out } = await run("--json", "codex", "send", "--when-busy", "queue", "t-1", "later please"); + assert.equal(code, 0, out); + const receipt = JSON.parse(out); + assert.equal(receipt.delivery, "queued"); + assert.equal(receipt.queuedSubmissionId, "q-1"); + assert.equal(receipt.threadStatus, "active"); + assert.match(receipt.messageId, /^[0-9a-f-]{36}$/); + assert.deepEqual(inits.at(-1).capabilities, { experimentalApi: true }); + const add = fake.received.find((m) => m.method === "thread/queue/add"); + assert.equal(add.params.clientUserMessageId, receipt.messageId); + assert.equal(add.params.input[0].text, "later please"); + assert.equal(fake.received.filter((m) => /turn\/(start|steer|interrupt)/.test(m.method)).length, 0); + + const queue = await run("--json", "codex", "queue", "t-1"); + assert.equal(queue.code, 0, queue.err); + assert.deepEqual(JSON.parse(queue.out), { threadId: "t-1", queued: [{ id: "q-1", clientUserMessageId: "m-1", text: "queued\u001b[31m body" }], nextCursor: null }); + const text = await run("codex", "queue", "t-1"); + assert.doesNotMatch(text.out, /\u001b/); + } finally { + await fake.close(); + } + const old = await fakeAppServer({ + ...baseHandlers, + "thread/resume": (params, ok) => ok({ thread: { id: params.threadId, status: { type: "active", activeFlags: [] } }, model: "m", cwd: "/", approvalPolicy: "never" }), + }); + try { + const { code, err } = await new Promise((resolve) => { + execFile(process.execPath, [CLI, "--json", "codex", "send", "--when-busy", "queue", "t-1", "x"], { encoding: "utf8", env: { ...process.env, CODEX_HOME: old.home, PATH: "/nonexistent", GROK_BOT_CODEX_EXPERIMENTAL: "1" } }, (error, out, err) => resolve({ code: error ? error.code : 0, out, err })); + }); + assert.equal(code, 1); + assert.equal(JSON.parse(err).reason, "unsupported"); + } finally { + await old.close(); + } +});