From ca95c41fdd3268571010640c31e6ae80731cf240 Mon Sep 17 00:00:00 2001 From: Donald Date: Thu, 20 Aug 2026 02:10:37 +0000 Subject: [PATCH] fix: keep large approval batches observable --- src/bridge.ts | 45 ++++++++++++++- src/xft-open-api.ts | 118 ++++++++++++++++++++++++++------------ test/bridge.test.ts | 41 ++++++++++++- test/xft-open-api.test.ts | 53 +++++++++++++++++ 4 files changed, 219 insertions(+), 38 deletions(-) diff --git a/src/bridge.ts b/src/bridge.ts index 6ba6f0e..1d5ff09 100644 --- a/src/bridge.ts +++ b/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) { 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) { + if (name !== "approval_process_batch") return undefined; + const requestedItems = Array.isArray(args.items) + ? args.items.filter((item): item is Record => 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 : {}; const status = Number(record.status ?? record.statusCode ?? record.code); diff --git a/src/xft-open-api.ts b/src/xft-open-api.ts index d0a4d7a..9466dfc 100644 --- a/src/xft-open-api.ts +++ b/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 = 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 { + private async findCurrentTodoIds(identity: XftIdentity, billIds: Set): Promise> { + const found = new Set(); + 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): Promise> { 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((resolve) => setTimeout(resolve, 1_000)); - current = await this.findCurrentTodo(identity, billId); + current = await this.findCurrentTodoIds(identity, billIds); } return current; } diff --git a/test/bridge.test.ts b/test/bridge.test.ts index 941ea5e..da8b11e 100644 --- a/test/bridge.test.ts +++ b/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; close(): Promise } }).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"); diff --git a/test/xft-open-api.test.ts b/test/xft-open-api.test.ts index c6f379c..749cace 100644 --- a/test/xft-open-api.test.ts +++ b/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).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[] = []; const fakeFetch: typeof fetch = async (input, init) => {