diff --git a/DESIGN.md b/DESIGN.md new file mode 100644 index 0000000000..c8f6cab833 --- /dev/null +++ b/DESIGN.md @@ -0,0 +1,73 @@ +--- +version: alpha +colors: + defaultInk: "#fcfcfc" + defaultSecondaryInk: "#fcfcfc99" + defaultInset: "#191919" + defaultHairline: "#333333" + primary: "#1084fe" +typography: + body: + fontFamily: 'Inter, -apple-system, BlinkMacSystemFont, "SF Pro Text", "Segoe UI", system-ui, sans-serif' +omitted: + - section: rounded + reason: Existing Tailwind component classes own geometry. + - section: spacing + reason: Existing chat and settings layouts own spacing. +--- + +## Overview + +OpenMausBot is a local chat workbench. New thread controls should feel like the +existing model and approval controls: compact, quiet, and explicit about what +will run. Connection status and the actor behind a reply carry meaning; they +are not decorative badges. + +## Colors + +`src/styles.css` is the canonical runtime token source. The frontmatter records +the default dark skin only. All skins override semantic tokens there. Components +use `ink`, `ink-secondary`, `inset`, `raised`, `hairline`, `accent`, and `danger` +utilities rather than copying these values. This feature changes no skin tokens. + +## Typography + +Inherit the current skin's sans font. Settings labels use the existing 12–13px +scale; exact conversation and delivery identifiers use the system monospace +stack and wrap instead of widening the chat. Reply text retains the normal chat +renderer and its locale-aware font fallbacks. + +## Layout + +Thread connection controls are an inline disclosure above the transcript. They +wrap at narrow widths, reserve a status line, and keep long result lists inside +a bounded scroll area. They do not create a second conversation window. + +## Elevation & Depth + +Use existing inset surfaces and hairline borders. Static connection information +has no new shadows or animation. + +## Shapes + +Reuse the existing field and rounded-lg action geometry. Keep touch controls +visible and keyboard focus explicit. + +## Components + +| Capability | Canonical owner | Source of truth | Allowed variants | Verification | +| --- | --- | --- | --- | --- | +| Select | `src/components/Select.tsx` | Existing application implementation | Native platform popup | Component and browser fixture | +| Confirmation | `src/components/ConfirmDialog.tsx` | Existing application implementation | Disconnect names its routing consequence | Browser fixture | +| Approval and questions | `ApprovalCard`, `QuestionCard` | Existing application implementation | Native requests, exact scoped verdicts | Server fixture and unit tests | +| Attachments | `AttachmentGallery`, server attachment store | Existing application implementation | Admitted inline images and supported files | Server and browser fixture | +| Feedback | Inline status and alert regions | Existing application implementation | Server-confirmed connection; uncertain delivery requires review | Component and server fixtures | +| Scrollbar | `src/styles.css` | Existing application implementation | Existing global baseline | Existing application contract | +| Locale | `src/lib/i18n.ts`, `src/locales/en.json` | Existing application implementation | Catalog keys; existing English fallback | `pnpm i18n:check` | + +## Do's and Don'ts + +Keep Codex provenance separate from its original response text. Say Connected or +Ready only after native resume and history reconciliation succeed. Keep failed +and disconnected bindings reserved until the user disconnects them. Do not add +normal bot instructions, model overrides, or silent replay to the native flow. diff --git a/docs/codex-conversation-binding.md b/docs/codex-conversation-binding.md new file mode 100644 index 0000000000..ef1bcd4a06 --- /dev/null +++ b/docs/codex-conversation-binding.md @@ -0,0 +1,185 @@ +# Connect an existing Codex conversation + +A local OpenMausBot thread can forward the person's text and images to an exact +existing local Codex conversation and display Codex's actual responses. The +OpenMausBot bot does not answer, add its instructions, or replay its transcript. + +## Requirements and limitations + +- Owner/admin-controlled local workspace. Cloud homes, shared hosted workspaces, + service loopback, and client-only sessions cannot discover/control conversations. +- Native Codex instance using the conversation's ChatGPT sign-in and `CODEX_HOME`. + Company, token-sharing ChatGPT variants, and API-key accounts are excluded. +- Codex **0.163.x**. Official 0.163.0 source and generated schemas were inspected. + Other minor versions fail before app-server launch. In particular, 0.159.2 has + a different legacy permission restoration path and is deliberately refused. +- The conversation must be visible to that local app-server. A UUID/title from + cloud-only ChatGPT/Codex Work does not make it accessible locally. Missing IDs + fail; no replacement, fork, unarchive, or reconstructed conversation is created. +- A thread-owned saved permission snapshot must be readable in the local Codex + store. Unsupported/remote-only history and old histories without that snapshot + fail before resume. The reader is version-pinned and never edits Codex files. +- Finish/stop work in the original Codex client before connecting. Use one writer + while connected. Separate clients/processes can race the same local history; + this feature cannot remotely attach to an arbitrary running application process. +- Outgoing structured input supports text and admitted images. Supported documents + in the attachment store are forwarded as their original file-reference text; + Codex reads them through its own tools and permissions. Inaccessible or + unsupported attachments are rejected before dispatch. Audio is not transcribed. +- Only command/file-change approvals and structured native questions have a user + interface. Other approval, permission-amendment, elicitation, and dynamic-tool + requests, root grants, or extra permission payloads fail closed with a visible + notice. Continue in Codex or cancel. + +## Setup, inspect, and disconnect + +Configure the native Codex engine through existing engine settings. Use its CLI +without profile, model, sandbox, approval, or `-c` overrides and the original +account/store. In the intended OpenMausBot thread, expand **Codex conversation**. +Choose the instance, find the title, inspect its exact ID, and select **Connect +this conversation**. An exact ID can also be entered directly. + +The server persists a reservation, connects, reads the exact thread, resumes with +only `threadId` and `excludeTurns`, and reconciles history. Only then does it show +**Ready**. The panel shows Codex's effective model, effort, working folder, +sandbox, and approval policy. Normal model/place/approval controls are hidden; +the saved native settings govern the conversation. A failed/disconnected binding +stays reserved, so messages never fall back to an OpenMausBot model. + +Native text/image replies have expandable provenance identifying conversation, +turn, and item. Existing native assistant messages/images are imported on connect. +The original history remains in Codex. No `thread/start` or `thread/fork` is used. + +The private `codex-bindings.json` in the workspace data directory persists +reservations, settings, and delivery receipts with mode 0600. Account/store +identity is hashed; credentials are not copied or logged. After restart the +reservation is **Disconnected** until explicitly reconnected. Reconnection checks +the original account/store and settings. Deliberate settings changes require +review in Codex, then disconnect/reselection to accept the new native settings. + +Finish/stop active work, cancel queued messages, and review uncertain deliveries. +Select **Disconnect Codex** and confirm. The thread returns to its normal bot; +both transcripts and the original Codex conversation are retained. Old send IDs +cannot be replayed by disconnecting/rebinding and retrying the same send. + +## Delivery and recovery + +Browser sends use the existing stable `sendId`. The durable outbox records text +before acknowledgement; retries return the original receipt, and different words +with the same ID are rejected. Messages run in order, one native turn at a time, +with up to 20 waiting messages. The panel can cancel queued work or interrupt the +native turn. Cancelled receipts remain cancelled. + +Dispatch is committed before `turn/start`. Codex's `clientUserMessageId` becomes +its user item's `clientId`; this identifies history, **not** a native exactly-once +submission guarantee. On reconnect, native turn/item history reconciles accepted +IDs and imports missed replies once. Reply deduplication uses conversation, turn, +and item ID in the durable message store, including a crash between message and +outbox commits. + +A dispatch whose outcome cannot be established is **Delivery uncertain**, never +automatically replayed. Following queued work stays held. Review the original +conversation in Codex, then select **Mark reviewed**. This closes that receipt and +permits following queued work; it never resends the uncertain message. Send a new +message with a new ID only if review establishes that it is needed. + +Only the owner/admin can answer a native approval card. Approve grants the +displayed request once. No session grants, remembered command rules, or permission +amendments are synthesized. Stale cards expire on disconnect/turn completion. +Codex's sandbox and approval policy remain authoritative; OpenMausBot Full access +does not grant native approvals. + +## Attachments + +Generated raster bytes and accessible image-view/file references are copied into +the existing private attachment store. PNG, JPEG, GIF, and WebP display in the +normal inline gallery. Document links use the existing file MIME allowlist and +limits. References must be in the native working folder or OpenMausBot attachment +store; existing canonical-path/file-handle checks apply. Remote-only/inaccessible +files, remote URLs, unsupported formats, and oversized files are not fetched using +another permission/credential context. A visible notice asks for a manual upload. + +## HTTP interface + +All routes are authenticated and admin scoped. Use personal IDs locally; the +placeholders below are intentionally not real account/conversation identifiers. + +| Method | Path | Body/query | +| --- | --- | --- | +| GET | `/api/codex/conversations` | Supported native instances | +| GET | `/api/codex/conversations?instanceId=codex&search=title` | Optional `cursor`, pages of 30 | +| GET | `/api/bots/BOT_ID/threads/THREAD_ID/codex` | Status, settings, delivery states | +| PUT | Same path | `{"instanceId":"codex","conversationId":"CONVERSATION_ID"}` | +| POST | Same path + `/reconnect` | `{}` | +| POST | Same path + `/cancel` | `{}` or `{"sendId":"..."}` | +| POST | Same path + `/acknowledge` | `{"sendId":"..."}` after review | +| DELETE | Same path | Disconnect idle/reviewed binding | + +Send through ordinary `POST /api/bots/BOT_ID/messages` with explicit `threadId`, +`text`, and stable 16–80 character `sendId`. Answer through existing +`/api/threads/THREAD_ID/respond`. Guarded relays, app/widget messages, quoted +replies, data-context injection, edits, compaction, and autonomous bot turns are +excluded from the native path. + +## Manual end-to-end smoke test + +This test changes a local binding and a harmless Codex text exchange may consume +normal Codex usage. Run only when separately authorized. Implementation fixtures +do not access the user's running app, target thread, conversation, or account. + +1. In Codex, record the exact existing conversation ID/title, latest message, + model, effort, cwd, sandbox, and approval policy. Locate an **existing** image + item with accessible bytes/path. Do not generate a new image for this test. +2. In a disposable OpenMausBot workspace, choose a dedicated thread and the native + instance for that conversation store. Connect by exact ID. Verify Connecting + becomes Ready and settings match. Verify the existing image displays inline + with original native conversation/turn/item provenance. +3. Send “Reply with the word connected. Do not use tools or change files.” Verify + the exact message appears in the original Codex conversation and its actual + reply appears once in OpenMausBot. Check Codex has no replacement/fork. +4. Retry the same HTTP send ID. Check the receipt ID is unchanged and neither side + has a duplicate. Reconnect and check text/image items still appear once. +5. If explicitly permitted, exercise a harmless native approval. Deny first and + verify denial remains denial. Approve only a second displayed request and + inspect Codex's actual outcome. Another thread must not answer that card. +6. Disconnect once settled. Check Codex retains original history/settings. Remove + only the disposable OpenMausBot workspace. + +If the intended conversation is cloud-only and absent from the local Codex list, +stop at step 2. The service owner must expose a supported exact-conversation +transport; this local bridge cannot manufacture one. + +## Verification and protocol evidence + +```sh +pnpm exec vitest run server/codex-bindings.test.ts server/drivers/codex-native-client.test.ts server/drivers/codex-saved-permissions.test.ts server/routes/codex-bindings.test.ts server/codex-binding.e2e.test.ts +pnpm typecheck +pnpm lint +pnpm i18n:check +``` + +Unit tests use an in-memory native peer and real durable message store. Stdio +tests use a dependency-free fake CLI. The end-to-end test launches a real isolated +OpenMausBot server and fake engines, covering selection/resume, exact text, SSE +provenance, an existing inline image, retry, approval, blocked compaction, and +disconnect. These do not prove live model output or remote Work availability. + +Installed verification inspected generated schemas from bundled 0.159.2 and a +separately installed disposable 0.163.0. A no-inference probe initialized 0.163.0; +creating a disposable native fixture thread timed out here. No model `turn/start` +was sent to either real binary. Real-account native resumption and the manual +smoke above remain unverified. The PR records actual checks and baseline failures. + +Official references: + +- [Codex app-server documentation](https://developers.openai.com/codex/app-server/) +- [0.163.0 resume implementation](https://github.com/openai/codex/blob/rust-v0.163.0/codex-rs/app-server/src/request_processors/thread_processor.rs) +- [0.163.0 persisted settings](https://github.com/openai/codex/blob/rust-v0.163.0/codex-rs/app-server/src/request_processors/persisted_resume_settings.rs) +- [0.159.2 resume implementation](https://github.com/openai/codex/blob/rust-v0.159.2/codex-rs/app-server/src/request_processors/thread_processor.rs) + +To inspect the real UI against fake engines without an account, run +`node --experimental-strip-types scripts/testing/verify-codex-binding-ui.ts`, +open its printed preview URL, select the fixture bot, find its fixture native +conversation, connect, and send harmless text. The existing fixture image appears +inline. Ctrl-C removes only the disposable fixture. Screenshots in +`docs/verification/images/codex-binding/` use synthetic data, not a live account. diff --git a/docs/verification/README.md b/docs/verification/README.md index a0edac04ec..e7e7a6a27a 100644 --- a/docs/verification/README.md +++ b/docs/verification/README.md @@ -315,3 +315,6 @@ state. A green unit test alone does not prove a user workflow. Interrupt the `launch` process with Ctrl-C. It stops the exact child it owns and removes only its temporary data directory. The server log remains at the printed path. Never kill processes by name and never delete a broad temp root. + +The [existing Codex conversation binding](../codex-conversation-binding.md) documents +its isolated fixtures, protocol-version boundary, and separately authorized manual smoke. diff --git a/docs/verification/images/codex-binding/after-connected.png b/docs/verification/images/codex-binding/after-connected.png new file mode 100644 index 0000000000..830efc7d8f Binary files /dev/null and b/docs/verification/images/codex-binding/after-connected.png differ diff --git a/docs/verification/images/codex-binding/after-narrow.png b/docs/verification/images/codex-binding/after-narrow.png new file mode 100644 index 0000000000..b64793cb2c Binary files /dev/null and b/docs/verification/images/codex-binding/after-narrow.png differ diff --git a/docs/verification/images/codex-binding/before-disconnected.png b/docs/verification/images/codex-binding/before-disconnected.png new file mode 100644 index 0000000000..f838a36847 Binary files /dev/null and b/docs/verification/images/codex-binding/before-disconnected.png differ diff --git a/scripts/testing/verify-codex-binding-ui.ts b/scripts/testing/verify-codex-binding-ui.ts new file mode 100644 index 0000000000..24a8f852a7 --- /dev/null +++ b/scripts/testing/verify-codex-binding-ui.ts @@ -0,0 +1,26 @@ +import { writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { launchVerificationServer } from "../control-omb.ts"; +import { fixtureApi, mountPreview, parkUntilSignal, type MountedPreview } from "./preview-fixture.ts"; + +const fixture = await launchVerificationServer(process.env, undefined, undefined, undefined, undefined, undefined, ["codex"]); +let preview: MountedPreview | undefined; +try { + const api = fixtureApi(fixture.info.url); + const wrapper = join(fixture.info.dataDir, "native-codex-ui-fixture.mjs"); + writeFileSync(wrapper, [ + `#!${process.execPath}`, + `process.env.FAKE_CODEX_NATIVE_DIR = ${JSON.stringify(fixture.info.dataDir)};`, + `await import(${JSON.stringify(pathToFileURL(fileURLToPath(new URL("../../server/testing/fake-codex-native.ts", import.meta.url))).href)});`, + ].join("\n"), { mode: 0o700 }); + await api("PATCH", "/api/instances/codex", { cli: wrapper }); + await api("PATCH", "/api/config", { language: "en" }); + const bot = (await api("POST", "/api/bots", { name: "Codex connection fixture" })).bot; + preview = await mountPreview(fixture, { entry: "/scripts/testing/threads-preview.tsx", route: "/__codex-binding.html", title: "Existing Codex connection fixture", logLevel: "warn" }); + console.log(JSON.stringify({ previewUrl: preview.previewUrl, url: fixture.info.url, botId: bot.id, threadId: bot.threadId, conversationId: "fixture-native-existing", logPath: fixture.info.logPath })); + await parkUntilSignal(); +} finally { + await preview?.close(); + await fixture.close(); +} diff --git a/server/codex-binding-attachments.ts b/server/codex-binding-attachments.ts new file mode 100644 index 0000000000..b5faf299e2 --- /dev/null +++ b/server/codex-binding-attachments.ts @@ -0,0 +1,59 @@ +import { lstatSync, realpathSync } from "node:fs"; +import { dirname, resolve } from "node:path"; +import { splitTranscriptAttachments, isPrivateAttachmentPath } from "../src/lib/composer-attachments.ts"; +import { ATTACHMENTS_DIR, FILE_MAX_BYTES, IMAGE_MAX_BYTES, extensionForFileMime, saveFile, saveImage } from "./attachments.ts"; +import { decodeGeneratedImage } from "./generated-image.ts"; +import { messageFileTargets, mimeFor, openMessageFile } from "./message-file.ts"; +import { extractTurnImages } from "./turn-images.ts"; +import type { Message } from "./store.ts"; + +export function codexUserInput(text: string): unknown[] { + const extracted = extractTurnImages(text); + const attachments = splitTranscriptAttachments(text, false); + if (attachments.images.some((image) => !extracted.images.some((admitted) => admitted.path === image.path))) throw Object.assign(new Error("An attached image is unavailable. Upload it again before sending."), { status: 422 }); + for (const file of attachments.files) { + try { + const path = resolve(file.path); + const stat = lstatSync(path); + if (!isPrivateAttachmentPath(path) || dirname(path) !== resolve(ATTACHMENTS_DIR) || dirname(realpathSync(path)) !== realpathSync(ATTACHMENTS_DIR) || !stat.isFile() || stat.isSymbolicLink() || stat.size <= 0 || stat.size > FILE_MAX_BYTES || !extensionForFileMime(mimeFor(path))) throw new Error("Unavailable attachment"); + } catch { throw Object.assign(new Error("An attached file is unavailable or unsupported. Upload it again before sending."), { status: 422 }); } + } + return [ + ...(extracted.text ? [{ type: "text", text: extracted.text, text_elements: [] }] : []), + ...extracted.images.map(({ path }) => ({ type: "localImage", path })), + ]; +} + +export async function codexItemAttachments(item: any, cwd: string): Promise<{ attachments: NonNullable; unavailable: boolean }> { + const attachments: NonNullable = []; + let unavailable = false; + if (item.type === "imageGeneration" && typeof item.result === "string" && item.result && item.result !== "[redacted]") { + try { + const decoded = decodeGeneratedImage(item.result); + const saved = saveImage(decoded.bytes, decoded.mime); + attachments.push({ kind: "image", path: saved.path, mime: saved.mime }); + return { attachments, unavailable }; + } catch { unavailable = true; } + } + const paths: string[] = item.type === "imageView" && typeof item.path === "string" ? [item.path] + : item.type === "imageGeneration" && typeof item.savedPath === "string" ? [item.savedPath] + : item.type === "agentMessage" && typeof item.text === "string" ? messageFileTargets(item.text.replace(/sandbox:(\/)/g, "$1")) : []; + for (const path of paths.slice(0, 20)) { + let opened: Awaited> | undefined; + try { + opened = await openMessageFile(path, [cwd, ATTACHMENTS_DIR]); + const bytes = await opened.handle.readFile(); + if (opened.mime.startsWith("image/")) { + if (bytes.length > IMAGE_MAX_BYTES) throw new Error("Image too large"); + const decoded = decodeGeneratedImage(bytes.toString("base64")); + const saved = saveImage(decoded.bytes, decoded.mime); + attachments.push({ kind: "image", path: saved.path, mime: saved.mime, name: opened.name }); + } else { + const saved = await saveFile((async function* () { yield bytes; })(), opened.name, opened.mime, { expectedBytes: bytes.length }); + attachments.push({ kind: "file", path: saved.path, mime: saved.mime, name: saved.name }); + } + } catch { unavailable = true; } + finally { await opened?.handle.close(); } + } + return { attachments, unavailable: unavailable || paths.length > 20 }; +} diff --git a/server/codex-binding.e2e.test.ts b/server/codex-binding.e2e.test.ts new file mode 100644 index 0000000000..9115679089 --- /dev/null +++ b/server/codex-binding.e2e.test.ts @@ -0,0 +1,67 @@ +import { readFileSync, writeFileSync } from "node:fs"; +import { basename, join } from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { expect, it } from "vitest"; +import { launchVerificationServer } from "../scripts/control-omb.ts"; +import { openSse } from "./testing/sse.ts"; + +it("relays through the real isolated server routes, SSE, approval cards and inline attachment URLs", async () => { + const fixture = await launchVerificationServer(process.env, undefined, undefined, undefined, undefined, undefined, ["codex"]); + const { url, dataDir, logPath } = fixture.info; + const events = await openSse(`${url}/api/events`); + const api = async (method: string, path: string, body?: unknown, expected = 200) => { + const response = await fetch(`${url}${path}`, { method, headers: { "content-type": "application/json" }, ...(body === undefined ? {} : { body: JSON.stringify(body) }), signal: AbortSignal.timeout(60_000) }); + const result = await response.json() as any; + expect(response.status, `${JSON.stringify(result)}; fixture log: ${logPath}`).toBe(expected); + return result; + }; + try { + const wrapper = join(dataDir, "native-codex-fixture.mjs"); + writeFileSync(wrapper, [ + `#!${process.execPath}`, + `process.env.FAKE_CODEX_NATIVE_DIR = ${JSON.stringify(dataDir)};`, + `await import(${JSON.stringify(pathToFileURL(fileURLToPath(new URL("./testing/fake-codex-native.ts", import.meta.url))).href)});`, + ].join("\n"), { mode: 0o700 }); + await api("PATCH", "/api/instances/codex", { cli: wrapper }); + const bot = (await api("POST", "/api/bots", { name: "Native Codex relay fixture" }, 201)).bot; + const path = `/api/bots/${bot.id}/threads/${bot.threadId}/codex`; + const list = await api("GET", "/api/codex/conversations?instanceId=codex"); + expect(list.conversations).toEqual([{ id: "fixture-native-existing", title: "Native fixture conversation", cwd: dataDir }]); + const connected = await api("PUT", path, { instanceId: "codex", conversationId: "fixture-native-existing" }); + expect(connected).toMatchObject({ connected: true, state: "ready", settings: { model: "fixture-native-model", reasoningEffort: "high", approvalPolicy: "on-request", sandbox: { type: "readOnly" } } }); + const messages = (await api("GET", `/api/threads/${bot.threadId}/messages?limit=100`)).messages; + const image = messages.find((message: any) => message.codex?.itemId === "existing-fixture-image"); + expect(image.attachments[0]).toMatchObject({ kind: "image", mime: "image/png", path: expect.stringMatching(/\.png$/) }); + const downloaded = await fetch(`${url}/api/attachments/${basename(image.attachments[0].path)}`); + expect(downloaded.headers.get("content-type")).toContain("image/png"); + expect((await downloaded.arrayBuffer()).byteLength).toBeGreaterThan(0); + const text = " harmless exact text 😀\n "; + const receipt = await api("POST", `/api/bots/${bot.id}/messages`, { threadId: bot.threadId, text, sendId: "native-fixture-send-001" }, 202); + expect(receipt.message.text).toBe(text); + const reply = await events.until((event) => event.kind === "message" && event.threadId === bot.threadId && event.message?.text === `Native fixture reply: ${text}`); + expect(reply.message.codex.conversationId).toBe("fixture-native-existing"); + const retried = await api("POST", `/api/bots/${bot.id}/messages`, { threadId: bot.threadId, text, sendId: "native-fixture-send-001" }, 202); + expect(retried.message.id).toBe(receipt.message.id); + await api("POST", `/api/bots/${bot.id}/messages`, { threadId: bot.threadId, text: "approval fixture", sendId: "native-fixture-send-002" }, 202); + const ask = await events.until((event) => event.kind === "message" && event.threadId === bot.threadId && event.message?.card?.tool === "Codex"); + expect((await api("GET", path)).state).toBe("awaiting-approval"); + await api("POST", `/api/threads/${bot.threadId}/respond`, { requestId: ask.message.card.requestId, behavior: "allow", always: true }, 400); + await api("POST", `/api/threads/${bot.threadId}/respond`, { requestId: ask.message.card.requestId, behavior: "allow" }); + await events.until((event) => event.kind === "message" && event.threadId === bot.threadId && event.message?.text === "Native fixture reply: approval fixture"); + await api("POST", `/api/bots/${bot.id}/compact`, { threadId: bot.threadId }, 409); + await api("DELETE", path); + expect((await api("GET", path)).configured).toBe(false); + const calls = readFileSync(join(dataDir, "native-fixture-requests.jsonl"), "utf8").trim().split("\n").map((line) => JSON.parse(line)); + expect(calls.filter((call) => call.method === "turn/start")).toHaveLength(2); + expect(calls.some((call) => call.method === "thread/start" || call.method === "thread/fork")).toBe(false); + expect(calls.find((call) => call.reply)?.reply.result).toEqual({ decision: "accept" }); + await api("PATCH", "/api/instances/codex", { cli: fileURLToPath(new URL("./testing/fake-codex-app-server.ts", import.meta.url)) }); + const incompatible = await api("PUT", path, { instanceId: "codex", conversationId: "fixture-native-existing" }, 409); + expect(incompatible).toMatchObject({ configured: true, connected: false, state: "failed" }); + await api("POST", `/api/bots/${bot.id}/messages`, { threadId: bot.threadId, text: "must not get a fallback answer", sendId: "native-fixture-send-003" }, 409); + const failedMessages = (await api("GET", `/api/threads/${bot.threadId}/messages?limit=100`)).messages; + expect(failedMessages.some((message: any) => message.text === "must not get a fallback answer")).toBe(false); + await api("DELETE", path); + console.log(`Native Codex fixture verification passed; log: ${logPath}`); + } finally { events.close(); await fixture.close(); } +}, 120_000); diff --git a/server/codex-bindings.test.ts b/server/codex-bindings.test.ts new file mode 100644 index 0000000000..f2e44ac87b --- /dev/null +++ b/server/codex-bindings.test.ts @@ -0,0 +1,348 @@ +import { mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { DATA_DIR } from "./config.ts"; +import { Store } from "./store.ts"; +import { CodexBindings } from "./codex-bindings.ts"; +import { CodexNativeRejection, type CodexNativeClient } from "./drivers/codex-native-client.ts"; +import { saveFile, saveImage } from "./attachments.ts"; +import { closeMessageDb } from "./message-db.ts"; + +const CONVERSATION = "fixture-existing-conversation"; +const SEND = "fixture-send-0123456789"; +const SECOND = "fixture-send-abcdefghij"; +const PNG = Buffer.from("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+a9xkAAAAASUVORK5CYII=", "base64"); + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason: unknown) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +} + +class Native implements CodexNativeClient { + verifySavedPermissions = async () => {}; + home = "/fixture/codex-home"; + email = "owner@example.test"; + calls: Array<{ method: string; params: any }> = []; + replies: unknown[] = []; + rejected: unknown[] = []; + turns: any[] = []; + cwd = DATA_DIR; + model = "fixture-saved-model"; + effort = "high"; + approvalPolicy = "on-request"; + sandbox = { type: "readOnly" }; + nativeId = CONVERSATION; + readId = CONVERSATION; + readStatus = "idle"; + historyError = false; + startError: Error | undefined; + startGate: ReturnType | undefined; + connectGate: ReturnType | undefined; + listener: (message: any) => void | Promise = () => {}; + closers: Array<() => void> = []; + waiters: Array<{ method: string; count: number; resolve(value: any): void }> = []; + closed = false; + request = async (method: string, params: any): Promise => { + const call = { method, params }; + this.calls.push(call); + for (const waiter of this.waiters) if (waiter.method === method && this.calls.filter((call) => call.method === method).length >= waiter.count) waiter.resolve(call); + if (method === "account/read") return { account: { type: "chatgpt", email: this.email } }; + if (method === "config/read") return { config: { model_provider: "openai" } }; + if (method === "thread/list") return { data: [{ id: CONVERSATION, name: "Existing fixture", cwd: this.cwd }], nextCursor: null }; + if (method === "thread/read") return { thread: { id: this.readId, name: "Existing fixture", model: this.model, reasoningEffort: this.effort, cwd: this.cwd, status: { type: this.readStatus } } }; + if (method === "thread/resume") { + await this.connectGate?.promise; + return { thread: { id: this.nativeId, name: "Existing fixture" }, model: this.model, modelProvider: "openai", reasoningEffort: this.effort, cwd: this.cwd, approvalPolicy: this.approvalPolicy, approvalsReviewer: "user", sandbox: this.sandbox, collaborationMode: null }; + } + if (method === "thread/turns/list") { + if (this.historyError) throw new Error("History unavailable"); + return { data: this.turns, nextCursor: null }; + } + if (method === "turn/start") { + if (this.startError) throw this.startError; + const turn = { id: `native-turn-${this.calls.filter((call) => call.method === method).length}`, status: "inProgress", items: [{ id: `user-${params.clientUserMessageId}`, type: "userMessage", clientId: params.clientUserMessageId, content: params.input }] }; + this.turns.push(turn); + await this.emit("turn/started", { threadId: CONVERSATION, turn }); + await this.startGate?.promise; + return { turn: { ...turn, status: "inProgress" } }; + } + if (method === "turn/interrupt") { await this.finish("interrupted"); return {}; } + throw new Error(`Unexpected method ${method}`); + }; + next(method: string, count = 1): Promise { + const calls = this.calls.filter((call) => call.method === method); + if (calls.length >= count) return Promise.resolve(calls[count - 1]); + return new Promise((resolve) => this.waiters.push({ method, count, resolve })); + } + reply(id: string | number, result: unknown) { this.replies.push({ id, result }); } + reject(id: string | number) { this.rejected.push(id); } + onMessage(listener: (message: any) => void | Promise) { this.listener = listener; } + onClose(listener: () => void) { this.closers.push(listener); } + async close() { if (!this.closed) { this.closed = true; for (const listener of this.closers) listener(); } } + async emit(method: string, params: any, id?: string | number) { await this.listener({ method, params, ...(id === undefined ? {} : { id }) }); } + async finish(status = "completed", items: any[] = [{ id: "actual-reply", type: "agentMessage", text: "Actual native response" }]) { + const turn = this.turns.at(-1)!; + turn.status = status; + turn.items.push(...items); + await this.emit("turn/completed", { threadId: CONVERSATION, turn }); + } +} + +let dir: string; +const bridges: CodexBindings[] = []; +beforeEach(() => { rmSync(DATA_DIR, { recursive: true, force: true }); dir = mkdtempSync(join(tmpdir(), "omb-codex-binding-")); }); +afterEach(async () => { await Promise.all(bridges.splice(0).map((bridge) => bridge.close())); rmSync(dir, { recursive: true, force: true }); }); + +function fixture() { + const store = new Store(() => ({ instanceId: "unrelated-model", model: "must-not-answer" })); + const bot = store.createBot({ name: "Relay fixture" }); + const thread = bot.threadId; + const native = new Native(); + let current = native; + const make = () => { + const bridge = new CodexBindings({ dataDir: dir, store, connect: async () => current, busy: () => false }); + bridges.push(bridge); + return bridge; + }; + const bridge = make(); + const bind = () => bridge.bind(bot.id, thread, "native-codex", CONVERSATION); + return { store, bot, thread, native, bridge, bind, make, use: (client: Native) => { current = client; }, view: () => bridge.view(bot.id, thread), messages: () => store.messagesFor(thread) }; +} + +describe("existing native Codex bindings", () => { + it("resumes only the exact selected conversation, without configuration or prompt injection", async () => { + const h = fixture(); + const view = await h.bind(); + expect(view).toMatchObject({ configured: true, connected: true, state: "ready", settings: { model: h.native.model, reasoningEffort: "high", cwd: DATA_DIR, approvalPolicy: "on-request", sandbox: { type: "readOnly" } } }); + expect(h.native.calls.find((call) => call.method === "thread/resume")?.params).toEqual({ threadId: CONVERSATION, excludeTurns: true }); + const text = " Exact text\n日本語 😀\n "; + await h.bridge.send(h.bot.id, h.thread, text, SEND); + const { params } = await h.native.next("turn/start"); + expect(params).toEqual({ threadId: CONVERSATION, clientUserMessageId: SEND, input: [{ type: "text", text, text_elements: [] }] }); + expect(h.native.calls.some((call) => ["thread/start", "thread/fork"].includes(call.method))).toBe(false); + expect(() => h.bridge.assertUnbound(h.thread)).toThrow(/reserved/); + expect(() => h.bridge.assertUnbound("another-thread")).not.toThrow(); + await h.native.finish(); + expect(h.messages().find((message) => message.text === "Actual native response")?.codex).toEqual({ conversationId: CONVERSATION, turnId: "native-turn-1", itemId: "actual-reply" }); + }); + + it("never claims ready while connecting, or after a different native thread was returned", async () => { + const h = fixture(); h.native.connectGate = deferred(); + const pending = h.bind(); + await h.native.next("thread/resume"); + expect(h.view()).toMatchObject({ configured: true, connected: false, state: "connecting" }); + h.native.nativeId = "different-conversation"; h.native.connectGate.resolve(undefined); + expect(await pending).toMatchObject({ connected: false, state: "failed" }); + await expect(h.bridge.send(h.bot.id, h.thread, "hello", SEND)).rejects.toThrow(/Reconnect/); + expect(h.native.calls.some((call) => call.method === "turn/start")).toBe(false); + expect(() => h.bridge.assertUnbound(h.thread)).toThrow(); + }); + + it("rejects missing, active and foreign conversations and prevents cross-thread reservations", async () => { + const h = fixture(); h.native.readId = "missing"; + expect(await h.bind()).toMatchObject({ state: "failed", connected: false }); + await h.bridge.disconnect(h.bot.id, h.thread); + h.use(new Native()); + const second = h.store.createTask(h.bot.id, "Second")!.threadId; + await h.bridge.bind(h.bot.id, second, "native-codex", CONVERSATION); + await expect(h.bind()).rejects.toThrow(/already reserved/); + await expect(h.bridge.send("another-bot", second, "hello", SEND)).rejects.toThrow(/No such/); + }); + + it("deduplicates concurrent sends and native replies, preserving FIFO delivery", async () => { + const h = fixture(); await h.bind(); + const [a, b] = await Promise.all([h.bridge.send(h.bot.id, h.thread, "first", SEND), h.bridge.send(h.bot.id, h.thread, "first", SEND)]); + expect(a.id).toBe(b.id); + await h.native.next("turn/start"); + await h.bridge.send(h.bot.id, h.thread, "second", SECOND); + expect(h.native.calls.filter((call) => call.method === "turn/start")).toHaveLength(1); + await h.native.finish(); + await h.native.next("turn/start", 2); + await h.native.finish("completed", [{ id: "actual-reply", type: "agentMessage", text: "Second response" }]); + await h.native.emit("item/completed", { threadId: CONVERSATION, turnId: "native-turn-2", item: { id: "actual-reply", type: "agentMessage", text: "Second response" } }); + expect(h.messages().filter((message) => message.text === "Second response")).toHaveLength(1); + expect(h.view().deliveries.map((delivery) => delivery.state)).toEqual(["completed", "completed"]); + await expect(h.bridge.send(h.bot.id, h.thread, "different", SEND)).rejects.toThrow(/sendId/); + }); + + it("handles completion before the turn/start acknowledgement", async () => { + const h = fixture(); await h.bind(); h.native.startGate = deferred(); + await h.bridge.send(h.bot.id, h.thread, "first", SEND); + await h.native.next("turn/start"); + await h.native.finish(); h.native.startGate.resolve(undefined); + await h.bridge.send(h.bot.id, h.thread, "second", SECOND); + await h.native.next("turn/start", 2); + expect(h.view().deliveries[0]?.state).toBe("completed"); + await h.native.finish(); + expect(h.view().running).toBe(false); + }); + + it("persists binding and receipts across restarts, then reconciles native IDs without replay", async () => { + const h = fixture(); await h.bind(); + await h.bridge.send(h.bot.id, h.thread, "first", SEND); await h.native.next("turn/start"); + await h.bridge.close(); + const replacement = new Native(); replacement.turns = structuredClone(h.native.turns); replacement.turns[0].status = "completed"; + replacement.turns[0].items.push({ id: "after-crash", type: "agentMessage", text: "Recovered native response" }); + h.use(replacement); const restarted = h.make(); + expect(restarted.view(h.bot.id, h.thread)).toMatchObject({ configured: true, connected: false, state: "disconnected" }); + expect(await restarted.reconnect(h.bot.id, h.thread)).toMatchObject({ connected: true, deliveries: [{ sendId: SEND, state: "completed" }] }); + const receipt = await restarted.send(h.bot.id, h.thread, "first", SEND); + expect(receipt.id).toBe(h.messages().find((message) => message.sendId === SEND)?.id); + expect(replacement.calls.filter((call) => call.method === "turn/start")).toHaveLength(0); + expect(h.messages().filter((message) => message.text === "Recovered native response")).toHaveLength(1); + const saved = JSON.parse(readFileSync(join(dir, "codex-bindings.json"), "utf8")); + expect(saved.bindings[0].identity).not.toContain("owner@example.test"); + if (process.platform !== "win32") expect(statSync(join(dir, "codex-bindings.json")).mode & 0o777).toBe(0o600); + }); + + it("holds uncertain dispatch and queued successors through recovery until explicit review", async () => { + const h = fixture(); await h.bind(); h.native.startError = new Error("transport lost"); + await h.bridge.send(h.bot.id, h.thread, "may have arrived", SEND); + await h.native.next("turn/start"); await h.native.close(); + const next = new Native(); h.use(next); + await h.bridge.reconnect(h.bot.id, h.thread); + expect(h.view().deliveries[0]?.state).toBe("uncertain"); + await h.bridge.send(h.bot.id, h.thread, "follow-up", SECOND); + expect(next.calls.filter((call) => call.method === "turn/start")).toHaveLength(0); + h.bridge.acknowledge(h.bot.id, h.thread, SEND); + const { params } = await next.next("turn/start"); + expect(params.clientUserMessageId).toBe(SECOND); + expect(h.messages().filter((message) => message.sendId === SEND)).toHaveLength(1); + }); + + it("deduplicates from a newly opened durable message store after a reply/outbox crash window", async () => { + const h = fixture(); await h.bind(); + await h.bridge.send(h.bot.id, h.thread, "durable", SEND); await h.native.next("turn/start"); + const beforeCompletion = readFileSync(join(dir, "codex-bindings.json"), "utf8"); + await h.native.finish(); await h.bridge.close(); closeMessageDb(); + writeFileSync(join(dir, "codex-bindings.json"), beforeCompletion); + const reopened = new Store(() => ({ instanceId: "unrelated", model: "unrelated" })); + const next = new Native(); next.turns = structuredClone(h.native.turns); + const restarted = new CodexBindings({ dataDir: dir, store: reopened, connect: async () => next, busy: () => false }); bridges.push(restarted); + await restarted.reconnect(h.bot.id, h.thread); + const receipt = await restarted.send(h.bot.id, h.thread, "durable", SEND); + expect(receipt.sendId).toBe(SEND); + expect(reopened.messagesFor(h.thread).filter((message) => message.text === "Actual native response")).toHaveLength(1); + expect(next.calls.some((call) => call.method === "turn/start")).toBe(false); + expect(restarted.view(h.bot.id, h.thread).deliveries[0]?.state).toBe("completed"); + }); + + it("prevents alias instances for the same account/store from binding the same native conversation", async () => { + const h = fixture(); await h.bind(); + const other = h.store.createTask(h.bot.id, "Alias")!.threadId; + const next = new Native(); h.use(next); + expect(await h.bridge.bind(h.bot.id, other, "alias-codex", CONVERSATION)).toMatchObject({ connected: false, state: "failed" }); + expect(next.calls.some((call) => call.method === "thread/resume")).toBe(false); + }); + + it("does not repeat a rejected native request, and exposes history failures truthfully", async () => { + const h = fixture(); await h.bind(); h.native.startError = new CodexNativeRejection("Native rejected"); + await h.bridge.send(h.bot.id, h.thread, "rejected", SEND); await h.native.next("turn/start"); + await h.native.close(); + const next = new Native(); next.historyError = true; h.use(next); + expect(await h.bridge.reconnect(h.bot.id, h.thread)).toMatchObject({ connected: false, state: "failed" }); + expect(next.calls.some((call) => call.method === "turn/start")).toBe(false); + }); + + it("refuses changed account, conversation store and saved settings on reconnect", async () => { + const h = fixture(); await h.bind(); await h.native.close(); + for (const change of ["email", "home", "model"] as const) { + const next = new Native(); next[change] = "changed"; h.use(next); + expect(await h.bridge.reconnect(h.bot.id, h.thread)).toMatchObject({ connected: false, state: "failed" }); + expect(next.calls.some((call) => call.method === "turn/start")).toBe(false); + } + }); + + it("requires explicit scoped approvals and expires stale requests", async () => { + const h = fixture(); await h.bind(); await h.bridge.send(h.bot.id, h.thread, "approval", SEND); await h.native.next("turn/start"); + await h.native.emit("item/commandExecution/requestApproval", { threadId: CONVERSATION, turnId: "native-turn-1", itemId: "command-1", command: "pwd", cwd: DATA_DIR, reason: "Read the current folder" }, 45); + expect(h.view().state).toBe("awaiting-approval"); expect(h.native.replies).toHaveLength(0); + const card = h.messages().find((message) => message.card)?.card; + expect(card).toMatchObject({ tool: "Codex", requestType: "permission" }); + expect(card?.allowSession).toBeUndefined(); + await expect(h.bridge.answer("foreign-bot", h.thread, card!.requestId!, "allow")).rejects.toThrow(); + expect(await h.bridge.answer(h.bot.id, h.thread, card!.requestId!, "allow")).toBe("allowed-once"); + expect(h.native.replies).toEqual([{ id: 45, result: { decision: "accept" } }]); + expect(await h.bridge.answer(h.bot.id, h.thread, card!.requestId!, "allow")).toBe("unavailable"); + await h.native.emit("item/fileChange/requestApproval", { threadId: CONVERSATION, turnId: "native-turn-1", itemId: "file-1", reason: "Change one fixture file" }, 46); + await h.native.close(); + expect(h.messages().find((message) => message.card?.requestId !== card!.requestId && message.card)?.card).toMatchObject({ expired: true, answered: "unavailable" }); + expect(h.native.replies).toHaveLength(1); + }); + + it("fails closed on foreign or unsupported approval methods", async () => { + const h = fixture(); await h.bind(); + await h.bridge.send(h.bot.id, h.thread, "approval scope", SEND); await h.native.next("turn/start"); + await h.native.emit("item/commandExecution/requestApproval", { threadId: "foreign", turnId: "turn", itemId: "item" }, 1); + await h.native.emit("item/permissions/requestApproval", { threadId: CONVERSATION, turnId: "native-turn-1", permissions: { fileSystem: { write: ["/"] } } }, 2); + await h.native.emit("item/fileChange/requestApproval", { threadId: CONVERSATION, turnId: "native-turn-1", grantRoot: "/" }, 3); + await h.native.emit("item/commandExecution/requestApproval", { threadId: CONVERSATION, turnId: "native-turn-1", additionalPermissions: { network: true } }, 4); + await h.native.emit("item/commandExecution/requestApproval", { threadId: CONVERSATION, turnId: "native-turn-1", availableDecisions: ["acceptForSession", "decline"] }, 5); + expect(h.native.rejected).toEqual([1, 2, 3, 4, 5]); expect(h.native.replies).toHaveLength(0); + expect(h.messages().some((message) => message.tool?.name.includes("No permission was granted"))).toBe(true); + }); + + it("routes native questions through structured answer cards", async () => { + const h = fixture(); await h.bind(); await h.bridge.send(h.bot.id, h.thread, "question", SEND); await h.native.next("turn/start"); + await h.native.emit("item/tool/requestUserInput", { threadId: CONVERSATION, turnId: "native-turn-1", questions: [{ id: "choice", header: "Pick", question: "Which option?", options: [{ label: "A", description: "First" }, { label: "B", description: "Second" }] }] }, "question-id"); + const card = h.messages().find((message) => message.card)?.card; + expect(card?.questionRequest?.questions[0]?.question).toBe("Which option?"); + await h.bridge.answer(h.bot.id, h.thread, card!.requestId!, "answer", "A"); + expect(h.native.replies).toEqual([{ id: "question-id", result: { answers: { choice: { answers: ["A"] } } } }]); + }); + + it("forwards admitted images and imports actual generated and existing images inline once", async () => { + const h = fixture(); const image = saveImage(PNG, "image/png"); + h.native.turns = [{ id: "old-native-turn", status: "completed", items: [{ id: "existing-image", type: "imageView", path: image.path }] }]; + await h.bind(); + expect(h.messages().find((message) => message.codex?.itemId === "existing-image")?.attachments?.[0]).toMatchObject({ kind: "image", mime: "image/png" }); + await h.bridge.send(h.bot.id, h.thread, `Look\n\n`, SEND); + const { params } = await h.native.next("turn/start"); + expect(params.input).toEqual([{ type: "text", text: "Look\n\n", text_elements: [] }, { type: "localImage", path: image.path }]); + await h.native.finish("completed", [{ id: "generated", type: "imageGeneration", result: PNG.toString("base64") }]); + const generated = h.messages().find((message) => message.codex?.itemId === "generated")!; + expect(generated.attachments?.[0]).toMatchObject({ kind: "image", mime: "image/png" }); + expect(readFileSync(generated.attachments![0]!.path)).toEqual(PNG); + await h.native.close(); h.use(new Native()); await h.bridge.reconnect(h.bot.id, h.thread); + expect(h.messages().filter((message) => message.codex?.itemId === "existing-image")).toHaveLength(1); + }); + + it("reports inaccessible files and rejects an unavailable user attachment before dispatch", async () => { + const h = fixture(); h.native.turns = [{ id: "old", status: "completed", items: [{ id: "missing-image", type: "imageView", path: join(dir, "missing.png") }] }]; + await h.bind(); + expect(h.messages().some((message) => message.tool?.name.includes("could not be accessed"))).toBe(true); + await expect(h.bridge.send(h.bot.id, h.thread, '', SEND)).rejects.toThrow(/unavailable/); + expect(h.native.calls.some((call) => call.method === "turn/start")).toBe(false); + }); + + it("preserves attachment syntax examples and forwards admitted document references without changing permissions", async () => { + const h = fixture(); await h.bind(); + const document = await saveFile((async function* () { yield Buffer.from("Harmless document"); })(), "fixture.txt", "text/plain"); + const text = `Read this if your native permissions allow it.\n\n\n\nExample: \`\``; + await h.bridge.send(h.bot.id, h.thread, text, SEND); + expect((await h.native.next("turn/start")).params.input).toEqual([{ type: "text", text, text_elements: [] }]); + await h.native.finish("completed", [{ id: "document-reply", type: "agentMessage", text: `[Existing document](${document.path})` }]); + expect(h.messages().find((message) => message.codex?.itemId === "document-reply")?.attachments?.[0]).toMatchObject({ kind: "file", mime: "text/plain", name: expect.stringMatching(/\.txt$/) }); + }); + + it("cancels queued and running turns without discarding receipts", async () => { + const h = fixture(); await h.bind(); await h.bridge.send(h.bot.id, h.thread, "first", SEND); await h.native.next("turn/start"); + await h.bridge.send(h.bot.id, h.thread, "second", SECOND); + await h.bridge.cancel(h.bot.id, h.thread, SECOND); + await h.bridge.cancel(h.bot.id, h.thread); + expect(h.view().deliveries.map((delivery) => delivery.state)).toEqual(["interrupted", "cancelled"]); + await h.bridge.disconnect(h.bot.id, h.thread); + expect(h.view()).toMatchObject({ configured: false, state: "disconnected" }); + const next = new Native(); h.use(next); await h.bind(); + await h.bridge.send(h.bot.id, h.thread, "first", SEND); + expect(next.calls.some((call) => call.method === "turn/start")).toBe(false); + }); + + it("fails startup closed if the durable binding ledger is corrupted", () => { + writeFileSync(join(dir, "codex-bindings.json"), "{broken"); + expect(() => fixture()).toThrow(/could not be read safely/); + }); +}); diff --git a/server/codex-bindings.ts b/server/codex-bindings.ts new file mode 100644 index 0000000000..ee6741badc --- /dev/null +++ b/server/codex-bindings.ts @@ -0,0 +1,444 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdirSync, readFileSync } from "node:fs"; +import { join } from "node:path"; +import type { CodexBindingView, CodexConnectionState, CodexDeliveryState } from "../shared/codex-binding.ts"; +import { parseProtocolAskQuestions } from "../shared/ask-question.ts"; +import { writeFileAtomic } from "./atomic.ts"; +import { codexItemAttachments, codexUserInput } from "./codex-binding-attachments.ts"; +import { codexQuestionReply } from "./drivers/codex-questions.ts"; +import { CodexNativeRejection, type CodexNativeClient } from "./drivers/codex-native-client.ts"; +import type { Message, Store } from "./store.ts"; + +interface Delivery { + sendId: string; + text: string; + state: CodexDeliveryState; + turnId?: string; +} + +interface Binding { + botId: string; + threadId: string; + instanceId: string; + conversationId: string; + title?: string; + identity?: string; + settings?: NonNullable & { modelProvider: string; approvalsReviewer: unknown; collaborationMode: unknown }; + deliveries: Delivery[]; +} + +interface Connection { + state: CodexConnectionState; + client?: CodexNativeClient; + error?: string; + running?: string; + work: Promise; + asks: Map; + pumping: boolean; + terminalTurns: Set; +} + +export interface CodexBindingsDeps { + dataDir: string; + store: Pick; + connect(instanceId: string): Promise; + busy(botId: string, threadId: string): boolean; +} + +function refuse(message: string, status = 409): never { throw Object.assign(new Error(message), { status }); } +function same(a: unknown, b: unknown): boolean { + const stable = (value: any): any => Array.isArray(value) ? value.map(stable) + : value && typeof value === "object" ? Object.fromEntries(Object.keys(value).sort().map((key) => [key, stable(value[key])])) : value; + return JSON.stringify(stable(a)) === JSON.stringify(stable(b)); +} + +/** Private, single-server outbox. A dispatch with no native acknowledgement is never automatically replayed. */ +export class CodexBindings { + private readonly deps: CodexBindingsDeps; + private readonly bindings = new Map(); + private readonly connections = new Map(); + private readonly file: string; + private stopped = false; + + constructor(deps: CodexBindingsDeps) { + this.deps = deps; + this.file = join(deps.dataDir, "codex-bindings.json"); + try { + const data = JSON.parse(readFileSync(this.file, "utf8")); + if (data.version !== 1 || !Array.isArray(data.bindings)) throw new Error("Unsupported Codex binding data"); + for (const binding of data.bindings as Binding[]) { + if (!binding.threadId || !binding.botId || !binding.instanceId || !binding.conversationId || !Array.isArray(binding.deliveries) || this.bindings.has(binding.threadId)) throw new Error("Invalid Codex binding data"); + for (const delivery of binding.deliveries) { + if (delivery.state === "dispatching" || delivery.state === "delivered") delivery.state = "uncertain"; + } + this.bindings.set(binding.threadId, binding); + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw new Error("Codex binding data could not be read safely. Restore it before starting OpenMausBot."); + } + } + + private persist(): void { + mkdirSync(this.deps.dataDir, { recursive: true, mode: 0o700 }); + writeFileAtomic(this.file, JSON.stringify({ version: 1, bindings: [...this.bindings.values()] }), { mode: 0o600 }); + } + + configured(threadId: string): boolean { return this.bindings.has(threadId); } + + assertUnbound(threadId: string): void { + if (this.configured(threadId)) refuse("This thread is reserved for its existing Codex conversation. Use its connection controls or disconnect it first."); + } + + private owned(botId: string, threadId: string): Binding { + const binding = this.bindings.get(threadId); + if (!binding || binding.botId !== botId || !this.deps.store.taskByThread(botId, threadId)) refuse("No such Codex connection.", 404); + return binding; + } + + view(botId: string, threadId: string): CodexBindingView { + if (!this.deps.store.taskByThread(botId, threadId)) refuse("No such thread.", 404); + const binding = this.bindings.get(threadId); + if (!binding) return { configured: false, connected: false, state: "disconnected", deliveries: [] }; + this.owned(botId, threadId); + const connection = this.connections.get(threadId); + const state = connection?.state ?? "disconnected"; + return { + configured: true, connected: state === "ready" || state === "awaiting-approval", state, + instanceId: binding.instanceId, conversationId: binding.conversationId, title: binding.title, + settings: binding.settings && { model: binding.settings.model, reasoningEffort: binding.settings.reasoningEffort, cwd: binding.settings.cwd, approvalPolicy: binding.settings.approvalPolicy, sandbox: binding.settings.sandbox }, error: connection?.error, running: Boolean(connection?.running), + deliveries: binding.deliveries.map(({ sendId, state, turnId }) => ({ sendId, state, turnId })), + }; + } + + async list(instanceId: string, searchTerm = "", cursor?: string): Promise { + const client = await this.deps.connect(instanceId); + try { + await this.identity(client); + const result = await client.request("thread/list", { limit: 30, searchTerm, ...(cursor ? { cursor } : {}) }); + if (!Array.isArray(result?.data)) throw new Error("Codex returned an unsupported conversation list."); + return { conversations: result.data.map((thread: any) => ({ id: thread.id, title: thread.name ?? thread.preview ?? thread.id, cwd: thread.cwd })), nextCursor: result.nextCursor ?? null }; + } finally { await client.close(); } + } + + private async identity(client: CodexNativeClient): Promise { + const result = await client.request("account/read", { refreshToken: false }); + if (result?.account?.type !== "chatgpt" || typeof result.account.email !== "string") throw new Error("Sign in to Codex with the ChatGPT account that owns the conversation. API-key accounts are not supported by this connection."); + const initialized = await client.request("config/read", { includeLayers: false }); + if (typeof initialized?.config?.model_provider === "string" && initialized.config.model_provider !== "openai") throw new Error("An existing conversation connection requires Codex's native OpenAI provider."); + return createHash("sha256").update(JSON.stringify([client.home, result.account.email.toLowerCase()])).digest("hex"); + } + + async bind(botId: string, threadId: string, instanceId: string, conversationId: string): Promise { + if (this.stopped) refuse("OpenMausBot is stopping."); + if (!this.deps.store.taskByThread(botId, threadId)) refuse("No such thread.", 404); + if (!/^[\w-]{1,128}$/.test(conversationId)) refuse("Select an exact Codex conversation ID.", 400); + if (this.bindings.has(threadId)) refuse("Disconnect the current Codex conversation before selecting another."); + if (this.deps.busy(botId, threadId)) refuse("Stop this thread's current turn before connecting Codex."); + if ([...this.bindings.values()].some((binding) => binding.instanceId === instanceId && binding.conversationId === conversationId)) refuse("That Codex conversation is already reserved by another thread."); + const binding: Binding = { botId, threadId, instanceId, conversationId, deliveries: [] }; + this.bindings.set(threadId, binding); + try { this.persist(); } catch (error) { this.bindings.delete(threadId); throw error; } + return this.reconnect(botId, threadId); + } + + async reconnect(botId: string, threadId: string): Promise { + const binding = this.owned(botId, threadId); + const previous = this.connections.get(threadId); + if (previous?.state === "connecting" || previous?.client) refuse("The Codex connection is already open or connecting."); + const connection: Connection = { state: "connecting", work: Promise.resolve(), asks: new Map(), pumping: false, terminalTurns: new Set() }; + this.connections.set(threadId, connection); + try { + const client = await this.deps.connect(binding.instanceId); + connection.client = client; + client.onClose(() => this.closed(binding, connection)); + const identity = await this.identity(client); + if (binding.identity && binding.identity !== identity) throw new Error("The signed-in Codex account changed. Restore the original account before reconnecting."); + if ([...this.bindings.values()].some((other) => other !== binding && other.identity === identity && other.conversationId === binding.conversationId)) throw new Error("This Codex conversation is already reserved through another instance."); + binding.identity = identity; + this.persist(); + const read = await client.request("thread/read", { threadId: binding.conversationId }); + if (read?.thread?.id !== binding.conversationId) throw new Error("Codex did not return the selected existing conversation."); + if (read.thread.status?.type === "active") throw new Error("This Codex conversation is already running. Finish or stop it in Codex before reconnecting."); + await client.verifySavedPermissions(read.thread); + const resumed = await client.request("thread/resume", { threadId: binding.conversationId, excludeTurns: true }); + if (resumed?.thread?.id !== binding.conversationId) throw new Error("Codex resumed a different conversation. No message was sent."); + const settings = this.settings(resumed); + if (typeof read.thread.model !== "string" || read.thread.model !== settings.model || read.thread.cwd !== settings.cwd || read.thread.reasoningEffort !== settings.reasoningEffort) throw new Error("Codex could not preserve the selected conversation's saved model, effort, or working folder. No message was sent."); + if (binding.settings && !same(binding.settings, settings)) throw new Error("Codex's saved settings changed. No message was sent. Review them in Codex, then disconnect and select the conversation again to accept its current settings."); + binding.identity = identity; + binding.title = resumed.thread.name ?? read.thread.name ?? binding.conversationId; + binding.settings = settings; + this.persist(); + client.onMessage((message) => { + connection.work = connection.work.then(() => this.receive(binding, connection, message)).catch(() => { + connection.state = "failed"; + connection.error = "A Codex response could not be saved. Reconnect to recover it from native history."; + void client.close().catch(() => {}); + }); + return connection.work; + }); + await this.reconcile(binding, connection); + if (this.stopped || this.bindings.get(threadId) !== binding || connection.state !== "connecting") throw new Error("The Codex connection closed before it was ready."); + connection.state = "ready"; + void this.pump(binding, connection); + } catch (error) { + connection.state = "failed"; + connection.error = error instanceof Error ? error.message : "Codex could not connect."; + const client = connection.client; + connection.client = undefined; + await client?.close().catch(() => {}); + } + return this.view(botId, threadId); + } + + private settings(result: any): NonNullable { + if (typeof result.model !== "string" || typeof result.cwd !== "string" || !result.sandbox || result.approvalPolicy === undefined || !(result.reasoningEffort === null || typeof result.reasoningEffort === "string")) throw new Error("Codex did not report its effective saved settings. No message was sent."); + return { model: result.model, modelProvider: result.modelProvider, reasoningEffort: result.reasoningEffort, cwd: result.cwd, sandbox: result.sandbox, approvalPolicy: result.approvalPolicy, approvalsReviewer: result.approvalsReviewer, collaborationMode: result.collaborationMode }; + } + + private notice(binding: Binding, itemId: string, text: string, turnId = "connection"): void { + const existing = this.deps.store.messagesFor(binding.threadId).some((message) => message.codex?.conversationId === binding.conversationId && message.codex.itemId === itemId && message.codex.turnId === turnId); + if (!existing) this.deps.store.appendMessage(binding.threadId, { role: "bot", kind: "activity", tool: { name: text, ok: false }, codex: { conversationId: binding.conversationId, turnId, itemId } }); + } + + private closed(binding: Binding, connection: Connection): void { + connection.client = undefined; + if (connection.state !== "failed") connection.state = "disconnected"; + for (const delivery of binding.deliveries) if (delivery.state === "dispatching" || delivery.state === "delivered") delivery.state = "uncertain"; + for (const ask of connection.asks.values()) this.expire(binding, ask.messageId); + connection.asks.clear(); + connection.running = undefined; + this.persist(); + } + + private expire(binding: Binding, messageId: string): void { + const message = this.deps.store.messagesFor(binding.threadId).find((message) => message.id === messageId); + if (message?.card && !message.card.answered) this.deps.store.patchMessage(binding.threadId, message.id, { card: { ...message.card, answered: "unavailable", expired: true } }); + } + + private async reconcile(binding: Binding, connection: Connection): Promise { + const client = connection.client!; + let cursor: string | undefined; + const cursors = new Set(); + const turns: any[] = []; + do { + const page = await client.request("thread/turns/list", { threadId: binding.conversationId, limit: 100, sortDirection: "asc", itemsView: "full", ...(cursor ? { cursor } : {}) }); + if (!Array.isArray(page?.data)) throw new Error("Codex history is unavailable. No message was sent."); + turns.push(...page.data); + cursor = page.nextCursor ?? undefined; + if (cursor && cursors.has(cursor) || turns.length > 10_000) throw new Error("Codex history could not be reconciled safely."); + if (cursor) cursors.add(cursor); + } while (cursor); + for (const turn of turns) { + if (!Array.isArray(turn.items) || typeof turn.id !== "string") throw new Error("Codex returned unsupported history."); + for (const item of turn.items) { + if (item.type === "userMessage" && item.clientId) { + const delivery = binding.deliveries.find((delivery) => delivery.sendId === item.clientId); + if (delivery) { delivery.turnId = turn.id; delivery.state = this.terminal(turn.status) ?? "uncertain"; } + } + await this.item(binding, turn.id, item); + } + if (turn.status === "inProgress") connection.running = turn.id; + } + for (const message of this.deps.store.messagesFor(binding.threadId)) if (message.codex?.conversationId === binding.conversationId && message.card && !message.card.answered) this.expire(binding, message.id); + this.persist(); + } + + private terminal(status: string): CodexDeliveryState | undefined { + return status === "completed" ? "completed" : status === "interrupted" ? "interrupted" : status === "failed" ? "failed" : undefined; + } + + private async item(binding: Binding, turnId: string, item: any): Promise { + if (typeof item?.id !== "string" || !["agentMessage", "imageGeneration", "imageView"].includes(item.type)) return; + if (this.deps.store.messagesFor(binding.threadId).some((message) => message.codex?.conversationId === binding.conversationId && message.codex.turnId === turnId && message.codex.itemId === item.id)) return; + const { attachments, unavailable } = await codexItemAttachments(item, binding.settings!.cwd); + const text = item.type === "agentMessage" && typeof item.text === "string" ? item.text : undefined; + if (text || attachments.length) this.deps.store.appendMessage(binding.threadId, { + role: "bot", kind: "text", text, ...(attachments.length ? { attachments } : {}), + codex: { conversationId: binding.conversationId, turnId, itemId: item.id }, + }); + if (unavailable) this.notice(binding, `${item.id}:unavailable`, "A Codex attachment could not be accessed or is unsupported. Upload it here from its original location.", turnId); + } + + async send(botId: string, threadId: string, text: string, sendId: string): Promise { + const binding = this.owned(botId, threadId); + if (!/^[A-Za-z0-9_-]{16,80}$/.test(sendId)) refuse("A stable sendId is required for Codex delivery.", 400); + if (!text.trim() || text.length > 1_000_000) refuse("Enter a message of at most 1,000,000 characters.", 400); + const previous = binding.deliveries.find((delivery) => delivery.sendId === sendId); + if (previous && previous.text !== text) refuse("sendId already belongs to another message."); + const accepted = this.deps.store.messagesFor(threadId).find((message) => message.role === "user" && message.sendId === sendId); + if (accepted && !previous) { + if (accepted.text !== text || accepted.codex?.conversationId !== binding.conversationId) refuse("sendId already belongs to another message."); + return accepted; + } + if (!previous) { + const connection = this.connections.get(threadId); + if (!connection?.client || !["ready", "awaiting-approval"].includes(connection.state)) refuse("Reconnect Codex before sending. This message has not been delivered."); + codexUserInput(text); + if (binding.deliveries.filter((delivery) => delivery.state === "queued").length >= 20) refuse("The Codex message queue is full.", 429); + binding.deliveries.push({ sendId, text, state: "queued" }); + this.persist(); + } + let message = this.deps.store.messagesFor(threadId).find((message) => message.role === "user" && message.sendId === sendId); + if (!message) message = this.deps.store.appendMessage(threadId, { role: "user", kind: "text", text, sendId, codex: { conversationId: binding.conversationId, turnId: previous?.turnId ?? "pending", itemId: sendId } }); + const connection = this.connections.get(threadId); + if (connection) void this.pump(binding, connection); + return message; + } + + private async pump(binding: Binding, connection: Connection): Promise { + if (connection.pumping || connection.running || connection.state !== "ready" || !connection.client || this.stopped) return; + if (binding.deliveries.some((delivery) => delivery.state === "uncertain")) return; + const delivery = binding.deliveries.find((delivery) => delivery.state === "queued"); + if (!delivery) return; + connection.pumping = true; + try { + const client = connection.client; + if (await this.identity(client) !== binding.identity) throw new Error("The signed-in Codex account changed. Restore the original account before reconnecting."); + const input = codexUserInput(delivery.text); + if (connection.client !== client || connection.state !== "ready" || this.stopped) return; + delivery.state = "dispatching"; + this.persist(); + const result = await client.request("turn/start", { threadId: binding.conversationId, clientUserMessageId: delivery.sendId, input }); + if (typeof result?.turn?.id !== "string") throw new Error("Codex did not acknowledge a native turn."); + delivery.turnId = result.turn.id; + if (delivery.state === "dispatching") delivery.state = this.terminal(result.turn.status) ?? "delivered"; + if (!this.terminal(result.turn.status) && !connection.terminalTurns.has(result.turn.id)) connection.running = result.turn.id; + this.persist(); + } catch (error) { + if (delivery.state === "dispatching") delivery.state = error instanceof CodexNativeRejection ? "failed" : "uncertain"; + connection.state = "failed"; + connection.error = error instanceof Error ? error.message : "Codex delivery failed."; + this.persist(); + this.notice(binding, `delivery:${delivery.sendId}`, delivery.state === "uncertain" ? "Codex delivery is uncertain. Reconnect and review the native conversation before sending it again." : "Codex did not accept this message. Reconnect before trying again."); + const client = connection.client; + connection.client = undefined; + await client?.close().catch(() => {}); + } finally { + connection.pumping = false; + if (!connection.running) void this.pump(binding, connection); + } + } + + private async receive(binding: Binding, connection: Connection, message: any): Promise { + if (this.bindings.get(binding.threadId) !== binding) return; + const params = message.params; + if (params?.threadId !== binding.conversationId) { + if (message.id !== undefined) connection.client?.reject(message.id); + return; + } + if (message.id !== undefined) { this.ask(binding, connection, message); return; } + if (message.method === "item/completed" && typeof params.turnId === "string") await this.item(binding, params.turnId, params.item); + if (message.method === "turn/started" && typeof params.turn?.id === "string") connection.running = params.turn.id; + if (message.method === "serverRequest/resolved") { + for (const [id, ask] of connection.asks) if (ask.id === params.requestId) { this.expire(binding, ask.messageId); connection.asks.delete(id); } + if (!connection.asks.size && connection.state === "awaiting-approval") connection.state = "ready"; + } + if (message.method === "turn/completed" && typeof params.turn?.id === "string") { + connection.terminalTurns.add(params.turn.id); + for (const item of params.turn.items ?? []) await this.item(binding, params.turn.id, item); + for (const delivery of binding.deliveries) if (delivery.turnId === params.turn.id || delivery.state === "dispatching" && !delivery.turnId) { + delivery.turnId = params.turn.id; + delivery.state = this.terminal(params.turn.status) ?? "uncertain"; + } + if (connection.running === params.turn.id) connection.running = undefined; + for (const [id, ask] of connection.asks) if (ask.params.turnId === params.turn.id) { this.expire(binding, ask.messageId); connection.asks.delete(id); } + if (!connection.asks.size && connection.state === "awaiting-approval") connection.state = "ready"; + this.persist(); + void this.pump(binding, connection); + } + if (message.method === "thread/settings/updated") { + connection.state = "failed"; + connection.error = "Codex settings changed while connected. Reconnect and review the conversation before sending again."; + await connection.client?.close(); + } + if (message.method === "thread/closed" || message.method === "error" && !params.willRetry) { + connection.state = "failed"; + connection.error = "Codex reported a connection or turn failure. Reconnect to recover native history."; + await connection.client?.close(); + } + } + + private ask(binding: Binding, connection: Connection, message: any): void { + const { method, params, id } = message; + const question = method === "item/tool/requestUserInput" ? parseProtocolAskQuestions(params.questions) : null; + const permission = (method === "item/commandExecution/requestApproval" || method === "item/fileChange/requestApproval") && + params.grantRoot == null && params.additionalPermissions == null && + (!Array.isArray(params.availableDecisions) || params.availableDecisions.includes("accept") && params.availableDecisions.includes("decline")); + if ((!permission && !question) || typeof params.turnId !== "string" || params.turnId !== connection.running) { + connection.client?.reject(id); + this.notice(binding, `unsupported:${String(id)}`, "Codex requested an unsupported approval or input interface. No permission was granted. Continue in Codex or cancel this turn."); + return; + } + if ([...connection.asks.values()].some((ask) => ask.id === id)) return; + const requestId = `codex-${randomUUID()}`; + const subtitle = [params.reason, params.command, params.cwd, params.grantRoot && `Write access: ${params.grantRoot}`, params.networkApprovalContext && JSON.stringify(params.networkApprovalContext)].filter((value) => typeof value === "string").join("\n"); + const saved = this.deps.store.appendMessage(binding.threadId, { + role: "bot", kind: "options", codex: { conversationId: binding.conversationId, turnId: params.turnId, itemId: `request:${requestId}` }, + card: { + title: question ? "Codex needs your answer" : "Codex needs your approval", subtitle, requestId, + requestType: question ? "question" : "permission", tool: "Codex", options: question ? [] : ["Allow", "Deny"], + ...(question ? { questionRequest: { version: 1, questions: question.map(({ question }) => question) } } : {}), + }, + }); + connection.asks.set(requestId, { id, method, params, messageId: saved.id }); + connection.state = "awaiting-approval"; + } + + async answer(botId: string, threadId: string, requestId: string, behavior: string, message?: string): Promise { + const binding = this.owned(botId, threadId); + const connection = this.connections.get(threadId); + const ask = connection?.asks.get(requestId); + if (!connection?.client || !ask) return "unavailable"; + if (await this.identity(connection.client) !== binding.identity) refuse("The Codex account changed. No approval was sent.", 403); + if (!connection.asks.has(requestId)) return "unavailable"; + const question = ask.method === "item/tool/requestUserInput" ? parseProtocolAskQuestions(ask.params.questions) : null; + if (question ? !["answer", "deny"].includes(behavior) : !["allow", "deny"].includes(behavior)) refuse("This answer does not match the native request.", 400); + connection.client.reply(ask.id, question ? codexQuestionReply(behavior as "answer" | "deny", message, "user", question) : { decision: behavior === "allow" ? "accept" : "decline" }); + connection.asks.delete(requestId); + const saved = this.deps.store.messagesFor(threadId).find((item) => item.id === ask.messageId); + if (saved?.card) this.deps.store.patchMessage(threadId, saved.id, { card: { ...saved.card, answered: behavior, ...(question && message ? { answeredText: message } : {}) } }); + if (!connection.asks.size) connection.state = "ready"; + return behavior === "allow" ? "allowed-once" : behavior === "answer" ? "answered" : "rejected"; + } + + async cancel(botId: string, threadId: string, sendId?: string): Promise { + const binding = this.owned(botId, threadId); + const connection = this.connections.get(threadId); + if (sendId) { + const delivery = binding.deliveries.find((delivery) => delivery.sendId === sendId); + if (!delivery) refuse("No such Codex delivery.", 404); + if (delivery.state === "queued") { delivery.state = "cancelled"; this.persist(); return; } + if (delivery.turnId !== connection?.running) refuse("Only a queued or currently running delivery can be cancelled."); + } + if (!connection?.client || !connection.running) refuse("No connected native turn is running. Reconnect and inspect Codex before retrying."); + await connection.client.request("turn/interrupt", { threadId: binding.conversationId, turnId: connection.running }); + } + + acknowledge(botId: string, threadId: string, sendId: string): void { + const binding = this.owned(botId, threadId); + const delivery = binding.deliveries.find((delivery) => delivery.sendId === sendId); + if (delivery?.state !== "uncertain") refuse("Only an uncertain delivery can be acknowledged."); + delivery.state = "cancelled"; + this.persist(); + const connection = this.connections.get(threadId); + if (connection) void this.pump(binding, connection); + } + + async disconnect(botId: string, threadId: string): Promise { + const binding = this.owned(botId, threadId); + const connection = this.connections.get(threadId); + if (connection?.state === "connecting" || connection?.running || binding.deliveries.some((delivery) => ["queued", "dispatching", "delivered", "uncertain"].includes(delivery.state))) refuse("Cancel queued work and finish or review the native turn before disconnecting."); + await connection?.client?.close(); + await connection?.work; + this.bindings.delete(threadId); + this.connections.delete(threadId); + this.persist(); + } + + async close(): Promise { + this.stopped = true; + await Promise.all([...this.connections.values()].map((connection) => connection.client?.close())); + await Promise.all([...this.connections.values()].map((connection) => connection.work)); + } +} diff --git a/server/contracts.ts b/server/contracts.ts index 137ddf1b0e..97e630e78b 100644 --- a/server/contracts.ts +++ b/server/contracts.ts @@ -417,6 +417,7 @@ export interface TextGenerationOptions { } export interface ProviderInstance { + readonly nativeCodex?: () => Promise; readonly instanceId: InstanceId; readonly driverKind: DriverKind; readonly displayName: string | undefined; diff --git a/server/drivers/codex-native-client.test.ts b/server/drivers/codex-native-client.test.ts new file mode 100644 index 0000000000..fd10f58093 --- /dev/null +++ b/server/drivers/codex-native-client.test.ts @@ -0,0 +1,28 @@ +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { afterEach, expect, it } from "vitest"; +import { connectCodexNative, type CodexNativeClient } from "./codex-native-client.ts"; + +const clients: CodexNativeClient[] = []; +const dirs: string[] = []; +afterEach(async () => { await Promise.all(clients.splice(0).map((client) => client.close())); for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); }); +const cli = fileURLToPath(new URL("../testing/fake-codex-native.ts", import.meta.url)); + +it("uses the native stdio protocol without model, sandbox, approval or MCP overrides", async () => { + const dir = mkdtempSync(join(tmpdir(), "omb-native-cli-")); dirs.push(dir); + const client = await connectCodexNative(cli, { ...process.env, HOME: dir, USERPROFILE: dir, FAKE_CODEX_NATIVE_DIR: dir }); clients.push(client); + expect(client.home).toBe(dir); + const result = await client.request("thread/resume", { threadId: "fixture-native-existing", excludeTurns: true }); + expect(result.thread.id).toBe("fixture-native-existing"); + const calls = readFileSync(join(dir, "native-fixture-requests.jsonl"), "utf8").trim().split("\n").map((line) => JSON.parse(line)); + expect(calls.map((call) => call.method)).toEqual(["initialize", "initialized", "thread/resume"]); + expect(calls.every((call) => call.argv.includes("app-server") && !call.argv.includes("-c"))).toBe(true); +}); + +it("refuses installed versions without verified settings preservation before any app-server runs", async () => { + const dir = mkdtempSync(join(tmpdir(), "omb-native-version-")); dirs.push(dir); + await expect(connectCodexNative(cli, { ...process.env, HOME: dir, USERPROFILE: dir, FAKE_CODEX_NATIVE_DIR: dir, FAKE_CODEX_NATIVE_VERSION: "codex-cli 0.159.2" })).rejects.toThrow(/0.163/); + expect(() => readFileSync(join(dir, "native-fixture-requests.jsonl"))).toThrow(); +}); diff --git a/server/drivers/codex-native-client.ts b/server/drivers/codex-native-client.ts new file mode 100644 index 0000000000..476e8a88b9 --- /dev/null +++ b/server/drivers/codex-native-client.ts @@ -0,0 +1,127 @@ +import { homedir } from "node:os"; +import { realpath } from "node:fs/promises"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { execCli, killCliTree, spawnCli } from "../procs.ts"; +import { codexLaunch } from "./codex-process.ts"; +import { verifyCodexSavedPermissions } from "./codex-saved-permissions.ts"; + +export interface CodexNativeClient { + readonly home: string; + verifySavedPermissions(thread: { id: string; path?: unknown }): Promise; + request(method: string, params: unknown): Promise; + reply(id: string | number, result: unknown): void; + reject(id: string | number): void; + onMessage(listener: (message: any) => void | Promise): void; + onClose(listener: () => void): void; + close(): Promise; +} + +export class CodexNativeRejection extends Error {} + +/** Codex 0.163.x app-server, generated locally with `app-server generate-ts`; no configuration overrides. */ +export async function connectCodexNative(cli: string, env: Record): Promise { + const version = await new Promise((resolve, reject) => { + execCli(cli, ["--version"], { env, timeout: 8_000 }, (error, stdout) => error ? reject(new Error("Codex CLI is unavailable.")) : resolve(stdout.trim())); + }); + if (!/^codex-cli 0\.163\.\d+$/.test(version)) throw new Error("Existing conversation connections require Codex 0.163.x. This CLI version has not been verified."); + if (/(?:^|\s)(?:-c|--config|--profile|-p|--model|-m|--sandbox|--ask-for-approval)(?:\s|=|$)/.test(cli)) { + throw new Error("Use a Codex CLI without configuration overrides for an existing conversation."); + } + const child = spawnCli(...codexLaunch(cli, ["app-server"]), { + env, cwd: env.HOME || env.USERPROFILE || homedir(), stdio: ["pipe", "pipe", "pipe"], + }) as ChildProcessWithoutNullStreams; + let sequence = 0; + let closed = false; + let stopping: Promise | undefined; + let buffer = ""; + let messageListener: (message: any) => void = () => {}; + const closeListeners = new Set<() => void>(); + let home = ""; + const pending = new Map }>(); + const fail = () => { + if (closed) return; + closed = true; + for (const request of pending.values()) { + clearTimeout(request.timer); + request.reject(new Error("Codex connection closed; delivery may be uncertain.")); + } + pending.clear(); + for (const listener of closeListeners) listener(); + }; + const send = (message: unknown) => { + if (closed || !child.stdin.writable) throw new Error("Codex is disconnected."); + child.stdin.write(`${JSON.stringify(message)}\n`); + }; + const close = () => stopping ??= (async () => { + fail(); + if (!await killCliTree(child, 3_000)) throw new Error("The Codex process could not be confirmed stopped."); + })(); + child.stderr.resume(); + child.on("error", fail); + child.on("close", fail); + child.stdin.on("error", fail); + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { + buffer += chunk; + if (Buffer.byteLength(buffer) > 64 * 1024 * 1024) { void close().catch(() => {}); return; } + let newline: number; + while (!closed && (newline = buffer.indexOf("\n")) >= 0) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + if (!line.trim()) continue; + let message: any; + try { message = JSON.parse(line); } catch { void close().catch(() => {}); return; } + if (message && typeof message.id === "number" && !message.method) { + const request = pending.get(message.id); + if (!request) continue; + pending.delete(message.id); + clearTimeout(request.timer); + if (message.error) request.reject(new CodexNativeRejection("Codex rejected the request. Check the conversation and account in Codex.")); + else request.resolve(message.result); + } else if (message?.method) messageListener(message); + } + }); + const client: CodexNativeClient = { + get home() { return home; }, + verifySavedPermissions: (thread) => verifyCodexSavedPermissions(home, thread), + request: (method, params) => new Promise((resolve, reject) => { + if (closed) { reject(new Error("Codex is disconnected.")); return; } + const id = ++sequence; + const timer = setTimeout(() => { + pending.delete(id); + reject(new Error("Codex request timed out; delivery may be uncertain.")); + void close().catch(() => {}); + }, 30_000); + timer.unref(); + pending.set(id, { resolve, reject, timer }); + try { send({ id, method, params }); } catch { void close().catch(() => {}); } + }), + reply: (id, result) => send({ id, result }), + reject: (id) => send({ id, error: { code: -32601, message: "Unsupported OpenMausBot approval interface; no action was approved." } }), + onMessage: (listener) => { messageListener = listener; }, + onClose: (listener) => { closeListeners.add(listener); if (closed) listener(); }, + close, + }; + try { + const initialized = await client.request("initialize", { clientInfo: { name: "openmausbot_native_bridge", version: "1" } }); + if (typeof initialized?.codexHome !== "string" || !initialized.codexHome) throw new Error("Codex did not identify its local conversation store."); + home = await realpath(initialized.codexHome); + send({ method: "initialized", params: {} }); + return client; + } catch (error) { await close(); throw error; } +} + +export function codexNativeAccess(cli: string, env: () => Record, disposed: () => boolean) { + const clients = new Set(); + return { + connect: async () => { + if (disposed()) throw new Error("This Codex instance was removed."); + const client = await connectCodexNative(cli, env()); + if (disposed()) { await client.close(); throw new Error("This Codex instance was removed."); } + clients.add(client); + client.onClose(() => clients.delete(client)); + return client; + }, + close: async () => { await Promise.all([...clients].map((client) => client.close())); }, + }; +} diff --git a/server/drivers/codex-saved-permissions.test.ts b/server/drivers/codex-saved-permissions.test.ts new file mode 100644 index 0000000000..615751403a --- /dev/null +++ b/server/drivers/codex-saved-permissions.test.ts @@ -0,0 +1,38 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, expect, it } from "vitest"; +import { verifyCodexSavedPermissions } from "./codex-saved-permissions.ts"; + +const dirs: string[] = []; +afterEach(() => { for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); }); +function fixture() { + const home = mkdtempSync(join(tmpdir(), "omb-saved-codex-")); dirs.push(home); + mkdirSync(join(home, "sessions")); + const path = join(home, "sessions", "fixture.jsonl"); + const meta = { type: "session_meta", payload: { id: "selected-thread" } }; + const snapshot = { type: "event_msg", payload: { type: "thread_settings_applied", thread_id: "selected-thread", thread_settings: { cwd: home, approval_policy: "on-request", permission_profile: { fixture: "read-only" } } } }; + const write = (items: unknown[]) => writeFileSync(path, items.map((item) => JSON.stringify(item)).join("\n")); + return { home, path, meta, snapshot, write }; +} + +it("requires a thread-owned saved permission snapshot before resuming", async () => { + const h = fixture(); h.write([h.meta, h.snapshot]); + await expect(verifyCodexSavedPermissions(h.home, { id: "selected-thread", path: h.path })).resolves.toBeUndefined(); +}); + +it("refuses old histories and copied snapshots from another thread", async () => { + const h = fixture(); + for (const items of [[h.meta], [h.meta, { ...h.snapshot, payload: { ...h.snapshot.payload, thread_id: "foreign-thread" } }], [{ ...h.meta, payload: { id: "foreign-thread" } }, h.snapshot]]) { + h.write(items); + await expect(verifyCodexSavedPermissions(h.home, { id: "selected-thread", path: h.path })).rejects.toThrow(/could not be verified/); + } +}); + +it("refuses missing, remote-only, corrupt and outside-store histories", async () => { + const h = fixture(); h.write([h.meta, h.snapshot]); + await expect(verifyCodexSavedPermissions(h.home, { id: "selected-thread" })).rejects.toThrow(); + await expect(verifyCodexSavedPermissions(join(h.home, "another-account"), { id: "selected-thread", path: h.path })).rejects.toThrow(); + writeFileSync(h.path, "{corrupt"); + await expect(verifyCodexSavedPermissions(h.home, { id: "selected-thread", path: h.path })).rejects.toThrow(); +}); diff --git a/server/drivers/codex-saved-permissions.ts b/server/drivers/codex-saved-permissions.ts new file mode 100644 index 0000000000..a14bc46f7b --- /dev/null +++ b/server/drivers/codex-saved-permissions.ts @@ -0,0 +1,37 @@ +import { constants } from "node:fs"; +import { open, realpath } from "node:fs/promises"; +import { isAbsolute, join, relative, sep } from "node:path"; + +function contained(root: string, file: string): boolean { + const suffix = relative(root, file); + return suffix !== "" && suffix !== ".." && !suffix.startsWith(`..${sep}`) && !isAbsolute(suffix); +} + +/** Codex 0.163.0 restores permissions from a thread-owned ThreadSettingsApplied rollout snapshot. */ +export async function verifyCodexSavedPermissions(home: string, thread: { id: string; path?: unknown }): Promise { + let handle: Awaited> | undefined; + try { + if (typeof thread.path !== "string" || !isAbsolute(thread.path)) throw new Error(); + const canonicalFile = await realpath(thread.path); + const roots = await Promise.all(["sessions", "archived_sessions"].map((directory) => realpath(join(home, directory)).catch(() => null))); + if (!roots.some((root) => root && contained(root, canonicalFile))) throw new Error(); + handle = await open(canonicalFile, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)); + const before = await handle.stat(); + if (!before.isFile() || before.size <= 0 || before.size > 64 * 1024 * 1024) throw new Error(); + let sessionId: string | undefined; + let snapshot = false; + for (const line of (await handle.readFile("utf8")).split("\n")) { + if (!line.trim()) continue; + const item = JSON.parse(line); + if (item.type === "session_meta") sessionId = item.payload?.id; + if (item.type === "event_msg" && item.payload?.type === "thread_settings_applied" && item.payload.thread_id === thread.id) { + const settings = item.payload.thread_settings; + snapshot = Boolean(settings && typeof settings.cwd === "string" && settings.approval_policy !== undefined && settings.permission_profile && typeof settings.permission_profile === "object"); + } + } + const after = await handle.stat(); + if (sessionId !== thread.id || !snapshot || before.size !== after.size || before.mtimeMs !== after.mtimeMs || await realpath(thread.path) !== canonicalFile) throw new Error(); + } catch { + throw new Error("Codex's saved permission snapshot could not be verified. No conversation was resumed or message sent. Review and save the original conversation's settings in Codex 0.163.x before connecting; remote-only or unsupported history stores cannot be bound."); + } finally { await handle?.close(); } +} diff --git a/server/drivers/codex.ts b/server/drivers/codex.ts index af5d9db88f..1aa7fe965b 100644 --- a/server/drivers/codex.ts +++ b/server/drivers/codex.ts @@ -8,7 +8,6 @@ import { createHash } from "node:crypto"; import { join } from "node:path"; import { homedir } from "node:os"; import { codexConfigMcp, codexConfigRunsServer, mountedMcpServerName } from "./codex-mcp-names.ts"; - import { DATA_DIR, stripWorkspaceCredentialEnv } from "../config.ts"; import { laterStartup, openStartupModelCatalog, writeStartupModelCache } from "../startup-model-catalog.ts"; import { hostedWorkspaceConfigured } from "../enterprise.ts"; @@ -45,6 +44,7 @@ import { codexDeveloperInstructions, syncCodexInstructions } from "./codex-instr import { volatileContextNote, withContextNote } from "./prompt-split.ts"; import type { ApprovalMode } from "../../shared/approval-mode.ts"; import { CodexDeviceAuthController } from "./codex-device-auth.ts"; +import { codexNativeAccess } from "./codex-native-client.ts"; import { codexAccountEmail, codexHome } from "./codex-identity.ts"; import { CODEX_MISSING_CREDENTIALS, CODEX_SIGN_IN_EXPIRED, codexLimitFacts, codexNotSignedIn, codexSignInExpired, codexSignInRefused, codexUserError, type CodexRoute } from "./codex-errors.ts"; import { codexLoginCheck, codexLoginStatus, storedCodexLogin } from "./codex-login.ts"; @@ -557,7 +557,7 @@ export const CodexDriver: ProviderDriver = { const planAuth = plan ? new ChatGptPlanAuthController({ directory: planDirectory }) : null; let planGeneration = 0; let planSigningOut = false; - let disposed = false; + let disposed = false; const nativeAccess = codexNativeAccess(config.cli, childEnv, () => disposed); let planWarning: string | undefined; let models = plan ? { default: "", options: [] } : config.managed ? { default: config.managed.models[0], options: config.managed.models.map(id => ({ id, label: id })) } : STATIC_CODEX_MODELS; const refreshModels = async (preserveExisting = false) => { @@ -1781,7 +1781,7 @@ export const CodexDriver: ProviderDriver = { }, refreshModels, ...(startupModelRefresh ? { startupModelRefresh } : {}), - generateMemoryText, + generateMemoryText, ...(!plan && !config.managed ? { nativeCodex: nativeAccess.connect } : {}), ...(plan ? { authenticationMethod: "browser-pkce" as const } : {}), startAuthentication: async () => { if (planUnavailable) throw new Error(planUnavailable); @@ -1796,7 +1796,7 @@ export const CodexDriver: ProviderDriver = { }, cancelAuthentication: () => (planAuth ?? authentication).cancel(), signOut: async () => { - if (!planAuth) { + if (!planAuth) { await nativeAccess.close(); const mark = loginMark(); await authentication.signOut(); loginAccepted(mark); @@ -1860,7 +1860,7 @@ export const CodexDriver: ProviderDriver = { }, }, dispose: async () => { - disposed = true; + disposed = true; await nativeAccess.close(); planGeneration++; await authentication.dispose(); await planAuth?.dispose(); diff --git a/server/host/codex-bindings.ts b/server/host/codex-bindings.ts new file mode 100644 index 0000000000..c58bba9355 --- /dev/null +++ b/server/host/codex-bindings.ts @@ -0,0 +1,15 @@ +import { CodexBindings, type CodexBindingsDeps } from "../codex-bindings.ts"; +import type { ProviderRegistry } from "../harness/registry.ts"; +import { createCodexBindingRoutes } from "../routes/codex-bindings.ts"; + +export function createCodexBindingsHost(deps: Omit & { registry: ProviderRegistry; enabled: boolean }) { + const bridge = new CodexBindings({ ...deps, connect: (instanceId) => { + if (!deps.enabled) throw new Error("Existing Codex connections require a local workspace."); + const connect = deps.registry.get(instanceId)?.nativeCodex; + if (!connect) throw new Error("This Codex instance does not support existing native conversations."); + return connect(); + } }); + const routes = createCodexBindingRoutes({ bridge, enabled: deps.enabled, instances: () => deps.registry.instances() + .filter((instance) => instance.enabled && instance.nativeCodex).map((instance) => ({ id: instance.instanceId, name: instance.displayName ?? instance.instanceId })) }); + return { bridge, routes }; +} diff --git a/server/index.ts b/server/index.ts index 2ac3a0a889..63e24f73e4 100644 --- a/server/index.ts +++ b/server/index.ts @@ -1,5 +1,6 @@ // Import first: it reads a Cloud home's secrets off the launcher's pipe before any module can start a process. import { BOOT_CLOUD_SECRETS } from "./cloud-secrets-boot.ts"; +import { createCodexBindingsHost } from "./host/codex-bindings.ts"; import { AsyncLocalStorage } from "node:async_hooks"; import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto"; import { existsSync, readFileSync, realpathSync, rmSync } from "node:fs"; @@ -49,7 +50,6 @@ import { type CredentialTargetId, } from "../shared/credential-request.ts"; import { isPendingSignInRequest, normalizeSignInSite, signInFallbackText, signInOutcome, signInResumePrompt } from "../shared/sign-in-request.ts"; - import { createRoomTurnLevels } from "./room-turn-level.ts"; import { RoomMemberLock } from "./room-member-lock.ts"; import { HELD_NOTE, approvalHeldNote, approvalHeldReason, approvalModeForOrigin, autoVerdict, deliverFullAccessApproval, delegatedApprovalMode, undeliveredApprovalTool } from "./auto-approve.ts"; @@ -235,7 +235,6 @@ import { type GoalRunMember, } from "./group-goal-run.ts"; import type { GroupGoalRunCardData, GroupGoalRunStatus } from "../shared/group-goal-run.ts"; - import { BUILT_IN_DRIVERS } from "./drivers/builtIn.ts"; import { openCodeProviderKeysAllowed, setOpenCodeProviderKeyPolicy } from "./drivers/acp/opencode-go.ts"; import { readMessageText, recallMessages, recentMessages, closeMessageSearch, closeMessageDb, chatFollowups, cancelledChatFollowup, settleChatFollowups, threadsReferencing, logNotification, lastRowAt } from "./message-db.ts"; @@ -3046,6 +3045,7 @@ const store = new Store( (instanceId) => registry.cliTarget(instanceId)?.driverKind, () => appLanguage() ?? Intl.DateTimeFormat().resolvedOptions().locale, ); +const nativeCodexHost = createCodexBindingsHost({ dataDir: DATA_DIR, store, registry, busy: threadBusy, enabled: !CLOUD_HOME && !HOSTED_WORKSPACE }); const creditMoves = cloudCredit && createCreditMoves({ credit: cloudCredit, store, context: selectionContext, threadBusy, saveDefault: saveNewBotModel, defaults: () => [cfg.defaultModelSelection, cfg.newBotDefaults?.profile.modelSelection], engineName: id => registry.get(id)?.displayName ?? id, log: line => console.warn(line) }); @@ -8094,7 +8094,7 @@ async function startTurn( workspaceMaintenance.assertAvailable(); moveFreeze.assertTurns(); const profile = store.bot(botId); assertBotHere(profile); if (!profile) throw Object.assign(new Error("no such bot"), { status: 404 }); - const threadId = opts?.threadId ?? profile.threadId; + const threadId = opts?.threadId ?? profile.threadId; nativeCodexHost.bridge.assertUnbound(threadId); const continuingRoutine = opts?.cardContinuation ? activeRoutineRunForThread(threadId) : null; if (continuingRoutine) { const onDispatchError = opts?.onDispatchError; @@ -13160,10 +13160,10 @@ ROUTES.push(createBotPresetRoutes({ presets: presetStore, orgStatuses: orgInstal ROUTES.push(createSharedComputerRoutes({ enabled: lendingEnabled, computers: sharedComputers, sessionLive: (id) => sessions.isLive(id), environmentId: ENVIRONMENT_ID, cloudHomeLender: CLOUD_HOME ? CLOUD_HOME_LENDER : null, principal: sessionLendingPrincipal, personKey, power: () => cloudPower, links: CLOUD_HOME ? deviceLinks : null, upgrades })); // `routines` is assigned at boot, so it is read per request. -ROUTES.push(createBotMessageRoutes({ +ROUTES.push(nativeCodexHost.routes); ROUTES.push(createBotMessageRoutes({ store, registry, handoffs, phoneSecretSubmissions, directTurnDispatchClaims, groupSpeakers, routines: () => routines, - desktopManaged: DESKTOP_MANAGED, cloudHome: Boolean(CLOUD_HOME), DirectSendRefused, + desktopManaged: DESKTOP_MANAGED, cloudHome: Boolean(CLOUD_HOME), DirectSendRefused, codexBindings: nativeCodexHost.bridge, acceptDirectSend, directSendRefusal, startTurn, usageTriggerFor, messageSender, cloudGuestSendRefusal, cloudThreadRefusal, cardAnswerRefusal, commandRememberRefusal, requestBehavior, requestedTaskBot, botForThread, threadBusy, botAtThreadCapacity, parksBehindCoordination, activeGroupTurnForBot, runningTurnInstance, engineSteersImages, resolveReplyTarget, @@ -16875,7 +16875,7 @@ const gracefulShutdown = createGracefulShutdown({ webhookIngress?.server.close(); tunnelListener?.close(); stopCloudHomeConnectedApps(); }, - async () => { await managedDesktop.close(); await registry.disposeAll(); }, + async () => { await nativeCodexHost.bridge.close(); await managedDesktop.close(); await registry.disposeAll(); }, async () => { await dataEngine.closeAll().catch((error) => console.warn("data: close failed", error)); }, async () => { const temporaries = [...temporaryBrowserSessions.keys()].map((botId) => forgetTemporaryBrowser(botId)); diff --git a/server/routes/bot-messages.ts b/server/routes/bot-messages.ts index c637ac50d9..f5a5fa9196 100644 --- a/server/routes/bot-messages.ts +++ b/server/routes/bot-messages.ts @@ -18,6 +18,8 @@ import { spokenLineFields } from "../live-call-record.ts"; import type { PhoneSecretSubmissionRegistry } from "../phone-secret.ts"; import type { ReplyTarget } from "../replies.ts"; import type { RequestAuth } from "../request-auth.ts"; +import { ownerOrAdmin } from "../request-auth.ts"; +import type { CodexBindings } from "../codex-bindings.ts"; import type { RoutineManager } from "../routines.ts"; import { acceptedSendMatch, parseSendId } from "../send-idempotency.ts"; import { cancelSteeredMessage, holdSteeredQueue, queuedSteeredMessage, restoreHeldSteeredQueue, settleHeldSteeredQueue } from "../steer-queue.ts"; @@ -43,6 +45,7 @@ type CardAnswer = { botId: string; botName?: string; threadId: string; requestId type Behavior = "allow" | "deny" | "answer"; export interface BotMessageRouteDeps { + codexBindings?: CodexBindings; store: Pick; @@ -191,6 +194,12 @@ export function createBotMessageRoutes(deps: BotMessageRouteDeps): RouteHandler const threadId = body.threadId ?? bot.threadId; const notYours = cloudGuestSendRefusal(auth, threadId); if (notYours) return json(res, 403, { error: notYours }); + if (deps.codexBindings?.configured(threadId)) { + if (!ownerOrAdmin(auth)) return json(res, 403, { error: "Only the local workspace owner can send to this Codex conversation." }); + if (guarded || pressed || fromAnApp || body.replyToId || body.dataContext) return json(res, 409, { error: "This Codex connection accepts direct text and images only. No message was sent." }); + const message = await deps.codexBindings.send(bot.id, threadId, String(body.text ?? ""), parseSendId(body.sendId) ?? ""); + return json(res, 202, { ok: true, threadId, message }); + } const onBehalfOf = guardedBody?.data?.onBehalfOf; const trigger: UsageTrigger = onBehalfOf ? { kind: "user", ...(onBehalfOf.email ? { email: onBehalfOf.email } : {}), ...(onBehalfOf.name ? { label: onBehalfOf.name } : {}) } @@ -447,6 +456,11 @@ export function createBotMessageRoutes(deps: BotMessageRouteDeps): RouteHandler const behavior = requestBehavior(body.behavior); const reviewedSha256 = typeof body.reviewedSha256 === "string" ? body.reviewedSha256 : undefined; if (!behavior) return json(res, 400, { error: "behavior must be allow, deny, or answer" }); + if (deps.codexBindings?.configured(bot.threadId) && String(body.requestId).startsWith("codex-")) { + if (!ownerOrAdmin(auth)) return json(res, 403, { error: "Only the local workspace owner can answer Codex." }); + if (body.always || body.rememberCommand) return json(res, 400, { error: "This connection supports approval of the displayed request only." }); + return json(res, 200, { ok: true, outcome: await deps.codexBindings.answer(bot.id, bot.threadId, String(body.requestId), behavior, body.message) }); + } const refusal = cardAnswerRefusal(auth, selected.threadId, String(body.requestId), behavior) ?? (selected.threadId === bot.threadId ? null : cardAnswerRefusal(auth, bot.threadId, String(body.requestId), behavior)); if (refusal) return json(res, 403, { error: refusal }); @@ -517,6 +531,13 @@ export function createBotMessageRoutes(deps: BotMessageRouteDeps): RouteHandler if (!behavior) return json(res, 400, { error: "behavior must be allow, deny, or answer" }); const requestId = String(body.requestId); const threadId = answeredIn(m[1], requestId); + if (deps.codexBindings?.configured(threadId) && requestId.startsWith("codex-")) { + if (!ownerOrAdmin(auth)) return json(res, 403, { error: "Only the local workspace owner can answer Codex." }); + if (body.always || body.rememberCommand) return json(res, 400, { error: "This connection supports approval of the displayed request only." }); + const owner = store.botByThread(threadId); + if (!owner) return json(res, 404, { error: "No such thread." }); + return json(res, 200, { ok: true, outcome: await deps.codexBindings.answer(owner.id, threadId, requestId, behavior, body.message) }); + } const refusal = cardAnswerRefusal(auth, m[1], requestId, behavior) ?? (threadId === m[1] ? null : cardAnswerRefusal(auth, threadId, requestId, behavior)); if (refusal) return json(res, 403, { error: refusal }); if (body.rememberCommand === true) { @@ -643,6 +664,11 @@ export function createBotMessageRoutes(deps: BotMessageRouteDeps): RouteHandler if (expectedThreadId !== undefined && (typeof expectedThreadId !== "string" || !/^[\w-]+$/.test(expectedThreadId))) { return json(res, 400, { error: "threadId must be a task id" }); } + if (deps.codexBindings?.configured(expectedThreadId ?? bot.threadId)) { + if (!ownerOrAdmin(auth)) return json(res, 403, { error: "Only the local workspace owner can stop Codex." }); + await deps.codexBindings.cancel(bot.id, expectedThreadId ?? bot.threadId); + return json(res, 200, { ok: true }); + } if (typeof expectedThreadId === "string" && store.taskByThread(bot.id, expectedThreadId)) { const routine = routines!.activeBotRunForBot(bot.id); if (routine?.threadId === expectedThreadId) await routines!.cancelRun(routine.id); diff --git a/server/routes/codex-bindings.test.ts b/server/routes/codex-bindings.test.ts new file mode 100644 index 0000000000..418835b3c9 --- /dev/null +++ b/server/routes/codex-bindings.test.ts @@ -0,0 +1,71 @@ +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, expect, it, vi } from "vitest"; +import { json, readBody } from "../harness/http.ts"; +import { requiredScope, type RequestAuth } from "../request-auth.ts"; +import { createCodexBindingRoutes } from "./codex-bindings.ts"; +import { dispatchRoutes } from "./table.ts"; + +const servers: Server[] = []; +afterEach(async () => { await Promise.all(servers.splice(0).map((server) => new Promise((resolve) => server.close(() => resolve())))); }); +async function fixture(auth: RequestAuth = { kind: "loopback", scopes: ["admin"] }, enabled = true) { + const disconnected = { configured: false, connected: false, state: "disconnected" as const, deliveries: [] }; + const bridge = { + view: vi.fn(() => disconnected), list: vi.fn(async () => ({ conversations: [] })), + bind: vi.fn(async () => ({ ...disconnected, configured: true, state: "failed" as const })), + reconnect: vi.fn(async () => disconnected), cancel: vi.fn(async () => {}), acknowledge: vi.fn(), disconnect: vi.fn(async () => {}), + }; + const route = createCodexBindingRoutes({ bridge, enabled, instances: () => [{ id: "codex", name: "Native Codex" }] }); + const server = createServer(async (req, res) => { + const url = new URL(req.url!, "http://localhost"); + try { + if (!await dispatchRoutes([route], { req, res, url, path: url.pathname, method: req.method!, auth, json, readBody })) json(res, 404, {}); + } catch (error) { json(res, (error as { status?: number }).status ?? 500, { error: "refused" }); } + }); + servers.push(server); await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const base = `http://127.0.0.1:${(server.address() as AddressInfo).port}`; + const request = async (method: string, path: string, body?: unknown) => { + const response = await fetch(`${base}${path}`, { method, headers: { "content-type": "application/json" }, ...(body === undefined ? {} : { body: JSON.stringify(body) }) }); + return { status: response.status, body: await response.json() }; + }; + return { bridge, request }; +} +const path = "/api/bots/owner-bot/threads/owner-thread/codex"; + +it("keeps conversation discovery, inspection and mutation admin scoped", () => { + for (const [method, url] of [["GET", "/api/codex/conversations"], ["GET", path], ["PUT", path], ["DELETE", path], ["POST", `${path}/cancel`]]) expect(requiredScope(method!, url!)).toBe("admin"); +}); + +it("refuses service loopback and hosted environments before reading native history", async () => { + for (const [auth, enabled] of [[{ kind: "loopback", trust: "service", scopes: ["client"] }, true], [{ kind: "loopback", scopes: ["admin"] }, false]] as const) { + const h = await fixture(auth as RequestAuth, enabled); + expect((await h.request("GET", "/api/codex/conversations?instanceId=codex")).status).toBe(403); + expect((await h.request("GET", path)).status).toBe(403); + expect(h.bridge.list).not.toHaveBeenCalled(); expect(h.bridge.view).not.toHaveBeenCalled(); + } +}); + +it("validates instance ownership and forbids configuration overrides", async () => { + const h = await fixture(); + expect((await h.request("GET", "/api/codex/conversations?instanceId=foreign")).status).toBe(404); + expect((await h.request("PUT", path, { instanceId: "codex", conversationId: "existing", approvalPolicy: "never" })).status).toBe(400); + expect((await h.request("PUT", path, { instanceId: "foreign", conversationId: "existing" })).status).toBe(404); + expect(h.bridge.bind).not.toHaveBeenCalled(); +}); + +it("returns a failed connection as configured but inactive, and exposes reconnect and disconnect", async () => { + const h = await fixture(); + expect(await h.request("PUT", path, { instanceId: "codex", conversationId: "existing" })).toMatchObject({ status: 409, body: { configured: true, connected: false, state: "failed" } }); + expect(h.bridge.bind).toHaveBeenCalledWith("owner-bot", "owner-thread", "codex", "existing"); + expect((await h.request("POST", `${path}/reconnect`, {})).status).toBe(200); + expect((await h.request("DELETE", path)).status).toBe(200); + expect(h.bridge.disconnect).toHaveBeenCalledWith("owner-bot", "owner-thread"); +}); + +it("scopes cancellation and uncertain-delivery review to the exact bot and thread", async () => { + const h = await fixture(); + expect((await h.request("POST", `${path}/cancel`, { sendId: "stable-send-identifier" })).status).toBe(200); + expect(h.bridge.cancel).toHaveBeenCalledWith("owner-bot", "owner-thread", "stable-send-identifier"); + expect((await h.request("POST", `${path}/acknowledge`, { sendId: "stable-send-identifier" })).status).toBe(200); + expect(h.bridge.acknowledge).toHaveBeenCalledWith("owner-bot", "owner-thread", "stable-send-identifier"); +}); diff --git a/server/routes/codex-bindings.ts b/server/routes/codex-bindings.ts new file mode 100644 index 0000000000..0a01948780 --- /dev/null +++ b/server/routes/codex-bindings.ts @@ -0,0 +1,44 @@ +import type { CodexBindings } from "../codex-bindings.ts"; +import { ownerOrAdmin } from "../request-auth.ts"; +import { PASS, type RouteHandler } from "./table.ts"; + +export function createCodexBindingRoutes(deps: { bridge: Pick; enabled: boolean; instances(): Array<{ id: string; name: string }> }): RouteHandler { + return async ({ req, res, url, path, method, auth, json, readBody }) => { + const match = path.match(/^\/api\/bots\/([\w-]+)\/threads\/([\w-]+)\/codex(?:\/(reconnect|cancel|acknowledge))?$/); + if (path !== "/api/codex/conversations" && !match) return PASS; + if (!ownerOrAdmin(auth) || !deps.enabled) return json(res, 403, { error: "Existing Codex conversations are available only to the owner of a local OpenMausBot workspace." }); + if (path === "/api/codex/conversations" && method === "GET") { + const instances = deps.instances(); + const instanceId = url.searchParams.get("instanceId"); + if (!instanceId) return json(res, 200, { instances }); + if (!instances.some((instance) => instance.id === instanceId)) return json(res, 404, { error: "No supported native Codex instance." }); + return json(res, 200, await deps.bridge.list(instanceId, (url.searchParams.get("search") ?? "").slice(0, 200), url.searchParams.get("cursor") ?? undefined)); + } + if (!match) return json(res, 405, { error: "Method not supported." }); + const [, botId, threadId, action] = match as [string, string, string, string?]; + deps.bridge.view(botId, threadId); + if (method === "GET" && !action) return json(res, 200, deps.bridge.view(botId, threadId)); + if (method === "DELETE" && !action) { await deps.bridge.disconnect(botId, threadId); return json(res, 200, deps.bridge.view(botId, threadId)); } + if (method !== "POST" && method !== "PUT") return json(res, 405, { error: "Method not supported." }); + const body = await readBody(req); + if (!body || typeof body !== "object" || Array.isArray(body)) return json(res, 400, { error: "Send a JSON object." }); + const keys = action === "cancel" || action === "acknowledge" ? ["sendId"] : action === "reconnect" ? [] : ["instanceId", "conversationId"]; + if (Object.keys(body).some((key) => !keys.includes(key))) return json(res, 400, { error: "Unsupported connection setting. Codex's saved settings cannot be overridden here." }); + if (action === "reconnect" && method === "POST") return json(res, 200, await deps.bridge.reconnect(botId, threadId)); + if (action === "cancel" && method === "POST") { + if (body.sendId !== undefined && typeof body.sendId !== "string") return json(res, 400, { error: "sendId must be a string." }); + await deps.bridge.cancel(botId, threadId, body.sendId); + return json(res, 200, deps.bridge.view(botId, threadId)); + } + if (action === "acknowledge" && method === "POST" && typeof body.sendId === "string") { + deps.bridge.acknowledge(botId, threadId, body.sendId); + return json(res, 200, deps.bridge.view(botId, threadId)); + } + if (!action && method === "PUT" && typeof body.instanceId === "string" && typeof body.conversationId === "string") { + if (!deps.instances().some((instance) => instance.id === body.instanceId)) return json(res, 404, { error: "No supported native Codex instance." }); + const result = await deps.bridge.bind(botId, threadId, body.instanceId, body.conversationId); + return json(res, result.connected ? 200 : 409, result); + } + return json(res, 400, { error: "Select a Codex instance and an exact existing conversation." }); + }; +} diff --git a/server/testing/fake-codex-native.ts b/server/testing/fake-codex-native.ts new file mode 100755 index 0000000000..3e05f235a6 --- /dev/null +++ b/server/testing/fake-codex-native.ts @@ -0,0 +1,66 @@ +#!/usr/bin/env -S node --experimental-strip-types +import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { createInterface } from "node:readline"; +import { join } from "node:path"; + +const root = process.env.FAKE_CODEX_NATIVE_DIR ?? process.env.HOME!; +const stateFile = join(root, "native-fixture-history.json"); +const logFile = join(root, "native-fixture-requests.jsonl"); +const nativeId = "fixture-native-existing"; +const rollout = join(root, "sessions", "native-fixture.jsonl"); +const png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+a9xkAAAAASUVORK5CYII="; +if (process.argv.includes("--version")) { process.stdout.write(`${process.env.FAKE_CODEX_NATIVE_VERSION ?? "codex-cli 0.163.0"}\n`); process.exit(0); } +if (process.argv.includes("login")) { process.stderr.write("Logged in using ChatGPT\n"); process.exit(0); } +mkdirSync(join(root, "sessions"), { recursive: true }); +writeFileSync(rollout, [ + { type: "session_meta", payload: { id: nativeId } }, + { type: "event_msg", payload: { type: "thread_settings_applied", thread_id: nativeId, thread_settings: { cwd: root, approval_policy: "on-request", permission_profile: { fixture: "read-only" } } } }, +].map((item) => JSON.stringify(item)).join("\n")); +const send = (message: unknown) => process.stdout.write(`${JSON.stringify(message)}\n`); +const turns: any[] = existsSync(stateFile) ? JSON.parse(readFileSync(stateFile, "utf8")) : [{ id: "existing-fixture-turn", status: "completed", items: [{ id: "existing-fixture-image", type: "imageGeneration", result: png, status: "completed" }] }]; +const persist = () => writeFileSync(stateFile, JSON.stringify(turns)); +const thread = () => ({ id: nativeId, path: rollout, name: "Native fixture conversation", model: "fixture-native-model", modelProvider: "openai", reasoningEffort: "high", cwd: root, status: { type: "idle" }, turns: [] }); +const notify = (method: string, params: any) => send({ method, params: { threadId: nativeId, ...params } }); +const finish = (status = "completed") => { + const turn = turns.at(-1)!; + turn.status = status; + if (status === "completed") turn.items.push({ id: `reply-${turn.id}`, type: "agentMessage", text: `Native fixture reply: ${turn.items[0].content.filter((input: any) => input.type === "text").map((input: any) => input.text).join("")}` }); + persist(); + for (const item of turn.items.slice(1)) notify("item/completed", { turnId: turn.id, item }); + notify("turn/completed", { turn }); +}; +const lines = createInterface({ input: process.stdin }); +lines.on("line", (line) => { + const message = JSON.parse(line); + if (!message.method) { + appendFileSync(logFile, `${JSON.stringify({ reply: message })}\n`); + if (message.id === "native-approval") finish(message.result?.decision === "accept" ? "completed" : "interrupted"); + return; + } + appendFileSync(logFile, `${JSON.stringify({ method: message.method, params: message.params, argv: process.argv.slice(2) })}\n`); + if (message.id === undefined) return; + const reply = (result: unknown) => send({ id: message.id, result }); + const params = message.params ?? {}; + switch (message.method) { + case "initialize": reply({ userAgent: "codex/0.163.0", codexHome: root, platformFamily: "unix", platformOs: "linux" }); break; + case "account/read": reply({ account: { type: "chatgpt", email: "fixture@example.test", planType: "pro" }, requiresOpenaiAuth: true }); break; + case "config/read": reply({ config: { model_provider: "openai" } }); break; + case "model/list": reply({ data: [{ id: "fixture-native-model", model: "fixture-native-model", displayName: "Fixture native", defaultReasoningEffort: "high", supportedReasoningEfforts: [{ reasoningEffort: "high", description: "High" }] }], nextCursor: null }); break; + case "thread/list": reply({ data: [thread()], nextCursor: null }); break; + case "thread/read": reply({ thread: thread() }); break; + case "thread/resume": reply({ thread: thread(), model: "fixture-native-model", modelProvider: "openai", cwd: root, approvalPolicy: "on-request", approvalsReviewer: "user", sandbox: { type: "readOnly" }, reasoningEffort: "high", collaborationMode: null }); break; + case "thread/turns/list": reply({ data: turns, nextCursor: null }); break; + case "turn/start": { + if (params.threadId !== nativeId || Object.keys(params).some((key) => !["threadId", "input", "clientUserMessageId"].includes(key))) { send({ id: message.id, error: { code: -32600, message: "Unexpected native target or overrides" } }); break; } + const turn = { id: `native-fixture-turn-${turns.length}`, status: "inProgress", items: [{ id: `user-${params.clientUserMessageId}`, type: "userMessage", clientId: params.clientUserMessageId, content: params.input }] }; + turns.push(turn); persist(); + reply({ turn }); notify("turn/started", { turn }); + const text = JSON.stringify(params.input); + if (text.includes("approval fixture")) send({ id: "native-approval", method: "item/commandExecution/requestApproval", params: { threadId: nativeId, turnId: turn.id, itemId: "fixture-command", command: "pwd", cwd: root, reason: "Harmless fixture approval" } }); + else if (!text.includes("hold fixture")) finish(); + break; + } + case "turn/interrupt": finish("interrupted"); reply({}); break; + default: send({ id: message.id, error: { code: -32601, message: "Unsupported fixture operation" } }); + } +}); diff --git a/shared/codex-binding.ts b/shared/codex-binding.ts new file mode 100644 index 0000000000..30d46a011d --- /dev/null +++ b/shared/codex-binding.ts @@ -0,0 +1,27 @@ +export type CodexConnectionState = "disconnected" | "connecting" | "ready" | "awaiting-approval" | "failed"; +export type CodexDeliveryState = "queued" | "dispatching" | "delivered" | "completed" | "interrupted" | "failed" | "uncertain" | "cancelled"; + +export interface CodexProvenance { + conversationId: string; + turnId: string; + itemId: string; +} + +export interface CodexBindingView { + configured: boolean; + connected: boolean; + state: CodexConnectionState; + instanceId?: string; + conversationId?: string; + title?: string; + error?: string; + running?: boolean; + settings?: { + model: string; + reasoningEffort: string | null; + cwd: string; + approvalPolicy: unknown; + sandbox: unknown; + }; + deliveries: Array<{ sendId: string; state: CodexDeliveryState; turnId?: string }>; +} diff --git a/shared/wire.ts b/shared/wire.ts index f53b54a754..d98850d738 100644 --- a/shared/wire.ts +++ b/shared/wire.ts @@ -424,6 +424,7 @@ export type CardAnswerer = ( /** One transcript line, serialized as stored. */ export interface WireMessage { + codex?: import("./codex-binding.ts").CodexProvenance; roomRequest?: { id: string; phase: "request" | "result" }; id: string; role: "bot" | "user"; diff --git a/src/components/ChatView.tsx b/src/components/ChatView.tsx index b0b3369b20..43a8c52b09 100644 --- a/src/components/ChatView.tsx +++ b/src/components/ChatView.tsx @@ -1,3 +1,4 @@ +import { CodexConnection, CodexSource } from "./CodexConnection"; import { createContext, memo, useCallback, useContext, useEffect, useLayoutEffect, useMemo, useRef, useState, type Dispatch } from "react"; import { useAdvancedMode } from "@/lib/interface-mode"; import { ThreadLinkContext } from "@/components/ThreadLinkMenu"; @@ -120,11 +121,9 @@ import { latestFailure, latestReply, type TranscriptSnapshot } from "@/lib/trans import { pendingApprovals } from "./PendingApproval"; import { TranscriptAnnouncer } from "./TranscriptAnnouncer"; import { JumpToLatestButton } from "./JumpToLatestButton"; - const USER_COLLAPSE_CHARS = 600; const USER_COLLAPSE_LINES = 8; export const SIMPLE_BUBBLE = "rounded-[23px] font-normal"; - interface ChatRows { botId: string; threadId: string; @@ -145,15 +144,12 @@ interface ChatRows { conversational: boolean; simple?: boolean; canReact: boolean; openedAt: number; } - const ChatRowsContext = createContext(null); - function useChatRows(): ChatRows { const rows = useContext(ChatRowsContext); if (!rows) throw new Error("a chat row outside ChatView"); return rows; } - const drawnBot = (bot: Bot) => [bot.id, bot.name, bot.hidden, bot.color, bot.mascotBody, bot.mascotExpression, bot.avatarUrl, bot.avatarCrop, bot.avatarZoom, bot.avatarFocusX, bot.avatarFocusY].join("\u0001"); @@ -569,6 +565,7 @@ const Bubble = memo(function Bubble({ ))} )} + {message.codex && } {!group && } {split ? ( (); + const nativeChanged = useCallback((view: import("../../shared/codex-binding").CodexBindingView) => setNativeState({ threadId: bot.threadId, view }), [bot.threadId]); + const nativeView = nativeState?.threadId === bot.threadId ? nativeState.view : undefined; const computerStarting = computerStartLine(state.computerStarts[bot.id], bot.name); const mascotMotion = state.mascotMotion?.botId === bot.id ? state.mascotMotion : null; @@ -1242,7 +1242,7 @@ export function ChatView({ bot: profile }: { bot: Bot }) { {/* Simple mode moves the model, the computer and ⋯ into the bot panel; a remote client's panel has neither. */} - {advanced && !remoteClient && } + {advanced && !remoteClient && !nativeView?.configured && } openThread(dispatch, { botId, threadId }, state, showThreads)} /> {(advanced || remoteClient) && } @@ -1398,7 +1398,7 @@ export function ChatView({ bot: profile }: { bot: Bot }) { {canWrite === false ? ( dispatch({ type: "newTask", botId: bot.id })} /> ) : ( - t(CONNECTION_LABELS[state]); +const DELIVERY_LABELS = { queued: "codex.queued", dispatching: "codex.dispatching", delivered: "codex.delivered", completed: "codex.completed", interrupted: "codex.interrupted", failed: "codex.failed", uncertain: "codex.uncertain", cancelled: "codex.cancelled" } as const; +const deliveryLabel = (state: CodexDeliveryState) => t(DELIVERY_LABELS[state]); + +export function CodexSource({ conversationId, turnId, itemId }: { conversationId: string; turnId: string; itemId: string }) { + return
+ {t("codex.replySource")} +
{t("codex.conversation")}
{conversationId}
{t("codex.turn")}
{turnId}
{t("codex.item")}
{itemId}
+
; +} + +export function CodexConnection({ botId, threadId, onChange }: { botId: string; threadId: string; onChange?: (view: CodexBindingView) => void }) { + const owner = useOwnerOrAdmin(); + const id = useId(); + const [view, setView] = useState(); + const [instances, setInstances] = useState>([]); + const [instanceId, setInstanceId] = useState(""); + const [conversationId, setConversationId] = useState(""); + const [search, setSearch] = useState(""); + const [results, setResults] = useState>(); + const [cursor, setCursor] = useState(null); + const [error, setError] = useState(""); + const [pending, setPending] = useState(false); + const [confirm, setConfirm] = useState(false); + const [unavailable, setUnavailable] = useState(false); + const alive = useRef(true); + const composing = useRef(false); + const epoch = useRef(0); + const endpoint = `/api/bots/${botId}/threads/${threadId}/codex`; + useEffect(() => { if (view) onChange?.(view); }, [view, onChange]); + useEffect(() => { + if (!owner) return; + alive.current = true; + const abort = new AbortController(); + const load = async () => { + const generation = epoch.current; + try { + const next = await api(endpoint, { signal: abort.signal, background: true }); + if (alive.current && generation === epoch.current) { setView(next); setUnavailable(false); } + } catch (error) { + if (error instanceof ApiError && error.status === 403) setUnavailable(true); + else if (!abort.signal.aborted && alive.current) setError(t("codex.statusUnavailable")); + } + }; + void load(); + void api<{ instances: Array<{ id: string; name: string }> }>("/api/codex/conversations", { signal: abort.signal, background: true }).then((next) => { + if (!alive.current) return; + setInstances(next.instances); + setInstanceId(next.instances[0]?.id ?? ""); + }).catch(() => {}); + const timer = setInterval(() => void load(), 2_000); + return () => { alive.current = false; abort.abort(); clearInterval(timer); }; + }, [endpoint, owner]); + + if (!owner || unavailable || !view || !view.configured && !instances.length) return null; + const perform = async (work: () => Promise) => { + if (pending) return; + epoch.current++; setPending(true); setError(""); + try { await work(); } + catch (error) { if (alive.current) setError(error instanceof Error ? error.message : t("codex.failed")); } + finally { if (alive.current) setPending(false); } + }; + const mutate = (method: string, action = "", body?: unknown) => perform(async () => { + try { + const next = await api(`${endpoint}${action}`, { method, body: body === undefined ? undefined : JSON.stringify(body), timeoutMs: 120_000 }); + if (alive.current) { setView(next); setConfirm(false); } + } catch (error) { + if (error instanceof ApiError && typeof error.body?.state === "string" && alive.current) setView(error.body as unknown as CodexBindingView); + throw error; + } + }); + const find = (more = false) => perform(async () => { + const query = new URLSearchParams({ instanceId, search, ...(more && cursor ? { cursor } : {}) }); + const found = await api<{ conversations: Array<{ id: string; title: string }>; nextCursor: string | null }>(`/api/codex/conversations?${query}`, { timeoutMs: 60_000 }); + if (alive.current) { setResults((previous) => more ? [...previous ?? [], ...found.conversations] : found.conversations); setCursor(found.nextCursor); } + }); + return
+ + {t("codex.connection")} · {stateLabel(view.state)}{view.running ? ` · ${t("codex.running")}` : ""} + +
+

{t("codex.connectionHint")}

+ {view.configured ? <> +

{view.title}
{view.conversationId}

+ {view.settings &&
+
{t("codex.model")}
{view.settings.model} · {view.settings.reasoningEffort ?? t("codex.defaultEffort")}
+
{t("codex.folder")}
{view.settings.cwd}
+
{t("codex.permissions")}
{JSON.stringify(view.settings.approvalPolicy)} · {JSON.stringify(view.settings.sandbox)}
+
} +
+ {!view.connected && } + {view.running && } + +
+ {view.deliveries.length > 0 &&
    + {view.deliveries.slice(-20).map((delivery) =>
  • + {delivery.sendId.slice(0, 12)}{deliveryLabel(delivery.state)} + {delivery.state === "queued" && } + {delivery.state === "uncertain" && <>{t("codex.reviewHint")}} +
    • )} +
} + :
{ event.preventDefault(); if (!composing.current) void find(); }} className="space-y-2"> + + + +
{ composing.current = true; }} onCompositionEnd={() => { composing.current = false; }} className={`${fieldClass} flex-1`} value={search} onChange={(event) => setSearch(event.target.value)} disabled={pending} /> + {search && } +
+ {results &&
    {results.map((result) =>
    • )}{!results.length &&
  • {t("codex.noResults")}
    • }
} + {cursor && results && } + + setConversationId(event.target.value)} disabled={pending} /> + +
} + {pending &&

{t("codex.working")}

} + {(error || view.error) &&

{error || view.error}

} +
+ setConfirm(false)} onConfirm={() => void mutate("DELETE")} /> +
; +} diff --git a/src/components/Composer.tsx b/src/components/Composer.tsx index 8f8f955642..4eb49fc379 100644 --- a/src/components/Composer.tsx +++ b/src/components/Composer.tsx @@ -98,7 +98,7 @@ export function Composer({ onClearReply, onConsumeReply, onRestoreReply, - locked: setupLocked = false, + locked: setupLocked = false, nativeConversation = false, nativeDisconnected = false, }: { bot?: Bot; group?: Group; @@ -108,10 +108,10 @@ export function Composer({ onClearReply?: () => void; onConsumeReply?: () => void; onRestoreReply?: (message: Message, threadId: string) => void; - locked?: boolean; + locked?: boolean; nativeConversation?: boolean; nativeDisconnected?: boolean; }) { const bot = profile ? currentTaskBot(profile) : undefined; - const locked = setupLocked || Boolean(bot?.awaitingThreadSnapshot); + const locked = setupLocked || nativeDisconnected || Boolean(bot?.awaitingThreadSnapshot); const { state, dispatch } = useStore(); const ownerOrAdmin = useOwnerOrAdmin(); const { threads, currentBotId } = useThreadRefs(); @@ -268,7 +268,7 @@ export function Composer({ const typedGoalText = group && !group.dm ? goalTextFromComposer(text) : null; const effectiveText = typedGoalText ?? text; const effectiveChannelMode = typedGoalText !== null ? "goal" : channelMode; - const engineSupportsImages = imageTargetsSupport(effectiveText, effectiveChannelMode); + const engineSupportsImages = nativeConversation || imageTargetsSupport(effectiveText, effectiveChannelMode); const slash = composerSlashTrigger(text, caret); const locale = activeLocale(); @@ -854,8 +854,8 @@ export function Composer({ } {/* Simple mode keeps the chip only while Finish together is on, so the + menu's choice stays visible. */} {goal && (advanced || goal.on) && } - {advanced && approvalChoice && } - {modeBot && !remoteClient && advanced && ( + {advanced && !nativeConversation && approvalChoice && } + {modeBot && !remoteClient && advanced && !nativeConversation && ( )} - {bot && !group && } + {bot && !group && !nativeConversation && } {group && !advanced && } {hasContent && !locked && (