fix(server): withdraw aborted Claude questions#22803

@@ -0,0 +1,648 @@
+// @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalDate:off globalFetch:off globalConsole:off preferSchemaOverJson:off - Host-side evidence driver retains raw wire evidence and owns deadline timers.
+// Runtime evidence for withdrawn Claude AskUserQuestion requests. Starts a real
+// server from --worktree against an isolated --home, backs the claudeAgent
+// instance with testFixtures/claudeQuestionMockCli.mjs, and records what the
+// database, the subscribeThread stream and the fixture saw for each row:
+//   node apps/server/scripts/claude-question-withdrawal-evidence.ts \
+//     --worktree <tree> --home /tmp/<rev>/home --out /tmp/<rev> --row all
+import * as NodeAssert from "node:assert/strict";
+import * as NodeChildProcess from "node:child_process";
+import * as NodeCrypto from "node:crypto";
+import * as NodeFS from "node:fs";
+import * as NodeHttp from "node:http";
+import * as NodeOS from "node:os";
+import * as NodePath from "node:path";
+import * as NodeSqlite from "node:sqlite";
+import * as NodeUtil from "node:util";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import * as Stream from "effect/Stream";
+import { Socket } from "effect/socket";
+import { RpcClient, RpcSerialization } from "effect/rpc";
+import {
+  ORCHESTRATION_PROTOCOL_VERSION,
+  ORCHESTRATION_PROTOCOL_HEADER,
+  CommandId,
+  ProjectId,
+  ProviderInstanceId,
+  RunId,
+  RuntimeRequestId,
+  ThreadId,
+  WsRpcGroup,
+  type OrchestrationV2ThreadStreamItem,
+} from "@t3tools/contracts";
+
+const ROWS = ["A1", "A2", "A3", "A4", "B", "C"] as const;
+type Row = (typeof ROWS)[number];
+const { values } = NodeUtil.parseArgs({
+  options: {
+    worktree: { type: "string" },
+    home: { type: "string" },
+    out: { type: "string" },
+    row: { type: "string", default: "all" },
+    timeout: { type: "string", default: "60" },
+  },
+});
+NodeAssert.ok(
+  values.worktree && values.home && values.out,
+  "--worktree, --home and --out are required",
+);
+const worktree = NodePath.resolve(values.worktree);
+const home = NodePath.resolve(values.home);
+const out = NodePath.resolve(values.out);
+NodeAssert.ok(
+  !home.startsWith(NodePath.join(NodeOS.homedir(), ".t3")),
+  "--home must not point at the live T3 home",
+);
+const rows: ReadonlyArray<Row> =
+  values.row === "all"
+    ? ROWS
+    : [ROWS.find((row) => row === values.row) ?? NodeAssert.fail("bad --row")];
+const timeoutMs = Number(values.timeout) * 1000;
+const fixture = NodePath.resolve(
+  import.meta.dirname,
+  "../src/provider/testFixtures/claudeQuestionMockCli.mjs",
+);
+const traces = NodePath.join(out, "traces");
+
+NodeFS.rmSync(home, { recursive: true, force: true });

