my-pi/pi-compaction/test/extension-runtime.test.ts

import test from "node:test";
import assert from "node:assert/strict";
import { performance } from "node:perf_hooks";
import piCompaction from "../index.ts";

function harness() {
  const handlers = new Map<string, Function[]>();
  const tools = new Map<string, any>();
  const commands = new Map<string, any>();
  const entries: Array<{ customType: string; data: unknown }> = [];
  const toolDefinitions = [{ name: "bash", description: "shell", parameters: { type: "object" } }];
  const pi: any = {
    on(name: string, handler: Function) {
      const list = handlers.get(name) ?? [];
      list.push(handler);
      handlers.set(name, list);
    },
    appendEntry(customType: string, data: unknown) { entries.push({ customType, data }); },
    registerTool(tool: any) { tools.set(tool.name, tool); },
    registerCommand(name: string, command: any) { commands.set(name, command); },
    getActiveTools() { return ["bash"]; },
    getAllTools() { return toolDefinitions; },
  };
  return { pi, handlers, tools, commands, entries };
}

function context(branch: any[] = []) {
  return {
    sessionManager: {
      getSessionId: () => "session-test",
      getBranch: () => branch,
    },
    model: { provider: "test", api: "test", id: "model", contextWindow: 100_000 },
    modelRegistry: {},
    getSystemPrompt: () => "system",
    getContextUsage: () => ({ tokens: 1_000, contextWindow: 100_000, percent: 1 }),
    hasUI: false,
    abort() {},
  } as any;
}

test("one runtime projects tool evidence and recovers it through context_tree_query", async () => {
  const h = harness();
  piCompaction(h.pi);
  const ctx = context();
  await h.handlers.get("session_start")![0]!({}, ctx);

  const assistant = {
    role: "assistant",
    content: [{ type: "toolCall", id: "call-1", name: "bash", arguments: { command: "npm test" } }],
    timestamp: 10,
  };
  const result = {
    role: "toolResult",
    toolCallId: "call-1",
    toolName: "bash",
    content: [{ type: "text", text: `failure evidence\n${"x".repeat(12_000)}\nstack tail` }],
    isError: true,
    timestamp: 11,
  };
  await h.handlers.get("turn_end")![0]!({ message: assistant, toolResults: [result] }, ctx);
  const projected = await h.handlers.get("context")![0]!({ messages: [assistant, result] }, ctx);

  assert.match(projected.messages[1].content[0].text, /pi-compaction tool=bash/);
  const cacheStart = performance.now();
  for (let iteration = 0; iteration < 1_000; iteration++) {
    await h.handlers.get("context")![0]!({ messages: [assistant, result] }, ctx);
  }
  const cacheAverageMs = (performance.now() - cacheStart) / 1_000;
  assert.ok(cacheAverageMs < 2, `full context handler cache-hit average ${cacheAverageMs.toFixed(3)}ms exceeded 2ms`);
  assert.equal(h.entries.length, 1);
  assert.equal(h.entries[0]!.customType, "pi-compaction");

  const query = h.tools.get("context_tree_query");
  const recovered = await query.execute("q", { toolCallIds: ["call-1"] }, new AbortController().signal, undefined, ctx);
  assert.match(recovered.content[0].text, /failure evidence/);

  await h.handlers.get("session_shutdown")![0]!({}, ctx);
});

test("small tool results are passed through without a stub wrapper", async () => {
  const h = harness();
  piCompaction(h.pi);
  const ctx = context();
  await h.handlers.get("session_start")![0]!({}, ctx);

  const assistant = {
    role: "assistant",
    content: [{ type: "toolCall", id: "call-small", name: "bash", arguments: { command: "ls" } }],
    timestamp: 20,
  };
  const result = {
    role: "toolResult",
    toolCallId: "call-small",
    toolName: "bash",
    content: [{ type: "text", text: "ok\nfile-a.ts\nfile-b.ts" }],
    timestamp: 21,
  };

  await h.handlers.get("turn_end")![0]!({ message: assistant, toolResults: [result] }, ctx);
  const projected = await h.handlers.get("context")![0]!({ messages: [assistant, result] }, ctx);

  assert.strictEqual(projected.messages[1].content[0].text, "ok\nfile-a.ts\nfile-b.ts");
  assert.doesNotMatch(projected.messages[1].content[0].text, /pi-compaction tool=/);

  // Original evidence stays recoverable even when the message is passed through.
  const query = h.tools.get("context_tree_query")!;
  const recovered = await query.execute("q", { toolCallIds: ["call-small"] }, new AbortController().signal, undefined, ctx);
  assert.match(recovered.content[0].text, /file-a\.ts/);

  await h.handlers.get("session_shutdown")![0]!({}, ctx);
});

