diff --git a/cmd/hwlab-workbench-worker/main.ts b/cmd/hwlab-workbench-worker/main.ts index 8d898452..8bce9ce7 100644 --- a/cmd/hwlab-workbench-worker/main.ts +++ b/cmd/hwlab-workbench-worker/main.ts @@ -13,7 +13,7 @@ const worker = await Worker.create({ namespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue, workflowsPath: path.resolve(import.meta.dir, "../../internal/workbench/workflows.ts"), - activities: createWorkbenchActivities(runtime.application) + activities: createWorkbenchActivities(runtime.application, { activity: runtime.activity!, eventPublisher: runtime.eventPublisher }) }); const healthPort = positivePort(process.env.WORKBENCH_WORKER_HEALTH_PORT, 6678); const health = Bun.serve({ diff --git a/internal/workbench/activities.ts b/internal/workbench/activities.ts index 0c676e45..12504837 100644 --- a/internal/workbench/activities.ts +++ b/internal/workbench/activities.ts @@ -1,12 +1,107 @@ -import type { WorkbenchApplication, WorkbenchTurnInput } from "./contracts.ts"; +import { ApplicationFailure, Context } from "@temporalio/activity"; -export function createWorkbenchActivities(application: WorkbenchApplication) { +import type { WorkbenchActivityConfig, WorkbenchApplication, WorkbenchEventPublisher, WorkbenchTurnInput } from "./contracts.ts"; + +export function createWorkbenchActivities(application: WorkbenchApplication, options: { + activity: WorkbenchActivityConfig; + eventPublisher?: WorkbenchEventPublisher | null; + attempt?: () => number; +}) { return { async dispatchWorkbenchTurn(input: WorkbenchTurnInput) { - return application.dispatchTurn(input); + try { + return await application.dispatchTurn(input); + } catch (error) { + const attempt = positiveInteger(options.attempt?.() ?? currentActivityAttempt(), 1); + const retryable = (error as any)?.retryable !== false; + const exhausted = !retryable || attempt >= options.activity.retryMaximumAttempts; + await publishDispatchFailure(options.eventPublisher, input, error, { + attempt, + maxAttempts: options.activity.retryMaximumAttempts, + retryable, + exhausted, + backoffMs: exhausted ? null : retryDelayMs(options.activity, attempt) + }); + throw ApplicationFailure.create({ + message: errorMessage(error), + type: errorCode(error), + nonRetryable: !retryable, + details: [{ + code: errorCode(error), + layer: (error as any)?.layer ?? null, + status: (error as any)?.status ?? null, + retryable, + attempt, + maxAttempts: options.activity.retryMaximumAttempts, + valuesPrinted: false + }] + }); + } }, async cancelWorkbenchTurn(input: { actor: WorkbenchTurnInput["actor"]; traceId: string; params: Record }) { return application.cancelTurn(input); } }; } + +async function publishDispatchFailure(publisher: WorkbenchEventPublisher | null | undefined, input: WorkbenchTurnInput, error: unknown, state: { attempt: number; maxAttempts: number; retryable: boolean; exhausted: boolean; backoffMs: number | null }) { + if (!publisher) return; + const code = errorCode(error); + const message = userErrorMessage(error); + const createdAt = new Date().toISOString(); + await publisher.publish({ + traceId: input.traceId, + sessionId: input.sessionId, + event: { + sourceEventId: `workbench:${input.traceId}:dispatch-failure:${state.attempt}`, + type: state.exhausted ? "error" : "backend_status", + phase: state.exhausted ? "dispatch-failed" : "dispatch-retry-scheduled", + failureDomain: (error as any)?.layer === "billing" ? "billing" : "infrastructure", + component: "hwlab-cloud-api", + code, + failureKind: code, + summary: message, + message, + retryable: state.retryable, + willRetry: !state.exhausted, + retryPhase: state.exhausted ? "retryExhausted" : "retryScheduled", + attempt: state.attempt, + maxAttempts: state.maxAttempts, + backoffMs: state.backoffMs, + createdAt, + terminal: false, + valuesPrinted: false + } + }); + if (!state.exhausted) return; + await publisher.publish({ + traceId: input.traceId, + sessionId: input.sessionId, + event: { + sourceEventId: `workbench:${input.traceId}:terminal-failed`, + type: "terminal_status", + terminalStatus: "failed", + status: "failed", + failureDomain: (error as any)?.layer === "billing" ? "billing" : "infrastructure", + code, + failureKind: code, + errorCode: code, + message, + retryable: state.retryable, + attempt: state.attempt, + maxAttempts: state.maxAttempts, + createdAt, + terminal: true, + valuesPrinted: false + } + }); +} + +function currentActivityAttempt() { + try { return Context.current().info.attempt; } catch { return 1; } +} +function retryDelayMs(activity: WorkbenchActivityConfig, attempt: number) { return Math.min(activity.retryMaximumIntervalMs, activity.retryInitialIntervalMs * (2 ** Math.max(0, attempt - 1))); } +function positiveInteger(value: unknown, fallback: number) { const parsed = Number(value); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; } +function errorCode(error: unknown) { return String((error as any)?.code ?? "workbench_activity_failed").trim() || "workbench_activity_failed"; } +function errorMessage(error: unknown) { return String((error as any)?.message ?? "Workbench activity failed").trim() || "Workbench activity failed"; } +function userErrorMessage(error: unknown) { return String((error as any)?.userMessage ?? (error as any)?.message ?? "Workbench activity failed").trim() || "Workbench activity failed"; } diff --git a/internal/workbench/cloud-application.ts b/internal/workbench/cloud-application.ts index d3005130..661fe666 100644 --- a/internal/workbench/cloud-application.ts +++ b/internal/workbench/cloud-application.ts @@ -1,10 +1,11 @@ -import type { WorkbenchApplication } from "./contracts.ts"; +import type { WorkbenchApplication, WorkbenchEventPublisher } from "./contracts.ts"; import { normalizeWorkbenchCloudAuthorization } from "./cloud-authorization.ts"; export function createCloudWorkbenchApplication(options: { baseUrl: string; authorization: string; fetchImpl?: typeof fetch; + eventPublisher?: WorkbenchEventPublisher | null; }): WorkbenchApplication { if (!options.baseUrl) throw codedError("workbench_cloud_api_url_required", "WORKBENCH_CLOUD_API_URL is required"); const authorization = normalizeWorkbenchCloudAuthorization(options.authorization); @@ -23,7 +24,16 @@ export function createCloudWorkbenchApplication(options: { body: body ? JSON.stringify(body) : undefined }); const payload = await response.json().catch(() => null) as Record | null; - if (!response.ok || !payload) throw codedError(String((payload?.error as any)?.code ?? "workbench_cloud_api_failed"), String((payload?.error as any)?.message ?? `Cloud API returned ${response.status}`), { status: response.status, route }); + if (!response.ok || !payload) { + const error = payload?.error && typeof payload.error === "object" ? payload.error as Record : {}; + throw codedError(String(error.code ?? "workbench_cloud_api_failed"), String(error.message ?? `Cloud API returned ${response.status}`), { + status: response.status, + route, + layer: error.layer, + retryable: error.retryable, + userMessage: error.userMessage + }); + } return payload; }; return { @@ -38,8 +48,18 @@ export function createCloudWorkbenchApplication(options: { }, async cancelTurn(input) { return request("POST", "/v1/agent/chat/cancel", { ...input.params, traceId: input.traceId }, input.actor); - } + }, + async close() { await options.eventPublisher?.close?.(); } }; } -function codedError(code: string, message: string, details?: unknown) { return Object.assign(new Error(message), { code, details }); } +function codedError(code: string, message: string, details: Record = {}) { + return Object.assign(new Error(message), { + code, + details, + status: details.status, + layer: details.layer, + retryable: details.retryable, + userMessage: details.userMessage + }); +} diff --git a/internal/workbench/runtime.ts b/internal/workbench/runtime.ts index a7d6bb8e..f4970e94 100644 --- a/internal/workbench/runtime.ts +++ b/internal/workbench/runtime.ts @@ -13,11 +13,12 @@ export function workbenchRuntime(env: Record = proce const mode = String(env.WORKBENCH_MODE ?? "").trim(); if (mode !== "native-test" && mode !== "agentrun-native" && mode !== "temporal") throw codedError("workbench_mode_required", "WORKBENCH_MODE must be native-test, agentrun-native, or temporal"); const stateFile = path.resolve(String(env.WORKBENCH_NATIVE_STATE_FILE ?? ".state/workbench-native/state.json")); + const eventPublisher = mode === "native-test" ? null : options.eventPublisher ?? createWorkbenchKafkaEventPublisher({ env }); const application = mode === "native-test" ? createNativeTestWorkbenchApplication({ stateFile }) : mode === "agentrun-native" - ? createNativeAgentRunWorkbenchApplication({ stateFile, env, eventPublisher: options.eventPublisher ?? createWorkbenchKafkaEventPublisher({ env }) }) - : createCloudWorkbenchApplication({ baseUrl: String(env.WORKBENCH_CLOUD_API_URL ?? ""), authorization: String(env.WORKBENCH_CLOUD_API_AUTHORIZATION ?? "") }); + ? createNativeAgentRunWorkbenchApplication({ stateFile, env, eventPublisher }) + : createCloudWorkbenchApplication({ baseUrl: String(env.WORKBENCH_CLOUD_API_URL ?? ""), authorization: String(env.WORKBENCH_CLOUD_API_AUTHORIZATION ?? ""), eventPublisher }); const temporalAddress = String(env.WORKBENCH_TEMPORAL_ADDRESS ?? env.TEMPORAL_ADDRESS ?? ""); const temporalNamespace = String(env.WORKBENCH_TEMPORAL_NAMESPACE ?? "unidesk"); const taskQueue = String(env.WORKBENCH_TEMPORAL_TASK_QUEUE ?? "hwlab-v03-workbench"); @@ -34,6 +35,7 @@ export function workbenchRuntime(env: Record = proce temporalAddress, temporalNamespace, taskQueue, activity, + eventPublisher, application, temporal, dispatch: createWorkbenchDispatcher({ application, temporal, mode }), diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 1c83462d..00901d8d 100644 --- a/internal/workbench/workbench.test.ts +++ b/internal/workbench/workbench.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import type { WorkbenchApplication, WorkbenchTemporalGateway } from "./contracts.ts"; +import { createWorkbenchActivities } from "./activities.ts"; import { createWorkbenchDispatcher } from "./dispatcher.ts"; import { createCloudWorkbenchApplication } from "./cloud-application.ts"; import { createWorkbenchHttpApp } from "./http.ts"; @@ -72,6 +73,59 @@ describe("Workbench Cloud application adapter", () => { await application.createSession({ actor: { id: "usr_test" }, params: {} }); expect(authorization).toBe("Bearer unified-internal-key"); }); + + test("preserves Cloud API retryability and billing failure semantics", async () => { + const application = createCloudWorkbenchApplication({ + baseUrl: "http://cloud.internal", + authorization: "unified-internal-key", + fetchImpl: (async () => Response.json({ + ok: false, + error: { code: "insufficient_credits", layer: "billing", retryable: false, message: "credits exhausted", userMessage: "余额不足" } + }, { status: 402 })) as typeof fetch + }); + const input = { actor: { id: "usr_test" }, traceId: "trc_billing", sessionId: "ses_billing", params: { message: "hi" } }; + await expect(application.dispatchTurn(input)).rejects.toMatchObject({ + code: "insufficient_credits", + layer: "billing", + retryable: false, + status: 402, + userMessage: "余额不足" + }); + }); +}); + +describe("Workbench Temporal activities", () => { + const activity = { startToCloseTimeoutMs: 1000, retryInitialIntervalMs: 100, retryMaximumIntervalMs: 500, retryMaximumAttempts: 3 }; + const input = { actor: { id: "usr_test" }, traceId: "trc_activity", sessionId: "ses_activity", params: { message: "hi" } }; + + test("publishes semantic Kafka failure and stops retrying non-retryable billing errors", async () => { + const published: any[] = []; + const application = stubApplication(); + application.dispatchTurn = async () => { throw Object.assign(new Error("credits exhausted"), { code: "insufficient_credits", layer: "billing", retryable: false, userMessage: "余额不足" }); }; + const activities = createWorkbenchActivities(application, { + activity, + attempt: () => 1, + eventPublisher: { async publish(event) { published.push(event); return { published: true }; } } + }); + await expect(activities.dispatchWorkbenchTurn(input)).rejects.toMatchObject({ type: "insufficient_credits", nonRetryable: true }); + expect(published).toHaveLength(2); + expect(published[0]).toMatchObject({ event: { type: "error", failureDomain: "billing", code: "insufficient_credits", retryable: false, willRetry: false, attempt: 1, maxAttempts: 3, terminal: false } }); + expect(published[1]).toMatchObject({ event: { type: "terminal_status", terminalStatus: "failed", errorCode: "insufficient_credits", message: "余额不足", terminal: true } }); + }); + + test("publishes the YAML-controlled exponential retry without sealing the turn early", async () => { + const published: any[] = []; + const application = stubApplication(); + application.dispatchTurn = async () => { throw Object.assign(new Error("billing unavailable"), { code: "user_billing_preflight_failed", layer: "billing", retryable: true }); }; + const activities = createWorkbenchActivities(application, { + activity, + attempt: () => 2, + eventPublisher: { async publish(event) { published.push(event); return { published: true }; } } + }); + await expect(activities.dispatchWorkbenchTurn(input)).rejects.toMatchObject({ type: "user_billing_preflight_failed", nonRetryable: false }); + expect(published).toHaveLength(1); + expect(published[0]).toMatchObject({ event: { type: "backend_status", retryPhase: "retryScheduled", attempt: 2, maxAttempts: 3, backoffMs: 200, terminal: false } }); + }); }); describe("Workbench native HTTP adapter", () => {