ThothII/backend/test/pi-process-manager.test.ts

import { test, expect, vi } from "vitest";
import { spawn } from "node:child_process";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { EventEmitter } from "node:events";
import { chmodSync, writeFileSync } from "node:fs";
import { PiProcessManager } from "../src/pi/pi-process-manager.js";
import { loadConfig } from "../src/config.js";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const FAKE = path.resolve(__dirname, "../../harness/tests/fake_pi/fake_pi_rpc.mjs");
const SCRIPT = path.resolve(__dirname, "../../harness/tests/fake_pi/scripts/f1_disambiguation.json");

test("spawnFor avvia un runtime e il bridge emette il widget F1", async () => {
  const cfg = loadConfig({ THT_HARNESS_DIR: "../harness" });
  const mgr = new PiProcessManager(cfg, { spawnFn: () => spawn("node", [FAKE, SCRIPT]) as any });
  const rt = await mgr.spawnFor("2026-06-27-100000-x", {});
  const widget = await new Promise<any>((res) => rt.bridge.onClientEvent((e) => e.type === "ui_request" && res(e)));
  expect(widget.ui_request.widget).toBe("select");
  mgr.teardown("2026-06-27-100000-x");
  expect(mgr.count()).toBe(0);
});

test("l'exit del child rimuove il runtime dalla mappa (exit handler)", async () => {
  const cfg = loadConfig({ THT_HARNESS_DIR: "../harness" });
  const mgr = new PiProcessManager(cfg, { spawnFn: () => spawn("node", [FAKE, SCRIPT]) as any });
  const rt = await mgr.spawnFor("exit-test", {});
  expect(mgr.count()).toBe(1);
  // Cause the child to exit on its own and await the 'exit' event (no teardown call).
  const exited = new Promise<void>((res) => rt.child.on("exit", () => res()));
  rt.child.kill();
  await exited;
  // Let the manager's registered exit handler run.
  await new Promise((res) => setImmediate(res));
  expect(mgr.count()).toBe(0);
  expect(mgr.get("exit-test")).toBeUndefined();
});

test("spawnFor sullo STESSO id rifiuta il duplicato senza interrompere il runtime attivo", async () => {
  const cfg = loadConfig({ THT_HARNESS_DIR: "../harness" });
  const mgr = new PiProcessManager(cfg, { spawnFn: () => spawn("node", [FAKE, SCRIPT]) as any });
  const first = await mgr.spawnFor("dup-id", {});
  expect(mgr.count()).toBe(1);
  await expect(mgr.spawnFor("dup-id", {})).rejects.toThrow(
    "session runtime already active: dup-id",
  );
  expect(mgr.count()).toBe(1);
  expect(mgr.get("dup-id")).toBe(first);
  mgr.teardown("dup-id");
});

test("l'exit del VECCHIO child non elimina il nuovo runtime (exit identity-checked)", async () => {
  const cfg = loadConfig({ THT_HARNESS_DIR: "../harness" });
  const firstChild = recordingChild();
  const secondChild = recordingChild();
  const children = [firstChild, secondChild];
  const mgr = new PiProcessManager(cfg, { spawnFn: () => children.shift() as any });
  const first = mgr.createFor("respawn-id", {});
  mgr.teardown("respawn-id");
  const second = mgr.createFor("respawn-id", {});
  // The old child's exit handler fires after the respawn; it must NOT evict `second`.
  firstChild.emit("exit", 0);
  expect(mgr.get("respawn-id")).toBe(second);
  expect(mgr.count()).toBe(1);
  void first;
  mgr.teardown("respawn-id");
});

test("identity-checked teardown cannot kill a replacement runtime", () => {
  const cfg = loadConfig({ THT_HARNESS_DIR: "../harness" });
  const firstChild = recordingChild();
  const secondChild = recordingChild();
  firstChild.kill = vi.fn();
  secondChild.kill = vi.fn();
  const children = [firstChild, secondChild];
  const mgr = new PiProcessManager(cfg, { spawnFn: () => children.shift() as any });
  const first = mgr.createFor("replace-id", {});
  mgr.teardown("replace-id");
  const second = mgr.createFor("replace-id", {});

  expect(mgr.teardownIfCurrent("replace-id", first)).toBe(false);
  expect(mgr.get("replace-id")).toBe(second);
  expect(secondChild.kill).not.toHaveBeenCalled();

  expect(mgr.teardownIfCurrent("replace-id", second)).toBe(true);
  expect(mgr.get("replace-id")).toBeUndefined();
  expect(secondChild.kill).toHaveBeenCalledOnce();
});

