feat: support native persistent session queues

.github/workflows/compiled-process-smoke.yml

@@ -0,0 +1,29 @@
+name: Compiled POSIX process smoke
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+  workflow_dispatch:
+
+permissions: {}
+
+jobs:
+  compiled-process-smoke:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-node@v7
+        with:
+          node-version: '24'
+      - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6
+        with:
+          bun-version: 1.3.11
+      - run: npm ci
+      - name: Compile and exercise native override without Node on child PATH
+        timeout-minutes: 3
+        run: npm run test:compiled-process

.github/workflows/native-queue.yml

@@ -0,0 +1,43 @@
+name: Native queue E2E
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+  workflow_dispatch:
+
+permissions: {}
+
+jobs:
+  native-queue:
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 10
+    permissions:
+      contents: read
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-node@v7
+        with:
+          node-version: '24'
+      - uses: actions/setup-python@v6
+        with:
+          python-version: '3.13'
+      - name: Install locked dependencies (including native optional package)
+        run: npm ci
+      - name: Fresh build and real native queue tests with loopback replies
+        timeout-minutes: 5
+        run: npm run test:native-queue
+      - name: Preserve build hashes, RPC, model, spawn and failure evidence
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: native-queue-${{ matrix.os }}
+          path: tmp/native-queue/
+          include-hidden-files: true
+          if-no-files-found: warn
+          retention-days: 7

README.md

@@ -8,6 +8,7 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol]
 
 ## Features
 