test("model switching after a native checkpoint projects the portable checkpoint", async () => {
  const h = harness();
  piCompaction(h.pi);
  const details = {
    kind: "pi-compaction",
    version: 1,
    checkpoint: "native",
    modelKey: "openai-codex/openai-codex-responses/gpt-old",
    replacementHistory: [{ type: "compaction", encrypted_content: "opaque" }],
    portableSummary: "decisions and files from the old model",
    sourceFingerprint: "old",
  };
  const branch = [
    { type: "compaction", id: "c1", parentId: null, timestamp: new Date().toISOString(), summary: "marker", firstKeptEntryId: "m0", tokensBefore: 1_000, details },
    { type: "message", id: "m1", parentId: "c1", timestamp: new Date().toISOString(), message: { role: "user", content: "continue", timestamp: 20 } },
  ] as any[];
  const ctx = context(branch);
  await h.handlers.get("session_start")![0]!({}, ctx);
  const projected = await h.handlers.get("context")![0]!({ messages: [] }, ctx);
  assert.match(projected.messages[0].content[0].text, /decisions and files/);
  assert.equal(projected.messages[1].content, "continue");
  await h.handlers.get("session_shutdown")![0]!({}, ctx);
});

for (const provider of ["sub-api", "openai-codex"]) {
  test(`${provider} native compaction uses the selected model and restores its checkpoint`, async (t) => {
    const h = harness();
    piCompaction(h.pi);
    const branch: any[] = [{
      type: "message", id: "u1", parentId: null,
      message: { role: "user", content: "Remember the test decision.", timestamp: 1 },
    }];
    const ctx = context(branch);
    ctx.model = {
      provider, api: provider === "sub-api" ? "openai-responses" : "openai-codex-responses",
      id: "selected-model", input: ["text"], contextWindow: 100_000,
      baseUrl: provider === "sub-api" ? "https://gateway.example/v1" : "https://codex.example/codex",
    };
    const jwt = Buffer.from(JSON.stringify({
      "https://api.openai.com/auth": { chatgpt_account_id: "test-account" },
    })).toString("base64url");
    const apiKey = provider === "sub-api" ? "ordinary-test-key" : `header.${jwt}.signature`;
    ctx.modelRegistry.getApiKeyAndHeaders = async (model: unknown) => {
      assert.equal(model, ctx.model);
      return { ok: true, apiKey, headers: { "x-test-route": "preserved" } };
    };
    const requests: Array<{ url: string; headers: Headers; body: any }> = [];
    t.mock.method(globalThis, "fetch", async (url: string, init: RequestInit) => {
      requests.push({ url: String(url), headers: new Headers(init.headers), body: JSON.parse(String(init.body)) });
      const events = [
        { type: "response.output_item.done", item: { type: "compaction", encrypted_content: "opaque-checkpoint" } },
        { type: "response.completed", response: { usage: { input_tokens: 12, output_tokens: 3 } } },
      ];
      return new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), {
        headers: { "content-type": "text/event-stream" },
      });
    });
    try {
      await h.handlers.get("context")![0]!({ messages: branch.map((entry) => entry.message) }, ctx);
      const headers: Record<string, string> = { "x-codex-beta-features": "existing-feature" };
      await h.handlers.get("before_provider_headers")![0]!({ headers }, ctx);
      assert.match(headers["x-codex-beta-features"]!, /existing-feature/);
      assert.notEqual(headers["x-codex-beta-features"], "existing-feature");
      await h.handlers.get("before_provider_request")![0]!({ payload: {
        model: "stale-cached-model", input: [{ role: "user", content: "STALE-RAW-HISTORY" }],
        store: true, stream: false,
      } }, ctx);
      const result = await h.handlers.get("session_before_compact")![0]!({
        reason: "manual", branchEntries: branch, signal: new AbortController().signal,
        preparation: { firstKeptEntryId: "u1" },
      }, ctx);
      assert.ok(result?.compaction, "native compaction should return a checkpoint");
      assert.equal(requests.length, 1);
      const request = requests[0]!;
      assert.equal(request.url, provider === "sub-api"
        ? "https://gateway.example/v1/responses" : "https://codex.example/codex/responses");
      assert.equal(request.headers.get("authorization"), `Bearer ${apiKey}`);
      assert.equal(request.headers.get("x-test-route"), "preserved");
      assert.equal(request.headers.get("chatgpt-account-id"), provider === "sub-api" ? null : "test-account");
      assert.equal(request.body.model, "selected-model");
      assert.equal(request.body.stream, true);
      assert.equal(request.body.store, false);
      assert.deepEqual(request.body.input.at(-1), { type: "compaction_trigger" });
      assert.match(JSON.stringify(request.body.input), /Remember the test decision/);
      assert.doesNotMatch(JSON.stringify(request.body.input), /STALE-RAW-HISTORY/);
      const details = result.compaction.details;
      assert.equal(details.checkpoint, "native");
      assert.equal(details.modelKey, `${provider}:${ctx.model.api}:selected-model`);
      assert.ok(details.replacementHistory.some((item: any) => item.encrypted_content === "opaque-checkpoint"));
      assert.ok(details.portableSummary.length > 0);
      branch.push(
        { type: "compaction", id: "c1", parentId: "u1", ...result.compaction },
        { type: "message", id: "u2", parentId: "c1", message: { role: "user", content: "Continue after compaction.", timestamp: 2 } },
      );
      const projected = await h.handlers.get("context")![0]!({ messages: [] }, ctx);
      assert.doesNotMatch(JSON.stringify(projected.messages), /opaque-checkpoint/);
      const payload = await h.handlers.get("before_provider_request")![0]!({
        payload: { model: ctx.model.id, input: [], messages: [], previous_response_id: "stale-response" },
      }, ctx);
      assert.ok(payload.input.some((item: any) => item.encrypted_content === "opaque-checkpoint"));
      assert.match(JSON.stringify(payload.input), /Continue after compaction/);
      assert.equal(payload.previous_response_id, undefined);
      assert.equal(payload.messages, undefined);
      ctx.model = { ...ctx.model, id: "another-model" };
      const switched = await h.handlers.get("context")![0]!({ messages: [] }, ctx);
      assert.match(JSON.stringify(switched.messages), /Remember the test decision/);
      assert.equal(await h.handlers.get("before_provider_request")![0]!({ payload: { input: [] } }, ctx), undefined);
    } finally {
      await h.handlers.get("session_shutdown")![0]!({}, ctx);
    }
  });
}