@@ -0,0 +1,248 @@
+#!/usr/bin/env node
+// Stand-in for the Claude Code CLI speaking the Agent SDK's stream-json control
+// protocol. It asks one AskUserQuestion and then follows the row named in the
+// first prompt (`ROW:<id> TRACE:<dir>`). Driven by
+// apps/server/scripts/claude-question-withdrawal-evidence.ts.
+import * as NodeCrypto from "node:crypto";
+import * as NodeFS from "node:fs";
+import * as NodePath from "node:path";
+import * as NodeReadline from "node:readline";
+
+if (process.argv.includes("--version")) {
+  process.stdout.write("2.1.0 (Claude Code)\n");
+  process.exit(0);
+}
+
+const sessionId = NodeCrypto.randomUUID();
+const model = "claude-sonnet-4-6";
+const usage = { input_tokens: 1, output_tokens: 1 };
+const questions = [
+  {
+    header: "Color",
+    question: "Red or Blue?",
+    options: [
+      { label: "Red", description: "Choose red." },
+      { label: "Blue", description: "Choose blue." },
+    ],
+    multiSelect: false,
+  },
+];
+
+let row = null;
+let traceFile = null;
+let traceDir = null;
+const pendingTrace = [];
+const trace = (entry) => {
+  const line = JSON.stringify({ t: performance.timeOrigin + performance.now(), ...entry }) + "\n";
+  if (traceFile) NodeFS.appendFileSync(traceFile, line);
+  else pendingTrace.push(line);
+};
+const write = (frame) => {
+  trace({ dir: "out", frame });
+  process.stdout.write(`${JSON.stringify(frame)}\n`);
+};
+const exit = (code) => {
+  trace({ event: "exit", code });
+  process.exit(code);
+};
+process.on("SIGTERM", () => exit(0));
+
+const waiters = new Map();
+const waitForResponse = (requestId) => new Promise((resolve) => waiters.set(requestId, resolve));
+const waitForFile = (path) =>
+  new Promise((resolve) => {
+    const poll = setInterval(() => {
+      if (NodeFS.existsSync(path)) {
+        clearInterval(poll);
+        resolve();
+      }
+    }, 50);
+  });
+
+const initializeResponse = {
+  commands: [],
+  agents: [],
+  models: [],
+  output_style: "default",
+  available_output_styles: ["default"],
+  account: { email: "fixture@example.invalid", subscriptionType: "max", tokenSource: "fixture" },
+};
+
+function ask(n) {
+  const toolUseId = `toolu_q${n}`;
+  const requestId = `cu_q${n}`;
+  write({
+    type: "assistant",
+    parent_tool_use_id: null,
+    session_id: sessionId,
+    uuid: NodeCrypto.randomUUID(),
+    message: {
+      id: `msg_q${n}`,
+      type: "message",
+      role: "assistant",
+      model,
+      content: [{ type: "tool_use", id: toolUseId, name: "AskUserQuestion", input: { questions } }],
+      stop_reason: "tool_use",
+      stop_sequence: null,
+      usage,
+    },
+  });
+  const response = waitForResponse(requestId);
+  write({
+    type: "control_request",
+    request_id: requestId,
+    request: {
+      subtype: "can_use_tool",
+      tool_name: "AskUserQuestion",
+      input: { questions },
+      tool_use_id: toolUseId,
+      permission_suggestions: [],
+    },
+  });
+  return { toolUseId, requestId, response };
+}
+
+function finish(toolUseId, decision) {
+  const answers = decision?.updatedInput?.answers ?? {};
+  write({
+    type: "user",
+    parent_tool_use_id: null,
+    session_id: sessionId,
+    uuid: NodeCrypto.randomUUID(),
+    message: {
+      role: "user",
+      content: [
+        {
+          type: "tool_result",
+          tool_use_id: toolUseId,
+          content:
+            decision?.behavior === "allow"
+              ? `User answered: ${JSON.stringify(answers)}`
+              : String(decision?.message ?? "denied"),
+        },
+      ],
+    },
+  });
+  const text = `Answer received: ${Object.values(answers).join(", ") || "none"}`;
+  write({
+    type: "assistant",
+    parent_tool_use_id: null,
+    session_id: sessionId,
+    uuid: NodeCrypto.randomUUID(),
+    message: {
+      id: "msg_done",
+      type: "message",
+      role: "assistant",
+      model,
+      content: [{ type: "text", text }],
+      stop_reason: "end_turn",
+      stop_sequence: null,
+      usage,
+    },
+  });
+  write({
+    type: "result",
+    subtype: "success",
+    is_error: false,
+    num_turns: 1,
+    result: text,
+    session_id: sessionId,
+    stop_reason: "end_turn",
+    usage,
+    total_cost_usd: 0,
+    duration_ms: 1,
+    duration_api_ms: 1,
+    uuid: NodeCrypto.randomUUID(),
+  });
+}
+
+async function runTurn(prompt) {
+  write({
+    type: "system",
+    subtype: "init",
+    session_id: sessionId,
+    cwd: process.cwd(),
+    tools: ["AskUserQuestion"],
+    model,
+    permissionMode: "bypassPermissions",
+    mcp_servers: [],
+    slash_commands: [],
+    uuid: NodeCrypto.randomUUID(),
+  });
+  if (typeof prompt.uuid === "string") {
+    write({ type: "command_lifecycle", command_uuid: prompt.uuid, state: "started" });
+  }
+  const q1 = ask(1);
+  switch (row) {
+    case "A1":
+    case "A2":
+      // The adapter interrupts then closes the query; the SDK's cleanup aborts
+      // the pending callback, so this process only waits to be shut down.
+      return;
+    case "A3":
+      await waitForFile(NodePath.join(traceDir, `${row}.die`));
+      process.stderr.write("fatal: fixture transport failure\n");
+      exit(1);
+      return;
+    case "A4": {
+      await waitForFile(NodePath.join(traceDir, `${row}.cancel`));
+      write({ type: "control_cancel_request", request_id: q1.requestId });
+      await q1.response;
+      const q2 = ask(2);
+      finish(q2.toolUseId, await q2.response);
+      return;
+    }
+    case "C": {
+      const decision = await q1.response;
+      // Stale: the SDK already settled this request when it sent the answer.
+      write({ type: "control_cancel_request", request_id: q1.requestId });
+      finish(q1.toolUseId, decision);
+      return;
+    }
+    default:
+      finish(q1.toolUseId, await q1.response);
+  }
+}
+
+for await (const line of NodeReadline.createInterface({ input: process.stdin })) {
+  if (!line.trim()) continue;
+  const frame = JSON.parse(line);
+  if (row === null && frame.type === "user") {
+    const text = JSON.stringify(frame.message?.content ?? "");
+    row = /ROW:([A-Z0-9]+)/.exec(text)?.[1] ?? "B";
+    traceDir = /TRACE:(\S+?)(?:\\n|\\"|"|\s|$)/.exec(text)?.[1] ?? null;

@@ -0,0 +1,648 @@
+// @effect-diagnostics nodeBuiltinImport:off globalTimers:off globalDate:off globalFetch:off globalConsole:off preferSchemaOverJson:off - Host-side evidence driver retains raw wire evidence and owns deadline timers.
+// Runtime evidence for withdrawn Claude AskUserQuestion requests. Starts a real
+// server from --worktree against an isolated --home, backs the claudeAgent
+// instance with testFixtures/claudeQuestionMockCli.mjs, and records what the
+// database, the subscribeThread stream and the fixture saw for each row:
+//   node apps/server/scripts/claude-question-withdrawal-evidence.ts \
+//     --worktree <tree> --home /tmp/<rev>/home --out /tmp/<rev> --row all
+import * as NodeAssert from "node:assert/strict";
+import * as NodeChildProcess from "node:child_process";
+import * as NodeCrypto from "node:crypto";
+import * as NodeFS from "node:fs";
+import * as NodeHttp from "node:http";
+import * as NodeOS from "node:os";
+import * as NodePath from "node:path";
+import * as NodeSqlite from "node:sqlite";
+import * as NodeUtil from "node:util";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+import * as Stream from "effect/Stream";
+import { Socket } from "effect/socket";
+import { RpcClient, RpcSerialization } from "effect/rpc";
+import {
+  ORCHESTRATION_PROTOCOL_VERSION,
+  ORCHESTRATION_PROTOCOL_HEADER,
+  CommandId,
+  ProjectId,
+  ProviderInstanceId,
+  RunId,
+  RuntimeRequestId,
+  ThreadId,
+  WsRpcGroup,
+  type OrchestrationV2ThreadStreamItem,
+} from "@t3tools/contracts";
+
+const ROWS = ["A1", "A2", "A3", "A4", "B", "C"] as const;
+type Row = (typeof ROWS)[number];
+const { values } = NodeUtil.parseArgs({
+  options: {
+    worktree: { type: "string" },
+    home: { type: "string" },
+    out: { type: "string" },
+    row: { type: "string", default: "all" },
+    timeout: { type: "string", default: "60" },
+  },
+});
+NodeAssert.ok(
+  values.worktree && values.home && values.out,
+  "--worktree, --home and --out are required",
+);
+const worktree = NodePath.resolve(values.worktree);
+const home = NodePath.resolve(values.home);
+const out = NodePath.resolve(values.out);
+NodeAssert.ok(
+  !home.startsWith(NodePath.join(NodeOS.homedir(), ".t3")),
+  "--home must not point at the live T3 home",
+);

@@ -0,0 +1,248 @@
+#!/usr/bin/env node
+// Stand-in for the Claude Code CLI speaking the Agent SDK's stream-json control
+// protocol. It asks one AskUserQuestion and then follows the row named in the
+// first prompt (`ROW:<id> TRACE:<dir>`). Driven by
+// apps/server/scripts/claude-question-withdrawal-evidence.ts.
+import * as NodeCrypto from "node:crypto";
+import * as NodeFS from "node:fs";
+import * as NodePath from "node:path";
+import * as NodeReadline from "node:readline";
+
+if (process.argv.includes("--version")) {
+  process.stdout.write("2.1.0 (Claude Code)\n");
+  process.exit(0);
+}
+
+const sessionId = NodeCrypto.randomUUID();
+const model = "claude-sonnet-4-6";
+const usage = { input_tokens: 1, output_tokens: 1 };
+const questions = [
+  {
+    header: "Color",
+    question: "Red or Blue?",
+    options: [
+      { label: "Red", description: "Choose red." },
+      { label: "Blue", description: "Choose blue." },
+    ],
+    multiSelect: false,
+  },
+];
+
+let row = null;
+let traceFile = null;
+let traceDir = null;
+const pendingTrace = [];
+const trace = (entry) => {
+  const line = JSON.stringify({ t: performance.timeOrigin + performance.now(), ...entry }) + "\n";
+  if (traceFile) NodeFS.appendFileSync(traceFile, line);
+  else pendingTrace.push(line);
+};
+const write = (frame) => {
+  trace({ dir: "out", frame });
+  process.stdout.write(`${JSON.stringify(frame)}\n`);
+};
+const exit = (code) => {
+  trace({ event: "exit", code });
+  process.exit(code);
+};
+process.on("SIGTERM", () => exit(0));
+
+const waiters = new Map();
+const waitForResponse = (requestId) => new Promise((resolve) => waiters.set(requestId, resolve));
+const waitForFile = (path) =>
+  new Promise((resolve) => {
+    const poll = setInterval(() => {
+      if (NodeFS.existsSync(path)) {
+        clearInterval(poll);
+        resolve();
+      }
+    }, 50);
+  });
+
+const initializeResponse = {
+  commands: [],
+  agents: [],
+  models: [],
+  output_style: "default",
+  available_output_styles: ["default"],
+  account: { email: "fixture@example.invalid", subscriptionType: "max", tokenSource: "fixture" },
+};
+
+function ask(n) {
+  const toolUseId = `toolu_q${n}`;
+  const requestId = `cu_q${n}`;
+  write({
+    type: "assistant",
+    parent_tool_use_id: null,
+    session_id: sessionId,
+    uuid: NodeCrypto.randomUUID(),
+    message: {
+      id: `msg_q${n}`,
+      type: "message",
+      role: "assistant",
+      model,
+      content: [{ type: "tool_use", id: toolUseId, name: "AskUserQuestion", input: { questions } }],
+      stop_reason: "tool_use",
+      stop_sequence: null,
+      usage,
+    },
+  });
+  const response = waitForResponse(requestId);
+  write({
+    type: "control_request",
+    request_id: requestId,
+    request: {
+      subtype: "can_use_tool",
+      tool_name: "AskUserQuestion",
+      input: { questions },
+      tool_use_id: toolUseId,
+      permission_suggestions: [],
+    },
+  });
+  return { toolUseId, requestId, response };
+}
+
+function finish(toolUseId, decision) {
+  const answers = decision?.updatedInput?.answers ?? {};
+  write({
+    type: "user",
+    parent_tool_use_id: null,
+    session_id: sessionId,
+    uuid: NodeCrypto.randomUUID(),
+    message: {
+      role: "user",
+      content: [
+        {
+          type: "tool_result",
+          tool_use_id: toolUseId,
+          content:
+            decision?.behavior === "allow"
+              ? `User answered: ${JSON.stringify(answers)}`
+              : String(decision?.message ?? "denied"),
+        },
+      ],
+    },
+  });
+  const text = `Answer received: ${Object.values(answers).join(", ") || "none"}`;
+  write({
+    type: "assistant",
+    parent_tool_use_id: null,
+    session_id: sessionId,
+    uuid: NodeCrypto.randomUUID(),
+    message: {
+      id: "msg_done",
+      type: "message",
+      role: "assistant",
+      model,
+      content: [{ type: "text", text }],
+      stop_reason: "end_turn",
+      stop_sequence: null,
+      usage,
+    },
+  });
+  write({
+    type: "result",
+    subtype: "success",
+    is_error: false,
+    num_turns: 1,
+    result: text,
+    session_id: sessionId,
+    stop_reason: "end_turn",
+    usage,
+    total_cost_usd: 0,
+    duration_ms: 1,
+    duration_api_ms: 1,
+    uuid: NodeCrypto.randomUUID(),
+  });
+}
+
+async function runTurn(prompt) {
+  write({
+    type: "system",
+    subtype: "init",
+    session_id: sessionId,
+    cwd: process.cwd(),
+    tools: ["AskUserQuestion"],
+    model,
+    permissionMode: "bypassPermissions",
+    mcp_servers: [],
+    slash_commands: [],
+    uuid: NodeCrypto.randomUUID(),
+  });
+  if (typeof prompt.uuid === "string") {
+    write({ type: "command_lifecycle", command_uuid: prompt.uuid, state: "started" });
+  }
+  const q1 = ask(1);
+  switch (row) {
+    case "A1":
+    case "A2":
+      // The adapter interrupts then closes the query; the SDK's cleanup aborts
+      // the pending callback, so this process only waits to be shut down.
+      return;
+    case "A3":
+      await waitForFile(NodePath.join(traceDir, `${row}.die`));
+      process.stderr.write("fatal: fixture transport failure\n");
+      exit(1);
+      return;
+    case "A4": {
+      await waitForFile(NodePath.join(traceDir, `${row}.cancel`));
+      write({ type: "control_cancel_request", request_id: q1.requestId });
+      await q1.response;
+      const q2 = ask(2);
+      finish(q2.toolUseId, await q2.response);
+      return;
+    }
+    case "C": {
+      const decision = await q1.response;
+      // Stale: the SDK already settled this request when it sent the answer.
+      write({ type: "control_cancel_request", request_id: q1.requestId });
+      finish(q1.toolUseId, decision);
+      return;
+    }
+    default:
+      finish(q1.toolUseId, await q1.response);
+  }
+}