test("oltre maxPiProcesses solleva errore", async () => {
  const cfg = { ...loadConfig({}), maxPiProcesses: 1 };
  const mgr = new PiProcessManager(cfg, { spawnFn: () => spawn("node", [FAKE, SCRIPT]) as any });
  await mgr.spawnFor("a", {});
  await expect(mgr.spawnFor("b", {})).rejects.toThrow(/max/i);
  mgr.teardown("a");
});

function recordingChild() {
  const ch: any = new EventEmitter();
  ch.stdout = new EventEmitter();
  ch.stderr = new EventEmitter();
  ch._writes = [] as string[];
  ch.stdin = { write: (d: any) => { ch._writes.push(String(d)); return true; } };
  ch.kill = () => {};
  return ch;
}

test("createFor kills a spawned child when post-spawn initialization throws", () => {
  const child = recordingChild();
  child.kill = vi.fn();
  child.stdout = {
    on: () => { throw new Error("READER_INIT_SENTINEL"); },
  };
  const mgr = new PiProcessManager(loadConfig({}), { spawnFn: () => child as any });

  expect(() => mgr.createFor("broken-init", {})).toThrow("READER_INIT_SENTINEL");

  expect(child.kill).toHaveBeenCalledOnce();
  expect(mgr.get("broken-init")).toBeUndefined();
  expect(mgr.count()).toBe(0);
});

test("createFor kills a spawned child when spawn boundary initialization throws", () => {
  const child = recordingChild();
  child.kill = vi.fn();
  child.stderr = {
    on: () => { throw new Error("STDERR_INIT_SENTINEL"); },
  };
  const mgr = new PiProcessManager(loadConfig({}), { spawnFn: () => child as any });

  expect(() => mgr.createFor("broken-spawn-init", {})).toThrow("STDERR_INIT_SENTINEL");

  expect(child.kill).toHaveBeenCalledOnce();
  expect(mgr.get("broken-spawn-init")).toBeUndefined();
  expect(mgr.count()).toBe(0);
});

test("un exit INATTESO del child notifica il client (info error + agent_end)", async () => {
  const cfg = loadConfig({});
  const child = recordingChild();
  const mgr = new PiProcessManager(cfg, { spawnFn: () => child as any });
  const rt = await mgr.spawnFor("crash-id", {});
  const seen: any[] = [];
  rt.bridge.onClientEvent((e) => seen.push(e));
  child.emit("exit", 137);
  expect(rt.bridge.turnState()).toBe("failed");
  expect(seen).toEqual([
    { type: "info", level: "error", text: expect.stringContaining("137") },
    { type: "system_event", event: "session_failed" },
    { type: "system_event", event: "agent_end" },
  ]);
  expect(mgr.count()).toBe(0);
});

test("l'exit dopo teardown NON emette eventi al client (uscita attesa)", async () => {
  const cfg = loadConfig({});
  const child = recordingChild();
  const mgr = new PiProcessManager(cfg, { spawnFn: () => child as any });
  const rt = await mgr.spawnFor("stop-id", {});
  const seen: any[] = [];
  rt.bridge.onClientEvent((e) => seen.push(e));
  mgr.teardown("stop-id");
  child.emit("exit", 0);
  expect(seen).toEqual([]);
});

test("spawnFor resume mode sends /riprendi-sessione <id>", async () => {
  const cfg = loadConfig({}); // no provider/model/thinking -> no rpc.request handshakes
  const child = recordingChild();
  const mgr = new PiProcessManager(cfg, { spawnFn: () => child as any });
  await mgr.spawnFor("sid-9", { mode: "resume" });
  expect(child._writes.join("")).toContain("/riprendi-sessione sid-9");
  expect(child._writes.join("")).not.toContain("/nuova-domanda");
  mgr.teardown("sid-9");
});

test("spawnFor default (new) mode sends /nuova-domanda", async () => {
  const cfg = loadConfig({});
  const child = recordingChild();
  const mgr = new PiProcessManager(cfg, { spawnFn: () => child as any });
  await mgr.spawnFor("sid-10", {});
  expect(child._writes.join("")).toContain("/nuova-domanda");
  mgr.teardown("sid-10");
});

test("spawnFor new mode forwards the real question instead of kickoff", async () => {
  const cfg = loadConfig({});
  const child = recordingChild();
  const mgr = new PiProcessManager(cfg, { spawnFn: () => child as any });
  await mgr.spawnFor("sid-question", { question: "pazienti con cardioversione e ILR" });
  const prompt = JSON.parse(child._writes.at(-1)!);
  expect(prompt.message).toBe('/nuova-domanda "pazienti con cardioversione e ILR"');
  expect(prompt.message).not.toContain("kickoff");
  mgr.teardown("sid-question");
});

