diff --git a/.github/workflows/codex-steering-experiment.yml b/.github/workflows/codex-steering-experiment.yml new file mode 100644 index 0000000..5f2c177 --- /dev/null +++ b/.github/workflows/codex-steering-experiment.yml @@ -0,0 +1,36 @@ +name: Codex steering experiment + +on: + pull_request: + paths: + - 'experiments/codex_steering/**' + - '.github/workflows/codex-steering-experiment.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + fixture-tests: + name: Fixtures (${{ matrix.os }}, Node ${{ matrix.node }}) + runs-on: ${{ matrix.os }} + timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest] + node: [22, 24] + defaults: + run: + working-directory: experiments/codex_steering + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: ${{ matrix.node }} + - run: npm ci --ignore-scripts + - run: npm run typecheck + - run: npm test + env: + TASK_TEMP: ${{ runner.temp }}/codex-steering-fixtures + - run: npm audit diff --git a/experiments/codex_steering/README.md b/experiments/codex_steering/README.md new file mode 100644 index 0000000..3921484 --- /dev/null +++ b/experiments/codex_steering/README.md @@ -0,0 +1,92 @@ +# Opt-in Codex steering experiment + +This is an unreleased, local TypeScript experiment for issue #141. It is not an +installed `session-peer` command and is excluded from the npm package. Ordinary +queue delivery, runtime behavior, SSH, and the Side Session experiment remain +unchanged. + +## Purpose and limits + +The prototype connects to an operator-selected, private Unix-domain WebSocket +of an existing Codex app-server. It submits user text using `turn/steer` with a +full thread UUID and the operator's exact `expectedTurnId`. It does not start a +server or session. The exact advertised server version candidate is **0.159.2**, +drawn from the Python experiment; that is not a qualification of native Codex +0.159.2 for this TypeScript implementation or of newer Codex versions. + +Opt-in, absolute socket/home paths, and an exact home match are required. Socket +ownership/privacy and canonical home-path checks are local guards. They do not establish atomic +owner generation or exclude same-user endpoint impersonation. Loaded target, +active status, direct-input capability, and a thread-specific feature snapshot +are inspected. These snapshots can race with state changes. The actual +`expectedTurnId` guard is applied by native `turn/steer`, not by a dry-run. + +`--require-instant` requires an affirmative `instant_interrupt` snapshot. +Disabled or unknown evidence is refused rather than treated as enabled. Without +that option, unknown/disabled feature evidence does not imply instant input +handling. The sender never enables a feature or overrides approvals or target +permissions. + +A metadata-only dry-run performs no `turn/steer` call and does not prove that the +expected turn matches. An `accepted` result means that the native endpoint accepted +the input; it does not mean consumption, ACK, completion, or termination of +running tools. A refusal and an uncertain post-attempt outcome are distinct. +Neither causes queue fallback, retry, or resend. The prototype never calls +queue-add, `turn/start`, `turn/interrupt`, resume, fork, or wake. It does not +answer approval requests. + +## Privacy and resource boundary + +Supply text through private stdin, not a secret-bearing command line. Results +and diagnostics omit message text, transcript text, and raw server errors. +Bounded incoming notifications are discarded, but may themselves contain +content. `thread/read` with `includeTurns: false` can still return a preview; +this experiment does **not** promise body-free native access. It makes no ACK +collector or receipt capability and offers no generic completion wait. + +RPC/frame/input bounds and cleanup cover this experiment's owned connection and +temporary test fixtures. The transport's monotonic deadline starts after CLI +stdin has completed; waiting for interactive stdin is not covered by that deadline. +They do not stop the existing owner or undo native side +effects. Timeout/disconnect after a steering call is not proof of non-delivery +and must not trigger an automatic second submission. + +## Isolated usage + +Use Node 22 or 24 for the separate experiment package. Never point the example +at an existing user session without an explicit, separate test authorization. +The values below are placeholders; no native test is authorized by this README. + +```sh +cd experiments/codex_steering +npm ci --ignore-scripts +npm run typecheck +npm test +printf '%s' 'Change the test review focus' | node --experimental-strip-types steer.ts \ + --experimental-steering --socket /private-test/app.sock \ + --expected-home /private-test/home \ + --to 00000000-0000-4000-8000-000000000001 \ + --expected-turn-id turn-example --message - --dry-run \ + --require-instant --timeout 10 +``` + +The experiment has its own dependencies, typecheck, and fixture tests. Root +runtime build/package tests do not qualify it. Its CI uses fake Unix sockets on +macOS/Linux; it invokes no native agent, provider, or existing user session and +uses no account credentials. + +## Qualification still required + +Native queue versus steer with `instant_interrupt` ON/OFF, real-provider input +handling, code-mode behavior, current native versions, owner lifecycle, +generation/authorization, and snapshot/acceptance races remain unqualified. +SSH and Windows are not implemented by this local Unix-socket prototype. Linux +fixture CI is not a native Linux steering test. Python's historical synthetic +native observations must not be presented as TypeScript test results. See +[RESULTS.md](RESULTS.md) for the exact executed scope. + +References: + +- [Official App Server: steer an active turn](https://learn.chatgpt.com/docs/app-server#steer-an-active-turn) +- [Python issue #247](https://github.com/abruption/session-peer/issues/247) +- [Python Draft PR #248](https://github.com/abruption/session-peer/pull/248), reference commit `6ebedde` (historical experimental evidence, not TS qualification) diff --git a/experiments/codex_steering/RESULTS.md b/experiments/codex_steering/RESULTS.md new file mode 100644 index 0000000..997f97d --- /dev/null +++ b/experiments/codex_steering/RESULTS.md @@ -0,0 +1,62 @@ +# TypeScript Codex steering experiment results + +## Status + +Unreleased experiment. Local checks below ran on 2026-10-12 KST on macOS +arm64, Node 24.16.0/npm 11.15.0. Current-head CI results are recorded in the +Draft PR rather than assumed from these local results. No native Codex, real provider, or existing user session was invoked +for this initial record. There is no merge, release, or general-support claim. + +## Evidence categories + +| Check | Current evidence | +| --- | --- | +| Separate package typecheck and fake-socket tests | PASS: strict TypeScript; 24 tests passed, 0 failed/skipped | +| macOS/Linux Node 22/24 fixture CI | Pending execution/report | +| Root npm artifact exclusion | PASS: npm pack dry-run lists 57 files, none under experiments/ | +| Actual native Codex 0.159.2 | Not run; exact-version candidate only | +| Native queue/steer × instant_interrupt ON/OFF | Not run | +| Real provider or model | Not run | +| Existing user session or live message | Not run | +| Owner-generation, restart, same-user impersonation resistance | Not established | +| Native Linux/Windows or SSH steering | Not qualified | +| Consumption, ACK, completion, running-tool cancellation | Not established | + +Tests using fake sockets can establish request construction, local guards, +bounded parsing, refusal/uncertainty classification, and owned-connection cleanup. +They cannot establish native acceptance, actual active-turn handling, or +instant-input latency. Metadata-only dry-run is not an expected-turn proof. + +## Remaining gates + +1. Independent current-head review and completed fixture/platform CI. +2. Explicit authorization and isolated native tests before any actual steering. +3. Native ON/OFF ordering and real-provider/code-mode behavior; no latency SLA. +4. Safe owner endpoint discovery/generation, authorization, and lifecycle. +5. Submission races, interruption/disconnect after acceptance, and concurrency. +6. Any SSH/Windows design, stable installed CLI contract, packaging, and release + require separate implementation and approval. + +The Python experiment at reference commit `6ebedde` reports historical native +observations for its own harness. Those results are not copied into the TS +counts or treated as proof that this implementation works against a native +server. No automatic fallback/resubmission is permitted after an attempted +effect, including when observation is incomplete. + +## Local commands and checks + +- In the experiment package: `npm ci --ignore-scripts`, `npm run typecheck`, + `npm test`, `npm audit` (zero vulnerabilities; ws 8.22.0). +- In the root package: `npm run build`, `node scripts/check-repository.mjs`, + `npm pack --ignore-scripts --dry-run --json`; inspected all packaged paths. +- Fake tests cover private socket guards, exact request/turn matching, dry-run, + active/direct/home/feature guards, malformed wire/UTF-8/binary/server requests, + disconnect and post-attempt timeout, concurrent/repeated attempt fencing, + cursor cycles, event flood, frame limits, and fixture cleanup. +- Read-only internal adapter review found no blocker for the Draft scope. + This is not native qualification or an external model review. + +The root installed runtime, lockfile, version, parity pins, and existing +workflows are unchanged. A separate fake-only experiment workflow was added. +The input byte limit includes UTF-8 bytes; interactive stdin has no elapsed-time +bound. The transport's total deadline starts after stdin completes. diff --git a/experiments/codex_steering/package-lock.json b/experiments/codex_steering/package-lock.json new file mode 100644 index 0000000..b0a7743 --- /dev/null +++ b/experiments/codex_steering/package-lock.json @@ -0,0 +1,82 @@ +{ + "name": "session-peer-ts-steering-experiment", + "version": "0.0.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "session-peer-ts-steering-experiment", + "version": "0.0.0", + "dependencies": { + "ws": "8.22.0" + }, + "devDependencies": { + "@types/node": "22.13.10", + "@types/ws": "8.18.2", + "typescript": "5.9.3" + } + }, + "node_modules/@types/node": { + "version": "22.13.10", + "resolved": "https://registry.npmjs.org/@types/node/-/node-22.13.10.tgz", + "integrity": "sha512-I6LPUvlRH+O6VRUqYOcMudhaIdUVWfsjnZavnsraHvpBwaEyMN29ry+0UVJhImYL16xsscu0aske3yA+uPOWfw==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~6.20.0" + } + }, + "node_modules/@types/ws": { + "version": "8.18.2", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.2.tgz", + "integrity": "sha512-67MQl+fpWKVTT1NYdnmo3U4sc/xPo/zQBncVnI74qmQa0z/b+1g6iYqNmGCPbxO+zz2aklb08a0oHfegiVd0/w==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, + "node_modules/typescript": { + "version": "5.9.3", + "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", + "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", + "dev": true, + "license": "Apache-2.0", + "bin": { + "tsc": "bin/tsc", + "tsserver": "bin/tsserver" + }, + "engines": { + "node": ">=14.17" + } + }, + "node_modules/undici-types": { + "version": "6.20.0", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.20.0.tgz", + "integrity": "sha512-Ny6QZ2Nju20vw1SRHe3d9jVu6gJ+4e3+MMpqu7pqE5HT6WsTSlce++GQmK5UXS8mzV8DSYHrQH+Xrf2jVcuKNg==", + "dev": true, + "license": "MIT" + }, + "node_modules/ws": { + "version": "8.22.0", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.22.0.tgz", + "integrity": "sha512-Ydggc987+RO0AnWtZ/7Wq9FtNvcrL1b/RO0ud9mWjUPgDrsAAwQSF51sm2hm1XofbU/4jkpGEsLFsZZxU+1DOg==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + } + } +} diff --git a/experiments/codex_steering/package.json b/experiments/codex_steering/package.json new file mode 100644 index 0000000..7455b76 --- /dev/null +++ b/experiments/codex_steering/package.json @@ -0,0 +1,18 @@ +{ + "name": "session-peer-ts-steering-experiment", + "version": "0.0.0", + "private": true, + "type": "module", + "scripts": { + "test": "node --experimental-strip-types --test test_steer.ts", + "typecheck": "tsc --noEmit --strict --module NodeNext --target ES2022 --allowImportingTsExtensions --skipLibCheck steer.ts test_steer.ts" + }, + "dependencies": { + "ws": "8.22.0" + }, + "devDependencies": { + "@types/node": "22.13.10", + "@types/ws": "8.18.2", + "typescript": "5.9.3" + } +} diff --git a/experiments/codex_steering/steer.ts b/experiments/codex_steering/steer.ts new file mode 100644 index 0000000..4f53a56 --- /dev/null +++ b/experiments/codex_steering/steer.ts @@ -0,0 +1,257 @@ +/** Local, opt-in research adapter. Never imported by the installed CLI. */ +import fs from 'node:fs'; +import path from 'node:path'; +import net from 'node:net'; +import { randomUUID } from 'node:crypto'; +import { performance } from 'node:perf_hooks'; +import { fileURLToPath } from 'node:url'; +import WebSocket, { type RawData } from 'ws'; + +export type ObjectValue = Record; +export interface Caller { readonly info: ObjectValue; call(method: string, params: ObjectValue): Promise } +export class Refused extends Error {} +export class Unknown extends Error {} +export class RpcRejected extends Refused { + readonly code: number; + constructor(method: string, code: number) { super(`rpc_rejected:${method}`); this.code = code; } +} +export const MAX_FRAME = 1_048_576; +const METHODS = new Set(['initialize', 'thread/loaded/list', 'thread/read', 'experimentalFeature/list', 'turn/steer']); +const decoder = new TextDecoder('utf-8', { fatal: true }); +const object = (value: unknown): value is ObjectValue => value !== null && typeof value === 'object' && !Array.isArray(value); +const uuid = (value: unknown): string | null => typeof value === 'string' && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value) ? value.toLowerCase() : null; +function canonicalHome(value: unknown): string { + if (typeof value !== 'string' || !path.isAbsolute(value)) throw new Refused('server_home_mismatch'); + try { const result = fs.realpathSync(value); if (!fs.statSync(result).isDirectory()) throw new Error(); return result; } + catch { throw new Refused('server_home_mismatch'); } +} +function socketIdentity(socketPath: string): fs.Stats { + if (process.platform === 'win32') throw new Refused('unsupported_platform'); + if (!path.isAbsolute(socketPath)) throw new Refused('absolute_socket_required'); + const uid = process.getuid?.(); + if (uid === undefined) throw new Refused('owned_socket_required'); + try { + // Inspect components without following a link. The immediate directory and + // endpoint are private; system ancestor directories need not be private. + let component = path.parse(socketPath).root; + for (const part of socketPath.slice(component.length).split(path.sep).filter(Boolean)) { + component = path.join(component, part); + if (fs.lstatSync(component).isSymbolicLink()) throw new Refused('symlink_socket_path'); + } + const parent = fs.lstatSync(path.dirname(socketPath)); + const endpoint = fs.lstatSync(socketPath); + if (!parent.isDirectory() || parent.uid !== uid || (parent.mode & 0o077) !== 0) throw new Refused('private_directory_required'); + if (!endpoint.isSocket() || endpoint.uid !== uid) throw new Refused('owned_socket_required'); + if ((endpoint.mode & 0o077) !== 0) throw new Refused('private_socket_required'); + return endpoint; + } catch (error) { if (error instanceof Refused) throw error; throw new Refused('socket_unavailable'); } +} +interface Pending { id: number; method: string; resolve(value: ObjectValue): void; reject(error: Error): void; timer: ReturnType } +export class Rpc implements Caller { + private ws: WebSocket; + private deadline: number; + private pending: Pending | undefined; + private sequence = 0; + private notifications = 0; + private traffic = 0; + private closed = false; + private initialized = false; + private initializeInfo: ObjectValue = {}; + get info(): ObjectValue { return this.initializeInfo; } + private constructor(ws: WebSocket, deadline: number) { + this.ws = ws; this.deadline = deadline; + ws.on('message', (data: RawData, binary: boolean) => this.message(data, binary)); + ws.on('error', () => this.fail(new Refused('transport_closed'))); + ws.on('close', () => this.fail(new Refused('transport_closed'))); + } + static async connect(socketPath: string, timeoutMs = 10_000): Promise { + if (!Number.isInteger(timeoutMs) || timeoutMs < 1 || timeoutMs > 30_000) throw new Refused('invalid_timeout'); + const deadline = performance.now() + timeoutMs; + const before = socketIdentity(socketPath); + const ws = new WebSocket('ws://localhost/', { + createConnection: () => net.createConnection({ path: socketPath }), + maxPayload: MAX_FRAME, perMessageDeflate: false, + handshakeTimeout: Math.max(1, Math.ceil(deadline - performance.now())), + }); + const rpc = new Rpc(ws, deadline); + try { + await new Promise((resolve, reject) => { + const timer = setTimeout(() => finish(new Refused('operation_timeout')), Math.max(1, Math.ceil(deadline - performance.now()))); + const opened = () => finish(); + const failed = () => finish(new Refused('connection_failed')); + const finish = (error?: Error) => { clearTimeout(timer); ws.off('open', opened); ws.off('error', failed); ws.off('close', failed); error ? reject(error) : resolve(); }; + ws.once('open', opened); ws.once('error', failed); ws.once('close', failed); + }); + const after = socketIdentity(socketPath); + if (before.dev !== after.dev || before.ino !== after.ino) throw new Refused('socket_changed'); + rpc.initializeInfo = await rpc.call('initialize', { + clientInfo: { name: 'session_peer_steer_experiment', version: '0.0.0' }, + capabilities: { experimentalApi: true }, + }); + if (typeof rpc.info.userAgent !== 'string' || !/^(?:codex-tui|codex_cli_rs|codex-cli|session_peer_steer_experiment)\/0\.159\.2(?:\s|$)/.test(rpc.info.userAgent)) throw new Refused('unsupported_server_version'); + canonicalHome(rpc.info.codexHome); + rpc.initialized = true; + rpc.write({ method: 'initialized' }); + return rpc; + } catch (error) { rpc.close(); throw error; } + } + close(): void { + this.fail(new Refused('transport_closed')); + this.ws.terminate(); + } + private fail(error: Error): void { + this.closed = true; + const pending = this.pending; this.pending = undefined; + if (pending) { clearTimeout(pending.timer); pending.reject(error); } + } + private write(value: ObjectValue): void { + if (this.closed || this.ws.readyState !== WebSocket.OPEN) throw new Refused('transport_closed'); + if (performance.now() >= this.deadline) throw new Refused('operation_timeout'); + const frame = Buffer.from(JSON.stringify(value), 'utf8'); + if (frame.length > MAX_FRAME) throw new Refused('frame_too_large'); + this.traffic += frame.length; + if (this.traffic > 4 * MAX_FRAME) throw new Refused('traffic_limit'); + this.ws.send(frame, { binary: false }, error => { if (error) this.fail(new Refused('transport_closed')); }); + } + private message(data: RawData, binary: boolean): void { + if (this.closed) return; + try { + const bytes = Array.isArray(data) ? Buffer.concat(data) : data instanceof ArrayBuffer ? Buffer.from(data) : data; + this.traffic += bytes.length; + if (binary || bytes.length > MAX_FRAME || this.traffic > 4 * MAX_FRAME) throw new Refused('invalid_frame'); + let value: unknown; + try { value = JSON.parse(decoder.decode(bytes)); } catch { throw new Refused('invalid_rpc_response'); } + if (!object(value)) throw new Refused('invalid_rpc_response'); + if ('method' in value) { + if ('id' in value) throw new Refused('server_interaction_required'); + if (typeof value.method !== 'string' || 'result' in value || 'error' in value) throw new Refused('invalid_rpc_response'); + if (++this.notifications > 100) throw new Refused('event_limit'); + return; // Never answer server requests or retain body-bearing events. + } + if (!this.pending || value.id !== this.pending.id) throw new Refused('invalid_rpc_response'); + if (('result' in value) === ('error' in value)) throw new Refused('invalid_rpc_response'); + const pending = this.pending; + if ('error' in value) { + if (!object(value.error) || !Number.isSafeInteger(value.error.code)) throw new Refused('invalid_rpc_response'); + this.pending = undefined; clearTimeout(pending.timer); + pending.reject(new RpcRejected(pending.method, value.error.code as number)); + } else { + if (!object(value.result)) throw new Refused('invalid_rpc_response'); + this.pending = undefined; clearTimeout(pending.timer); pending.resolve(value.result); + } + } catch (error) { this.fail(error instanceof Refused ? error : new Refused('invalid_rpc_response')); this.ws.terminate(); } + } + async call(method: string, params: ObjectValue): Promise { + if (!METHODS.has(method) || (method === 'initialize' && (this.initialized || this.sequence > 0))) throw new Refused('method_not_allowed'); + if (this.pending) throw new Refused('concurrent_rpc'); + if (this.closed || this.ws.readyState !== WebSocket.OPEN) throw new Refused('transport_closed'); + if (++this.sequence > 80) throw new Refused('rpc_limit'); + const remaining = this.deadline - performance.now(); + if (remaining <= 0) throw new Refused('operation_timeout'); + return await new Promise((resolve, reject) => { + const timer = setTimeout(() => { this.fail(new Refused('operation_timeout')); this.ws.terminate(); }, Math.ceil(remaining)); + this.pending = { id: this.sequence, method, resolve, reject, timer }; + try { this.write({ id: this.sequence, method, params }); } + catch (error) { this.fail(error instanceof Error ? error : new Refused('transport_closed')); } + }); + } +} +async function pages(rpc: Caller, method: string, params: ObjectValue): Promise { + const all: unknown[] = []; const seen = new Set(); let cursor: string | undefined; + for (let count = 0; count < 32; count++) { + const result = await rpc.call(method, { ...params, limit: 100, ...(cursor === undefined ? {} : { cursor }) }); + if (!Array.isArray(result.data) || result.data.length > 100) throw new Refused('invalid_inventory'); + all.push(...result.data); + if (result.nextCursor === null || result.nextCursor === undefined) return all; + if (typeof result.nextCursor !== 'string' || !result.nextCursor || Buffer.byteLength(result.nextCursor) > 1024 || seen.has(result.nextCursor)) throw new Refused('invalid_cursor'); + cursor = result.nextCursor; seen.add(cursor); + } + throw new Refused('inventory_limit'); +} +const busy = new WeakSet(); +const attempted = new WeakSet(); +export async function steer(rpc: Caller, target: string, expectedTurn: string, text: string, + options: { expectedHome: string; optIn?: boolean; requireInstant?: boolean; dryRun?: boolean }): Promise { + if (!options.optIn) throw new Refused('steering_opt_in_required'); + if (busy.has(rpc)) throw new Refused('concurrent_steering'); + if (attempted.has(rpc)) throw new Refused('submission_already_attempted'); + busy.add(rpc); + try { + const id = uuid(target.startsWith('codex:') ? target.slice(6) : target); + if (!id) throw new Refused('full_thread_uuid_required'); + if (typeof expectedTurn !== 'string' || !expectedTurn.trim() || Buffer.byteLength(expectedTurn) > 128 || /[\x00-\x1f\x7f]/.test(expectedTurn)) throw new Refused('expected_turn_required'); + if (typeof text !== 'string' || !text || text.includes('\0') || Buffer.byteLength(text) > 65_536 || /[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(? uuid(item) !== null)) throw new Refused('invalid_inventory'); + if (loaded.filter(item => uuid(item) === id).length !== 1) throw new Refused('target_not_loaded'); + const reply = await rpc.call('thread/read', { threadId: id, includeTurns: false }); + const thread = reply.thread; + if (!object(thread) || uuid(thread.id) !== id) throw new Refused('thread_mismatch'); + const status = thread.status; + if (!object(status) || status.type !== 'active') throw new Refused('target_not_active'); + if (!Array.isArray(status.activeFlags) || status.activeFlags.length !== 0) throw new Refused('active_interaction_pending'); + if (thread.canAcceptDirectInput !== true) throw new Refused('direct_input_unverified'); + const features = await pages(rpc, 'experimentalFeature/list', { threadId: id }); + if (!features.every(item => object(item) && typeof item.name === 'string')) throw new Refused('invalid_inventory'); + const matching = features.filter(item => object(item) && item.name === 'instant_interrupt') as ObjectValue[]; + const flag = matching.length === 1 && typeof matching[0]!.enabled === 'boolean' ? matching[0]!.enabled : null; + if (options.requireInstant && flag !== true) throw new Refused('instant_interrupt_not_enabled_or_verified'); + const result: ObjectValue = { schemaVersion: 1, ok: true, experimental: true, mode: 'steer', submitted: false, + consumptionConfirmed: false, retryAllowed: false, dryRun: !!options.dryRun, + instantInterruptEnabledSnapshot: flag, status: options.dryRun ? 'validated' : 'accepted' }; + if (options.dryRun) return { ...result, expectedTurnValidated: false }; + attempted.add(rpc); // Fence before the first potentially effectful call, even if it fails. + let accepted: ObjectValue; + try { + accepted = await rpc.call('turn/steer', { threadId: id, expectedTurnId: expectedTurn, + input: [{ type: 'text', text }], clientUserMessageId: randomUUID() }); + } catch (error) { + if (error instanceof RpcRejected && (error.code === -32600 || error.code === -32601)) throw error; + throw new Unknown('submission_outcome_unknown'); + } + if (!object(accepted) || accepted.turnId !== expectedTurn) throw new Unknown('returned_turn_mismatch'); + return { ...result, submitted: true, turnId: expectedTurn }; + } finally { busy.delete(rpc); } +} + +async function stdinMessage(): Promise { + const chunks: Buffer[] = []; let size = 0; + for await (const chunk of process.stdin) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + size += bytes.length; + if (size > 65_536) throw new Refused('invalid_message'); + chunks.push(bytes); + } + try { return decoder.decode(Buffer.concat(chunks)); } catch { throw new Refused('invalid_message'); } +} +export async function main(argv = process.argv.slice(2)): Promise { + let rpc: Rpc | undefined; + try { + const flags = new Set(['--experimental-steering', '--require-instant', '--dry-run']); + const values = new Set(['--socket', '--expected-home', '--to', '--expected-turn-id', '--message', '--timeout']); + const parsed = new Map(); + for (let i = 0; i < argv.length; i++) { + const key = argv[i]!; + if (parsed.has(key)) throw new Refused('invalid_arguments'); + if (flags.has(key)) parsed.set(key, true); + else if (values.has(key) && i + 1 < argv.length && !argv[i + 1]!.startsWith('--')) parsed.set(key, argv[++i]!); + else throw new Refused('invalid_arguments'); + } + if (!parsed.has('--experimental-steering')) throw new Refused('steering_opt_in_required'); + const get = (key: string): string => { const value = parsed.get(key); if (typeof value !== 'string' || !value) throw new Refused('invalid_arguments'); return value; }; + const timeout = parsed.has('--timeout') ? get('--timeout') : '10'; + if (!/^(?:[1-9]|[12][0-9]|30)$/.test(timeout)) throw new Refused('invalid_timeout'); + const socket = get('--socket'), home = get('--expected-home'), target = get('--to'), turn = get('--expected-turn-id'), message = get('--message'); + const text = message === '-' ? await stdinMessage() : message; + rpc = await Rpc.connect(socket, Number(timeout) * 1000); + const result = await steer(rpc, target, turn, text, { expectedHome: home, optIn: true, requireInstant: parsed.has('--require-instant'), dryRun: parsed.has('--dry-run') }); + console.log(JSON.stringify(result)); return 0; + } catch (error) { + const unknown = error instanceof Unknown; + const reason = error instanceof Refused || unknown ? error.message : 'connection_failed'; + console.log(JSON.stringify({ schemaVersion: 1, ok: false, experimental: true, mode: 'steer', status: unknown ? 'unknown' : 'refused', submitted: unknown ? null : false, consumptionConfirmed: false, retryAllowed: false, reason })); + return 1; + } finally { rpc?.close(); } +} +if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) process.exitCode = await main(); diff --git a/experiments/codex_steering/test_steer.ts b/experiments/codex_steering/test_steer.ts new file mode 100644 index 0000000..273505a --- /dev/null +++ b/experiments/codex_steering/test_steer.ts @@ -0,0 +1,251 @@ +import test, { type TestContext } from 'node:test'; +import assert from 'node:assert/strict'; +import { chmodSync, mkdirSync, mkdtempSync, realpathSync, rmSync, symlinkSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createServer } from 'node:http'; +import { performance } from 'node:perf_hooks'; +import { WebSocketServer, type WebSocket } from 'ws'; +import { Rpc, RpcRejected, Refused, Unknown, steer, type Caller, type ObjectValue } from './steer.ts'; + +const THREAD = '00000000-0000-4000-8000-000000000001', TURN = 'fixture-turn'; +function fixture(t: TestContext, cleanup = true) { + const base = process.env.TASK_TEMP ?? tmpdir(); + mkdirSync(base, { recursive: true, mode: 0o700 }); + const root = realpathSync(mkdtempSync(join(base, 'codex-ts-steer-'))); + chmodSync(root, 0o700); + const home = join(root, 'home'); mkdirSync(home, { mode: 0o700 }); + if (cleanup) t.after(() => rmSync(root, { recursive: true, force: true })); + return { root, home }; +} +class Fake implements Caller { + info: ObjectValue; + calls: { method: string; params: ObjectValue }[] = []; + active = true; loaded = true; direct = true; flag: unknown = true; + activeFlags: unknown = []; effectError?: Error; returnedTurn = TURN; + constructor(home: string) { this.info = { codexHome: home }; } + async call(method: string, params: ObjectValue): Promise { + this.calls.push({ method, params }); + if (method === 'thread/loaded/list') return { data: this.loaded ? [THREAD] : [], nextCursor: null }; + if (method === 'thread/read') return { thread: { id: THREAD, + status: { type: this.active ? 'active' : 'idle', activeFlags: this.activeFlags }, canAcceptDirectInput: this.direct } }; + if (method === 'experimentalFeature/list') return { data: [{ name: 'instant_interrupt', enabled: this.flag }], nextCursor: null }; + assert.equal(method, 'turn/steer'); + if (this.effectError) throw this.effectError; + return { turnId: this.returnedTurn }; + } +} +const input = 'private fixture input'; +function send(rpc: Caller, home: string, extra: Partial[4]> = {}) { + return steer(rpc, THREAD, TURN, input, { expectedHome: home, optIn: true, ...extra }); +} +const effects = (rpc: Fake) => rpc.calls.filter(c => c.method === 'turn/steer'); + +test('successful input targets the exact original turn once and is never an ACK', async t => { + const { home } = fixture(t), rpc = new Fake(home), result = await send(rpc, home); + assert.equal(result.status, 'accepted'); assert.equal(result.submitted, true); + assert.equal(result.turnId, TURN); assert.equal(result.consumptionConfirmed, false); + assert.equal(result.retryAllowed, false); + assert.equal(effects(rpc).length, 1); + const params = effects(rpc)[0]!.params; + assert.equal(params.threadId, THREAD); assert.equal(params.expectedTurnId, TURN); + assert.deepEqual(params.input, [{ type: 'text', text: input }]); + for (const key of ['model', 'cwd', 'sandboxPolicy', 'approvalPolicy']) assert.equal(key in params, false); + assert.equal(JSON.stringify(result).includes(input), false); + await assert.rejects(send(rpc, home), Refused); + assert.equal(effects(rpc).length, 1); +}); + +test('dry-run checks metadata but makes zero submissions and cannot prove the expected turn', async t => { + const { home } = fixture(t), rpc = new Fake(home), result = await send(rpc, home, { dryRun: true }); + assert.equal(result.status, 'validated'); assert.equal(result.submitted, false); + assert.equal(result.expectedTurnValidated, false); assert.equal(result.consumptionConfirmed, false); + assert.equal(effects(rpc).length, 0); + assert.deepEqual(rpc.calls.find(c => c.method === 'thread/read')?.params, { threadId: THREAD, includeTurns: false }); +}); + +test('explicit opt-in, valid message, full thread and expected turn are required before effect', async t => { + const { home } = fixture(t); + for (const [target, turn, text, optIn] of [ + [THREAD, TURN, input, false], ['prefix', TURN, input, true], + [THREAD, '', input, true], [THREAD, 'x'.repeat(129), input, true], + [THREAD, TURN, '', true], [THREAD, TURN, '\0', true], + [THREAD, TURN, 'é'.repeat(32769), true], + ] as const) { + const rpc = new Fake(home); + await assert.rejects(steer(rpc, target, turn, text, { expectedHome: home, optIn }), Refused); + assert.equal(effects(rpc).length, 0); + } +}); + +test('home, loaded/active target and direct input are pre-effect guards', async t => { + const { root, home } = fixture(t), other = join(root, 'other'); mkdirSync(other, { mode: 0o700 }); + for (const change of [ + (r: Fake) => { r.info.codexHome = other; }, (r: Fake) => { r.loaded = false; }, + (r: Fake) => { r.active = false; }, (r: Fake) => { r.direct = false; }, + (r: Fake) => { r.activeFlags = ['waitingOnApproval']; }, + ]) { + const rpc = new Fake(home); change(rpc); + await assert.rejects(send(rpc, home), Refused); assert.equal(effects(rpc).length, 0); + } +}); + +test('instant feature is a boolean snapshot and required unsupported snapshots refuse', async t => { + const { home } = fixture(t); + for (const flag of [false, null, undefined, 'true', 1]) { + const rpc = new Fake(home); rpc.flag = flag; + const result = await send(rpc, home, { dryRun: true }); + assert.equal(result.instantInterruptEnabledSnapshot, flag === false ? false : null); + await assert.rejects(send(rpc, home, { requireInstant: true }), Refused); + assert.equal(effects(rpc).length, 0); + } +}); + +test('native wrong-turn/method rejection is refused; other failures after attempt are unknown without fallback', async t => { + const { home } = fixture(t); + for (const [error, expected] of [ + [new RpcRejected('turn/steer', -32600), Refused], [new RpcRejected('turn/steer', -32601), Refused], + [new RpcRejected('turn/steer', -32603), Unknown], [new Error('private provider detail'), Unknown], + [new Refused('rpc_timeout'), Unknown], + ] as const) { + const rpc = new Fake(home); rpc.effectError = error; + await assert.rejects(send(rpc, home), expected); + await assert.rejects(send(rpc, home), Refused); + assert.equal(effects(rpc).length, 1); + assert.equal(rpc.calls.some(c => ['turn/start', 'turn/interrupt', 'thread/queue/add'].includes(c.method)), false); + } + const rpc = new Fake(home); rpc.returnedTurn = 'replaced-turn'; + await assert.rejects(send(rpc, home), Unknown); assert.equal(effects(rpc).length, 1); +}); + +test('a simultaneous operation cannot submit through the same caller', async t => { + const { home } = fixture(t), rpc = new Fake(home), original = rpc.call.bind(rpc); + let release!: () => void, entered!: () => void; + const gate = new Promise(r => { release = r; }), started = new Promise(r => { entered = r; }); + rpc.call = async (method, params) => { if (method === 'thread/loaded/list') { entered(); await gate; } return original(method, params); }; + const first = send(rpc, home); await started; + try { await assert.rejects(send(rpc, home), Refused); } finally { release(); } + await first; assert.equal(effects(rpc).length, 1); +}); + +type Handler = (request: ObjectValue, ws: WebSocket, base: ObjectValue) => void; +async function owner(t: TestContext, change?: Handler) { + const f = fixture(t, false), socket = join(f.root, 'owner.sock'); + const http = createServer(), wss = new WebSocketServer({ server: http, perMessageDeflate: false }); + const requests: ObjectValue[] = [], sockets = new Set(); + wss.on('connection', ws => { + sockets.add(ws); ws.on('close', () => sockets.delete(ws)); + ws.on('message', async bytes => { + const request = JSON.parse(bytes.toString()) as ObjectValue; + requests.push(request); if (!('id' in request)) return; + const model = new Fake(f.home); + const result = request.method === 'initialize' + ? { userAgent: 'codex-tui/0.159.2 fixture', codexHome: f.home } + : await model.call(request.method as string, request.params as ObjectValue); + const base = { id: request.id, result }; + if (change) change(request, ws, base); else ws.send(JSON.stringify(base)); + }); + }); + // Register before listening so a failed setup also cleans its own resources. + t.after(async () => { + try { + for (const ws of sockets) ws.terminate(); + await new Promise(done => wss.close(() => done())); + if (http.listening) await new Promise((done, reject) => http.close(error => error ? reject(error) : done())); + } finally { rmSync(f.root, { recursive: true, force: true }); } + }); + await new Promise((done, reject) => { http.once('error', reject); http.listen(socket, done); }); + chmodSync(socket, 0o600); + return { ...f, socket, requests }; +} + +test('real Unix WebSocket transport sends one exact steering request and refuses unsafe methods', async t => { + const f = await owner(t), rpc = await Rpc.connect(f.socket, 1000); t.after(() => rpc.close()); + try { + await assert.rejects(rpc.call('turn/start', {}), Refused); + const result = await send(rpc, f.home); + assert.equal(result.submitted, true); assert.equal(result.consumptionConfirmed, false); + const effect = f.requests.filter(r => r.method === 'turn/steer'); assert.equal(effect.length, 1); + assert.equal((effect[0]!.params as ObjectValue).expectedTurnId, TURN); + assert.equal(f.requests.some(r => r.method === 'turn/start'), false); + } finally { rpc.close(); } +}); + +test('private owned socket checks refuse permissive directories/socket and symlink entries', async t => { + const f = await owner(t); + chmodSync(f.socket, 0o666); await assert.rejects(Rpc.connect(f.socket, 1000), Refused); + chmodSync(f.socket, 0o600); chmodSync(f.root, 0o755); await assert.rejects(Rpc.connect(f.socket, 1000), Refused); + chmodSync(f.root, 0o700); const alias = join(f.root, 'alias.sock'); symlinkSync(f.socket, alias); + await assert.rejects(Rpc.connect(alias, 1000), Refused); +}); + +test('wrong server version fails during initialization with zero steering', async t => { + const f = await owner(t, (request, ws, base) => { + if (request.method === 'initialize') base.result = { userAgent: 'codex-tui/0.999.0 fixture', codexHome: 'private-home' }; + ws.send(JSON.stringify(base)); + }); + await assert.rejects(Rpc.connect(f.socket, 1000), Refused); + assert.equal(f.requests.filter(r => r.method === 'turn/steer').length, 0); +}); + +test('wire schema failure, binary/invalid UTF8, response loss and server interaction after attempt stay unknown', async t => { + const changes: Handler[] = [ + (r, ws, base) => ws.send(JSON.stringify({ ...base, error: { code: '-32600', message: 'private' } })), + (r, ws, base) => ws.send(JSON.stringify({ ...base, error: { code: -32600 }, result: base.result })), + (r, ws) => ws.send(Buffer.from([0xc3, 0x28]), { binary: false }), + (r, ws) => ws.send(Buffer.from('binary fixture')), + (r, ws) => ws.send(JSON.stringify({ id: 99, method: 'item/commandExecution/requestApproval', params: {} })), + (r, ws) => ws.terminate(), + (r, ws) => ws.send(JSON.stringify({ id: r.id, result: { turnId: 'wrong-turn' } })), + ]; + for (let i = 0; i < changes.length; i++) await t.test('case ' + i, async st => { + const f = await owner(st, (r, ws, base) => r.method === 'turn/steer' ? changes[i]!(r, ws, base) : ws.send(JSON.stringify(base))); + const rpc = await Rpc.connect(f.socket, 1000); + try { await assert.rejects(send(rpc, f.home), Unknown); } + finally { rpc.close(); } + assert.equal(f.requests.filter(r => r.method === 'turn/steer').length, 1); + assert.equal(f.requests.some(r => ['turn/start', 'turn/interrupt', 'thread/queue/add'].includes(r.method as string)), false); + }); +}); + +test('total deadline expires on unanswered initialization and cleans the connection', async t => { + const f = await owner(t, () => {}), start = performance.now(); + await assert.rejects(Rpc.connect(f.socket, 100), Refused); + assert.ok(performance.now() - start < 2000); + assert.equal(f.requests.filter(r => r.method === 'turn/steer').length, 0); +}); + +test('a cyclic inventory cursor refuses before steering', async t => { + const { home } = fixture(t), rpc = new Fake(home); + rpc.call = async (method, params) => { + rpc.calls.push({ method, params }); + assert.equal(method, 'thread/loaded/list'); + return { data: [THREAD], nextCursor: 'same-cursor' }; + }; + await assert.rejects(send(rpc, home), /invalid_cursor/); + assert.equal(rpc.calls.length, 2); assert.equal(effects(rpc).length, 0); +}); + +test('post-attempt deadline expires without a second submission', async t => { + const f = await owner(t, (r, ws, base) => { + if (r.method !== 'turn/steer') ws.send(JSON.stringify(base)); + }); + const start = performance.now(), rpc = await Rpc.connect(f.socket, 200); + try { + await assert.rejects(send(rpc, f.home), Unknown); + await assert.rejects(send(rpc, f.home), Refused); + } finally { rpc.close(); } + assert.ok(performance.now() - start < 2000); + assert.equal(f.requests.filter(r => r.method === 'turn/steer').length, 1); +}); + +test('notification flood and oversized frame fail closed before submission', async t => { + for (const mode of ['notifications', 'oversize']) await t.test(mode, async st => { + const f = await owner(st, (_r, ws) => { + if (mode === 'oversize') ws.send(' '.repeat(1_048_577)); + else for (let i = 0; i < 101; i++) ws.send(JSON.stringify({ method: 'fixture/event', params: {} })); + }); + await assert.rejects(Rpc.connect(f.socket, 1000), Refused); + assert.equal(f.requests.filter(r => r.method === 'turn/steer').length, 0); + }); +});