fix: surface Temporal dispatch failures

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({

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<string, unknown> }) {
       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"; }

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<string, unknown> | 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<string, unknown> : {};
+      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<string, unknown> = {}) {
+  return Object.assign(new Error(message), {
+    code,
+    details,
+    status: details.status,
+    layer: details.layer,
+    retryable: details.retryable,
+    userMessage: details.userMessage
+  });
+}

internal/workbench/runtime.ts

@@ -13,11 +13,12 @@ export function workbenchRuntime(env: Record<string, string | undefined> = 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<string, string | undefined> = proce
     temporalAddress,
     temporalNamespace,
     taskQueue, activity,
+    eventPublisher,
     application,
     temporal,
     dispatch: createWorkbenchDispatcher({ application, temporal, mode }),

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", () => {