fix(acp): clear abandoned approvals without leaving providers blocked (#8548)

* fix(acp): clear permission requests a finished turn abandoned

An ACP agent can give up on a permission request itself and end its turn
normally. Paseo kept that request pending, so the approval card stayed up
with nothing listening, and the next prompt reported the agent as waiting.

Cancel a turn's open requests whenever that turn ends, cancel a request
when its tool reaches a terminal status, and publish the resolved event in
both cases. Requests raised outside a turn stay answerable.

Closes #7664

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(acp): keep permission cleanup consistent after failed turns

Failed and canceled turns clear every permission in the agent manager. Answer that same set in the ACP session so a request raised outside the foreground turn cannot leave the provider waiting after its approval disappears.

Co-authored-by: 3ae3ae <ogyshjsm@naver.com>

---------

Co-authored-by: 3ae3ae <ogyshjsm@naver.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

packages/plugin/src/client/host.test.ts

@@ -24,17 +24,23 @@ describe("plugin RPC contracts", () => {
     expectTypeOf<RpcOutput<typeof contract>>().toEqualTypeOf<{ value: number }>();
   });
 
-  it("validates input before invoking the daemon", async () => {
-    let invocations = 0;
-    const invoke = async () => {
-      invocations += 1;
-      return { value: 2 };
+  it("sends input unchanged for the plugin to parse", async () => {
+    let parses = 0;
+    const transformed = defineRpc({
+      name: "transformed",
+      input: z.string().transform((text) => {
+        parses += 1;
+        return Number(text);
+      }),
+      output: z.number(),
+    });
+    const invoke = async (_method: string, input: unknown) => {
+      expect(input).toBe("2");
+      expect(parses).toBe(0);
+      return transformed.input.parseAsync(input);
     };
-
-    await expect(callPluginRpc(contract, invoke, { value: "wrong" })).rejects.toBeInstanceOf(
-      z.ZodError,
-    );
-    expect(invocations).toBe(0);
+    await expect(callPluginRpc(transformed, invoke, "2")).resolves.toBe(2);
+    expect(parses).toBe(1);
   });
 
   it("validates output before resolving", async () => {

packages/plugin/src/rpc.ts

@@ -39,8 +39,7 @@ export async function callPluginRpc<InputSchema extends ZodType, OutputSchema ex
   invoke: (method: string, input: unknown) => Promise<unknown>,
   input: PluginWireInput<ZodInput<InputSchema>>,
 ): Promise<ZodOutput<OutputSchema>> {
-  const parsedInput = await contract.input.parseAsync(input);
-  const output = await invoke(contract.name, parsedInput);
+  const output = await invoke(contract.name, input);
   return contract.output.parseAsync(output);
 }
 

packages/protocol/src/messages.ts

@@ -1603,7 +1603,7 @@ export const PluginRpcInvokeRequestSchema = z.object({
   requestId: z.string(),
   pluginId: PluginIdSchema,
   method: z.string().min(1),
-  input: z.unknown(),
+  input: z.unknown().optional(),
 });
 
 export const PluginRpcSubscribeRequestSchema = z.object({
@@ -6969,7 +6969,7 @@ export const PluginRpcInvokeResponseSchema = z.object({
   type: z.literal("plugin.rpc.invoke.response"),
   payload: z.object({
     requestId: z.string(),
-    output: z.unknown(),
+    output: z.unknown().optional(),
   }),
 });
 

packages/protocol/tests/validation/ws-outbound.test.ts

@@ -5,7 +5,7 @@ import { fileURLToPath, pathToFileURL } from "node:url";
 import { createJiti } from "jiti";
 import type { z } from "zod";
 import { describe, expect, it } from "vitest";
-import { WSOutboundMessageSchema } from "../../src/messages.js";
+import { PluginRpcInvokeResponseSchema, WSOutboundMessageSchema } from "../../src/messages.js";
 import { validateWSOutboundMessage } from "../../src/validation/ws-outbound.js";
 
 interface GeneratedSchema {
@@ -401,3 +401,8 @@ const SourceSchema = z.object({
     });
   });
 });
+
+it("accepts omitted RPC output so the receiving client can apply schema defaults", () => {
+  const message = { type: "plugin.rpc.invoke.response", payload: { requestId: "default-output" } };
+  expect(PluginRpcInvokeResponseSchema.safeParse(message).success).toBe(true);
+});

packages/server/src/server/agent/providers/acp-agent.ts

@@ -643,6 +643,7 @@ interface PendingPermission {
   resolve: (response: RequestPermissionResponse) => void;
   reject: (error: Error) => void;
   turnId: string | null;
+  toolCallId: string;
 }
 
 interface PendingUserMessage {
@@ -2644,6 +2645,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
         resolve,
         reject,
         turnId: this.activeForegroundTurnId,
+        toolCallId: params.toolCall.toolCallId,
       });
     });
 
@@ -3207,6 +3209,9 @@ export class ACPAgentSession implements AgentSession, ACPClient {
       snapshot = this.toolSnapshotTransformer(snapshot);
     }
     this.toolCalls.set(toolCallId, snapshot);
+    if (update.status === "completed" || update.status === "failed") {
+      this.cancelPendingPermissions((pending) => pending.toolCallId === toolCallId);
+    }
     return [this.wrapTimeline(mapToolSnapshotToTimeline(snapshot, this.terminalEntries))];
   }
 
