server/acp-recovery.e2e.test.ts

Rama main · réplica de lectura del repositorio original.

import { spawn, type ChildProcess } from "node:child_process";
import { closeSync, existsSync, openSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { expect, it } from "vitest";
import { launchVerificationServer, runControlOmb, verificationServerEnvironment } from "../scripts/control-omb.ts";
import { AUTO_RETRY_LEAD } from "./incidents.ts";
import { waitForExit } from "./testing/cleanup.ts";

it.each([false, true])("recovers the same conversation after an ACP internal error without replay (output: %s)", async (afterOutput) => {
  const fixture = await launchVerificationServer();
  const { dataDir, url, logPath } = fixture.info;
  let server: ChildProcess | undefined;
  const failureFile = join(dataDir, "rpc-failure.json");
  const rpcFile = join(dataDir, "rpc.jsonl");
  const dumpFile = join(dataDir, "acp-dump.json");
  const evidence: unknown[] = [{ url, dataDir, afterOutput }];
  const api = async (method: string, path: string, body?: unknown, status = 200) => {
    const response = await fetch(url + path, {
      method, headers: { "content-type": "application/json", origin: url },
      ...(body === undefined ? {} : { body: JSON.stringify(body) }), signal: AbortSignal.timeout(30_000),
    });
    const value = await response.json() as any;
    expect(response.status, `${method} ${path}: ${JSON.stringify(value)}`).toBe(status);
    return value;
  };
  const control = async (...args: string[]) => {
    const result = await runControlOmb([...args, "--url", url]) as any;
    evidence.push({ command: args, result });
    return result;
  };
  const calls = () => existsSync(rpcFile)
    ? readFileSync(rpcFile, "utf8").trim().split("\n").filter(Boolean).map((line) => JSON.parse(line) as { pid: number; method: string }) : [];
  const prompts = () => calls().filter((call) => call.method === "session/prompt");
  try {
    // Add only the repository-owned synthetic provider to the launcher's
    // temporary config while stopped. Never discover the person's CLI/home.
    await waitForExit(fixture.child, { signal: "SIGTERM" });
    const configPath = join(dataDir, "config.json");
    const config = JSON.parse(readFileSync(configPath, "utf8"));
    config.instances.opencodeGo = {
      driver: "opencodeGo", displayName: "Recovery fixture",
      config: { cli: fileURLToPath(new URL("./testing/fake-acp-cli.ts", import.meta.url)) },
      environment: {
        FAKE_ACP_MODELS: "fixture/recovery", FAKE_ACP_DUMP: dumpFile, FAKE_ACP_DUMP_PROMPT: "1",
        FAKE_ACP_RPC_APPEND_FILE: rpcFile, FAKE_ACP_RPC_FAILURE_FILE: failureFile,
        ...(afterOutput ? { FAKE_ACP_RPC_FAILURE_AFTER_OUTPUT: "1" } : {}),
      },
    };
    writeFileSync(configPath, JSON.stringify(config));
    const log = openSync(logPath, "a", 0o600);
    server = spawn(process.execPath, ["--experimental-strip-types", fileURLToPath(new URL("./index.ts", import.meta.url))], {
      cwd: fileURLToPath(new URL("..", import.meta.url)),
      env: verificationServerEnvironment({}, dataDir, Number(new URL(url).port)), stdio: ["ignore", log, log],
    });
    closeSync(log);
    await expect.poll(async () => {
      try { return (await fetch(url + "/api/health", { signal: AbortSignal.timeout(1_000) })).ok; } catch { return false; }
    }, { timeout: 20_000 }).toBe(true);

    const { bot } = await control("new-bot", "--name", "Recovery bot");
    const threadId = bot.activeTaskId;
    await api("PATCH", `/api/bots/${bot.id}/tasks/${threadId}`, { modelSelection: { instanceId: "opencodeGo", model: "fixture/recovery" } });
    const messages = async () => (await api("GET", `/api/threads/${threadId}/messages?limit=100`)).messages as any[];
    const task = async () => (await api("GET", "/api/bots")).bots.find((entry: any) => entry.id === bot.id)
      .tasks.find((entry: any) => entry.threadId === threadId);
    const send = (text: string) => control("send", "--bot", bot.id, "--task", threadId, "--text", text);
    const wait = () => control("wait", "--bot", bot.id, "--task", threadId, "--timeout", "30");
    const firstText = "REMEMBER_RECOVERY_7Q";
    const failedText = "FAILED_WORK_8R";
    const retryText = "EXPLICIT_RETRY_9S";

    await send(firstText);
    expect((await wait()).status).toBe("settled");
    expect(prompts()).toHaveLength(1);
    const firstProcess = prompts()[0].pid;

    writeFileSync(failureFile, JSON.stringify({ code: -32603, message: "Internal error", data: { details: "OpenCode service failure" } }));
    await send(failedText);
    // OpenMausBot retries the failed run once by itself, on a replacement process that loads the same session; the fake
    // still refuses, so that retry fails too.
    const failures = async () => (await messages()).filter((message) => message.tool?.ok === false && /OpenCode service failure/.test(message.tool.name));
    await expect.poll(async () => (await failures()).length, { timeout: 20_000 }).toBe(2);
    await expect.poll(async () => (await task()).busy, { timeout: 20_000 }).toBe(false);
    expect((await messages()).filter((message) => message.role === "user" && message.text?.startsWith(AUTO_RETRY_LEAD))).toHaveLength(1);
    expect(prompts()).toHaveLength(3);
    expect(prompts()[1].pid).toBe(firstProcess);
    const retryProcess = prompts()[2].pid;
    expect(retryProcess).not.toBe(firstProcess);
    expect(calls().filter((call) => call.pid === retryProcess).map((call) => call.method)).toContain("session/load");
    expect(calls().filter((call) => call.method === "session/prompt.error")).toHaveLength(2);
    evidence.push({ failedTask: await task(), promptCallsBeforeRetry: prompts() });

    // The fake keeps the error cached in its process even after removal.
    // Only a replacement process can answer this explicit next user message.
    rmSync(failureFile);
    await send(retryText);
    expect((await wait()).status).toBe("settled");
    expect(prompts()).toHaveLength(4);
    expect([firstProcess, retryProcess]).not.toContain(prompts()[3].pid);
    const prompt = JSON.stringify(JSON.parse(readFileSync(`${dumpFile}.prompt.json`, "utf8")));
    expect(prompt.split(retryText).length - 1).toBe(1);
    expect(prompt).not.toContain(firstText);
    expect(prompt).not.toContain(failedText);
    const finalMessages = await messages();
    for (const text of [firstText, failedText, retryText]) {
      expect(finalMessages.filter((message) => message.role === "user" && message.text === text)).toHaveLength(1);
    }
    expect(finalMessages.findLast((message) => message.role === "bot" && message.kind === "text")?.turnSucceeded).toBe(true);
    expect((await task()).busy).toBe(false);
    evidence.push({ task: await task(), calls: calls(), messages: finalMessages, prompt });
  } finally {
    const evidencePath = `${logPath}.acp-recovery.json`;
    writeFileSync(evidencePath, JSON.stringify(evidence, null, 2));
    await waitForExit(server, { signal: "SIGTERM" });
    await fixture.close();
    console.info(JSON.stringify({ logPath, evidencePath, fixtureRemoved: !existsSync(dataDir) }));
  }
}, 90_000);