test(server): verify peer messages preserve answerable questions

Exercise message delivery through the daemon and provider interfaces so the regression checks that the question survives and its answer reaches the provider. Retain the human-steer cancellation behavior.

Co-authored-by: Dominic Anaeto <dominic.anaeto@gmail.com>

packages/server/src/server/daemon-e2e/peer-message-permissions.test.ts

@@ -0,0 +1,129 @@
+import { mkdir, mkdtemp, rm } from "node:fs/promises";
+import path from "node:path";
+import type { FormInfo } from "@opencode/client";
+import { expect, test } from "vitest";
+
+import { createTestLogger } from "../../test-utils/test-logger.js";
+import { OpenCodeV2AgentClient } from "../agent/providers/opencode/v2/agent.js";
+import { V2Harness } from "../agent/providers/opencode/test-utils/v2-harness.js";
+import { createTestAgentClients } from "../test-utils/fake-agent-client.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+
+async function createQuestionHarness() {
+  const artifacts = path.resolve(".artifacts");
+  await mkdir(artifacts, { recursive: true });
+  const root = await mkdtemp(path.join(artifacts, "peer-permissions-"));
+  const runtime = new V2Harness();
+  runtime.autoComplete = false;
+  runtime.api.model.list = async () => ({ location: runtime.info.location, data: [] });
+  runtime.api.provider.list = async () => ({
+    location: runtime.info.location,
+    data: [{ id: "test", name: "Test", activation: "enabled", package: "test" }],
+  });
+  let form: FormInfo | null = null;
+  const answers: Array<Record<string, unknown>> = [];
+  runtime.api.session.form.list = async () => (form ? [form] : []);
+  runtime.api.session.form.cancel = async () => {
+    form = null;
+    runtime.finishExecution("interrupted");
+  };
+  runtime.api.session.form.reply = async ({ answer }) => {
+    answers.push(answer);
+    form = null;
+    runtime.finishExecution();
+  };
+  // A steer is admitted to the existing execution, rather than starting another one.
+  runtime.prompt = async (input) => {
+    runtime.prompts.push(input.text);
+    if (runtime.prompts.length === 1) runtime.startExecution();
+  };
+  const provider = new OpenCodeV2AgentClient({
+    logger: createTestLogger(),
+    runtime: runtime.runtime,
+  });
+  const daemon = await createTestPaseoDaemon({
+    agentClients: { ...createTestAgentClients(), opencode: provider },
+    paseoHomeRoot: path.join(root, "daemon"),
+    staticDir: path.join(root, "static"),
+    cleanup: false,
+  });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  try {
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    const recipient = await client.createAgent({
+      provider: "opencode",
+      cwd: root,
+      title: "Waiting for a human answer",
+    });
+    const sender = await client.createAgent({ provider: "claude", cwd: root });
+    await client.sendMessage(recipient.id, "Ask which color to use");
+    await expect.poll(() => runtime.prompts).toHaveLength(1);
+    form = {
+      id: "color-question",
+      sessionID: "session",
+      title: "Color",
+      fields: [{ key: "color", title: "Color", type: "string" }],
+    };
+    runtime.push({ id: "question-created", created: 2, type: "form.created", data: { form } });
+    await client.waitForAgentUpsert(recipient.id, (agent) =>
+      (agent.pendingPermissions ?? []).some((request) => request.id === "color-question"),
+    );
+    return {
+      client,
+      recipient,
+      sender,
+      answers,
+      async close() {
+        await client.close();
+        await daemon.close();
+        await rm(root, { recursive: true, force: true });
+      },
+    };
+  } catch (error) {
+    await client.close();
+    await daemon.close();
+    await rm(root, { recursive: true, force: true });
+    throw error;
+  }
+}
+
+test("an agent-sourced steer leaves the question pending and answerable", async () => {
+  const { client, recipient, sender, answers, close } = await createQuestionHarness();
+  try {
+    await client.sendAgentMessage(recipient.id, "Peer status check", {
+      sourceAgentId: sender.id,
+    });
+    // The public wait includes admitted message delivery before reporting permission status.
+    const waiting = await client.waitForFinish(recipient.id);
+    expect(waiting.status).toBe("permission");
+    expect(waiting.final?.pendingPermissions).toMatchObject([
+      { id: "color-question", kind: "question" },
+    ]);
+    await client.respondToPermission(recipient.id, "color-question", {
+      behavior: "allow",
+      updatedInput: { answers: { Color: "Blue" } },
+    });
+    await client.waitForAgentUpsert(
+      recipient.id,
+      (agent) => (agent.pendingPermissions ?? []).length === 0 && agent.status === "idle",
+    );
+    expect(answers).toEqual([{ color: "Blue" }]);
+  } finally {
+    await close();
+  }
+});
+
+test("a human steer still dismisses the question", async () => {
+  const { client, recipient, answers, close } = await createQuestionHarness();
+  try {
+    await client.sendAgentMessage(recipient.id, "Use red instead", { activeTurnBehavior: "steer" });
+    const finished = await client.waitForFinish(recipient.id);
+    expect(finished.status).toBe("idle");
+    expect(finished.final?.pendingPermissions ?? []).toEqual([]);
+    expect(answers).toEqual([]);
+  } finally {
+    await close();
+  }
+});