test("createFor does not prompt until start is called", async () => {
  const cfg = loadConfig({});
  const child = recordingChild();
  const mgr = new PiProcessManager(cfg, { spawnFn: () => child as any });
  const rt = mgr.createFor("sid-deferred", { question: "q" });
  expect(child._writes).toEqual([]);
  await mgr.configure(rt, {});
  expect(child._writes).toEqual([]);
  mgr.start("sid-deferred", rt, { question: "q" });
  expect(JSON.parse(child._writes.at(-1)!).message).toBe('/nuova-domanda "q"');
  mgr.teardown("sid-deferred");
});

test("a created runtime is active during configure/bootstrap", () => {
  const child = recordingChild();
  const mgr = new PiProcessManager(loadConfig({}), { spawnFn: () => child as any });
  const runtime = mgr.createFor("starting-id", {});
  expect(runtime.bridge.turnState()).toBe("running");
  mgr.teardown("starting-id");
});

test("start reactivates the runtime before sending the prompt", () => {
  const child = recordingChild();
  const mgr = new PiProcessManager(loadConfig({}), { spawnFn: () => child as any });
  const runtime = mgr.createFor("restart-id", {});
  child.stdout.emit("data", `${JSON.stringify({ type: "agent_end", messages: [] })}\n`);
  expect(runtime.bridge.turnState()).toBe("idle");
  let stateAtWrite = runtime.bridge.turnState();
  child.stdin.write = (data: unknown) => {
    stateAtWrite = runtime.bridge.turnState();
    child._writes.push(String(data));
    return true;
  };

  mgr.start("restart-id", runtime, { question: "q" });

  expect(stateAtWrite).toBe("running");
  expect(runtime.bridge.turnState()).toBe("running");
  mgr.teardown("restart-id");
});

test("production spawn uses explicit Pi path and passes portable data root without rewriting PATH", async () => {
  vi.stubEnv("PATH", "/usr/local/bin:/usr/bin");
  vi.stubEnv("PI_PROVIDER_API_KEY", "provider-secret");
  vi.stubEnv("NODE_EXTRA_CA_CERTS", "/certs/company-ca.pem");
  const calls: any[][] = [];
  const child = recordingChild();
  child.stderr.resume = () => {};
  const spawnFn = (...args: any[]) => { calls.push(args); return child as any; };
  const cfg = loadConfig({
    THT_HARNESS_DIR: "/app/harness",
    PI_BIN: "/usr/local/bin/pi",
    THT_DATA_ROOT: "/data",
  });
  const mgr = new PiProcessManager(cfg, { spawnFn });

  try {
    await mgr.spawnFor("portable-session", { author: "user@example.test" });
    const [bin, args, options] = calls[0];
    expect(bin).toBe("/usr/local/bin/pi");
    expect(args).toEqual(["--mode", "rpc"]);
    expect(options.cwd).toBe("/app/harness");
    expect(options.env).toMatchObject({
      PATH: "/usr/local/bin:/usr/bin",
      NODE_EXTRA_CA_CERTS: "/certs/company-ca.pem",
      THT_DATA_ROOT: "/data",
      THT_SESSION: "portable-session",
      THT_AUTHOR: "user@example.test",
    });
    expect(options.env).not.toHaveProperty("PI_PROVIDER_API_KEY");
  } finally {
    mgr.teardown("portable-session");
    vi.unstubAllEnvs();
  }
});

test("session Pi spawn omits ambient THT_DATA_ROOT when config does not provide one", async () => {
  vi.stubEnv("THT_DATA_ROOT", "/ambient-must-not-leak");
  vi.stubEnv("PI_PROVIDER_API_KEY", "still-inherited");
  const calls: any[][] = [];
  const child = recordingChild();
  child.stderr.resume = () => {};
  const mgr = new PiProcessManager(loadConfig({ PI_BIN: "/usr/local/bin/pi" }), {
    spawnFn: (...args: any[]) => { calls.push(args); return child as any; },
  });

  try {
    await mgr.spawnFor("no-data-root", {});
    expect(calls[0][2].env).not.toHaveProperty("THT_DATA_ROOT");
    expect(calls[0][2].env).not.toHaveProperty("PI_PROVIDER_API_KEY");
  } finally {
    mgr.teardown("no-data-root");
    vi.unstubAllEnvs();
  }
});