test("SubAPI native failures and cancellation return control to Pi without a checkpoint", async (t) => {
  const h = harness();
  piCompaction(h.pi);
  const branch: any[] = [{
    type: "message", id: "u1", parentId: null,
    message: { role: "user", content: "Test only.", timestamp: 1 },
  }];
  const ctx = context(branch);
  ctx.model = { provider: "sub-api", api: "openai-responses", id: "test-model", input: ["text"], baseUrl: "https://gateway.example/v1" };
  ctx.modelRegistry.getApiKeyAndHeaders = async () => ({ ok: true, apiKey: "test-key" });
  const responses = [
    new Response('{"error":{"message":"unsupported"}}', { status: 400 }),
    new Response('data: {"type":"response.completed","response":{}}\n\n', { headers: { "content-type": "text/event-stream" } }),
  ];
  let calls = 0;
  t.mock.method(globalThis, "fetch", async (_url: string, init: RequestInit) => {
    init.signal?.throwIfAborted();
    calls++;
    return responses.shift()!;
  });
  try {
    for (let attempt = 0; attempt < 2; attempt++) {
      assert.equal(await h.handlers.get("session_before_compact")![0]!({
        reason: "manual", branchEntries: branch, preparation: { firstKeptEntryId: "u1" },
        signal: new AbortController().signal,
      }, ctx), undefined);
    }
    const controller = new AbortController();
    controller.abort();
    assert.equal(await h.handlers.get("session_before_compact")![0]!({
      reason: "manual", branchEntries: branch, preparation: { firstKeptEntryId: "u1" }, signal: controller.signal,
    }, ctx), undefined);
    assert.equal(calls, 2);
    assert.equal(branch.some((entry) => entry.type === "compaction"), false);
  } finally {
    await h.handlers.get("session_shutdown")![0]!({}, ctx);
  }
});

