fix: keep large approval batches observable

src/bridge.ts

@@ -37,6 +37,8 @@ export const APPROVAL_ONBOARDING_EXAMPLES = [
   "先列出待办，我确认后同意指定的几张",
 ] as const;
 
+const REMOTE_WRITE_TIMEOUT_MS = 45_000;
+
 class AuthRequired extends Error {
   constructor(readonly challenge: NonNullable<BridgeState["device"]>) { super("需要钉钉登录"); }
 }
@@ -83,7 +85,14 @@ export class RemoteBridge {
     return this.withCallSlot(async () => {
       try {
         const client = await this.remoteClient();
-        return this.appendOnboarding(await client.callTool({ name, arguments: args }));
+        try {
+          const options = name === "approval_process_batch" ? { timeout: REMOTE_WRITE_TIMEOUT_MS } : undefined;
+          return this.appendOnboarding(await client.callTool({ name, arguments: args }, undefined, options));
+        } catch (error) {
+          const uncertain = bridgeWriteOutcomeUncertainResult(name, args);
+          if (uncertain) return uncertain;
+          throw error;
+        }
       } catch (error) {
         if (error instanceof AuthRequired) return this.authResult(error.challenge);
         throw error;
@@ -362,6 +371,40 @@ export function bridgeAuthToolResult(result: BridgeLoginResult) {
   return { content, structuredContent: { result } };
 }
 
+export function bridgeWriteOutcomeUncertainResult(name: string, args: Record<string, unknown>) {
+  if (name !== "approval_process_batch") return undefined;
+  const requestedItems = Array.isArray(args.items)
+    ? args.items.filter((item): item is Record<string, unknown> => Boolean(item) && typeof item === "object" && !Array.isArray(item))
+    : [];
+  const selectionUsed = requestedItems.length === 0 && typeof args.selectionId === "string";
+  const message = "审批写请求超过 bridge 等待时间或连接中断，操作可能已经到达网关；禁止自动重试，请重新列出实时待办核对。";
+  const results = requestedItems.map((item, index) => ({
+    index: index + 1,
+    ...(typeof item.approvalId === "string" ? { approvalId: item.approvalId } : {}),
+    status: "outcome_uncertain",
+    outcomeUncertain: true,
+    success: false,
+    message,
+  }));
+  const result = {
+    action: args.action,
+    status: "outcome_uncertain",
+    outcomeUncertain: true,
+    total: selectionUsed ? null : requestedItems.length,
+    uncertain: selectionUsed ? null : requestedItems.length,
+    ...(selectionUsed ? { targetCount: "unknown" } : {}),
+    succeeded: 0,
+    failed: 0,
+    results,
+    requiresRelist: true,
+    retryable: false,
+  };
+  return {
+    content: [{ type: "text" as const, text: `${message}\n${JSON.stringify(result)}` }],
+    structuredContent: { result },
+  };
+}
+
 function isAuthenticationFailure(error: unknown): boolean {
   const record = error && typeof error === "object" ? error as Record<string, unknown> : {};
   const status = Number(record.status ?? record.statusCode ?? record.code);

src/xft-open-api.ts

@@ -250,13 +250,27 @@ export class XftOpenApiClient implements XftApprovalProvider {
     items: XftApprovalBatchItem[],
   ) {
     const identity = await this.resolveIdentity(user);
-    const results: XftApprovalBatchResultItem[] = [];
-    for (const item of items) {
+    const results: Array<XftApprovalBatchResultItem | undefined> = Array.from({ length: items.length }, () => undefined);
+    const candidates: Array<{ index: number; item: XftApprovalBatchItem }> = [];
+    for (const [index, item] of items.entries()) {
+      if (action === "return" && item.targetNodeId !== "OFFICIAL_DEFAULT") {
+        results[index] = {
+          billId: item.billId,
+          status: "failed",
+          success: false,
+          failureConfirmed: true,
+          error: "薪福通正式接口仅支持按当前流程规则退回，请重新获取可执行动作",
+        };
+      } else candidates.push({ index, item });
+    }
+
+    const accepted: Array<{ index: number; item: XftApprovalBatchItem }> = [];
+    for (const { index, item } of candidates) {
+      if (results[index]) continue;
       let writeStarted = false;
       try {
-        if (action === "return" && item.targetNodeId !== "OFFICIAL_DEFAULT") {
-          throw new ApprovalSafetyError("薪福通正式接口仅支持按当前流程规则退回，请重新获取可执行动作");
-        }
+        // Keep ownership validation immediately adjacent to every write. Only
+        // the slow post-write transition polling is shared by the batch.
         await this.requireCurrentTodo(identity, item.billId);
         writeStarted = true;
         await this.request(ACTION_ENDPOINTS[action], {
@@ -267,47 +281,67 @@ export class XftOpenApiClient implements XftApprovalProvider {
           }],
           empId: identity.enterpriseUserId,
         }, "POST", identity);
-        const stillPending = await this.waitForTodoTransition(identity, item.billId);
-        if (stillPending) {
-          results.push({
+        accepted.push({ index, item });
+      } catch (error) {
+        results[index] = writeStarted && isUncertainXftWriteError(error) ? {
+          billId: item.billId,
+          status: "outcome_uncertain",
+          success: false,
+          outcomeUncertain: true,
+          error: `${errorMessage(error)}；写请求可能已经到达薪福通，禁止自动重试`,
+        } : {
+          billId: item.billId,
+          status: "failed",
+          success: false,
+          failureConfirmed: true,
+          error: errorMessage(error),
+        };
+      }
+    }
+
+    if (accepted.length > 0) {
+      try {
+        const stillPending = await this.waitForTodoTransitions(
+          identity,
+          new Set(accepted.map(({ item }) => item.billId)),
+        );
+        for (const { index, item } of accepted) {
+          results[index] = stillPending.has(item.billId) ? {
             billId: item.billId,
             status: "outcome_uncertain",
             success: false,
             outcomeUncertain: true,
             error: "薪福通已接收操作，但单据仍在该用户实时待办中；禁止自动重试，请稍后重新查询待办",
-          });
-          continue;
+          } : { billId: item.billId, status: "succeeded", success: true };
         }
-        results.push({ billId: item.billId, status: "succeeded", success: true });
       } catch (error) {
-        if (writeStarted && isUncertainXftWriteError(error)) {
-          results.push({
-            billId: item.billId,
-            status: "outcome_uncertain",
-            success: false,
-            outcomeUncertain: true,
-            error: `${errorMessage(error)}；写请求可能已经到达薪福通，禁止自动重试`,
-          });
-        } else {
-          results.push({
-            billId: item.billId,
-            status: "failed",
-            success: false,
-            failureConfirmed: true,
-            error: errorMessage(error),
-          });
-        }
+        for (const { index, item } of accepted) results[index] = {
+          billId: item.billId,
+          status: "outcome_uncertain",
+          success: false,
+          outcomeUncertain: true,
+          error: `薪福通已接收操作，但实时状态复核失败：${errorMessage(error)}；禁止自动重试`,
+        };
       }
     }
+
+    const completed = results.map((result, index) => result ?? ({
+      billId: items[index]!.billId,
+      status: "failed" as const,
+      success: false as const,
+      failureConfirmed: true as const,
+      error: "薪福通未返回该单据的处理结果",
+    }));
     return {
       action,
-      total: results.length,
-      succeeded: results.filter((item) => item.success).length,
-      failed: results.filter((item) => !item.success).length,
-      results,
+      total: completed.length,
+      succeeded: completed.filter((item) => item.success).length,
+      failed: completed.filter((item) => !item.success).length,
+      results: completed,
       diagnostics: {
         source: "xft_official_openapi",
         realtimeOwnershipValidation: true,
+        batchedPostWriteVerification: true,
         transactional: false,
         persistedBusinessData: false,
       },
@@ -966,12 +1000,24 @@ export class XftOpenApiClient implements XftApprovalProvider {
     return item;
   }
 
-  private async waitForTodoTransition(identity: XftIdentity, billId: string): Promise<CurrentTodo | undefined> {
+  private async findCurrentTodoIds(identity: XftIdentity, billIds: Set<string>): Promise<Set<string>> {
+    const found = new Set<string>();
+    if (billIds.size === 0) return found;
+    const pageSize = 1000;
+    for (let currentPage = 1; currentPage <= 100; currentPage += 1) {
+      const page = await this.loadTodoPage(identity, currentPage, pageSize);
+      for (const item of page.items) if (billIds.has(item.billId)) found.add(item.billId);
+      if (found.size === billIds.size || page.pageRecordCount < pageSize) return found;
+    }
+    throw new ApprovalSafetyError("薪福通待办超过安全分页上限，无法确认单据归属");
+  }
+
+  private async waitForTodoTransitions(identity: XftIdentity, billIds: Set<string>): Promise<Set<string>> {
     const deadline = Date.now() + 20_000;
-    let current = await this.findCurrentTodo(identity, billId);
-    while (current && Date.now() < deadline) {
+    let current = await this.findCurrentTodoIds(identity, billIds);
+    while (current.size > 0 && Date.now() < deadline) {
       await new Promise<void>((resolve) => setTimeout(resolve, 1_000));
-      current = await this.findCurrentTodo(identity, billId);
+      current = await this.findCurrentTodoIds(identity, billIds);
     }
     return current;
   }

test/bridge.test.ts

@@ -6,10 +6,49 @@ import test from "node:test";
 import { Client } from "@modelcontextprotocol/sdk/client/index.js";
 import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js";
 import {
-  APPROVAL_ONBOARDING_EXAMPLES, RemoteBridge, bridgeAuthToolResult, createBridgeMcpServer,
+  APPROVAL_ONBOARDING_EXAMPLES, RemoteBridge, bridgeAuthToolResult, bridgeWriteOutcomeUncertainResult, createBridgeMcpServer,
   syncBundledSkillForLegacyUpdate,
 } from "../src/bridge.js";
 
+test("bridge write timeout fallback remains structured and forbids retries", () => {
+  const result = bridgeWriteOutcomeUncertainResult("approval_process_batch", {
+    action: "approve",
+    items: [{ approvalId: "approval-one" }, { approvalId: "approval-two" }],
+  });
+  assert.equal(result?.structuredContent.result.status, "outcome_uncertain");
+  assert.equal(result?.structuredContent.result.uncertain, 2);
+  assert.equal(result?.structuredContent.result.retryable, false);
+  assert.equal(result?.structuredContent.result.requiresRelist, true);
+  assert.equal(result?.structuredContent.result.results.length, 2);
+  assert.match(result?.content[0]?.text ?? "", /禁止自动重试/);
+  assert.equal(bridgeWriteOutcomeUncertainResult("approval_list_todos", {}), undefined);
+});
+
+test("stdio bridge converts a remote batch timeout into a structured uncertain outcome", async () => {
+  const root = mkdtempSync(join(tmpdir(), "ding-bridge-write-timeout-"));
+  const statePath = join(root, "session.json");
+  writeFileSync(statePath, JSON.stringify({
+    client: { client_id: "bridge-client" },
+    tokens: { access_token: "fresh-access", savedAt: Date.now(), expires_in: 900 },
+  }));
+  const bridge = new RemoteBridge("https://approval.example.test/mcp", statePath, async () => {
+    throw new Error("network should not be used by the injected client");
+  });
+  (bridge as unknown as { client: { callTool(): Promise<unknown>; close(): Promise<void> } }).client = {
+    async callTool() { throw new Error("Request timed out"); },
+    async close() {},
+  };
+
+  const result = await bridge.call("approval_process_batch", {
+    action: "approve",
+    items: [{ approvalId: "approval-one" }],
+  });
+  assert.equal(result.structuredContent.result.status, "outcome_uncertain");
+  assert.equal(result.structuredContent.result.uncertain, 1);
+  assert.equal(result.structuredContent.result.retryable, false);
+  assert.match(result.content[0].text, /重新列出实时待办核对/);
+});
+
 test("legacy updater post-install syncs the complete bundled Skill", () => {
   const root = mkdtempSync(join(tmpdir(), "ding-bridge-skill-sync-"));
   const bundled = join(root, "package", "skill", "dingtalk-approval");

test/xft-open-api.test.ts

@@ -329,6 +329,59 @@ test("Xft OpenAPI derives empId server-side and validates ownership before and a
   assert.equal(actionBodies[0]?.empId, "EMP-1");
 });
 
+test("Xft approval batches revalidate each write and share one post-write todo scan", async () => {
+  const billIds = Array.from({ length: 20 }, (_, index) => String(10_000_000_100 + index));
+  const pending = new Set(billIds);
+  let todoQueries = 0;
+  let actionCalls = 0;
+  const client = new XftOpenApiClient(config(), async (input, init) => {
+    const path = new URL(String(input)).pathname;
+    const body = init?.body ? decodedBody(init) : {};
+    if (path.includes("user-mapping")) {
+      return encrypted({ code: "0", data: { records: [{ thirdStaffId: "ding-u1", enterpriseUserId: "EMP-1" }] } });
+    }
+    if (path.endsWith("todo-task/page/by-user")) {
+      todoQueries += 1;
+      return encrypted({ code: "0", data: { records: [...pending].map((billId) => ({
+        todoTaskId: `TD-${billId}`, groupId: billId, appCode: "xft-itrip",
+      })) } });
+    }
+    if (path.endsWith("todo-task/detail/by-ids")) {
+      const taskIds = Array.isArray(body.todoTaskIds) ? body.todoTaskIds : [];
+      return encrypted({ code: "0", data: taskIds.map((taskId) => ({
+        todoTaskId: taskId, businessId: String(taskId).replace(/^TD-/, ""), appCode: "xft-itrip",
+      })) });
+    }
+    if (path.endsWith("bill-list")) {
+      const ids = Array.isArray(body.billIds) ? body.billIds.map(String) : [];
+      return encrypted({ code: "0", data: { list: ids.map((billId) => ({ billId, eventType: "COMMON_REIM" })) } });
+    }
+    if (path.endsWith("approve/pass")) {
+      actionCalls += 1;
+      const businessList = Array.isArray(body.businessList) ? body.businessList : [];
+      const businessId = businessList[0] && typeof businessList[0] === "object"
+        ? String((businessList[0] as Record<string, unknown>).businessId)
+        : "";
+      pending.delete(businessId);
+      return encrypted({ code: "0", data: {} });
+    }
+    throw new Error(`unexpected ${path}`);
+  });
+
+  const result = await client.processApprovalBatch(
+    { userId: "ding-u1" },
+    "agree",
+    "同意",
+    billIds.map((billId) => ({ billId })),
+  );
+
+  assert.equal(result.succeeded, 20);
+  assert.equal(result.failed, 0);
+  assert.equal(actionCalls, 20);
+  assert.equal(todoQueries, 21, "each write needs a live ownership scan, followed by one shared post-write scan");
+  assert.deepEqual(result.results.map((item) => item.billId), billIds);
+});
+
 test("Xft submission discovers live templates and derives protected create fields server-side", async () => {
   const createBodies: Record<string, unknown>[] = [];
   const fakeFetch: typeof fetch = async (input, init) => {