test.each([
  ["anthropic", "ANTHROPIC_API_KEY"],
  ["OpenAI", "OPENAI_API_KEY"],
  ["gemini", "GEMINI_API_KEY"],
  ["google", "GEMINI_API_KEY"],
  ["deepseek", "DEEPSEEK_API_KEY"],
  ["zai", "ZAI_API_KEY"],
  ["openrouter", "OPENROUTER_API_KEY"],
])("injects the generic file credential only as %s provider env", async (provider, expectedName) => {
  const secret = path.resolve(__dirname, `.model-key-${process.pid}-${provider}`);
  writeFileSync(secret, "provider-secret", { mode: 0o600 });
  const calls: any[][] = [];
  const child = recordingChild();
  child.stderr.resume = () => {};
  const mgr = new PiProcessManager(loadConfig({
    PI_BIN: "/usr/local/bin/pi", THT_MODEL_API_KEY_FILE: secret,
  }), {
    spawnFn: (...args: any[]) => { calls.push(args); return child as any; },
    // Empty auth store: exercise the managed-key injection path deterministically,
    // independent of whatever providers the developer's ~/.pi/agent/auth.json holds.
    authProviders: () => new Set(),
  });
  try {
    await mgr.spawnFor("credential-session", { provider });
    const env = calls[0][2].env;
    expect(env[expectedName]).toBe("provider-secret");
    expect(env).not.toHaveProperty("PI_PROVIDER_API_KEY");
    expect(env).not.toHaveProperty("THT_MODEL_API_KEY_FILE");
    expect(JSON.stringify(calls[0].slice(0, 2))).not.toContain("provider-secret");
  } finally {
    mgr.teardown("credential-session");
    await import("node:fs/promises").then((fs) => fs.unlink(secret));
  }
});

test("skips managed-key injection for a provider present in pi's auth store", async () => {
  const secret = path.resolve(__dirname, `.model-key-${process.pid}-authskip`);
  writeFileSync(secret, "provider-secret", { mode: 0o600 });
  const calls: any[][] = [];
  const child = recordingChild();
  child.stderr.resume = () => {};
  const mgr = new PiProcessManager(loadConfig({
    PI_BIN: "/usr/local/bin/pi", THT_MODEL_API_KEY_FILE: secret,
  }), {
    spawnFn: (...args: any[]) => { calls.push(args); return child as any; },
    authProviders: () => new Set(["deepseek"]),
  });
  try {
    await mgr.spawnFor("authskip-session", { provider: "deepseek" });
    const env = calls[0][2].env;
    // pi resolves deepseek from its own auth store, so no key is forced onto it.
    expect(env.DEEPSEEK_API_KEY).toBeUndefined();
  } finally {
    mgr.teardown("authskip-session");
    await import("node:fs/promises").then((fs) => fs.unlink(secret));
  }
});

test("session Pi spawn reads the single secret bundle and scrubs its path", async () => {
  const secret = path.resolve(__dirname, `.bundle-${process.pid}`);
  writeFileSync(secret, "THT_MODEL_API_KEY=bundle-secret\n", { mode: 0o600 });
  chmodSync(secret, 0o600);
  const calls: any[][] = [];
  const child = recordingChild();
  child.stderr.resume = () => {};
  const mgr = new PiProcessManager(loadConfig({
    PI_BIN: "/usr/local/bin/pi", THT_SECRETS_FILE: secret,
  }), { spawnFn: (...args: any[]) => { calls.push(args); return child as any; } });
  try {
    await mgr.spawnFor("bundle-session", { provider: "openai" });
    expect(calls[0][2].env.OPENAI_API_KEY).toBe("bundle-secret");
    expect(calls[0][2].env).not.toHaveProperty("THT_SECRETS_FILE");
  } finally {
    mgr.teardown("bundle-session");
    await import("node:fs/promises").then((fs) => fs.unlink(secret));
  }
});

test.each([["OpenAI", "openai"], ["gemini", "google"]])(
  "set_model uses canonical packaged provider ID for %s", async (provider, canonical) => {
    const secret = path.resolve(__dirname, `.canonical-key-${process.pid}-${provider}`);
    writeFileSync(secret, "provider-secret", { mode: 0o600 });
    const child = recordingChild();
    child.stderr.resume = () => {};
    child.stdin.write = (data: unknown) => {
      const request = JSON.parse(String(data));
      child._writes.push(String(data));
      if (request.id) {
        queueMicrotask(() => child.stdout.emit("data", `${JSON.stringify({
          type: "response", id: request.id, success: true,
        })}\n`));
      }
      return true;
    };
    const mgr = new PiProcessManager(loadConfig({ THT_MODEL_API_KEY_FILE: secret }), {
      spawnFn: () => child as any,
    });
    try {
      await mgr.spawnFor("canonical-provider", { provider, model: "model-id" });
      expect(child._writes.join("")).toContain(`\"provider\":\"${canonical}\"`);
    } finally {
      mgr.teardown("canonical-provider");
      await import("node:fs/promises").then((fs) => fs.unlink(secret));
    }
  },
);