test("other providers and SubAPI Chat Completions keep their provider requests unchanged", async () => {
  const h = harness();
  piCompaction(h.pi);
  const ctx = context();
  try {
    for (const model of [
      { provider: "other", api: "openai-responses", id: "test" },
      { provider: "sub-api", api: "openai-completions", id: "test" },
    ]) {
      ctx.model = model;
      const headers: Record<string, string> = {};
      await h.handlers.get("before_provider_headers")![0]!({ headers }, ctx);
      assert.equal(headers["x-codex-beta-features"], undefined);
      assert.equal(await h.handlers.get("before_provider_request")![0]!({ payload: { input: [] } }, ctx), undefined);
    }
  } finally {
    await h.handlers.get("session_shutdown")![0]!({}, ctx);
  }
});

test("native runtime delegates authentication to each provider without blocking SubAPI header-only auth", async (t) => {
  const cases: Array<{ name: string; provider?: string; auth: any; succeeds: boolean; authorization?: string; gatewayAuthorization?: string }> = [
    { name: "explicit header", auth: { ok: true, apiKey: "dummy-key", headers: { Authorization: "Bearer explicit" } }, succeeds: true, authorization: "Bearer explicit" },
    { name: "header only", auth: { ok: true, headers: { authorization: "Bearer header-only" } }, succeeds: true, authorization: "Bearer header-only" },
    { name: "proxy header only", auth: { ok: true, headers: { "CF-AIG-Authorization": "Bearer proxy" } }, succeeds: true, gatewayAuthorization: "Bearer proxy" },
    { name: "blank header", auth: { ok: true, headers: { Authorization: " " } }, succeeds: false },
    { name: "missing credentials", auth: { ok: true }, succeeds: false },
    { name: "registry failure", auth: { ok: false, error: "credentials unavailable" }, succeeds: false },
    { name: "Codex header only", provider: "openai-codex", auth: { ok: true, headers: { Authorization: "Bearer header-only" } }, succeeds: false },
    { name: "Codex invalid JWT", provider: "openai-codex", auth: { ok: true, apiKey: "invalid", headers: { Authorization: "Bearer header-only" } }, succeeds: false },
  ];
  for (const scenario of cases) {
    await t.test(scenario.name, async (t) => {
      const h = harness();
      piCompaction(h.pi);
      const branch: any[] = [{
        type: "message", id: "u1", parentId: null,
        message: { role: "user", content: "Synthetic authentication test.", timestamp: 1 },
      }];
      const ctx = context(branch);
      const provider = scenario.provider ?? "sub-api";
      ctx.model = {
        provider, api: provider === "sub-api" ? "openai-responses" : "openai-codex-responses",
        id: "test-model", input: ["text"], baseUrl: "https://gateway.example/v1",
      };
      ctx.modelRegistry.getApiKeyAndHeaders = async () => scenario.auth;
      let calls = 0;
      t.mock.method(globalThis, "fetch", async (_url: string, init: RequestInit) => {
        calls++;
        const headers = new Headers(init.headers);
        assert.equal(headers.get("authorization"), scenario.authorization ?? null);
        assert.equal(headers.get("cf-aig-authorization"), scenario.gatewayAuthorization ?? null);
        const events = [
          { type: "response.output_item.done", item: { type: "compaction", encrypted_content: "test-checkpoint" } },
          { type: "response.completed", response: {} },
        ];
        return new Response(events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""), {
          headers: { "content-type": "text/event-stream" },
        });
      });
      try {
        const result = await h.handlers.get("session_before_compact")![0]!({
          reason: "manual", branchEntries: branch, preparation: { firstKeptEntryId: "u1" },
          signal: new AbortController().signal,
        }, ctx);
        assert.equal(Boolean(result?.compaction), scenario.succeeds);
        assert.equal(calls, scenario.succeeds ? 1 : 0);
      } finally {
        await h.handlers.get("session_shutdown")![0]!({}, ctx);
      }
    });
  }
});