packages/server/src/server/session.test.ts

@@ -67,7 +67,6 @@ import {
   createProviderSnapshot,
   createProviderSnapshotManagerStub,
 } from "./test-utils/session-stubs.js";
-import { createTestAgentMessages } from "./test-utils/agent-messages.js";
 import { isPlatform } from "../test-utils/platform.js";
 import {
   GitHubAuthenticationError,
@@ -313,7 +312,6 @@ interface SessionForTestOptions {
   permissions?: readonly DaemonPermission[];
   agentManager?: { [K in keyof SessionOptions["agentManager"]]?: unknown };
   agentStorage?: { [K in keyof SessionOptions["agentStorage"]]?: unknown };
-  agentMessages?: SessionOptions["agentMessages"];
   github?: Partial<ForgeService & GitHubService>;
   checkoutDiffManager?: { scheduleRefreshForCwd: ReturnType<typeof vi.fn> };
   workspaceGitService?: {
@@ -390,7 +388,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
   const sessionOptions: SessionOptions = {
     workspaceRuntimeStore: options.workspaceRuntimeStore ?? createWorkspaceRuntimeStore(),
     messageReceipts: createMessageReceiptsStub(),
-    agentMessages: options.agentMessages ?? createAgentMessageDeliveryStub(),
+    agentMessages: createAgentMessageDeliveryStub(),
     creationService: createTestCreationService(),
     clientId: options.clientId ?? "test-client",
     onMessage: (message) => messages.push(message),
@@ -5678,71 +5676,6 @@ test("project.list returns every active project descriptor", async () => {
   ]);
 });
 
-describe("send_agent_message_request pending permissions", () => {
-  const recipientId = "22222222-2222-4222-8222-222222222222";
-  const senderId = "33333333-3333-4333-8333-333333333333";
-
-  function steerableRecipient() {
-    return {
-      getAgent: vi.fn((agentId: string) =>
-        agentId === recipientId
-          ? { id: recipientId, provider: "claude", lifecycle: "running" }
-          : null,
-      ),
-      waitForAgentClose: vi.fn().mockResolvedValue(undefined),
-      tryRunOutOfBand: vi.fn().mockResolvedValue(false),
-      reloadStalePluginSession: vi.fn().mockResolvedValue(undefined),
-      steerOrReplaceActiveTurn: vi.fn().mockResolvedValue({ status: "steered" }),
-    };
-  }
-
-  test("an agent-sourced message steers without clearing the recipient's pending permissions", async () => {
-    const agentManager = steerableRecipient();
-    const agentMessages = createTestAgentMessages({
-      agentManager: asAgentManager(agentManager),
-      agentStorage: asAgentStorage({ get: vi.fn().mockResolvedValue(undefined) }),
-      logger: pino({ level: "silent" }),
-    });
-    const session = createSessionForTest({ agentManager, agentMessages });
-
-    await session.handleMessage({
-      type: "send_agent_message_request",
-      requestId: "agent-send",
-      agentId: recipientId,
-      text: "Status check: reply when done.",
-      sourceAgentId: senderId,
-      messageId: "agent-message",
-    });
-    await agentMessages.close();
-
-    expect(agentManager.steerOrReplaceActiveTurn).toHaveBeenCalledWith(
-      recipientId,
-      expect.anything(),
-      { clientMessageId: "agent-message" },
-    );
-  });
-
-  test("a sourceless steer still clears the recipient's pending permissions", async () => {
-    const agentManager = steerableRecipient();
-    const session = createSessionForTest({ agentManager });
-
-    await session.handleMessage({
-      type: "send_agent_message_request",
-      requestId: "sourceless-send",
-      agentId: recipientId,
-      text: "Status check: reply when done.",
-      messageId: "sourceless-message",
-      activeTurnBehavior: "steer",
-    });
-
-    expect(agentManager.steerOrReplaceActiveTurn).toHaveBeenCalledWith(
-      recipientId,
-      expect.anything(),
-      { clientMessageId: "sourceless-message", clearPendingPermissions: true },
-    );
-  });
-});
-
 describe("agent config setters", () => {
   function liveAgentManager(overrides: { [K in keyof SessionOptions["agentManager"]]?: unknown }): {
     [K in keyof SessionOptions["agentManager"]]?: unknown;