diff --git a/AGENTS.md b/AGENTS.md index 8789d05e..7ba9842d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,6 +71,9 @@ with no silent fallback. (native in-process plugin injected via `OPENCODE_CONFIG_CONTENT`). - **`@cotal-ai/connector-hermes`** (`extensions/connector-hermes`): the Hermes (Nous Research) adapter; includes a Python sidecar. +- **`@cotal-ai/connector-codex`** (`extensions/connector-codex`): the OpenAI Codex adapter — + a headless host-mode peer that drives a `codex app-server` thread over JSON-RPC (wake / + steer / interrupt); resolved by agent type `codex-app-server`. - **`@cotal-ai/cmux`** (`extensions/cmux`): the cmux integration: a driver over the cmux CLI plus a self-registering `cmux` Runtime and `TerminalLayout` provider. - **`@cotal-ai/cli`** (`implementations/cli`): the mesh CLI: `up`, `join`, `watch`, `console`, diff --git a/bin/cotal.ts b/bin/cotal.ts index 047a0444..eb012b83 100644 --- a/bin/cotal.ts +++ b/bin/cotal.ts @@ -11,6 +11,7 @@ import "@cotal-ai/manager"; // self-registers supervise / cmux / start / stop / import "@cotal-ai/connector-claude-code"; // registers the `claude` connector that spawn / start resolve import "@cotal-ai/connector-opencode"; // registers the `opencode` connector (native in-process plugin) import "@cotal-ai/connector-hermes"; // registers the `hermes` connector (Nous Research gateway as a mesh peer) +import "@cotal-ai/connector-codex"; // registers the `codex-app-server` connector (headless host-mode peer driving codex app-server) import "@cotal-ai/cmux"; // opt into the cmux integration — registers the `cmux` runtime + TerminalLayout providers import { claudeConnector } from "@cotal-ai/connector-claude-code"; import { registry } from "@cotal-ai/core"; diff --git a/bin/package.json b/bin/package.json index d79e0b1e..cb613cee 100644 --- a/bin/package.json +++ b/bin/package.json @@ -34,6 +34,7 @@ "@cotal-ai/cli": "workspace:*", "@cotal-ai/cmux": "workspace:*", "@cotal-ai/connector-claude-code": "workspace:*", + "@cotal-ai/connector-codex": "workspace:*", "@cotal-ai/connector-hermes": "workspace:*", "@cotal-ai/connector-opencode": "workspace:*", "@cotal-ai/core": "workspace:*", diff --git a/docs/architecture.md b/docs/architecture.md index 7887dc0f..2f18d80e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -153,10 +153,12 @@ examples ──→ one-or-more implementations ──→ core ←(peer)── ex ``` The migration is done: `demos/` use-cases are now `examples/`. The connector is split into -`@cotal-ai/connector-core` (shared mesh runtime) plus two thin adapters, -`@cotal-ai/connector-claude-code` (`claudeConnector`) and `@cotal-ai/connector-opencode` -(`opencodeConnector`). Those are `extensions/` packages that **peer-depend** on core and export -a `Connector`. `@cotal-ai/cli` and `@cotal-ai/manager` are `implementations/` packages. +`@cotal-ai/connector-core` (shared mesh runtime) plus thin adapters — +`@cotal-ai/connector-claude-code` (`claudeConnector`), `@cotal-ai/connector-opencode` +(`opencodeConnector`), `@cotal-ai/connector-hermes`, and `@cotal-ai/connector-codex` +(`codexAppServerConnector`, a headless host-mode peer driving `codex app-server` — see +[codex-integration.md](codex-integration.md)). Those are `extensions/` packages that +**peer-depend** on core and export a `Connector`. `@cotal-ai/cli` and `@cotal-ai/manager` are `implementations/` packages. Assembly lives at the **composition root**: an example (`examples/01/src/manager.ts`) imports the manager plus the connectors it wants and hands them to the manager (`new Manager({ connectors: […] })`), which resolves one by agent type when spawning (unknown throws). diff --git a/docs/codex-integration.md b/docs/codex-integration.md new file mode 100644 index 00000000..075dee1f --- /dev/null +++ b/docs/codex-integration.md @@ -0,0 +1,55 @@ +# Codex connector + +> App Server: [developers.openai.com/codex/app-server](https://developers.openai.com/codex/app-server) +> (built + verified against codex-cli 0.137.0). + +The connector turns OpenAI Codex into a Cotal mesh peer over the shared runtime in +[`@cotal-ai/connector-core`](../extensions/connector-core). It runs **host-mode** (headless): +a peer that drives a `codex app-server` over JSON-RPC for true wake / steer / interrupt of a +*live* session — no native TUI. The human view comes via the manager's attach / WS. + +Identity-gated (`hasIdentity()`): launched with no `COTAL_*` env it stays off the mesh. The +manager resolves it by agent type **`codex-app-server`**. + +## How it works + +[`host.ts`](../extensions/connector-codex/src/host.ts) embeds a `MeshAgent` in the same process +as an `AppServerDriver` ([`app-server.ts`](../extensions/connector-codex/src/app-server.ts)), +which owns a `codex app-server` child and speaks the app-server **v2** JSON-RPC (JSONL over +stdio — the same protocol the TUI / VS Code extension use; the wire omits the `jsonrpc` field). +A mesh message becomes a real user turn: + +- idle → `turn/start` (wake); a turn already running → `turn/steer` (true mid-turn inject, no + abort — Codex exceeds Claude here); shutdown → `turn/interrupt`. +- Presence is read off the event stream: `turn/started`→working, `turn/completed`→idle, an + approval request→waiting (auto-accepted to stay autonomous). Each turn's final `agentMessage` + is routed back to whoever prompted it (channel→`send`, dm/anycast→`dm`). + +`approval_policy=never` + `sandbox_mode=workspace-write` make a spawned session autonomous (it +would otherwise hang on the first approval the host can't surface). The operator's `~/.codex` +(auth, model, their own servers) is never written — all config rides `-c` overrides. Codex auth +is the operator's ChatGPT login in `~/.codex/auth.json` (reachable via `HOME`) or `OPENAI_API_KEY` +(the only provider key forwarded into the child — no other operator env leaks, per `launchEnv`). + +The host is **directed-only**: it acts on DMs, anycasts, and @-mentions; ambient channel chatter +is dropped, not surfaced — keeping a headless peer focused and its inbox bounded. It drives off +the mesh inbox with **ack-on-completion**: a turn's surfaced messages are `drainInbox()`-acked +only when the turn ends un-interrupted (a `failed` turn acks to avoid a retry-loop; only +`interrupted`/crash redelivers), and a message is steered into the live turn only when it shares +that turn's scope (`channel:` vs `dm:`), so a DM never rides a channel broadcast. + +## Verified + +Built + typechecked against codex-cli 0.137.0 — the app-server protocol bindings were confirmed +via `codex app-server generate-ts` (the v2 methods `turn/start` / `turn/steer` / `turn/interrupt` +and notifications `turn/started` / `item/completed` / `turn/completed` match). A live turn (DM → +`PONG` round-trip) is exercised by `smoke:codex` against the real binary (needs an authenticated +`codex`; gated by `COTAL_E2E_CODEX=1`). + +## Not yet + +- **Agent-initiated `cotal_*` tools.** The host owns all mesh I/O so there is one mesh identity; + letting the Codex agent proactively `cotal_send`/`cotal_spawn` needs those tool calls routed + back through the host's single `MeshAgent` (else two presences). Today the peer is reply-driven. +- **Attach surface + remote-TUI** (a human watching the real `codex` TUI while the mesh drives it) + — rides Codex's newer `--remote` path; tracked upstream (codex#18203, codex#21551). diff --git a/extensions/connector-codex/package.json b/extensions/connector-codex/package.json new file mode 100644 index 00000000..bd80a01b --- /dev/null +++ b/extensions/connector-codex/package.json @@ -0,0 +1,41 @@ +{ + "name": "@cotal-ai/connector-codex", + "description": "Cotal connector for OpenAI Codex: a headless host-mode peer that drives a live `codex app-server` thread over JSON-RPC.", + "version": "0.1.0", + "license": "Apache-2.0", + "repository": { + "type": "git", + "url": "https://github.com/Cotal-AI/Cotal.git", + "directory": "extensions/connector-codex" + }, + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "scripts": { + "typecheck": "tsc -p tsconfig.json --noEmit", + "build": "tsc -p tsconfig.json", + "prepublishOnly": "pnpm run build" + }, + "dependencies": { + "@cotal-ai/connector-core": "workspace:*", + "tsx": "^4.22.4" + }, + "peerDependencies": { + "@cotal-ai/core": ">=0.1.0" + }, + "devDependencies": { + "@cotal-ai/core": "workspace:*" + }, + "files": [ + "dist" + ], + "publishConfig": { + "access": "public" + } +} diff --git a/extensions/connector-codex/smoke/host-turn.smoke.ts b/extensions/connector-codex/smoke/host-turn.smoke.ts new file mode 100644 index 00000000..63f9680a --- /dev/null +++ b/extensions/connector-codex/smoke/host-turn.smoke.ts @@ -0,0 +1,113 @@ +/** + * Live E2E for the Codex host-mode connector — drives the REAL `codex` binary end to end. + * Spins up its own nats-server + a real mesh, launches the host peer (tsx host-main.ts) as a + * child, then DMs it and asserts a model reply routes back over the mesh. Unlike the opencode + * turn-wedge smoke (fake client), this needs an AUTHENTICATED codex, so it is gated behind + * `COTAL_E2E_CODEX=1` — without the flag it skips (keeps `pnpm smoke`/CI green). + * + * 1. handshake — AppServerDriver.start() against the real binary returns a thread id (no model); + * 2. full turn — operator DMs the peer "reply PONG"; the codex turn's final message comes back + * as a DM. Proves wake (turn/start) + presence + reply routing over a real mesh. + * Run: COTAL_E2E_CODEX=1 pnpm smoke:codex + */ +import { strict as assert } from "node:assert"; +import { spawn } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { CotalEndpoint, seedChannelRegistry, isReachable } from "@cotal-ai/core"; +import { AppServerDriver } from "../src/app-server.js"; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +if (!/^(1|true|yes|on)$/i.test(process.env.COTAL_E2E_CODEX ?? "")) { + console.log("SKIP codex host E2E — set COTAL_E2E_CODEX=1 (needs an authenticated `codex` CLI) to run it"); + process.exit(0); +} + +const PORT = 14271; +const servers = `nats://127.0.0.1:${PORT}`; +const space = "codexe2e"; +const PEER = "codexpeer"; +let pass = 0; +const check = (name: string, cond: boolean, extra?: unknown) => { + assert.ok(cond, `${name}${extra !== undefined ? ` — ${JSON.stringify(extra)}` : ""}`); + pass++; + console.log(` ✓ ${name}`); +}; + +const HOST_ENTRY = fileURLToPath(new URL("../src/host-main.ts", import.meta.url)); +const TSX = fileURLToPath(new URL("../node_modules/.bin/tsx", import.meta.url)); + +const dir = mkdtempSync(join(tmpdir(), "cotal-codexe2e-")); +const nats = spawn("nats-server", ["-js", "-p", String(PORT), "-sd", join(dir, "js")], { stdio: "ignore" }); + +const operator = new CotalEndpoint({ + space, + servers, + card: { name: "operator", kind: "agent", id: "operator" }, + channels: ["team"], +}); +operator.on("error", () => {}); + +let reply = ""; +operator.on("message", (msg: { parts?: { kind: string; text?: string }[] }, _d: unknown, meta: { kind: string; historical: boolean }) => { + if (meta.historical || meta.kind !== "dm") return; + reply += (msg.parts ?? []).filter((p) => p.kind === "text").map((p) => p.text ?? "").join(""); +}); + +let online = false; +operator.on("presence", (e: { type: string; presence: { card: { id: string; name: string } } }) => { + const c = e.presence.card; + if ((c.id === PEER || c.name === PEER) && e.type !== "offline") online = true; +}); + +let host: ReturnType | undefined; +let handshakeDriver: AppServerDriver | undefined; +try { + // (1) handshake against the real binary — spawn → initialize → thread/start returns a thread id. + handshakeDriver = new AppServerDriver({ cwd: dir, log: () => {} }); + const threadId = await handshakeDriver.start(); + check("app-server handshake returns a thread id", typeof threadId === "string" && threadId.length > 0, threadId); + await handshakeDriver.stop(); + + // (2) full turn over a real mesh. + for (let i = 0; i < 50; i++) { if (await isReachable(servers)) break; await sleep(200); } + await seedChannelRegistry({ servers, space, file: { defaults: { replay: false }, channels: { team: { replay: false } } } }); + await operator.start(); + + host = spawn(TSX, [HOST_ENTRY], { + env: { + ...process.env, + COTAL_SPACE: space, + COTAL_NAME: PEER, + COTAL_ID: PEER, + COTAL_SERVERS: servers, + COTAL_SUBSCRIBE: "team", + COTAL_ROLE: "coder", + }, + stdio: ["ignore", "inherit", "inherit"], + }); + + // Wait for the peer to come online (mesh connected AND app-server thread up — the host only + // reports idle after driver.start()). + for (let i = 0; i < 600 && !online; i++) await sleep(100); // up to 60s + check("codex host peer comes online", online); + + await operator.unicast(PEER, "Reply with exactly the single word: PONG. Do not run any tools or commands."); + + // A model turn can take a while; poll up to 120s for the reply DM. + for (let i = 0; i < 1200 && !/PONG/i.test(reply); i++) await sleep(100); + check("codex peer replies to the DM with PONG", /PONG/i.test(reply), reply.trim()); + + console.log(`\nCODEX HOST E2E PASSED ✅ (${pass} checks) reply=${JSON.stringify(reply.trim())}`); +} finally { + host?.kill("SIGTERM"); + await handshakeDriver?.stop().catch(() => {}); + await operator.stop().catch(() => {}); + nats.kill("SIGKILL"); + await sleep(200); + rmSync(dir, { recursive: true, force: true }); +} +process.exit(0); diff --git a/extensions/connector-codex/src/app-server.ts b/extensions/connector-codex/src/app-server.ts new file mode 100644 index 00000000..59c42ea6 --- /dev/null +++ b/extensions/connector-codex/src/app-server.ts @@ -0,0 +1,247 @@ +/** + * Codex app-server driver — a JSON-RPC 2.0 (JSONL over stdio) client that owns a + * `codex app-server` child and drives a live thread: start a turn (wake), steer a + * turn already in flight (true mid-turn injection), or interrupt one. Presence + * falls out of the app-server's own event stream, so the host never guesses what + * the session is doing. + * + * This is the host-mode path: Cotal is the parent holding the pipe, so a peer + * message becomes a real user turn — unlike the pull-only MCP injection. Built to + * the app-server **v2** protocol the TUI/VS Code extension also speak; verified + * against codex-cli 0.137.0 (`codex app-server generate-ts`). The wire omits the + * `jsonrpc` field, matching the generated bindings. + */ +import { spawn, type ChildProcess } from "node:child_process"; +import { EventEmitter } from "node:events"; + +/** A text user-input item (the only kind we send). `text_elements` is required by the schema. */ +function textInput(text: string): { type: "text"; text: string; text_elements: never[] } { + return { type: "text", text, text_elements: [] }; +} + +interface Pending { + resolve: (v: unknown) => void; + reject: (e: Error) => void; +} + +/** One decoded line off the app-server: a response, a notification, or a server→client request. */ +interface RpcMessage { + id?: number | string; + method?: string; + params?: Record; + result?: unknown; + error?: { code?: number; message?: string }; +} + +/** + * Emits: + * - `"turnStarted"` (turnId) — a turn began (→ working) + * - `"turnCompleted"` ({turnId, text}) — a turn finished; `text` = final agent message (may be "") + * - `"waiting"` — an approval was requested (auto-answered) + * - `"closed"` (code) — the child exited + * - `"error"` (Error) + */ +export class AppServerDriver extends EventEmitter { + private child?: ChildProcess; + private nextId = 1; + private readonly pending = new Map(); + private buf = ""; + private threadId?: string; + private activeTurnId?: string; + /** Final agent-message text per turn (the last `agentMessage` item of a turn is the reply). */ + private readonly replyByTurn = new Map(); + private readonly cwd: string; + private readonly model?: string; + private readonly log: (m: string) => void; + + constructor(opts: { cwd: string; model?: string; log?: (m: string) => void }) { + super(); + this.cwd = opts.cwd; + this.model = opts.model; + this.log = opts.log ?? ((m) => process.stderr.write(`[cotal-codex-host] ${m}\n`)); + } + + get busy(): boolean { + return this.activeTurnId !== undefined; + } + + /** Spawn `codex app-server`, initialize, and start a thread. Resolves with the thread id. */ + async start(): Promise { + // approval_policy=never + sandbox=workspace-write make a spawned session autonomous; without + // them the first command would block on an approval the host can't surface (deadlock). + const child = spawn( + "codex", + ["app-server", "-c", 'approval_policy="never"', "-c", 'sandbox_mode="workspace-write"'], + { cwd: this.cwd, stdio: ["pipe", "pipe", "pipe"] }, + ); + this.child = child; + child.stdout!.setEncoding("utf8"); + child.stdout!.on("data", (d: string) => this.onData(d)); + child.stderr!.setEncoding("utf8"); + child.stderr!.on("data", (d: string) => this.log(`app-server: ${d.trimEnd()}`)); + child.on("exit", (code) => { + for (const p of this.pending.values()) p.reject(new Error("app-server exited")); + this.pending.clear(); + this.emit("closed", code ?? 0); + }); + child.on("error", (e) => this.emit("error", e)); + + await this.request("initialize", { + clientInfo: { name: "cotal", title: "Cotal", version: "0.0.0" }, + capabilities: { experimentalApi: true, requestAttestation: false }, + }); + this.notify("initialized"); + + const started = (await this.request("thread/start", { + cwd: this.cwd, + ...(this.model ? { model: this.model } : {}), + approvalPolicy: "never", + sandbox: "workspace-write", + })) as { thread?: { id?: string } }; + const id = started.thread?.id; + if (!id) throw new Error("thread/start returned no thread id"); + this.threadId = id; + this.log(`thread started: ${id}`); + return id; + } + + /** Begin a new user turn — wakes the session. */ + async startTurn(text: string): Promise { + if (!this.threadId) throw new Error("thread not started"); + await this.request("turn/start", { threadId: this.threadId, input: [textInput(text)] }); + } + + /** Inject input into the turn currently in flight (true mid-turn steer). Returns false if there + * is no active turn or the turn just ended (the caller then falls back to {@link startTurn}). */ + async steer(text: string): Promise { + if (!this.threadId || !this.activeTurnId) return false; + try { + await this.request("turn/steer", { + threadId: this.threadId, + input: [textInput(text)], + expectedTurnId: this.activeTurnId, + }); + return true; + } catch (e) { + this.log(`steer failed: ${(e as Error).message}`); + return false; + } + } + + /** Cancel the in-flight turn, if any. */ + async interrupt(): Promise { + if (!this.threadId || !this.activeTurnId) return; + try { + await this.request("turn/interrupt", { threadId: this.threadId, turnId: this.activeTurnId }); + } catch (e) { + this.log(`interrupt failed: ${(e as Error).message}`); + } + } + + async stop(): Promise { + try { + this.child?.stdin?.end(); + } catch { + /* ignore */ + } + this.child?.kill("SIGTERM"); + } + + // ---- JSON-RPC plumbing --------------------------------------------------- + + private request(method: string, params?: Record): Promise { + const id = this.nextId++; + return new Promise((resolve, reject) => { + if (!this.writeLine({ id, method, ...(params ? { params } : {}) })) + return reject(new Error("app-server not running")); + this.pending.set(id, { resolve, reject }); + }); + } + + private notify(method: string, params?: Record): void { + this.writeLine({ method, ...(params ? { params } : {}) }); + } + + private writeLine(obj: unknown): boolean { + const stdin = this.child?.stdin; + if (!stdin || !stdin.writable) return false; + stdin.write(JSON.stringify(obj) + "\n"); + return true; + } + + private onData(chunk: string): void { + this.buf += chunk; + let nl: number; + while ((nl = this.buf.indexOf("\n")) >= 0) { + const line = this.buf.slice(0, nl).trim(); + this.buf = this.buf.slice(nl + 1); + if (!line) continue; + let msg: RpcMessage; + try { + msg = JSON.parse(line) as RpcMessage; + } catch { + continue; + } + this.dispatch(msg); + } + } + + private dispatch(msg: RpcMessage): void { + // Response to one of our requests (id, no method). + if (msg.id !== undefined && msg.method === undefined) { + const p = this.pending.get(msg.id as number); + if (!p) return; + this.pending.delete(msg.id as number); + if (msg.error) p.reject(new Error(msg.error.message ?? "app-server error")); + else p.resolve(msg.result); + return; + } + // Server→client request (id AND method) — approvals etc. Keep the session autonomous. + if (msg.id !== undefined && msg.method) return this.answerServerRequest(msg.id, msg.method); + // Notification (method, no id). + if (msg.method) this.onNotification(msg.method, msg.params ?? {}); + } + + /** Auto-answer server-initiated requests so an unattended session never stalls. With + * approval_policy=never these shouldn't fire, but we accept defensively. */ + private answerServerRequest(id: number | string, method: string): void { + this.emit("waiting"); + if (method === "execCommandApproval" || method === "applyPatchApproval") + return void this.writeLine({ id, result: { decision: "approved" } }); // legacy: ReviewDecision + if (method.endsWith("/requestApproval")) + return void this.writeLine({ id, result: { decision: "accept" } }); // v2 approval decision + // Anything we don't implement (dynamic tool calls, elicitations): decline cleanly. + this.writeLine({ id, error: { code: -32601, message: "unsupported by cotal host" } }); + } + + private onNotification(method: string, params: Record): void { + switch (method) { + case "turn/started": { + const turn = params.turn as { id?: string } | undefined; + this.activeTurnId = turn?.id; + if (this.activeTurnId) this.emit("turnStarted", this.activeTurnId); + return; + } + case "item/completed": { + const item = params.item as { type?: string; text?: string } | undefined; + const turnId = params.turnId as string | undefined; + if (item?.type === "agentMessage" && typeof item.text === "string" && turnId) + this.replyByTurn.set(turnId, item.text); // last agentMessage of the turn = the reply + return; + } + case "turn/completed": { + const turn = params.turn as { id?: string; status?: string } | undefined; + const turnId = turn?.id ?? this.activeTurnId; + const text = (turnId && this.replyByTurn.get(turnId)) || ""; + if (turnId) this.replyByTurn.delete(turnId); + this.activeTurnId = undefined; + // status ∈ completed | interrupted | failed — the host acks the inbox only on a clean + // finish, so an interrupted/crashed turn redelivers its surfaced messages. + this.emit("turnCompleted", { turnId, text, status: turn?.status ?? "completed" }); + return; + } + default: + return; // deltas, reasoning, diffs, status pings — not needed for presence/reply + } + } +} diff --git a/extensions/connector-codex/src/extension.ts b/extensions/connector-codex/src/extension.ts new file mode 100644 index 00000000..1d7a1b0a --- /dev/null +++ b/extensions/connector-codex/src/extension.ts @@ -0,0 +1,42 @@ +import { fileURLToPath } from "node:url"; +import { registry, type Connector, type LaunchOpts, type LaunchSpec } from "@cotal-ai/core"; +import { launchEnv } from "@cotal-ai/connector-core"; + +/** The host loop runs via tsx (Codex host-mode has no plugin copy-install). Resolved to the sibling + * that actually exists next to THIS module: the compiled `.js` from dist/, the `.ts` source in dev. + * `../node_modules/.bin/tsx` sits at the package root, correct from dist or src alike. */ +const TSX = fileURLToPath(new URL("../node_modules/.bin/tsx", import.meta.url)); +const inDist = import.meta.url.includes("/dist/"); +const HOST_ENTRY = fileURLToPath(new URL(`./host-main.${inDist ? "js" : "ts"}`, import.meta.url)); + +/** + * The Codex host-mode connector: launches an embedded Cotal peer that drives a headless + * `codex app-server` over JSON-RPC. A mesh message becomes a real user turn — wake an idle + * session (turn/start), steer one already mid-turn (turn/steer), or interrupt it + * (turn/interrupt); presence is read off the app-server event stream rather than self-reported. + * No native TUI — the human view comes via the manager's attach. Self-registers; the manager + * resolves it by agent type "codex-app-server". + */ +export const codexAppServerConnector: Connector = { + kind: "connector", + name: "codex-app-server", + buildLaunch(opts: LaunchOpts): LaunchSpec { + // The host (tsx) spawns `codex app-server`, which needs codex auth: a ChatGPT login in + // ~/.codex/auth.json (reachable via HOME on the OS allow-list) or OPENAI_API_KEY. Forward only + // the named provider key (P3) — never the operator's unrelated env. Identity rides COTAL_* in + // the child env; the embedded MeshAgent reads it via configFromEnv. + const env: Record = { + ...launchEnv({ providerKeys: ["OPENAI_API_KEY"] }), + COTAL_SPACE: opts.space, + COTAL_NAME: opts.name, + }; + if (opts.role) env.COTAL_ROLE = opts.role; + if (opts.id) env.COTAL_ID = opts.id; + if (opts.creds) env.COTAL_CREDS = opts.creds; + if (opts.servers) env.COTAL_SERVERS = opts.servers; + if (opts.configPath) env.COTAL_AGENT_FILE = opts.configPath; + return { command: TSX, args: [HOST_ENTRY], env }; + }, +}; + +registry.register(codexAppServerConnector); diff --git a/extensions/connector-codex/src/host-main.ts b/extensions/connector-codex/src/host-main.ts new file mode 100644 index 00000000..cc6aab81 --- /dev/null +++ b/extensions/connector-codex/src/host-main.ts @@ -0,0 +1,10 @@ +/** + * Entry point for the Codex host-mode peer — what the `codex-app-server` connector + * launches via tsx. Runs the embedded MeshAgent + app-server driver loop. + */ +import { runCodexHost } from "./host.js"; + +runCodexHost().catch((e) => { + process.stderr.write(`[cotal-codex-host] fatal: ${(e as Error).stack ?? String(e)}\n`); + process.exit(1); +}); diff --git a/extensions/connector-codex/src/host.ts b/extensions/connector-codex/src/host.ts new file mode 100644 index 00000000..68bc340a --- /dev/null +++ b/extensions/connector-codex/src/host.ts @@ -0,0 +1,171 @@ +/** + * Codex host-mode peer: embeds a Cotal {@link MeshAgent} in the same process as an + * {@link AppServerDriver}, so a Codex session is a full lateral peer that can be + * DRIVEN live. Inbound mesh messages become real Codex turns — a directed message + * wakes an idle session (turn/start) or steers one already mid-turn (turn/steer); + * presence is derived from the app-server's turn events, not self-reported. Each + * turn's final agent message is routed back to whoever prompted it. + * + * The standalone host-mode path (no plugin, no cmux/send-key). Two correctness + * properties it holds (the ones the embed adapters converged on): + * - **ack-on-completion** — it drives off the mesh inbox and `drainInbox()`-acks a + * turn's surfaced messages when the turn ends un-interrupted, so an interrupt or a + * crash mid-turn leaves them on the stream to redeliver (never silently lost, never + * double-answered after a restart). A model-`failed` turn DOES ack (drop, don't + * retry-loop — matching opencode); only `interrupted` redelivers. + * - **scope isolation** — a message is only steered into the live turn when it shares + * that turn's reply audience (`scopeKey`), so a private DM is never folded into, and + * broadcast by, a channel turn (and vice-versa). + */ +import { MeshAgent, configFromEnv, fmtFrom, type InboxItem } from "@cotal-ai/connector-core"; +import { AppServerDriver } from "./app-server.js"; + +/** Render an inbox item as the prompt text fed into a turn (mirrors the cotal_inbox formatting). */ +function render(it: InboxItem): string { + const who = fmtFrom(it); + if (it.kind === "dm") return `[DM from ${who}] ${it.text}`; + if (it.kind === "anycast") return `[@${it.service} request from ${who}] ${it.text}`; + return `[#${it.channel} from ${who}] ${it.text}`; +} + +/** A message's reply audience. A channel mention's audience is the channel (which already saw it); + * a DM/anycast is private to its sender. Only same-scope messages may share a turn. */ +function scopeKey(it: InboxItem): string { + return it.kind === "channel" ? `channel:${it.channel ?? ""}` : `dm:${it.fromId}`; +} + +export async function runCodexHost(): Promise { + const config = configFromEnv(); + const mesh = new MeshAgent(config); + mesh.start(); // background connect with retry + + const driver = new AppServerDriver({ cwd: process.cwd(), model: process.env.COTAL_MODEL }); + + // Presence is meaningless before we've connected; never let a status push throw. + const status = (s: "idle" | "working" | "waiting", activity?: string): void => { + if (mesh.connected) void mesh.setStatus(s, activity).catch(() => {}); + }; + + // We only act on *directed* traffic (DM, anycast, or an @mention of us); ambient channel chatter + // isn't answered. A message we should process must be directed and not our own echo. + const actionable = (it: InboxItem): boolean => + it.fromId !== mesh.id && (it.kind === "dm" || it.kind === "anycast" || it.mentionsMe); + + // How to reply to a turn started by `it` — to its channel, or privately to its sender. + const deliverTo = (it: InboxItem) => (text: string): Promise => + it.kind === "channel" ? mesh.send(text, it.channel) : mesh.dm(it.fromId, text); + + // The turn in progress: its scope + where its reply goes. `surfaced` is the count of inbox + // front-items fed into it (the starting message + same-scope messages steered in), which is + // exactly what `drainInbox(surfaced)` acks on clean completion. The inbox IS the queue — we never + // copy messages into a parallel buffer, so the stream's ack is the single source of truth. + let active: { scope: string; deliver: (text: string) => Promise; originId: string } | undefined; + let surfaced = 0; + let absorbing = false; + let ready = false; // the app-server thread is up — don't surface a turn before then + + function pump(): void { + if (!ready || active || driver.busy) return; + const pending = mesh.peekInbox(); + let i = 0; + while (i < pending.length && !actionable(pending[i])) i++; + if (i >= pending.length) return; // nothing actionable; leave ambient on the stream (capped) + if (i > 0) mesh.drainInbox(i); // drop the leading non-actionable run so our item is front[0] + const item = pending[i]; + active = { scope: scopeKey(item), deliver: deliverTo(item), originId: item.id }; + surfaced = 1; + status("working", `handling ${item.kind} from ${item.fromName}`); + driver.startTurn(render(item)).catch((e) => { + process.stderr.write(`[cotal-codex-host] turn/start failed: ${(e as Error).message}\n`); + active = undefined; + surfaced = 0; + status("idle"); + }); + } + + // Steer every front-contiguous same-scope message (beyond the ones already surfaced) into the live + // turn. Stops at the first non-matching front-item, which keeps the surfaced run contiguous so the + // completion-time `drainInbox(surfaced)` acks exactly what was fed in. Serialised by `absorbing`. + async function absorb(): Promise { + if (absorbing) return; + absorbing = true; + try { + while (active && driver.busy) { + const it = mesh.peekInbox()[surfaced]; + if (!it || !actionable(it) || scopeKey(it) !== active.scope) break; + if (!(await driver.steer(render(it)))) break; // turn ended mid-absorb — next pump handles it + surfaced++; + } + } finally { + absorbing = false; + } + } + + driver.on("turnStarted", () => { + status("working"); + void absorb(); + }); + driver.on("waiting", () => status("waiting", "approval")); + driver.on("turnCompleted", (r: { text: string; status: string }) => { + const turn = active; + const consumed = surfaced; + // Ack (and reply) on any terminal status EXCEPT interrupt: a 'completed' or model-'failed' turn + // acks its surfaced run (failed drops rather than retry-loops, like opencode); only 'interrupted' + // skips the ack so the run redelivers. A crash before this event never acked → redelivers too. + const acked = r.status !== "interrupted"; + active = undefined; + surfaced = 0; + // Ack the surfaced front-run — but ONLY if our origin is still front[0]. MeshAgent force-evicts+ + // acks from the front at MAX_INBOX, so a 200+ msg burst mid-turn can evict our in-flight prefix; + // a position-based drainInbox(consumed) would then ack the wrong (newer) items. If the origin is + // gone, the overflow already acked the prefix → skip the drain (front-id guard, like hermes'). + if (turn && acked && consumed > 0 && mesh.peekInbox()[0]?.id === turn.originId) + mesh.drainInbox(consumed); // sole ack site + const text = r.text?.trim(); + process.stderr.write(`[cotal-codex-host] turn ${r.status}: reply ${text ? `${text.length}c` : "(empty)"}\n`); + void (async () => { + if (turn && acked && text) { + try { + await turn.deliver(text); + } catch (e) { + process.stderr.write(`[cotal-codex-host] reply route failed: ${(e as Error).message}\n`); + } + } + status("idle"); + pump(); + })(); + }); + driver.on("closed", () => void mesh.stop().then(() => process.exit(0)).catch(() => process.exit(1))); + + // New mesh traffic: feed it into the live turn if it fits the scope, else start the next turn. + mesh.on("incoming", () => { + void absorb(); + pump(); + }); + + await driver.start(); + ready = true; + status("idle"); + pump(); // drain anything that arrived while the app-server thread was starting + process.stderr.write( + `[cotal-codex-host] ready (app-server) — space="${config.space}" name="${config.name}"${config.role ? ` role="${config.role}"` : ""}\n`, + ); + + const shutdown = async (): Promise => { + try { + await driver.interrupt(); + await driver.stop(); + } catch { + /* ignore */ + } + try { + await mesh.stop(); + } finally { + process.exit(0); + } + }; + process.on("SIGINT", () => void shutdown()); + process.on("SIGTERM", () => void shutdown()); + + await new Promise(() => {}); // keep alive +} diff --git a/extensions/connector-codex/src/index.ts b/extensions/connector-codex/src/index.ts new file mode 100644 index 00000000..9bd9b262 --- /dev/null +++ b/extensions/connector-codex/src/index.ts @@ -0,0 +1,3 @@ +export * from "./extension.js"; // self-registers the `codex-app-server` connector on import +export { runCodexHost } from "./host.js"; +export { AppServerDriver } from "./app-server.js"; diff --git a/extensions/connector-codex/tsconfig.json b/extensions/connector-codex/tsconfig.json new file mode 100644 index 00000000..051d08e2 --- /dev/null +++ b/extensions/connector-codex/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "./src", + "outDir": "./dist" + }, + "include": ["src"] +} diff --git a/package.json b/package.json index 4c7754ac..89ee7563 100644 --- a/package.json +++ b/package.json @@ -47,6 +47,7 @@ "smoke:cross-path-dedup": "tsx extensions/connector-core/smoke/cross-path-dedup.smoke.ts", "smoke:feedback": "tsx extensions/connector-core/smoke/feedback.smoke.ts", "smoke:opencode": "tsx extensions/connector-opencode/smoke/turn-wedge.smoke.ts", + "smoke:codex": "tsx extensions/connector-codex/smoke/host-turn.smoke.ts", "smoke:attach": "tsx implementations/manager/smoke/attach.smoke.ts", "smoke:env-isolate": "tsx implementations/manager/smoke/env-isolate.smoke.ts", "smoke:install": "tsx implementations/cli/smoke/install.smoke.ts", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 28f7527b..27f2169c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -57,6 +57,9 @@ importers: '@cotal-ai/connector-claude-code': specifier: workspace:* version: link:../extensions/connector-claude-code + '@cotal-ai/connector-codex': + specifier: workspace:* + version: link:../extensions/connector-codex '@cotal-ai/connector-hermes': specifier: workspace:* version: link:../extensions/connector-hermes @@ -200,6 +203,19 @@ importers: specifier: ^4.22.4 version: 4.22.4 + extensions/connector-codex: + dependencies: + '@cotal-ai/connector-core': + specifier: workspace:* + version: link:../connector-core + tsx: + specifier: ^4.22.4 + version: 4.22.4 + devDependencies: + '@cotal-ai/core': + specifier: workspace:* + version: link:../../packages/core + extensions/connector-core: dependencies: '@modelcontextprotocol/sdk':