test.each(["ollama", "local-qwen"])(
  "local provider %s spawns without a model key and scrubs ambient credentials",
  async (provider) => {
    vi.stubEnv("PI_PROVIDER_API_KEY", "ambient-secret");
    vi.stubEnv("THT_MODEL_API_KEY_FILE", "/ambient/secret-path");
    vi.stubEnv("OPENAI_API_KEY", "unselected-provider-secret");
    const calls: any[][] = [];
    const child = recordingChild();
    child.stderr.resume = () => {};
    const mgr = new PiProcessManager(loadConfig({ PI_BIN: "/usr/local/bin/pi" }), {
      spawnFn: (...args: any[]) => { calls.push(args); return child as any; },
    });
    try {
      await mgr.spawnFor(`local-session-${provider}`, { provider });
      expect(calls[0][2].env).not.toHaveProperty("PI_PROVIDER_API_KEY");
      expect(calls[0][2].env).not.toHaveProperty("THT_MODEL_API_KEY_FILE");
      expect(calls[0][2].env).not.toHaveProperty("OPENAI_API_KEY");
    } finally {
      mgr.teardown(`local-session-${provider}`);
      vi.unstubAllEnvs();
    }
  },
);

test.each(["amazon-bedrock", "azure-openai-responses", "cloudflare-workers-ai", "cloudflare-ai-gateway"])(
  "session spawn rejects compound provider %s before spawning Pi", async (provider) => {
    const secret = path.resolve(__dirname, `.compound-key-${process.pid}-${provider}`);
    writeFileSync(secret, "provider-secret", { mode: 0o600 });
    let spawns = 0;
    const mgr = new PiProcessManager(loadConfig({ THT_MODEL_API_KEY_FILE: secret }), {
      spawnFn: () => { spawns += 1; throw new Error("must not spawn"); },
    });
    try {
      await expect(mgr.spawnFor("compound-provider", { provider })).rejects.toThrow(
        "compound credential bundles are unsupported by THT_MODEL_API_KEY_FILE; dedicated provider configuration is required",
      );
      expect(spawns).toBe(0);
    } finally {
      await import("node:fs/promises").then((fs) => fs.unlink(secret));
    }
  },
);

test.each(["missing", "permissive", "unreadable", "directory", "symlink", "unsupported"])(
  "hosted provider credential failure is sanitized: %s", async (kind) => {
    const target = path.resolve(__dirname, `.bad-model-key-${process.pid}-${kind}`);
    if (kind === "permissive") {
      writeFileSync(target, "DO_NOT_LEAK", { mode: 0o644 });
      chmodSync(target, 0o644);
    } else if (kind === "unreadable") {
      writeFileSync(target, "DO_NOT_LEAK", { mode: 0o000 });
    } else if (kind === "directory") {
      await import("node:fs/promises").then((fs) => fs.mkdir(target));
    } else if (kind === "symlink") {
      const source = `${target}-source`;
      writeFileSync(source, "DO_NOT_LEAK", { mode: 0o600 });
      await import("node:fs/promises").then((fs) => fs.symlink(source, target));
    }
    const cfg = loadConfig({ THT_MODEL_API_KEY_FILE: target });
    const mgr = new PiProcessManager(cfg, { spawnFn: () => {
      throw new Error("spawn must not occur");
    } });
    try {
      const provider = kind === "unsupported" ? "unknown-hosted" : "anthropic";
      await expect(mgr.spawnFor("bad-secret", { provider }))
        .rejects.toThrow("model provider credential is unavailable");
    } finally {
      if (kind === "permissive") await import("node:fs/promises").then((fs) => fs.unlink(target));
      if (kind === "unreadable") {
        chmodSync(target, 0o600);
        await import("node:fs/promises").then((fs) => fs.unlink(target));
      }
      if (kind === "directory") await import("node:fs/promises").then((fs) => fs.rmdir(target));
      if (kind === "symlink") {
        await import("node:fs/promises").then(async (fs) => {
          await fs.unlink(target);
          await fs.unlink(`${target}-source`);
        });
      }
    }
  },
);