@@ -3414,11 +3419,11 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     event: Extract<AgentStreamEvent, { type: "turn_completed" | "turn_failed" | "turn_canceled" }>,
   ): void {
     this.deliverTranslatedEvents(this.flushPendingUserMessage());
-    // A turn that ends without completing leaves no way to answer its open
-    // permission requests, so tell the agent they were cancelled.
-    if (event.type !== "turn_completed") {
-      this.cancelPendingPermissions();
-    }
+    // Failed and canceled turns clear every request in the manager. Answer the
+    // same set here so an outside-turn request cannot keep the provider blocked.
+    this.cancelPendingPermissions(
+      (pending) => event.type !== "turn_completed" || pending.turnId === event.turnId,
+    );
     this.activeForegroundTurnId = null;
     this.fallbackAssistantMessageId = null;
     if (this.submittedUserMessageTurnId === event.turnId) {
@@ -3427,11 +3432,21 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     this.pushEvent(event);
   }
 
-  private cancelPendingPermissions(): void {
-    for (const pending of this.pendingPermissions.values()) {
+  private cancelPendingPermissions(
+    matches: (pending: PendingPermission) => boolean = () => true,
+  ): void {
+    for (const [requestId, pending] of this.pendingPermissions) {
+      if (!matches(pending)) continue;
+      this.pendingPermissions.delete(requestId);
       pending.resolve({ outcome: { outcome: "cancelled" } });
+      this.pushEvent({
+        type: "permission_resolved",
+        provider: this.provider,
+        requestId,
+        resolution: { behavior: "deny", message: "Permission request canceled" },
+        turnId: pending.turnId ?? undefined,
+      });
     }
-    this.pendingPermissions.clear();
   }
 
   private emitBootstrapThreadEvent(): void {

packages/server/src/server/agent/providers/claude/agent.error-result.test.ts

@@ -13,6 +13,11 @@ interface ScriptedTurn {
   before?: string;
   after?: string;
   isError: boolean;
+  subtype?:
+    | "error_max_turns"
+    | "error_during_execution"
+    | "error_max_budget_usd"
+    | "error_max_structured_output_retries";
   usage?: {
     inputTokens: number;
     cachedInputTokens: number;
@@ -57,11 +62,12 @@ function scriptedQuery(turns: ScriptedTurn[]): ClaudeQueryFactory {
         if (turn.after) yield assistant(`after-${index}`, turn.after);
         yield {
           type: "result",
-          subtype: "success",
+          subtype: turn.subtype ?? "success",
           session_id: "error-session",
           duration_ms: 1,
           duration_api_ms: 0,
           is_error: turn.isError,
+          ...(turn.subtype ? { errors: ["Reached maximum number of turns (1)"] } : {}),
           num_turns: 1,
           result: turn.isError ? (turn.errorReply ?? "") : (turn.after ?? turn.errorReply ?? ""),
           stop_reason: "stop_sequence",
@@ -188,3 +194,33 @@ test("an error reply the turn moves past is still shown", async () => {
   expect(terminal(completed)).toMatchObject({ type: "turn_completed" });
   expect(assistantTexts(completed)).toEqual([LIMIT]);
 });
+
+test.each([
+  "error_max_turns",
+  "error_during_execution",
+  "error_max_budget_usd",
+  "error_max_structured_output_retries",
+] as const)("a %s result publishes final usage before failure", async (subtype) => {
+  const usage = {
+    inputTokens: 2,
+    cachedInputTokens: 12577,
+    outputTokens: 110,
+    totalCostUsd: 0.00365537,
+  };
+  const [failed, next] = await runTurns([
+    { before: "Running the command.", isError: true, subtype, usage },
+    { after: "Done.", isError: false },
+  ]);
+
+  // Collection ends at the terminal event, so this also checks ordering.
+  expect(failed).toContainEqual(
+    expect.objectContaining({ type: "usage_updated", usage: expect.objectContaining(usage) }),
+  );
+  expect(terminal(failed)).toMatchObject({
+    type: "turn_failed",
+    error: "Reached maximum number of turns (1)",
+  });
+  expect(assistantTexts(failed)).toEqual(["Running the command."]);
+  expect(terminal(next)).toMatchObject({ type: "turn_completed" });
+  expect(assistantTexts(next)).toEqual(["Done."]);
+});

packages/server/src/server/agent/providers/claude/agent.ts

@@ -5037,6 +5037,7 @@ class ClaudeAgentSession implements AgentSession {
       "errors" in message && Array.isArray(message.errors) && message.errors.length > 0
         ? message.errors.join("\n")
         : "Claude run failed";
+    if (usage) events.push({ type: "usage_updated", provider: "claude", usage });
     events.push(...this.sidechainTracker.finishAll("failed"));
     events.push(this.buildTurnFailedEvent(errorMessage));
   }

packages/server/src/server/agent/providers/generic-acp-agent.permission-turn-failure.test.ts

@@ -5,6 +5,7 @@ import path from "node:path";
 import { describe, expect, test } from "vitest";
 
 import { createTestLogger } from "../../../test-utils/test-logger.js";
+import { AgentManager } from "../agent-manager.js";
 import type { AgentSession, AgentStreamEvent } from "../agent-sdk-types.js";
 import { GenericACPAgentClient } from "./generic-acp-agent.js";
 
@@ -12,6 +13,7 @@ describe("GenericACPAgentClient permission requests when a turn ends early", ()
   test.each([
     { ending: "fail", terminalEvent: "turn_failed" },
     { ending: "cancel", terminalEvent: "turn_canceled" },
+    { ending: "end_turn", terminalEvent: "turn_completed" },
   ] as const)(
     "answers a permission request left open by a $terminalEvent turn with cancelled",
     async ({ ending, terminalEvent }) => {
@@ -25,6 +27,16 @@ describe("GenericACPAgentClient permission requests when a turn ends early", ()
           await waitForEvent(events, terminalEvent);
           expect(hasEvent(events, "permission_requested")).toBe(true);
           expect(session.getPendingPermissions()).toEqual([]);
+          const requested = permissionRequests(events)[0]!;
+          expect(resolvedEvents(events)).toEqual([
+            {
+              type: "permission_resolved",
+              provider: "acp",
+              requestId: requested.request.id,
+              resolution: { behavior: "deny", message: "Permission request canceled" },
+              turnId: requested.turnId,
+            },
+          ]);
 
           const replyStart = events.length;
           await session.startTurn("turn two");
@@ -39,6 +51,84 @@ describe("GenericACPAgentClient permission requests when a turn ends early", ()
       });
     },
   );
+
+  test.each(["completed", "failed"] as const)(
+    "cancels only the matching permission when its tool becomes %s",
+    async (status) => {
+      await withFakeACPAgent(status, async (command, cwd) => {
+        const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
+        const session = await client.createSession({ provider: "acp", cwd });
+        try {
+          const events = collectEvents(session);
+          await session.startTurn("turn one");
+          await waitForToolUpdate(events);
+          expect(pendingToolIds(session)).toEqual(["tool-2"]);
+          expect(resolvedEvents(events)).toEqual([
+            expect.objectContaining({ requestId: permissionRequests(events)[0]!.request.id }),
+          ]);
+        } finally {
+          await session.close();
+        }
+      });
+    },
+  );
+
+  test("keeps a permission raised outside a turn answerable after another turn completes", async () => {
+    await withFakeACPAgent("outside_turn", async (command, cwd) => {
+      const client = new GenericACPAgentClient({ logger: createTestLogger(), command });
+      const session = await client.createSession({ provider: "acp", cwd });
+      try {
+        const outside = session.getPendingPermissions();
+        expect(pendingToolIds(session)).toEqual(["tool-outside"]);
+        const events = collectEvents(session);
+        await session.startTurn("turn one");
+        await waitForEvent(events, "turn_completed");
+        expect(session.getPendingPermissions()).toEqual(outside);
+        await session.respondToPermission(outside[0]!.id, { behavior: "allow" });
+        expect(session.getPendingPermissions()).toEqual([]);
+      } finally {
+        await session.close();
+      }
+    });
+  });
+
+  test.each(["fail", "cancel"] as const)(
+    "settles an outside-turn permission that the manager clears when a turn ends with %s",
+    async (ending) => {
+      await withFakeACPAgent(`outside_${ending}`, async (command, cwd) => {
+        const logger = createTestLogger();
+        const client = new GenericACPAgentClient({
+          logger,
+          command,
+          waitForInitialCommands: false,
+        });
+        const manager = new AgentManager({ clients: { acp: client }, logger });
+        const agent = await manager.createAgent({ provider: "acp", cwd }, undefined, {
+          workspaceId: undefined,
+        });
+        try {
+          await waitForManagedPermission(manager, agent.id);
+          if (ending === "fail") {
+            await expect(manager.runAgent(agent.id, "turn one")).rejects.toThrow(
+              "turn failed with a permission open",
+            );
+          } else {
+            await expect(manager.runAgent(agent.id, "turn one")).resolves.toMatchObject({
+              canceled: true,
+            });
+          }
+          expect(manager.getPendingPermissions(agent.id)).toEqual([]);
+          const result = await manager.runAgent(agent.id, "turn two");
+          expect(result.finalText).toBe(
+            'perm-outside answered: {"outcome":{"outcome":"cancelled"}}',
+          );
+        } finally {
+          await manager.closeAgent(agent.id);
+          await manager.flush();
+        }
+      });
+    },
+  );
 });
 
 function collectEvents(session: AgentSession): AgentStreamEvent[] {
@@ -47,6 +137,10 @@ function collectEvents(session: AgentSession): AgentStreamEvent[] {
   return events;
 }
 
+async function waitForManagedPermission(manager: AgentManager, agentId: string): Promise<void> {
+  await expect.poll(() => manager.getPendingPermissions(agentId).length).toBe(1);
+}
+
 function hasEvent(
   events: AgentStreamEvent[],
   type: AgentStreamEvent["type"],
@@ -77,8 +171,32 @@ function assistantText(events: AgentStreamEvent[]): string {
     .join("");
 }
 
+function permissionRequests(events: AgentStreamEvent[]) {
+  return events.filter((event) => event.type === "permission_requested");
+}
+
+function resolvedEvents(events: AgentStreamEvent[]) {
+  return events.filter((event) => event.type === "permission_resolved");
+}
+
+function pendingToolIds(session: AgentSession) {
+  return session.getPendingPermissions().map((request) => request.metadata?.toolCallId);
+}
+
+async function waitForToolUpdate(events: AgentStreamEvent[]): Promise<void> {
+  await expect.poll(() => assistantText(events)).toBe("tool terminal");
+}
+
 async function withFakeACPAgent(
-  ending: "fail" | "cancel",
+  ending:
+    | "fail"
+    | "cancel"
+    | "end_turn"
+    | "completed"
+    | "failed"
+    | "outside_turn"
+    | "outside_fail"
+    | "outside_cancel",
   run: (command: [string, ...string[]], cwd: string) => Promise<void>,
 ): Promise<void> {
   const testDir = await mkdtemp(path.join(tmpdir(), "paseo-acp-permission-turn-failure-"));
@@ -91,8 +209,7 @@ async function withFakeACPAgent(
   }
 }
 
-// Turn one asks for permission, then fails or is cancelled while the request is
-// still open.
+// Turn one ends while its permission request is still open.
 // Later turns report whether that request was ever answered.
 const fakeACPAgentScript = `
 const readline = require("node:readline");
@@ -101,15 +218,34 @@ const ending = process.argv[2];
 const rl = readline.createInterface({ input: process.stdin });
 let promptCount = 0;
 let permissionAnswer = null;
+let outsidePermissionAnswer = null;
 
 function write(message) {
   process.stdout.write(JSON.stringify({ jsonrpc: "2.0", ...message }) + "\\n");
 }
 
+function ask(sessionId, requestId, toolCallId) {
+  const toolCall = { toolCallId, title: "echo hello", kind: "execute", status: "pending" };
+  write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "tool_call", ...toolCall } } });
+  write({
+    id: requestId,
+    method: "session/request_permission",
+    params: {
+      sessionId,
+      toolCall,
+      options: [
+        { optionId: "allow", name: "Allow once", kind: "allow_once" },
+        { optionId: "reject", name: "Reject", kind: "reject_once" },
+      ],
+    },
+  });
+}
+
 rl.on("line", (line) => {
   const message = JSON.parse(line);
-  if (message.method === undefined && message.id === "perm-1") {
-    permissionAnswer = message.result ?? { error: message.error };
+  if (message.method === undefined) {
+    if (message.id === "perm-1") permissionAnswer = message.result ?? { error: message.error };
+    if (message.id === "perm-outside") outsidePermissionAnswer = message.result ?? { error: message.error };
     return;
   }
   if (message.method === "initialize") {
@@ -117,6 +253,7 @@ rl.on("line", (line) => {
     return;
   }
   if (message.method === "session/new") {
+    if (ending.startsWith("outside_")) ask("session-1", "perm-outside", "tool-outside");
     write({ id: message.id, result: { sessionId: "session-1" } });
     return;
   }
@@ -124,35 +261,32 @@ rl.on("line", (line) => {
     const sessionId = message.params.sessionId;
     promptCount += 1;
     if (promptCount === 1) {
-      const toolCall = { toolCallId: "tool-1", title: "echo hello", kind: "execute", status: "pending" };
-      write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "tool_call", ...toolCall } } });
-      write({
-        id: "perm-1",
-        method: "session/request_permission",
-        params: {
-          sessionId,
-          toolCall,
-          options: [
-            { optionId: "allow", name: "Allow once", kind: "allow_once" },
-            { optionId: "reject", name: "Reject", kind: "reject_once" },
-          ],
-        },
-      });
+      ask(sessionId, "perm-1", "tool-1");
+      if (ending === "completed" || ending === "failed") {
+        ask(sessionId, "perm-2", "tool-2");
+        write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "tool_call_update", toolCallId: "tool-1", status: ending } } });
+        write({ method: "session/update", params: { sessionId, update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "tool terminal" } } } });
+        return;
+      }
       setTimeout(() => {
-        if (ending === "cancel") {
+        if (ending === "cancel" || ending === "outside_cancel") {
           write({ id: message.id, result: { stopReason: "cancelled" } });
+        } else if (ending === "end_turn" || ending === "outside_turn") {
+          write({ id: message.id, result: { stopReason: "end_turn" } });
         } else {
           write({ id: message.id, error: { code: -32603, message: "turn failed with a permission open" } });
         }
       }, 50);
       return;
     }
-    const state = permissionAnswer === null ? "unanswered" : "answered: " + JSON.stringify(permissionAnswer);
+    const answer = ending.startsWith("outside_") ? outsidePermissionAnswer : permissionAnswer;
+    const requestId = ending.startsWith("outside_") ? "perm-outside" : "perm-1";
+    const state = answer === null ? "unanswered" : "answered: " + JSON.stringify(answer);
     write({
       method: "session/update",
       params: {
         sessionId,
-        update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: "perm-1 " + state } },
+        update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: requestId + " " + state } },
       },
     });
     write({ id: message.id, result: { stopReason: "end_turn" } });

packages/server/src/server/plugins/plugin-process.ts

@@ -139,7 +139,7 @@ export function createPluginWorker(options: {
       registerSubscription(contract, handler as PluginSubscriptionHandler);
       return;
     }
-    register(contract, (input, context) => handler(contract.input.parse(input), context as never));
+    register(contract, (input, context) => handler(input, context as never));
   }) as PluginServerContext["handle"];
 
   function register(contract: PluginRpcContract, handler: RpcHandler): void {
@@ -593,7 +593,6 @@ export function createPluginWorker(options: {
         if (!paseo) throw new Error("Plugin Paseo API is unavailable");
         return registered.handler(input, { paseo });
       })
-      .then((output) => registered.contract.output.parseAsync(output))
       .then((output: unknown) => {
         assertPluginWireValue(output, "output");
         return output;

packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts

@@ -8,6 +8,9 @@ import {
   type WebSocketLike,
 } from "@getpaseo/client/internal/daemon-client";
 import { afterEach, expect, test } from "vitest";
+import { defineRpc } from "@getpaseo/plugin";
+import { callPluginRpc } from "@getpaseo/plugin/client/host";
+import { z } from "zod";
 import { resolveDaemonVersion } from "../daemon-version.js";
 import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js";
 
@@ -39,6 +42,13 @@ const repeated = defineSubscription({
   input: z.object({ times: z.string().transform(Number) }),
   event: z.string().transform((text) => text.length),
 });
+const transformed = defineRpc({
+  name: "transformed",
+  input: z.object({ when: z.string().transform(async (text) => new Date(text)) }),
+  output: z.object({ n: z.string().transform(async (text) => Number(text)) }),
+});
+const defaulted = defineRpc({ name: "defaulted", input: z.string().default("x"), output: z.string() });
+const defaultedOutput = defineRpc({ name: "defaulted-output", input: z.object({}), output: z.string().default("x") });
 const watchers = defineRpc({
   name: "watchers",
   input: z.object({}),
@@ -81,6 +91,9 @@ export default function contribute(server: PluginServerContext) {
   server.handle(repeated, async ({ times }, { push }) => {
     await push(typeof times === "number" ? "x".repeat(times) : "input arrived untransformed");
   });
+  server.handle(transformed, ({ when }) => ({ n: String(when.getUTCFullYear()) }));
+  server.handle(defaulted, (input) => input);
+  server.handle(defaultedOutput, () => undefined);
   server.handle(watchers, () => ({ starts, active }));
   server.handle(thumbnail, ({ size }) => Buffer.alloc(size, 7));
   server.handle(nestedOutput, () => ({ data: new Uint8Array(1) }));
@@ -352,3 +365,64 @@ export default function contribute(server: PluginServerContext) {
     await harness.close();
   }
 }, 60_000);
+
+test("plain RPC parses async input and output transforms once at the receiver", async () => {
+  const transformed = defineRpc({
+    name: "transformed",
+    input: z.object({ when: z.string().transform(async (text) => new Date(text)) }),
+    output: z.object({ n: z.string().transform(async (text) => Number(text)) }),
+  });
+
+  const harness = await startHarness();
+  try {
+    await expect(
+      callPluginRpc(
+        transformed,
+        (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input),
+        { when: "2026-01-01T00:00:00.000Z" },
+      ),
+    ).resolves.toEqual({ n: 2026 });
+  } finally {
+    await harness.close();
+  }
+});
+
+test("plain RPC applies a top-level input default in the plugin process", async () => {
+  const contract = defineRpc({
+    name: "defaulted",
+    input: z.string().default("x"),
+    output: z.string(),
+  });
+  const harness = await startHarness();
+  try {
+    await expect(
+      callPluginRpc(
+        contract,
+        (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input),
+        undefined,
+      ),
+    ).resolves.toBe("x");
+  } finally {
+    await harness.close();
+  }
+});
+
+test("plain RPC applies a top-level output default on the client", async () => {
+  const contract = defineRpc({
+    name: "defaulted-output",
+    input: z.object({}),
+    output: z.string().default("x"),
+  });
+  const harness = await startHarness();
+  try {
+    await expect(
+      callPluginRpc(
+        contract,
+        (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input),
+        {},
+      ),
+    ).resolves.toBe("x");
+  } finally {
+    await harness.close();
+  }
+}, 30_000);

packages/server/src/server/plugins/runtime.posix.test.ts

@@ -1853,7 +1853,7 @@ export function inspect() { void Surface; return {}; }`,
     await runtime.stopAll();
   });
 
-  it("rejects a handler result that does not match its RPC output schema", async () => {
+  it("forwards a handler result for the client to validate against its RPC output schema", async () => {
     const directory = await createPlugin(
       "invalid-output",
       `import { z } from "zod";
@@ -1872,7 +1872,9 @@ export default function contribute(plugin: any) {
 
     await runtime.startPlugin("invalid-output", directory);
 
-    await expect(runtime.invoke("invalid-output", "broken", {})).rejects.toThrow();
+    await expect(runtime.invoke("invalid-output", "broken", {})).resolves.toEqual({
+      value: "wrong",
+    });
     await runtime.stopAll();
   });
 

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

@@ -558,34 +558,31 @@ describe("runWorktreeSetupInBackground", () => {
   });
 
   test("runs setup from an exact workspace subdirectory", async () => {
-    const { tempDir, repoDir } = createGitRepo();
+    const tempDir = realpathSync.native(mkdtempSync(path.join(tmpdir(), "worktree-setup-test-")));
     cleanupPaths.push(tempDir);
+    const repoDir = path.join(tempDir, "source");
     const sourceWorkspaceCwd = path.join(repoDir, "packages", "app");
+    const worktree: WorktreeConfig = {
+      branchName: "feature-subdirectory-setup",
+      worktreePath: path.join(tempDir, "worktree"),
+    };
+    const workspaceCwd = path.join(worktree.worktreePath, "packages", "app");
+    // Setup consumes an existing workspace. Git creation is covered by the workflow tests.
     mkdirSync(sourceWorkspaceCwd, { recursive: true });
+    mkdirSync(workspaceCwd, { recursive: true });
+    execFileSync("git", ["init", "-b", worktree.branchName], {
+      cwd: worktree.worktreePath,
+      stdio: "pipe",
+    });
     writeFileSync(
-      path.join(sourceWorkspaceCwd, "paseo.json"),
+      path.join(workspaceCwd, "paseo.json"),
       JSON.stringify({
         worktree: {
           setup: ["pwd > setup-cwd.txt"],
         },
       }),
     );
-    execFileSync("git", ["add", "."], { cwd: repoDir, stdio: "pipe" });
-    execFileSync("git", ["-c", "commit.gpgsign=false", "commit", "-m", "add app setup"], {
-      cwd: repoDir,
-      stdio: "pipe",
-    });
-
     const paseoHome = path.join(tempDir, ".paseo");
-    const createdWorktree = await createLegacyWorktreeForTest({
-      branchName: "feature-subdirectory-setup",
-      cwd: repoDir,
-      baseBranch: "main",
-      worktreeSlug: "feature-subdirectory-setup",
-
-      paseoHome,
-    });
-    const workspaceCwd = path.join(createdWorktree.worktreePath, "packages", "app");
 
     await runWorktreeSetupInBackground(
       {
@@ -600,16 +597,16 @@ describe("runWorktreeSetupInBackground", () => {
         requestCwd: sourceWorkspaceCwd,
         repoRoot: repoDir,
         workspaceId: "ws-subdirectory-setup",
-        worktree: createdWorktree,
+        worktree,
         shouldBootstrap: true,
         slug: "feature-subdirectory-setup",
-        worktreePath: createdWorktree.worktreePath,
+        worktreePath: worktree.worktreePath,
         workspaceCwd,
       },
     );
 
     expect(existsSync(path.join(workspaceCwd, "setup-cwd.txt"))).toBe(true);
-    expect(existsSync(path.join(createdWorktree.worktreePath, "setup-cwd.txt"))).toBe(false);
+    expect(existsSync(path.join(worktree.worktreePath, "setup-cwd.txt"))).toBe(false);
   });
 
   test("emits running then completed snapshots for no-setup workspaces without auto-starting scripts", async () => {

public-docs/plugins/reference.md

@@ -2222,7 +2222,7 @@ export default function contribute(server: PluginServerContext) {
 }
 ```
 
-Inputs and outputs are validated on both sides. RPC names start with a lowercase letter and contain lowercase letters, numbers, dots, hyphens, or underscores. `useRpc()` returns a typed async function. Use TanStack Query for request state, caching, and mutations.
+Inputs are parsed once in the plugin process; outputs are parsed once on the client. Send schema inputs from the client and return output-schema inputs from the handler; schema transforms run at the receiving side. RPC names start with a lowercase letter and contain lowercase letters, numbers, dots, hyphens, or underscores. `useRpc()` returns a typed async function. Use TanStack Query for request state, caching, and mutations.
 
 Backend handlers receive the same `PaseoApi` as `{ paseo }`. Their connection belongs to the subprocess and closes when the plugin stops. It does not subscribe to timelines or catalog events until plugin code subscribes. Follow the [SDK event contract](/docs/sdk/events) for cleanup and timeline replacements. Backend code can use Node APIs and dependencies installed in the plugin directory.