From 97df10011a297292fe7202ba9ad4491270b86991 Mon Sep 17 00:00:00 2001 From: ework-agent Date: Fri, 11 Sep 2026 23:59:53 +0800 Subject: [PATCH] test: add deterministic fake-upstream real-codex compression E2E MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delivers the "real regression" asked for in #686: a test where compression ACTUALLY happens inside real codex, driven by a deterministic fake Responses-API upstream — no model, no tokens, no external secrets. - tests/e2e/fake-upstream.mjs: standalone fake serving the two contracts observed against codex-cli 0.147.0 (stream:true -> Responses SSE assistant turn; stream:false -> plain JSON that bili's summarization call parses via extractSummaryText). Recognises bili summarization calls by their exact TASK text and replies with a faithful summary that keeps sentinel numbers and drops bulk filler. Logs every /v1/responses request to FAKE_REQLOG as the oracle. - tests/e2e/e2e-codex-fake.test.ts: ACP_TEST_E2E_FAKE=1 gate. overflow scenario (window 12k) accumulates ~10KB filler per turn via `resume --last`; asserts real compression fires repeatedly, the forwarded payload stays bounded despite heavy injection, sentinels survive, and the summary block is present. control scenario (window 60k) asserts NO compression (no false positives). - package.json: npm run test:e2e:fake - .github/workflows/ci-e2e-fake.yml: pull_request + workflow_dispatch gate on ubuntu-latest, installs @openai/codex, zero tokens/secrets. Validated locally against codex-cli 0.147.0: typecheck clean, build ok, 2/2 pass. Refs #686 --- .github/workflows/ci-e2e-fake.yml | 47 ++++++ package.json | 3 +- tests/e2e/e2e-codex-fake.test.ts | 228 ++++++++++++++++++++++++++++++ tests/e2e/fake-upstream.mjs | 133 +++++++++++++++++ 4 files changed, 410 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/ci-e2e-fake.yml create mode 100644 tests/e2e/e2e-codex-fake.test.ts create mode 100644 tests/e2e/fake-upstream.mjs diff --git a/.github/workflows/ci-e2e-fake.yml b/.github/workflows/ci-e2e-fake.yml new file mode 100644 index 00000000..7fb668d1 --- /dev/null +++ b/.github/workflows/ci-e2e-fake.yml @@ -0,0 +1,47 @@ +name: e2e-fake-codex + +on: + pull_request: + workflow_dispatch: + +concurrency: + group: e2e-fake-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + e2e-fake: + name: real codex vs deterministic fake upstream (zero tokens) + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-node@v4 + with: + node-version: 22 + cache: npm + + - name: Install codex CLI + run: npm i -g @openai/codex@latest + + - name: Install deps + run: npm ci + + - name: Build + run: npm run build + + - name: Run deterministic real-codex compression E2E + run: ACP_TEST_E2E_FAKE=1 node --import tsx --test tests/e2e/e2e-codex-fake.test.ts + + - name: Collect E2E artifacts + if: failure() + run: | + mkdir -p tmp/e2e-artifacts + for d in tmp/e2e-codex-fake-*; do [ -d "$d" ] && cp -r "$d" tmp/e2e-artifacts/ || true; done + + - uses: actions/upload-artifact@v4 + if: failure() + with: + name: e2e-fake-artifacts + path: tmp/e2e-artifacts + if-no-files-found: ignore diff --git a/package.json b/package.json index 56c7c8c2..281ce9cb 100644 --- a/package.json +++ b/package.json @@ -15,7 +15,8 @@ "typecheck": "tsc --noEmit --project tsconfig.build.json", "test": "node --import tsx --test tests/*.test.ts", "start": "node dist/index.js", - "test:e2e": "node --import tsx --test tests/e2e/e2e-codex.test.ts" + "test:e2e": "node --import tsx --test tests/e2e/e2e-codex.test.ts", + "test:e2e:fake": "node --import tsx --test tests/e2e/e2e-codex-fake.test.ts" }, "keywords": [ "context", diff --git a/tests/e2e/e2e-codex-fake.test.ts b/tests/e2e/e2e-codex-fake.test.ts new file mode 100644 index 00000000..b2856e9b --- /dev/null +++ b/tests/e2e/e2e-codex-fake.test.ts @@ -0,0 +1,228 @@ +import { spawn, spawnSync } from "node:child_process"; +import fs from "node:fs"; +import net from "node:net"; +import path from "node:path"; +import { strict as assert } from "node:assert"; +import { test } from "node:test"; + +const CODEX_BIN = process.env.E2E_CODEX_BIN ?? "codex"; +const DIST = process.env.E2E_BILI_DIST ?? path.resolve(import.meta.dirname, "../../dist/index.js"); +const MODEL = process.env.E2E_MODEL ?? "qwen3.8-27b"; +const TMO = Number(process.env.E2E_TMO ?? 120_000); +const FAKE_UPSTREAM = path.join(import.meta.dirname, "fake-upstream.mjs"); +const WORK_ROOT = path.join(process.cwd(), "tmp"); +fs.mkdirSync(WORK_ROOT, { recursive: true }); + +function codexAvailable(): boolean { + try { return spawnSync(CODEX_BIN, ["--version"], { timeout: 15_000 }).status === 0; } catch { return false; } +} +const run = process.env.ACP_TEST_E2E_FAKE === "1"; +const skipReason = !run + ? "set ACP_TEST_E2E_FAKE=1 (real codex + local fake upstream; deterministic, zero tokens)" + : (!codexAvailable() ? `codex binary "${CODEX_BIN}" not found on PATH` : undefined); + +/** Deterministic filler: unique per index, bulky, carried verbatim in the prompt. */ +function filler(i: number, lines: number): string { + const out: string[] = []; + for (let n = 0; n < lines; n += 1) { + out.push(`doc#${String(i).padStart(2, "0")} line${String(n).padStart(4, "0")} checksum ${(n * 7919 + i * 104729) % 999983}`); + } + return out.join("\n"); +} + +type OracleEntry = { t: number; model: string; stream: boolean; isSummary: boolean; inputLen: number; input: unknown[] }; + +function flatContent(c: unknown): string { + if (typeof c === "string") return c; + if (Array.isArray(c)) return c.map((p) => (p && typeof p === "object" && "text" in p ? String((p as { text: unknown }).text) : "")).join(""); + return String(c ?? ""); +} +function allInputText(input: unknown[]): string { + return (input || []).map((it) => flatContent((it as { content?: unknown }).content)).join("\n"); +} +function readOracle(reqLog: string): OracleEntry[] { + if (!fs.existsSync(reqLog)) return []; + return fs.readFileSync(reqLog, "utf8").split("\n").filter((l) => l.trim()).map((l) => JSON.parse(l) as OracleEntry); +} + +function freePort(): Promise { + return new Promise((resolve) => { + const s = net.createServer(); + s.listen(0, "127.0.0.1", () => { + const p = (s.address() as net.AddressInfo).port; + s.close(() => resolve(p)); + }); + }); +} + +function windowEnv(contextWindow: number): Record { + return { BILI_LAUNCHER_MODEL_WINDOWS: JSON.stringify({ [MODEL]: contextWindow }) }; +} + +type Ctx = { + work: string; + codexHome: string; + xdg: { config: string; cache: string; state: string }; + port: number; + fakePort: number; + reqLog: string; + fakePid?: number; + proxyPid?: number; + resumed: boolean; + turnCount: number; +}; + +async function startCtx(contextWindow: number): Promise { + const work = fs.mkdtempSync(path.join(WORK_ROOT, "e2e-codex-fake-")); + const ctx: Ctx = { + work, + codexHome: path.join(work, "codex-home"), + xdg: { config: path.join(work, "xdg-config"), cache: path.join(work, "xdg-cache"), state: path.join(work, "xdg-state") }, + port: await freePort(), + fakePort: await freePort(), + reqLog: path.join(work, "fake-requests.jsonl"), + resumed: false, + turnCount: 0, + }; + for (const d of [ctx.codexHome, ctx.xdg.config, ctx.xdg.cache, ctx.xdg.state]) fs.mkdirSync(d, { recursive: true }); + + const fake = spawn(process.execPath, [FAKE_UPSTREAM], { + env: { ...process.env, FAKE_PORT: String(ctx.fakePort), FAKE_HOST: "127.0.0.1", FAKE_REQLOG: ctx.reqLog, FAKE_MODEL: MODEL }, + stdio: ["ignore", "pipe", "pipe"], + }); + ctx.fakePid = fake.pid; + await waitFor(`http://127.0.0.1:${ctx.fakePort}/v1/models`, 15_000); + + fs.writeFileSync(path.join(ctx.codexHome, "config.toml"), [ + `model = "${MODEL}"`, + 'model_provider = "e2e"', + "model_context_window = 60000", + "", + "[model_providers.e2e]", + 'name = "OpenAI"', + `base_url = "http://127.0.0.1:${ctx.port}/bili/http://127.0.0.1:${ctx.fakePort}/v1"`, + 'wire_api = "responses"', + 'env_key = "E2E_UPSTREAM_KEY"', + "", + ].join("\n")); + + const logPath = path.join(work, "bili.log"); + const proxy = spawn(process.execPath, [DIST, "start", "--port", String(ctx.port), "--no-auto-update"], { + env: { + ...process.env, + XDG_CONFIG_HOME: ctx.xdg.config, + XDG_CACHE_HOME: ctx.xdg.cache, + XDG_STATE_HOME: ctx.xdg.state, + BILLION_CONTEXT_NO_AUTO_UPDATE: "1", + ...windowEnv(contextWindow), + }, + stdio: ["ignore", "ignore", "pipe"], + }); + ctx.proxyPid = proxy.pid; + proxy.stderr!.on("data", (c: Buffer) => { try { fs.appendFileSync(logPath, c); } catch { /* noop */ } }); + await waitFor(`http://127.0.0.1:${ctx.port}/__bili/health`, 30_000, "bili proxy"); + return ctx; +} + +function waitFor(url: string, ms: number, label = "service"): Promise { + return new Promise((resolve, reject) => { + const started = Date.now(); + const poll = (): void => { + fetch(url).then((r) => (r.ok ? resolve() : retry())).catch(retry); + }; + const retry = (): void => { + if (Date.now() - started > ms) { reject(new Error(`${label} did not come up within ${ms}ms`)); return; } + setTimeout(poll, 250); + }; + poll(); + }); +} + +function teardown(ctx: Ctx): void { + for (const pid of [ctx.proxyPid, ctx.fakePid]) { + if (pid) { try { process.kill(pid, "SIGKILL"); } catch { /* already gone */ } } + } +} + +function logs(ctx: Ctx): string { + const parts: string[] = []; + const stateLog = path.join(ctx.xdg.state, "billion-context", "bili.log"); + if (fs.existsSync(stateLog)) parts.push(fs.readFileSync(stateLog, "utf8")); + try { parts.push(fs.readFileSync(path.join(ctx.work, "bili.log"), "utf8")); } catch { /* noop */ } + return parts.join(""); +} + +function turn(ctx: Ctx, prompt: string): Promise<{ code: number; last: string }> { + ctx.turnCount += 1; + const label = `t${ctx.turnCount}`; + const lastFile = path.join(ctx.work, `${label}.last`); + const args = ["exec", "--skip-git-repo-check", "--output-last-message", lastFile]; + if (ctx.resumed) args.push("resume", "--last"); + args.push(prompt); + return new Promise((resolve, reject) => { + const child = spawn(CODEX_BIN, args, { + cwd: ctx.work, + env: { ...process.env, CODEX_HOME: ctx.codexHome, E2E_UPSTREAM_KEY: "fake", RUST_LOG: "error" }, + stdio: ["ignore", "ignore", "pipe"], + }); + const timer = setTimeout(() => { + try { child.kill("SIGKILL"); } catch { /* noop */ } + reject(new Error(`turn ${label} timed out after ${TMO}ms`)); + }, TMO); + child.on("exit", (code) => { + clearTimeout(timer); + const last = fs.existsSync(lastFile) ? fs.readFileSync(lastFile, "utf8").trim() : ""; + ctx.resumed = true; + resolve({ code: code ?? -1, last }); + }); + }); +} + +test("overflow: compression really happens in codex; bulk folded, sentinels retained", { skip: skipReason }, async (t) => { + const ctx = await startCtx(12_000); + t.after(() => teardown(ctx)); + + const planted = [4781, 2903, 6577]; + const warm = await turn(ctx, `档案摘要:\n${planted.map((s) => `本档案哨兵值 = ${s}`).join("\n")}\n\n请确认收到, 只回复: 收到#1`); + assert.equal(warm.code, 0, `warmup failed (code=${warm.code}); log:\n${logs(ctx)}`); + + for (let k = 2; k <= 5; k += 1) { + const r = await turn(ctx, `${filler(k, 300)}\n\n请确认已读取档案#k, 只回复: 收到#${k}`); + assert.equal(r.code, 0, `load turn ${k} failed (code=${r.code}); log:\n${logs(ctx)}`); + } + + const log = logs(ctx); + assert.match(log, /preflight compressed|compress requested|\[Compressed m\d/, "a real compression event must occur once context exceeds the window"); + + const oracle = readOracle(ctx.reqLog); + assert.ok(oracle.some((o) => o.isSummary), "summarization must call the upstream at least once"); + const summaries = oracle.filter((o) => o.isSummary); + assert.ok(summaries.length >= 2, `expected repeated summarization as context accumulated (got ${summaries.length})`); + const mains = oracle.filter((o) => !o.isSummary); + assert.ok(mains.length >= 2, "expected several forwarded requests"); + const lens = mains.map((m) => m.inputLen); + const minLen = Math.min(...lens); + const peak = Math.max(...lens); + assert.ok(peak <= minLen * 1.3, `despite ~10KB filler injected on every load turn the forwarded payload must stay bounded (min=${minLen}, peak=${peak}); unbounded growth would mean compression is not folding the bulk`); + + const lastMain = allInputText(mains[mains.length - 1].input); + assert.match(lastMain, /\[Compressed conversation section\]/, "final payload must carry the summary block"); + for (const s of planted) { + assert.ok(lastMain.includes(String(s)), `sentinel ${s} must survive compression into the final payload`); + } +}); + +test("under-window: no compression occurs (control)", { skip: skipReason }, async (t) => { + const ctx = await startCtx(60_000); + t.after(() => teardown(ctx)); + + const r = await turn(ctx, "请只回复: 收到"); + assert.equal(r.code, 0, `codex exec should succeed (got ${r.code})\nbili log:\n${logs(ctx)}`); + + const log = logs(ctx); + assert.doesNotMatch(log, /preflight compressed|compress requested/, "control turn must NOT trigger compression"); + + const oracle = readOracle(ctx.reqLog); + assert.ok(oracle.length > 0, "fake upstream received no requests"); + assert.ok(oracle.every((o) => !o.isSummary), "control turn must make no summarization calls"); +}); diff --git a/tests/e2e/fake-upstream.mjs b/tests/e2e/fake-upstream.mjs new file mode 100644 index 00000000..3b3c9d21 --- /dev/null +++ b/tests/e2e/fake-upstream.mjs @@ -0,0 +1,133 @@ +// Deterministic fake Responses-API upstream for the real-codex E2E (#686): drives +// REAL `codex` through bili so compression ACTUALLY happens in-process, no model/network. +// Two contracts observed against codex-cli 0.147.0: +// stream:true -> Responses SSE assistant turn; stream:false -> plain JSON body, which +// bili's summarization call parses via extractSummaryText -> json.output[].content[].text. +// A summarization request is recognised by bili's exact TASK text in `instructions`; its +// reply is a faithful summary preserving the sentinel numbers found in the segment. +// Every /v1/responses request is appended to FAKE_REQLOG (JSONL) as the assertion oracle. + +import http from "node:http"; +import fs from "node:fs"; +import path from "node:path"; + +const PORT = Number(process.env.FAKE_PORT || 8199); +const HOST = process.env.FAKE_HOST || "127.0.0.1"; +const REQLOG = process.env.FAKE_REQLOG || path.join(process.cwd(), "tmp", "fake-requests.jsonl"); +const MODEL = process.env.FAKE_MODEL || "qwen3.8-27b"; +try { fs.mkdirSync(path.dirname(REQLOG), { recursive: true }); } catch { /* noop */ } + +let n = 0; +const uid = () => `resp_${(++n).toString(16).padStart(6, "0")}`; +const estIn = (s) => Math.max(1, Math.round(s.length / 4)); +const estOut = (s) => Math.max(1, Math.round(s.length / 4)); + +function flatContent(c) { + if (typeof c === "string") return c; + if (Array.isArray(c)) return c.map((p) => (p && p.text) || "").join(""); + return String(c ?? ""); +} +function allInputText(input) { + return (input || []).map((it) => flatContent(it.content)).join("\n"); +} +function lastUserText(input) { + return (input || []) + .filter((it) => it.role === "user") + .map((it) => flatContent(it.content)) + .join("\n"); +} + +// Chat turn: echo an ack tag so the test can confirm each round-trip landed. +function answerFor(userText) { + const m = userText.match(/收到#(\d+)/); + return m ? `收到#${m[1]}` : "收到"; +} + +// Faithful summarizer: keep the salient numbers (哨兵值 sentinels) from the +// segment being compressed, drop the bulk filler lines. >= MIN_SUMMARY_CHARS. +function buildFakeSummary(content) { + const sents = []; + const re = /哨兵值\s*[=:]\s*(\d+)/g; + let m; + while ((m = re.exec(content))) sents.push(m[1]); + const uniq = [...new Set(sents)]; + const docs = [...new Set(content.match(/doc#\d+/g) || [])]; + let text = + `[Compressed conversation section] — folded ${docs.length || "?"} archive fragment(s)` + + ` under window overflow. Preserved sentinel values: ${uniq.length ? uniq.join(", ") : "(none)"}.` + + ` Verbatim filler lines omitted to save tokens.`; + while (text.length < 50) text += " (context preserved)"; + return text; +} + +// Only bili's summarization calls carry this exact task text in instructions. +function isSummaryRequest(parsed) { + const ins = typeof parsed.instructions === "string" ? parsed.instructions : ""; + return /must be compressed because the session context exceeds|Write a \*?\*?(tier-1|\d+-tier) compression summary/.test(ins); +} + +function messageEvents(text) { + const id = uid(); + const msgId = `msg_${id}`; + const msg = { type: "message", id: msgId, role: "assistant", content: [{ type: "output_text", text }] }; + const inTok = 1, outTok = estOut(text); + return [ + ["response.created", { type: "response.created", response: { id, status: "in_progress", output: [] } }], + ["response.output_item.added", { type: "response.output_item.added", output_index: 0, item: { type: "message", id: msgId, role: "assistant", content: [] } }], + ["response.content_part.added", { type: "response.content_part.added", item_id: msgId, output_index: 0, content_index: 0, part: { type: "output_text", text: "" } }], + ["response.output_text.delta", { type: "response.output_text.delta", item_id: msgId, output_index: 0, content_index: 0, delta: text }], + ["response.output_text.done", { type: "response.output_text.done", item_id: msgId, output_index: 0, content_index: 0, text }], + ["response.content_part.done", { type: "response.content_part.done", item_id: msgId, output_index: 0, content_index: 0, part: { type: "output_text", text } }], + ["response.output_item.done", { type: "response.output_item.done", output_index: 0, item: msg }], + ["response.completed", { type: "response.completed", response: { id, status: "completed", output: [msg], usage: { input_tokens: inTok, output_tokens: outTok, total_tokens: inTok + outTok } } }], + ]; +} +function sse(res, events) { + res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache", connection: "keep-alive" }); + for (const [ev, data] of events) res.write(`event: ${ev}\ndata: ${JSON.stringify(data)}\n\n`); + res.write("data: [DONE]\n\n"); + res.end(); +} +function jsonResponse(res, text, inTok) { + const id = uid(); + const msg = { type: "message", id: `msg_${id}`, role: "assistant", content: [{ type: "output_text", text }] }; + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ + id, object: "response", status: "completed", model: MODEL, output: [msg], + usage: { input_tokens: inTok, output_tokens: estOut(text), total_tokens: inTok + estOut(text) }, + })); +} + +const server = http.createServer((req, res) => { + try { + if (req.method === "GET" && /\/models$/.test(req.url)) { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ object: "list", data: [{ id: MODEL, object: "model" }] })); + return; + } + if (req.method === "POST" && /\/responses$/.test(req.url)) { + let raw = ""; + req.on("data", (c) => (raw += c)); + req.on("end", () => { + let parsed = {}; + try { parsed = JSON.parse(raw || "{}"); } catch { /* noop */ } + const summary = isSummaryRequest(parsed); + let text; + if (summary) text = buildFakeSummary(allInputText(parsed.input)); + else text = answerFor(lastUserText(parsed.input)); + try { + fs.appendFileSync(REQLOG, JSON.stringify({ t: Date.now(), model: parsed.model, stream: !!parsed.stream, isSummary: summary, inputLen: raw.length, input: parsed.input }) + "\n"); + } catch { /* noop */ } + const inTok = estIn(raw); + if (parsed.stream) sse(res, messageEvents(text)); + else jsonResponse(res, text, inTok); + }); + return; + } + res.writeHead(404, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: { message: "not found" } })); + } catch (e) { + try { res.writeHead(500, { "content-type": "application/json" }); res.end(JSON.stringify({ error: { message: String(e) } })); } catch { /* noop */ } + } +}); +server.listen(PORT, HOST, () => console.log(`fake upstream listening on http://${HOST}:${PORT}/v1`));