repositories / pi-ext
pi-ext
bugabingas pi extensions
owned by admin
extensions/os-notifier/__tests__/harness.test.ts
Rawimport {
chmodSync,
mkdirSync,
mkdtempSync,
readdirSync,
readFileSync,
rmSync,
writeFileSync,
} from "node:fs";
import { tmpdir } from "node:os";
import path, { delimiter, join } from "node:path";
import { afterEach, describe, expect, it, vi } from "vitest";
import { createTestSession, type TestSession } from "../../../test/harness";
import {
agentDoneNotification,
agentWaitingNotification,
deliverNative,
nativeCommand,
oscBackend,
oscUserVar,
TurnEligibility,
} from "../index";
describe("os-notifier native commands", () => {
it("constructs Linux notify-send arguments", () => {
expect(nativeCommand("linux", "Title", "Body", "info")).toEqual({
command: "notify-send",
args: ["--app-name=Pi", "--urgency=low", "Title", "Body"],
});
});
it("constructs macOS osascript arguments without interpolating content", () => {
const command = nativeCommand(
"darwin",
`Title ' " $()`,
"Body\nline",
"warning",
);
expect(command.command).toBe("osascript");
expect(command.args.slice(-2)).toEqual([`Title ' " $()`, "Body\nline"]);
expect(command.args[1]).not.toContain("Title");
});
it("constructs Windows PowerShell arguments without interpolating content", () => {
const command = nativeCommand(
"win32",
`Title ' " $()`,
"Body\nline",
"error",
);
expect(command.command).toBe("powershell.exe");
expect(command.args).toContain("-EncodedCommand");
expect(command.args.join(" ")).not.toContain("Title");
expect(command.env).toMatchObject({
PI_OS_NOTIFIER_TITLE: `Title ' " $()`,
PI_OS_NOTIFIER_BODY: "Body\nline",
});
});
it("maps natural Linux urgency", () => {
expect(nativeCommand("linux", "T", "B", "info").args).toContain(
"--urgency=low",
);
expect(nativeCommand("linux", "T", "B", "warning").args).toContain(
"--urgency=normal",
);
expect(nativeCommand("linux", "T", "B", "error").args).toContain(
"--urgency=critical",
);
});
it("selects Kitty OSC 99 and generic OSC 777", () => {
expect(oscBackend({ KITTY_WINDOW_ID: "1" })).toBe("osc99");
expect(oscBackend({})).toBe("osc777");
});
it("encodes pane-scoped metadata as OSC user variables", () => {
expect(oscUserVar("agent", "pi")).toBe(
"\u001b]1337;SetUserVar=agent=cGk=\u0007",
);
expect(oscUserVar("agent_session", "")).toBe(
"\u001b]1337;SetUserVar=agent_session=\u0007",
);
});
it("falls back exactly once only when native delivery fails", async () => {
const writes: string[] = [];
expect(
await deliverNative(
{ command: process.execPath, args: ["-e", "process.exit(0)"] },
"Title",
"Body",
"info",
"osc777",
(value) => writes.push(value),
),
).toBe("native");
expect(writes).toEqual([]);
expect(
await deliverNative(
{ command: process.execPath, args: ["-e", "process.exit(7)"] },
"Title",
"Body",
"error",
"osc777",
(value) => writes.push(value),
),
).toBe("fallback");
expect(writes).toHaveLength(1);
expect(writes[0]).toContain("\u001b]777;notify;Title;❌ Body\u0007");
expect(
await deliverNative(undefined, "Title", "Body", "info", "osc777", () => {
throw new Error("terminal unavailable");
}),
).toBe("failed");
});
it("kills a hung native command and falls back", async () => {
const writes: string[] = [];
await deliverNative(
{
command: process.execPath,
args: [
"-e",
"process.on('SIGTERM', () => {}); setInterval(() => {}, 1000)",
],
},
"Title",
"Body",
"warning",
"osc777",
(value) => writes.push(value),
20,
);
expect(writes).toHaveLength(1);
});
});
describe("os-notifier turn eligibility", () => {
it("allows direct interactive TUI runs only", () => {
const state = new TurnEligibility();
state.input("interactive");
state.prepare();
state.start("tui");
expect(state.settle()).toBe(true);
state.input("rpc");
state.prepare();
state.start("tui");
expect(state.settle()).toBe(false);
state.input("interactive");
state.prepare();
state.start("rpc");
state.input("interactive", "followUp");
expect(state.settle()).toBe(false);
});
it("keeps eligibility bound to the accepted input that started the lifecycle", () => {
const extensionRun = new TurnEligibility();
extensionRun.input("extension");
extensionRun.prepare();
extensionRun.start("tui");
extensionRun.input("interactive", "followUp");
expect(extensionRun.settle()).toBe(false);
const interactiveRun = new TurnEligibility();
interactiveRun.input("interactive");
interactiveRun.prepare();
interactiveRun.start("tui");
interactiveRun.input("extension", "followUp");
expect(interactiveRun.settle()).toBe(true);
});
it("preserves eligibility across low-level retries until settlement", () => {
const state = new TurnEligibility();
state.input("interactive");
state.prepare();
state.start("tui");
expect(state.isEligible()).toBe(true);
state.start("tui");
expect(state.settle()).toBe(true);
expect(state.isEligible()).toBe(false);
});
it("accepts input prepared after a pre-prompt compaction run starts", () => {
const state = new TurnEligibility();
state.input("interactive");
state.start("tui");
state.prepare();
expect(state.settle()).toBe(true);
});
it("does not consume handled interactive input for an extension-triggered run", () => {
const state = new TurnEligibility();
state.input("interactive");
state.start("tui");
expect(state.settle()).toBe(false);
state.prepare();
state.start("tui");
expect(state.settle()).toBe(false);
});
});
describe("os-notifier payload", () => {
it("includes bounded context when Pi needs input", () => {
expect(
agentWaitingNotification(
{ kind: "confirm", title: "Allow destructive operation?" },
{
cwd: "/home/oli/Workspace/dotfiles",
branch: "trunk",
sessionName: "notifier work",
},
),
).toEqual({
title: "Pi needs input",
message:
"Confirmation • Allow destructive operation? • dotfiles • trunk • notifier work",
severity: "warning",
});
});
it("includes rich bounded context in completion body", () => {
expect(
agentDoneNotification(
{
messages: [
{ role: "user", content: "implement richer Pi notifications" },
{ role: "assistant", stopReason: "stop" },
],
},
{
cwd: "/home/oli/Workspace/dotfiles/pi-ext",
model: { name: "GPT-5.5", provider: "openai", id: "gpt-5.5" },
durationMs: 45_000,
branch: "trunk",
sessionName: "notifier work",
},
),
).toEqual({
title: "Pi finished",
message:
"Ready after 45s • pi-ext • trunk • notifier work • GPT-5.5\n↳ implement richer Pi notifications",
severity: "info",
});
});
it("sanitizes and bounds prompt and error detail", () => {
const notification = agentDoneNotification(
{
messages: [
{ role: "user", content: `prompt\u0000\u009d${"x".repeat(100)}` },
{
role: "assistant",
stopReason: "error",
errorMessage: `failure\n${"y".repeat(200)}`,
},
],
},
{},
);
expect(notification.title).toBe("Pi failed");
expect(notification.severity).toBe("error");
expect(notification.message).not.toContain("\u0000");
expect(notification.message).not.toContain("\u009d");
expect(notification.message.match(/\n/gu)).toHaveLength(1);
expect(notification.message.length).toBeLessThan(240);
expect(notification.message).toContain("…");
});
it("retains the accepted prompt when a final retry event has no user message", () => {
const notification = agentDoneNotification(
{ messages: [{ role: "assistant", stopReason: "stop" }] },
{ prompt: "original interactive prompt" },
);
expect(notification.message).toContain("↳ original interactive prompt");
});
it("classifies aborted runs as warnings", () => {
expect(
agentDoneNotification(
{ messages: [{ role: "assistant", stopReason: "aborted" }] },
{},
),
).toMatchObject({ title: "Pi stopped", severity: "warning" });
});
});
describe("os-notifier pi-test-harness", () => {
let t: TestSession | undefined;
const focusState = (
globalThis as unknown as {
__bugabinga_pi_focus: { focused: boolean };
}
).__bugabinga_pi_focus;
const originalFocus = focusState.focused;
afterEach(() => {
focusState.focused = originalFocus;
t?.dispose();
t = undefined;
});
it("writes enabled lifecycle debug records in an isolated sandbox", async () => {
const root = mkdtempSync(join(tmpdir(), "pi-ext-os-notifier-debug-"));
const original = {
debug: process.env.PI_OS_NOTIFIER_DEBUG,
state: process.env.XDG_STATE_HOME,
};
try {
process.env.PI_OS_NOTIFIER_DEBUG = "1";
process.env.XDG_STATE_HOME = join(root, "state");
vi.resetModules();
const { default: extension } = await import("../index.ts");
t = await createTestSession({
cwd: root,
env: {
PI_OS_NOTIFIER_DEBUG: "1",
XDG_STATE_HOME: join(root, "state"),
},
extensionFactories: [extension],
});
await t.session.extensionRunner.emit({
type: "session_start",
reason: "startup",
});
await t.session.extensionRunner.emit({
type: "session_shutdown",
reason: "quit",
});
const events = readdirSync(
join(root, "state", "pi-ext", "debug", "os-notifier"),
).flatMap((file) =>
readFileSync(
join(root, "state", "pi-ext", "debug", "os-notifier", file),
"utf8",
)
.trim()
.split("\n")
.map((line) => JSON.parse(line)),
);
expect(events).toEqual(
expect.arrayContaining([
expect.objectContaining({
extension: "os-notifier",
event: "session.start",
}),
expect.objectContaining({
extension: "os-notifier",
event: "session.shutdown",
}),
]),
);
} finally {
t?.dispose();
t = undefined;
if (original.debug === undefined) delete process.env.PI_OS_NOTIFIER_DEBUG;
else process.env.PI_OS_NOTIFIER_DEBUG = original.debug;
if (original.state === undefined) delete process.env.XDG_STATE_HOME;
else process.env.XDG_STATE_HOME = original.state;
rmSync(root, { recursive: true, force: true });
}
});
it.skipIf(process.platform !== "linux")(
"records detached fallback delivery after shutdown",
async () => {
const root = mkdtempSync(join(tmpdir(), "pi-ext-os-notifier-debug-"));
const original = {
debug: process.env.PI_OS_NOTIFIER_DEBUG,
state: process.env.XDG_STATE_HOME,
};
try {
process.env.PI_OS_NOTIFIER_DEBUG = "1";
process.env.XDG_STATE_HOME = join(root, "state");
vi.resetModules();
const { default: extension } = await import("../index.ts");
t = await createTestSession({
cwd: root,
env: {
PI_OS_NOTIFIER_DEBUG: "1",
XDG_STATE_HOME: join(root, "state"),
},
extensionFactories: [extension],
});
const bin = join(root, "bin");
mkdirSync(bin);
writeFileSync(
join(bin, "notify-send"),
`#!${process.execPath}\nsetTimeout(() => process.exit(1), 250);\n`,
);
chmodSync(join(bin, "notify-send"), 0o755);
t.env.PATH = `${bin}${delimiter}${t.env.PATH ?? ""}`;
t.session.extensionRunner.setUIContext(
t.session.extensionRunner.getUIContext(),
"tui",
);
focusState.focused = false;
await emitCompletion(t, "interactive");
const readEvents = () =>
readdirSync(
join(root, "state", "pi-ext", "debug", "os-notifier"),
).flatMap((file) =>
readFileSync(
join(root, "state", "pi-ext", "debug", "os-notifier", file),
"utf8",
)
.trim()
.split("\n")
.filter(Boolean)
.map((line) => JSON.parse(line)),
);
await t.session.extensionRunner.emit({
type: "session_shutdown",
reason: "quit",
});
expect(
readEvents().some(
(event) => event.event === "notification.deliver.finish",
),
).toBe(false);
await waitFor(() =>
readEvents().some(
(event) =>
event.event === "notification.deliver.finish" &&
event.outcome === "fallback",
),
);
const terminal = readEvents().filter(
(event) => event.event === "notification.deliver.finish",
);
expect(terminal).toEqual([
expect.objectContaining({ kind: "delivery", outcome: "fallback" }),
]);
} finally {
t?.dispose();
t = undefined;
if (original.debug === undefined)
delete process.env.PI_OS_NOTIFIER_DEBUG;
else process.env.PI_OS_NOTIFIER_DEBUG = original.debug;
if (original.state === undefined) delete process.env.XDG_STATE_HOME;
else process.env.XDG_STATE_HOME = original.state;
rmSync(root, { recursive: true, force: true });
}
},
);
it("loads in real Pi extension runtime", async () => {
const extensionPath = path.resolve(import.meta.dirname, "../index.ts");
t = await createTestSession({ extensions: [extensionPath] });
expect(t.session.extensionRunner.extensions).toHaveLength(1);
expect(t.session.extensionRunner.extensions[0].path).toBe(extensionPath);
});
it.skipIf(process.platform !== "linux")(
"notifies once per uninterrupted unfocused interactive prompt wait",
async () => {
const extensionPath = path.resolve(import.meta.dirname, "../index.ts");
t = await createTestSession({ extensions: [extensionPath] });
const bin = join(t.cwd, "bin");
const log = join(t.cwd, "notify.log");
const executable = join(bin, "notify-send");
mkdirSync(bin);
writeFileSync(
executable,
`#!${process.execPath}\nrequire("node:fs").appendFileSync(process.env.TEST_NOTIFY_LOG, JSON.stringify(process.argv.slice(2)) + "\\n");\n`,
);
chmodSync(executable, 0o755);
t.env.PATH = `${bin}${delimiter}${t.env.PATH ?? ""}`;
t.env.TEST_NOTIFY_LOG = log;
t.session.extensionRunner.setUIContext(
t.session.extensionRunner.getUIContext(),
"tui",
);
focusState.focused = false;
await t.session.extensionRunner.emit({
type: "input",
text: "test notification",
source: "interactive",
} as never);
await t.session.extensionRunner.emitBeforeAgentStart(
"test notification",
undefined,
"BASE",
{} as never,
);
await t.session.extensionRunner.emit({ type: "agent_start" } as never);
const prompt = {
type: "ui_prompt_start",
reason: "ui_prompt",
kind: "confirm",
title: "Continue?",
} as const;
await t.session.extensionRunner.emit(prompt as never);
await waitFor(
() => readFileSync(log, "utf8").trim().split("\n").length === 1,
);
await t.session.extensionRunner.emit(prompt as never);
await new Promise((resolve) => setTimeout(resolve, 50));
expect(readFileSync(log, "utf8").trim().split("\n")).toHaveLength(1);
await t.session.extensionRunner.emit({
type: "ui_prompt_end",
reason: "ui_prompt",
kind: "confirm",
title: "Continue?",
} as never);
await t.session.extensionRunner.emit(prompt as never);
await waitFor(
() => readFileSync(log, "utf8").trim().split("\n").length === 2,
);
const args = JSON.parse(readFileSync(log, "utf8").trim().split("\n")[0]);
expect(args).toContain("Pi needs input");
expect(args.join(" ")).toContain("Confirmation • Continue?");
},
);
it.skipIf(process.platform !== "linux")(
"uses native delivery only for unfocused interactive TUI runs",
async () => {
const extensionPath = path.resolve(import.meta.dirname, "../index.ts");
t = await createTestSession({ extensions: [extensionPath] });
const bin = join(t.cwd, "bin");
const log = join(t.cwd, "notify.log");
const executable = join(bin, "notify-send");
mkdirSync(bin);
writeFileSync(
executable,
`#!${process.execPath}\nrequire("node:fs").appendFileSync(process.env.TEST_NOTIFY_LOG, JSON.stringify(process.argv.slice(2)) + "\\n");\n`,
);
chmodSync(executable, 0o755);
t.env.PATH = `${bin}${delimiter}${t.env.PATH ?? ""}`;
t.env.TEST_NOTIFY_LOG = log;
t.session.extensionRunner.setUIContext(
t.session.extensionRunner.getUIContext(),
"tui",
);
focusState.focused = false;
await emitCompletion(t, "interactive", false);
await new Promise((resolve) => setTimeout(resolve, 50));
expect(() => readFileSync(log, "utf8")).toThrow();
await t.session.extensionRunner.emit({ type: "agent_settled" } as never);
await waitFor(() => readFileSync(log, "utf8").trim().length > 0);
const first = readFileSync(log, "utf8").trim().split("\n");
expect(JSON.parse(first[0])).toContain("Pi finished");
focusState.focused = true;
await emitCompletion(t, "interactive");
focusState.focused = false;
await emitCompletion(t, "extension");
t.session.extensionRunner.setUIContext(
t.session.extensionRunner.getUIContext(),
"rpc",
);
await emitCompletion(t, "interactive");
await new Promise((resolve) => setTimeout(resolve, 50));
expect(readFileSync(log, "utf8").trim().split("\n")).toHaveLength(1);
},
);
});
async function emitCompletion(
t: TestSession,
source: "interactive" | "rpc" | "extension",
settle = true,
): Promise<void> {
await t.session.extensionRunner.emit({
type: "input",
text: "test notification",
source,
} as never);
await t.session.extensionRunner.emitBeforeAgentStart(
"test notification",
undefined,
"BASE",
{} as never,
);
await t.session.extensionRunner.emit({ type: "agent_start" } as never);
await t.session.extensionRunner.emit({
type: "agent_end",
messages: [
{ role: "user", content: "test notification" },
{ role: "assistant", stopReason: "stop" },
],
} as never);
if (settle) {
await t.session.extensionRunner.emit({ type: "agent_settled" } as never);
}
}
async function waitFor(
assertion: () => boolean,
timeoutMs = 2000,
): Promise<void> {
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
try {
if (assertion()) return;
} catch {
// File may not exist until the child process starts.
}
await new Promise((resolve) => setTimeout(resolve, 10));
}
throw new Error("timed out waiting for native notification executable");
}