test(sessions): split pi session service specs

2026-07-04 23:53:58 +02:00
parent 04f9d0e764
commit 386386d2b5
7 changed files with 2134 additions and 2105 deletions
@@ -0,0 +1,299 @@
+import { createAssistantMessageEventStream, type AssistantMessage } from "@earendil-works/pi-ai";
+import type { StreamFn } from "@earendil-works/pi-agent-core";
+import { AuthStorage, ModelRegistry } from "@earendil-works/pi-coding-agent";
+import { describe, expect, it, vi } from "vitest";
+import { PiSessionService } from "./piSessionService.js";
+import { CapturingSessionEventHub, fakeRuntime, runtimeCreator, sessionGateway, sessionRecord, sessionRef, testModel, type RuntimeCreator } from "./piSessionService.testSupport.js";
+
+describe("PiSessionService prompt, queue, and auth warnings", () => {
+  it("sends prompts to an injected runtime without touching the SDK runtime", async () => {
+    const fake = fakeRuntime("prompt-session");
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("prompt-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("prompt-session"), "Build the thing");
+
+    expect(fake.calls.prompt).toEqual([{ text: "Build the thing", options: undefined }]);
+    await service.dispose();
+  });
+
+  it("echoes the user message for direct prompts but not command-forwarded ones", async () => {
+    const fake = fakeRuntime("echo-session", {
+      resourceLoader: { getSkills: () => ({ skills: [{ name: "skill-creator" }] }) },
+    });
+    const hub = new CapturingSessionEventHub();
+    const service = new PiSessionService(hub, {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("echo-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("echo-session"), "Build the thing");
+    expect(hub.sessionEvents.filter(({ event }) => event.type === "message.append")).toHaveLength(1);
+
+    // The client optimistically renders command-forwarded prompts (e.g. /skill:*),
+    // so the server must not publish a second copy via message.append.
+    await service.runCommand(sessionRef("echo-session"), "/skill:skill-creator");
+    expect(hub.sessionEvents.filter(({ event }) => event.type === "message.append")).toHaveLength(1);
+    expect(fake.calls.prompt).toEqual([
+      { text: "Build the thing", options: undefined },
+      { text: "/skill:skill-creator", options: undefined },
+    ]);
+
+    await service.dispose();
+  });
+
+  it("rejects malformed prompt text before opening the runtime", async () => {
+    const fake = fakeRuntime("prompt-session");
+    let createCalls = 0;
+    const createAgentRuntime: RuntimeCreator = async () => {
+      createCalls += 1;
+      await Promise.resolve();
+      return fake.runtime;
+    };
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime,
+      sessionManager: sessionGateway([sessionRecord("prompt-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await expect(service.prompt("prompt-session", undefined)).rejects.toThrow("Prompt text is required");
+
+    expect(createCalls).toBe(0);
+    expect(fake.calls.prompt).toEqual([]);
+    await service.dispose();
+  });
+
+  it("generates a session name for the first prompt via the session's agent.streamFn", async () => {
+    const model = testModel();
+    const streamCalls: unknown[] = [];
+    const streamFn: StreamFn = (streamModel, context, options) => {
+      streamCalls.push({ streamModel, context, options });
+      const stream = createAssistantMessageEventStream();
+      const message: AssistantMessage = {
+        role: "assistant",
+        content: [{ type: "text", text: "Fix login bug" }],
+        api: "anthropic-messages",
+        provider: "anthropic",
+        model: model.id,
+        usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, totalTokens: 0, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } },
+        stopReason: "stop",
+        timestamp: Date.now(),
+      };
+      stream.push({ type: "done", reason: "stop", message });
+      stream.end(message);
+      return stream;
+    };
+    const hub = new CapturingSessionEventHub();
+    const fake = fakeRuntime("name-session", { model, agent: { streamFn } });
+    const service = new PiSessionService(hub, {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("name-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("name-session"), "Please fix the login bug");
+    await vi.waitFor(() => { expect(fake.session.sessionName).toBe("Fix login bug"); });
+
+    expect(streamCalls).toHaveLength(1);
+    expect(hub.sessionEvents.some(({ event }) => event.type === "session.name" && event.name === "Fix login bug")).toBe(true);
+    await service.dispose();
+  });
+
+  it("includes queued message details in session status", async () => {
+    const fake = fakeRuntime("status-session", {
+      messages: [{ role: "user", content: "hello" }, { role: "assistant", content: "hi" }],
+      pendingMessageCount: 2,
+      getSteeringMessages: () => ["adjust this turn"],
+      getFollowUpMessages: () => ["then do this"],
+    });
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("status-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await expect(service.status(sessionRef("status-session"))).resolves.toMatchObject({
+      pendingMessageCount: 2,
+      queuedMessages: [{ kind: "steer", text: "adjust this turn" }, { kind: "followUp", text: "then do this" }],
+      messageCount: 2,
+    });
+    await service.dispose();
+  });
+
+  it("does not enqueue duplicate queued message text", async () => {
+    const fake = fakeRuntime("dedupe-session", {
+      isStreaming: true,
+      pendingMessageCount: 1,
+      getFollowUpMessages: () => ["already queued"],
+    });
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("dedupe-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("dedupe-session"), "already queued", "followUp");
+
+    expect(fake.calls.prompt).toEqual([]);
+    await service.dispose();
+  });
+
+  it("does not append queued prompts to the transcript before delivery", async () => {
+    const hub = new CapturingSessionEventHub();
+    const fake = fakeRuntime("queued-session", { isStreaming: true });
+    const service = new PiSessionService(hub, {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("queued-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("queued-session"), "Wait for the current turn", "followUp");
+
+    expect(fake.calls.prompt).toEqual([{ text: "Wait for the current turn", options: { streamingBehavior: "followUp" } }]);
+    expect(hub.sessionEvents.some(({ event }) => event.type === "message.append")).toBe(false);
+    await service.dispose();
+  });
+
+  it("holds prompts sent during compaction until compaction finishes", async () => {
+    const hub = new CapturingSessionEventHub();
+    const fake = fakeRuntime("compacting-session", { isCompacting: true });
+    let resolveFirstPrompt: (() => void) | undefined;
+    fake.session.prompt = (text: string, options?: { streamingBehavior?: "steer" | "followUp" }) => {
+      fake.calls.prompt.push({ text, options });
+      if (options === undefined) {
+        fake.session.isStreaming = true;
+        return new Promise<void>((resolve) => { resolveFirstPrompt = resolve; });
+      }
+      return Promise.resolve();
+    };
+    const service = new PiSessionService(hub, {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("compacting-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("compacting-session"), "Start task 1", "followUp");
+    await service.prompt(sessionRef("compacting-session"), "Then task 2", "followUp");
+
+    expect(fake.calls.prompt).toEqual([]);
+    expect(hub.sessionEvents.some(({ event }) => event.type === "message.append")).toBe(false);
+    await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({
+      pendingMessageCount: 2,
+      queuedMessages: [{ kind: "followUp", text: "Start task 1" }, { kind: "followUp", text: "Then task 2" }],
+    });
+
+    fake.session.isCompacting = false;
+    fake.emit({ type: "compaction_end" });
+    await new Promise((resolve) => setTimeout(resolve, 5));
+
+    expect(fake.calls.prompt).toEqual([{ text: "Start task 1", options: undefined }]);
+    expect(hub.sessionEvents.some(({ event }) => event.type === "message.append" && JSON.stringify(event.message).includes("Start task 1"))).toBe(true);
+    await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({
+      pendingMessageCount: 1,
+      queuedMessages: [{ kind: "followUp", text: "Then task 2" }],
+    });
+
+    fake.emit({ type: "agent_start" });
+    await new Promise((resolve) => setTimeout(resolve, 5));
+
+    expect(fake.calls.prompt).toEqual([
+      { text: "Start task 1", options: undefined },
+      { text: "Then task 2", options: { streamingBehavior: "followUp" } },
+    ]);
+    await expect(service.status(sessionRef("compacting-session"))).resolves.toMatchObject({
+      pendingMessageCount: 0,
+      queuedMessages: [],
+    });
+    resolveFirstPrompt?.();
+    await service.dispose();
+  });
+
+  it("clears queued messages when aborting active work", async () => {
+    const fake = fakeRuntime("abort-session");
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("abort-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.status(sessionRef("abort-session"));
+    await service.abort(sessionRef("abort-session"));
+
+    expect(fake.calls.clearQueue).toBe(1);
+    expect(fake.calls.abort).toBe(1);
+    await service.dispose();
+  });
+
+  it("clears prompts queued during compaction when aborting active work", async () => {
+    const fake = fakeRuntime("abort-compaction-session", { isCompacting: true });
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("abort-compaction-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.prompt(sessionRef("abort-compaction-session"), "Do not deliver after abort", "followUp");
+    await expect(service.status(sessionRef("abort-compaction-session"))).resolves.toMatchObject({ pendingMessageCount: 1 });
+    await service.abort(sessionRef("abort-compaction-session"));
+
+    expect(fake.calls.clearQueue).toBe(1);
+    expect(fake.calls.prompt).toEqual([]);
+    await expect(service.status(sessionRef("abort-compaction-session"))).resolves.toMatchObject({ pendingMessageCount: 0, queuedMessages: [] });
+    await service.dispose();
+  });
+
+  it("refreshes auth state and dedupes warnings when logout removes the current model's credentials", async () => {
+    const hub = new CapturingSessionEventHub();
+    const authStorage = AuthStorage.inMemory({ anthropic: { type: "api_key", key: "sk-test" } });
+    const modelRegistry = ModelRegistry.inMemory(authStorage);
+    const model = modelRegistry.find("anthropic", "claude-3-5-sonnet-20241022");
+    if (model === undefined) throw new Error("Expected Anthropic model fixture");
+    const fake = fakeRuntime("auth-session", { model, modelRegistry });
+
+    const service = new PiSessionService(hub, {
+      modelRegistry,
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("auth-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.status(sessionRef("auth-session"));
+    hub.sessionEvents.length = 0;
+    hub.globalEvents.length = 0;
+
+    authStorage.logout("anthropic");
+    service.applyAuthChange({ removedProviderId: "anthropic" });
+    service.applyAuthChange({ removedProviderId: "anthropic" });
+
+    const warningCount = () => hub.sessionEvents.filter(({ event }) => event.type === "command.output" && event.level === "error" && event.message.includes("anthropic/claude-3-5-sonnet-20241022")).length;
+    expect(warningCount()).toBe(1);
+    expect(hub.globalEvents.some((event) => event.type === "status.update" && event.status.sessionId === "auth-session")).toBe(true);
+
+    authStorage.set("anthropic", { type: "api_key", key: "sk-new" });
+    service.applyAuthChange();
+    authStorage.logout("anthropic");
+    service.applyAuthChange({ removedProviderId: "anthropic" });
+    expect(warningCount()).toBe(2);
+
+    await service.dispose();
+  });
+
+  it("clears queued messages when stopping a session runtime", async () => {
+    const fake = fakeRuntime("stop-session");
+    const service = new PiSessionService(new CapturingSessionEventHub(), {
+      createAgentRuntime: runtimeCreator(fake.runtime),
+      sessionManager: sessionGateway([sessionRecord("stop-session")]),
+      heartbeatIntervalMs: 60_000,
+    });
+
+    await service.status(sessionRef("stop-session"));
+    service.stop(sessionRef("stop-session"));
+
+    expect(fake.calls.clearQueue).toBe(1);
+    await service.dispose();
+  });
+});