+- Native persistent [message queue controls](docs/native-queue-controls.md), including edit, reorder, cancellation and explicit start.
 - ChatGPT, API key, and client-provided custom gateway authentication.
 - Model, reasoning effort, fast mode, approval, and sandbox mode configuration.
 - Concrete recommended model and reasoning-effort values through the opt-in [AIR recommended config values](docs/air-extensions.md#recommended-config-values) capability.

docs/native-queue-controls.md

@@ -0,0 +1,103 @@
+# Native persistent queue controls / 原生持久队列
+
+## English
+
+Inspect `initialize._meta.queue.actions`; only individually probed native methods
+are advertised. The tested floor is native 0.160.1, not an asserted first release.
+All requests use `_session/queue` and a `sessionId`:
+
+| action | Additional fields | Native result |
+| --- | --- | --- |
+| list | optional cursor, limit | data, nextCursor |
+| add | input, clientUserMessageId | queuedSubmission |
+| update | queuedSubmissionId, input | queuedSubmission |
+| delete | queuedSubmissionId | deleted |
+| reorder | queuedSubmissionIds | empty object |
+| start | optional queuedSubmissionId | turn |
+
+Responses are `{status:"ok",result:...}` or `{status:"unsupported"}`. Native
+failures stay errors. `input` uses native UserInput, including `text_elements`;
+this method does not convert ACP prompt blocks. `update` preserves the original
+client message ID. Reorder supplies every current pending ID exactly once; a
+stale set is rejected by native storage. List again after a conflict.
+
+Queue persistence and scheduling use Codex's original `thread/queue/*` methods.
+Adding to a loaded idle thread can start generation immediately. While a turn
+runs, additions remain pending. Successful or failed completion advances the
+queue. Consumption means accepted for execution, not successful completion.
+Explicit start can select a non-head entry; subsequent entries retain their
+order and drain automatically. It does not copy the entry into `session/prompt`.
+
+The adapter installs notifications and interactive delegates before add/start or
+native resume. Autonomous turns use fresh event and permission contexts.
+Ordinary ACP updates carry output/tools/usage; `_session/queue/turn` carries
+`{sessionId,turn}` at start and completion, including native status/error.
+`_session/queue/changed` carries `{sessionId}` as an invalidation, not a snapshot
+or an acknowledgement that generation has stopped. These notification names
+are also exposed in the capability. Hosts must register them before use.
+
+`session/cancel` interrupts the current owned turn. A native interrupted turn
+preserves pending messages and suppresses automatic advancement; use explicit
+start to continue. Cancel racing a completed turn is not an atomic pause of the
+whole queue. Queue entries can still be edited/reordered/deleted during active
+generation. Close keeps the closing fence while interrupting/unsubscribing.
+The native unsubscribe operation is not itself a durable pause.
+
+Unloaded queues can be listed, edited, reordered and deleted without resuming.
+Add without an ACP session requires native `notLoaded` status, because an
+unsubscribed but still loaded thread can auto-dispatch. Load an idle session
+before explicit start. Load/resume may automatically execute a non-interrupted
+persisted queue. Ordinary prompts are refused while a pending/active queue is
+observed; use the queue start/control operations or explicit steering.
+External native writers remain outside the adapter's admission fence.
+
+Queue mutations are serialized against local lifecycle/settings changes and
+bounded at 30 seconds. A timeout invalidates and terminates the provider using
+the existing runtime recovery path. An ambiguous mutation failure fences later
+queue writes until successful load/reconciliation. Never automatically retry an
+add/start: admission and durable queue deletion are not one atomic operation.
+No exactly-once execution, independent pause API or arbitrary native RPC tunnel
+is promised. Subagent behavior remains subject to existing negotiated support.
+
+Native source references: `codex-rs/app-server/src/request_processors/thread_queue_processor.rs`,
+`codex-rs/ext/queue/src/service.rs`, `codex-rs/state/src/runtime/queued_items.rs`
+at `e9e6cf63492da108a3cabaf3cc7b23d0640ef4cf`. Desktop calls the same six queue
+operations. Source semantics and executable verification are separate evidence.
+
+After an app-server crash, queue actions re-probe the replacement server. A session
+that was loaded in the dead server requires an explicit `session/load` before queue
+mutations; an uncertain write is never replayed. Lazy recovery of an ordinary prompt
+refuses to resume a nonempty queue. Explicit load subscribes before native resume and
+retains cancellation while history is buffered. Direct session-index rename/archive/
+unarchive requests share the queue reservation and cannot overlap a pending queue
+write or cancellation. Existing serialized index writes retain their ordering.
+
+## 中文
+
+客户端先读取 `initialize._meta.queue.actions`，仅调用逐项原生探测通过的动作。
+接口 `_session/queue` 统一带 `sessionId`，参数和返回见上表。新增/编辑采用原生
+UserInput（含 `text_elements`），不把 ACP prompt 格式直接混入。编辑保留原始
+clientUserMessageId；重排必须包含当前全部待发 ID 且各一次，集合陈旧则重读后处理。
+
+原生负责持久化和派发，不另建调度器。空闲已加载会话的 add 可能立即生成；生成中
+新增则排队，正常结束或失败后继续下一条。start 可选择非队首消息，该条结束后按
+剩余顺序继续。不能通过重发同一文本模拟 start，否则可能重复执行。
+
+ACP 在 resume/add/start 前注册事件和审批处理，自动轮次使用独立生命周期。
+输出/工具/用量沿用普通 session/update；`_session/queue/turn` 通知开始/结束及原生
+turn 状态，`_session/queue/changed` 提醒重读列表。前端应先注册这两个通知。
+
+session/cancel 中断当前轮次；原生 interrupted 状态保留剩余消息、暂停自动推进，
+显式 start 恢复。取消恰好遇到轮次完成不构成全队列原子暂停，unsubscribe 也不等于
+持久暂停。生成中允许编辑/重排/删除待发项。未加载队列可管理；无 ACP 本地会话时
+add 必须确认原生 notLoaded，避免未订阅但仍已加载的线程自动生成无人处理。
+load/resume 可能恢复非 interrupted 队列并自动执行。
+
+app-server 崩溃后重新探测替代进程的能力。旧进程中已加载的会话需显式
+session/load 后再修改队列；普通 prompt 的自动恢复不会隐式派发非空持久队列。
+原生恢复前先订阅，历史缓冲期间仍可取消。sessionIndex 的直接重命名、归档、
+取消归档接口也与队列写入及取消互斥，同时保留原有索引写入串行顺序。
+
+30 秒 mutation 超时沿用 provider 失效与终止流程；结果不明确时要求成功 load 对账。
+不自动重试 add/start、不承诺 exactly-once。外部原生客户端的并发写入不受 ACP
+本地保护锁控制。此次仅扩展 ACP，Codeg 界面接入和全局安装不在本次改动中。

package.json

@@ -40,6 +40,8 @@
     "release:preflight": "bash scripts/release-preflight.sh",
     "test": "vitest run --no-file-parallelism --retry=2",
     "test:e2e": "npm run build && RUN_E2E_TESTS=true vitest run --no-file-parallelism --retry=2 src/__tests__/CodexACPAgent/e2e",
+    "test:native-queue": "node scripts/native-queue.mjs",
+    "test:compiled-process": "node scripts/CodexProcessTermination-smoke.mjs",
     "test:watch": "vitest",
     "typecheck": "tsc --noEmit && tsc --noEmit -p examples/tsconfig.json",
     "codex-test": "tsx .claude/skills/run-codex/scripts/run-codex-test.ts"

scripts/CodexProcessTermination-smoke.mjs

@@ -0,0 +1,47 @@
+#!/usr/bin/env node
+// Linux/macOS: node scripts/CodexProcessTermination-smoke.mjs
+// Offline: append --compile-executable-path=/absolute/path/to/bun
+// Containers must use --init so killed orphan group members are reaped.
+import {spawnSync} from "node:child_process";
+import {mkdtempSync, mkdirSync, rmSync} from "node:fs";
+import {tmpdir} from "node:os";
+import path from "node:path";
+import {fileURLToPath} from "node:url";
+
+if (process.platform === "win32") throw new Error("Run this POSIX process-group smoke test on Linux/macOS (containers: --init)");
+const options = process.argv.slice(2);
+if (options.length > 1 || options.some(value => !value.startsWith("--compile-executable-path="))) {
+    throw new Error("Usage: node scripts/CodexProcessTermination-smoke.mjs [--compile-executable-path=/absolute/path/to/bun]");
+}
+const root = fileURLToPath(new URL("../", import.meta.url));
+const work = mkdtempSync(path.join(tmpdir(), "CodexProcessTermination-smoke-"));
+const noNode = path.join(work, "empty-path");
+mkdirSync(noNode);
+// Keep fixture runs isolated from the user's ACP configuration and log settings.
+const env = {...process.env, APP_SERVER_LOGS: "", CODEX_CONFIG: "{}", DEFAULT_AUTH_REQUEST: ""};
+function run(command, args, runEnv = env) {
+    const result = spawnSync(command, args, {
+        cwd: root, env: runEnv, stdio: "inherit", timeout: 60_000,
+        killSignal: "SIGKILL", windowsHide: true,
+    });
+    if (result.error) throw result.error;
+    if (result.status !== 0) throw new Error(`${path.basename(command)} failed: status=${result.status}, signal=${result.signal}`);
+}
+function compile(source, name) {
+    const output = path.join(work, name);
+    // Native host target avoids cross-runtime downloads; an explicit cached Bun
+    // executable may be supplied for fully offline reproduction.
+    run("bun", ["build", "--minify", "--compile", ...options, source, "--outfile", output]);
+    return output;
+}
+try {
+    const fixture = compile("src/__tests__/fixtures/CodexProcessTermination.cjs", "native-fixture");
+    const transport = compile("src/__tests__/fixtures/CodexProcessTermination.compiled.ts", "transport-smoke");
+    const acp = compile("src/index.ts", "codex-acp");
+    run(transport, [fixture, "native"], {...env, PATH: noNode});
+    run(process.execPath, ["src/__tests__/fixtures/CodexProcessTermination.entrypoint.cjs", acp, fixture],
+        {...env, PATH: noNode});
+    console.log("PASS compiled POSIX smoke: no Node on child PATH; group termination, EOF, timeout and ACP entrypoint verified");
+} finally {
+    rmSync(work, {recursive: true, force: true});
+}

scripts/native-queue.mjs

@@ -0,0 +1,106 @@
+// No account, external provider, globally installed Codex, or saved fixture is used.
+import { spawnSync } from 'node:child_process';
+import { createHash } from 'node:crypto';
+import { mkdirSync, mkdtempSync, readFileSync, readdirSync, realpathSync, writeFileSync } from 'node:fs';
+import { createRequire } from 'node:module';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const repo = path.resolve(path.dirname(fileURLToPath(import.meta.url)), '..');
+const args = process.argv.slice(2);
+let output, wrongReply = false;
+for (let i = 0; i < args.length; i++) {
+  if (args[i] === '--wrong-reply' && !wrongReply) wrongReply = true;
+  else if (args[i] === '--output' && !output && args[i + 1] && !args[i + 1].startsWith('--')) output = args[++i];
+  else {
+    console.error('Usage: npm run test:native-queue -- [--wrong-reply] [--output <artifact-parent-directory>]');
+    process.exit(2);
+  }
+}
+if (Number(process.versions.node.split('.')[0]) < 24) throw new Error('Node 24+ is required');
+const parent = path.resolve(output ?? path.join(repo, 'tmp', 'native-queue'));
+mkdirSync(parent, { recursive: true });
+const run = mkdtempSync(path.join(parent, 'run-'));
+for (const name of ['home', 'workspace', 'tmp', 'home/appdata', 'home/localappdata', 'home/.config', 'home/.cache']) {
+  mkdirSync(path.join(run, name), { recursive: true });
+}
+// A whitelist deliberately excludes credentials, proxies, NODE_OPTIONS, Git
+// overrides, Codex configuration, and the invoking user's profile directories.
+const env = Object.fromEntries(Object.entries(process.env).filter(([key]) =>
+  ['PATH', 'SYSTEMROOT', 'WINDIR', 'COMSPEC', 'PATHEXT'].includes(key.toUpperCase())));
+Object.assign(env, {
+  HOME: path.join(run, 'home'), USERPROFILE: path.join(run, 'home'),
+  CODEX_HOME: path.join(run, 'home'), APPDATA: path.join(run, 'home/appdata'),
+  LOCALAPPDATA: path.join(run, 'home/localappdata'),
+  XDG_CONFIG_HOME: path.join(run, 'home/.config'), XDG_CACHE_HOME: path.join(run, 'home/.cache'),
+  TMP: path.join(run, 'tmp'), TEMP: path.join(run, 'tmp'), TMPDIR: path.join(run, 'tmp'),
+  GIT_CONFIG_NOSYSTEM: '1', GIT_CONFIG_GLOBAL: path.join(run, 'home/.gitconfig'),
+  GIT_OPTIONAL_LOCKS: '0',
+});
+writeFileSync(env.GIT_CONFIG_GLOBAL, '');
+const hash = (file) => createHash('sha256').update(readFileSync(file)).digest('hex');
+function walk(dir) {
+  return readdirSync(dir, { withFileTypes: true }).flatMap((entry) => {
+    const file = path.join(dir, entry.name);
+    return entry.isDirectory() ? walk(file) : [file];
+  }).sort();
+}
+function sources() {
+  return Object.fromEntries([...walk(path.join(repo, 'src')),
+    ...walk(path.join(repo, 'scripts/native-queue')),
+    ...['package.json', 'package-lock.json', 'build.mjs', 'scripts/native-queue.mjs'].map((name) => path.join(repo, name))]
+    .map((file) => [path.relative(repo, file), hash(file)]));
+}
+function command(binary, argv) {
+  const result = spawnSync(binary, argv, {
+    cwd: repo, env, windowsHide: true, encoding: 'utf8', timeout: 120_000,
+  });
+  if (result.error || result.status !== 0) {
+    throw new Error(`${binary} failed: ${result.error ?? result.stderr ?? result.status}`);
+  }
+  return result.stdout.trim();
+}
+try {
+  const require = createRequire(path.join(repo, 'package.json'));
+  const codexManifest = require.resolve('@openai/codex/package.json');
+  const installed = JSON.parse(readFileSync(codexManifest, 'utf8'));
+  const version = installed.version.match(/^(\d+)\.(\d+)\.(\d+)$/)?.slice(1).map(Number);
+  if (!version || !(version[0] > 0 || version[1] > 160 || (version[1] === 160 && version[2] >= 1))) {
+    throw new Error(`Native queue requires @openai/codex 0.160.1+, found ${installed.version}`);
+  }
+  const packageName = `@openai/codex-${process.platform}-${process.arch}`;
+  if (!installed.optionalDependencies?.[packageName]) throw new Error(`Unsupported native platform: ${packageName}`);
+  const nativeManifest = createRequire(codexManifest).resolve(`${packageName}/package.json`);
+  // Discover the actual installed layout; do not assume a vendor triple or bin/codex subdirectory.
+  const candidates = walk(path.join(path.dirname(nativeManifest), 'vendor'))
+    .filter((file) => path.basename(file) === (process.platform === 'win32' ? 'codex.exe' : 'codex'));
+  if (candidates.length !== 1) throw new Error(`Expected one installed native executable, found ${candidates.length}`);
+  const native = realpathSync(candidates[0]);
+  const nativeVersion = command(native, ['--version']);
+  if (nativeVersion !== `codex-cli ${installed.version}`) throw new Error(`Native/package version mismatch: ${nativeVersion}`);
+  const before = sources();
+  writeFileSync(path.join(run, 'build.log'), command(process.execPath, [path.join(repo, 'build.mjs')]));
+  if (JSON.stringify(before) !== JSON.stringify(sources())) throw new Error('Sources changed during build; rerun after concurrent edits finish');
+  const metadata = {
+    repo, run, wrongReply, node: process.execPath, nodeVersion: process.version, native,
+    nativePackage: packageName, codexVersion: installed.version,
+    nativeVersion, nativeSha256: hash(native),
+    distSha256: hash(path.join(repo, 'dist/index.js')), sources: before,
+  };
+  const manifest = path.join(run, 'build.json');
+  writeFileSync(manifest, JSON.stringify(metadata, null, 2) + '\n');
+  const python = process.env.NATIVE_E2E_PYTHON || (process.platform === 'win32' ? 'python' : 'python3');
+  console.log(`Native queue E2E artifacts: ${run}`);
+  const result = spawnSync(python, ['-B', path.join(repo, 'scripts/native-queue/probe.py'), manifest], {
+    // Python bounds each RPC/child shutdown and cleans up in finally. Killing
+    // just Python on a launcher timeout would orphan its owned ACP/native tree.
+    cwd: repo, env, windowsHide: true, stdio: 'inherit',
+  });
+  if (result.error) throw result.error;
+  process.exitCode = result.status ?? 1;
+} catch (error) {
+  writeFileSync(path.join(run, 'launcher-failure.txt'), String(error.stack ?? error));
+  console.error(error);
+  console.error(`Native queue E2E artifacts: ${run}`);
+  process.exitCode = 1;
+}