Merge pull request #1623 from pikasTech/fix/1619-workbench-turn-admission

fix: Workbench turn admission survives preflight failure

internal/cloud/server-agent-chat.test.ts

@@ -832,6 +832,108 @@ test("cloud api /v1/agent/chat delegates v0.3 turns to AgentRun v0.1 over adapte
   }
 });
 
+test("cloud api keeps admitted user message when billing preflight fails before AgentRun dispatch (#1619)", async () => {
+  const traceId = "trc_issue1619_billing_after_admission";
+  const sessionId = "ses_issue1619_billing_after_admission";
+  const conversationId = "cnv_issue1619_billing_after_admission";
+  const ownerSessions = new Map([[sessionId, testAgentSessionRecord({ sessionId, conversationId, projectId: "prj_hwpod_workbench", status: "idle" })]]);
+  const ownerWrites = [];
+  const billingCalls = [];
+  const agentRunCalls = [];
+  const agentRunServer = createHttpServer(async (request, response) => {
+    agentRunCalls.push({ method: request.method, url: request.url });
+    response.writeHead(500, { "content-type": "application/json" });
+    response.end(`${JSON.stringify({ ok: false, message: "AgentRun must not be called when billing preflight failed" })}\n`);
+  });
+  await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve));
+  const agentRunPort = agentRunServer.address().port;
+  const traceStore = createCodeAgentTraceStore();
+  const server = createCloudApiServer({
+    traceStore,
+    env: {
+      HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
+      AGENTRUN_MGR_URL: `http://127.0.0.1:${agentRunPort}`,
+      HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
+      HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601",
+      HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT: "0123456789abcdef0123456789abcdef01234567",
+      HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "deepseek",
+      HWLAB_ENVIRONMENT: "v03",
+      HWLAB_GITOPS_PROFILE: "v03",
+      HWLAB_USER_BILLING_CODE_AGENT_ENABLED: "1"
+    },
+    userBillingAuth: { active: true },
+    userBillingClient: {
+      configured: true,
+      async billingPreflight(body) {
+        billingCalls.push(body);
+        assert.ok(ownerWrites.some((write) => write.traceId === traceId && write.status === "running"), "durable admission must be recorded before billing preflight");
+        return { ok: false, status: 503, error: { code: "billing_preflight_unavailable", message: "billing preflight unavailable" } };
+      }
+    },
+    accessController: {
+      required: false,
+      async authenticate() {
+        return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION, userBilling: { active: true } };
+      },
+      async recordAgentSessionOwner(input) {
+        ownerWrites.push(input);
+        const existing = ownerSessions.get(input.sessionId ?? input.id);
+        const record = testAgentSessionRecord({
+          ...existing,
+          ...input,
+          session: { ...(existing?.session ?? {}), ...(input.session ?? {}) }
+        });
+        ownerSessions.set(record.id, record);
+        return record;
+      },
+      async getAgentSession(requestedSessionId) {
+        return ownerSessions.get(requestedSessionId) ?? null;
+      }
+    }
+  });
+  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+
+  try {
+    const { port } = server.address();
+    const submit = await fetch(`http://127.0.0.1:${port}/v1/agent/chat`, {
+      method: "POST",
+      headers: { "content-type": "application/json", "x-trace-id": traceId, cookie: "hwlab_session=test-stub-session" },
+      body: JSON.stringify({
+        conversationId,
+        projectId: "prj_hwpod_workbench",
+        sessionId,
+        message: "测试一下和 cpython 对比的性能"
+      })
+    });
+    assert.equal(submit.status, 202);
+    const accepted = await submit.json();
+    assert.equal(accepted.accepted, true);
+    assert.equal(accepted.traceId, traceId);
+
+    const payload = await pollAgentResult(port, traceId);
+    validateCodeAgentChatSchema(payload);
+    assert.equal(payload.status, "failed");
+    assert.equal(payload.error.code, "billing_preflight_unavailable");
+    assert.match(payload.finalResponse.text, /billing preflight unavailable/u);
+    assert.equal(billingCalls.length, 1);
+    assert.equal(agentRunCalls.length, 0);
+    assert.ok(ownerWrites.some((write) => write.traceId === traceId && write.status === "running"));
+    assert.ok(ownerWrites.some((write) => write.traceId === traceId && write.status === "failed"));
+    const stored = ownerSessions.get(sessionId);
+    const userMessage = stored?.session?.messages?.find((message) => message.role === "user");
+    assert.match(userMessage?.text ?? "", /cpython/u);
+
+    const turn = await fetch(`http://127.0.0.1:${port}/v1/agent/turns/${traceId}`, { headers: { cookie: "hwlab_session=test-stub-session" } });
+    assert.equal(turn.status, 200);
+    const turnBody = await turn.json();
+    assert.equal(turnBody.status, "failed");
+    assert.equal(turnBody.error.code, "billing_preflight_unavailable");
+  } finally {
+    await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
+    await new Promise((resolve, reject) => agentRunServer.close((error) => (error ? reject(error) : resolve())));
+  }
+});
+
 test("AgentRun sync converts terminal command result even when run remains claimed (#1555)", async () => {
   const calls = [];
   const traceId = "trc_issue1555_terminal_command";

internal/cloud/server-code-agent-http.ts

@@ -108,17 +108,23 @@ export async function handleCodeAgentChatHttp(request, response, options) {
   };
   const manualSession = perf ? await perf.measure("manual_session", () => prepareManualCodeAgentSession({ params: chatParams, options, traceId, response })) : await prepareManualCodeAgentSession({ params: chatParams, options, traceId, response });
   if (manualSession?.blocked) return;
-  const billingPreflight = perf ? await perf.measure("billing_preflight", () => preflightCodeAgentBilling({ params: chatParams, options, traceId, response })) : await preflightCodeAgentBilling({ params: chatParams, options, traceId, response });
+  const nativeSessionChatParams = stripSyntheticConversationContext(chatParams, traceId, options);
-  if (billingPreflight?.blocked) return;
-  const nativeSessionChatParams = stripSyntheticConversationContext({ ...chatParams, userBillingReservation: billingPreflight?.reservation ?? null }, traceId, options);
 
   if (codeAgentChatShortConnectionRequested(request, params, options)) {
     if (perf) perf.recordPhase({ phase: "short_connection_accept", durationMs: 0, outcome: "ok" });
-    submitCodeAgentChatTurn({
+    try {
-      params: nativeSessionChatParams,
+      await submitCodeAgentChatTurn({
-      options,
+        params: nativeSessionChatParams,
-      traceId
+        options,
-    });
+        traceId
+      });
+    } catch (error) {
+      if (isCodeAgentAdmissionUnavailableError(error)) {
+        sendJson(response, error.statusCode ?? 503, error.payload);
+        return;
+      }
+      throw error;
+    }
     const traceUrl = `/v1/agent/traces/${encodeURIComponent(traceId)}`;
     const turnUrl = `/v1/agent/turns/${encodeURIComponent(traceId)}`;
     sendJson(response, 202, {
@@ -143,10 +149,35 @@ export async function handleCodeAgentChatHttp(request, response, options) {
     return;
   }
 
-  const payload = perf ? await perf.measure("code_agent_run", () => runCodeAgentChat(nativeSessionChatParams, options)) : await runCodeAgentChat(nativeSessionChatParams, options);
+  const admittedBase = codeAgentAdmittedFailureBasePayload({ params: nativeSessionChatParams, options, traceId });
-  await (perf ? perf.measure("billing_finalize", () => finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options })) : finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options }));
+  try {
-  await (perf ? perf.measure("session_owner_record", () => recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: codeAgentOwnerStatusForResult(payload) })) : recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: codeAgentOwnerStatusForResult(payload) }));
+    await recordCodeAgentTurnAdmission({ payload: admittedBase, params: nativeSessionChatParams, options, traceId });
-  const responsePayload = annotateOwner(payload, nativeSessionChatParams);
+  } catch (error) {
+    if (isCodeAgentAdmissionUnavailableError(error)) {
+      sendJson(response, error.statusCode ?? 503, error.payload);
+      return;
+    }
+    throw error;
+  }
+  const billingPreflight = perf ? await perf.measure("billing_preflight", () => preflightCodeAgentBilling({ params: nativeSessionChatParams, options, traceId, response: null, sendResponse: false })) : await preflightCodeAgentBilling({ params: nativeSessionChatParams, options, traceId, response: null, sendResponse: false });
+  if (billingPreflight?.blocked) {
+    const failed = await settleAdmittedCodeAgentFailure({
+      base: admittedBase,
+      params: nativeSessionChatParams,
+      options,
+      traceId,
+      results: options.codeAgentChatResults,
+      failure: billingPreflight,
+      traceLabel: "billing:preflight:failed"
+    });
+    sendJson(response, billingPreflight.statusCode ?? 503, failed);
+    return;
+  }
+  const nativeSessionChatParamsWithBilling = { ...nativeSessionChatParams, userBillingReservation: billingPreflight?.reservation ?? null };
+  const payload = perf ? await perf.measure("code_agent_run", () => runCodeAgentChat(nativeSessionChatParamsWithBilling, options)) : await runCodeAgentChat(nativeSessionChatParamsWithBilling, options);
+  await (perf ? perf.measure("billing_finalize", () => finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParamsWithBilling, options })) : finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParamsWithBilling, options }));
+  await (perf ? perf.measure("session_owner_record", () => recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParamsWithBilling, options, status: codeAgentOwnerStatusForResult(payload) })) : recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParamsWithBilling, options, status: codeAgentOwnerStatusForResult(payload) }));
+  const responsePayload = annotateOwner(payload, nativeSessionChatParamsWithBilling);
 
   sendJson(response, responsePayload.status === "failed" && responsePayload.error?.code === "invalid_params" ? 400 : 200, responsePayload);
 }
@@ -711,7 +742,7 @@ function firstNonEmptyValue(...values) {
   return null;
 }
 
-function submitCodeAgentChatTurn({ params, options, traceId }) {
+async function submitCodeAgentChatTurn({ params, options, traceId }) {
   const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
   const results = options.codeAgentChatResults ?? createCodeAgentChatResultStore();
   const acceptedPayload = {
@@ -725,59 +756,54 @@ function submitCodeAgentChatTurn({ params, options, traceId }) {
     projectId: params.projectId ?? null,
     updatedAt: new Date().toISOString()
   };
-  void recordCodeAgentSessionOwner({ payload: acceptedPayload, params, options, status: "running" });
+  await recordCodeAgentTurnAdmission({ payload: acceptedPayload, params, options, traceId });
+  results.set(traceId, annotateOwner(acceptedPayload, params));
   if (codeAgentAgentRunAdapterEnabled(options.env ?? process.env)) {
-    const initial = withCodeAgentBillingReservation(initialAgentRunChatResult({ params, options, traceId }), params);
-    results.set(traceId, annotateOwner(initial, params));
     const run = async () => {
+      let initial = null;
       let executionOptions = options;
       try {
-        executionOptions = { ...options, ...(await codeAgentChatExecutionOptions(options, params)) };
+        initial = initialAgentRunChatResult({ params, options, traceId });
-        const payload = await submitAgentRunChatTurn({ params, options: executionOptions, traceId, traceStore, results });
+        results.set(traceId, annotateOwner(initial, params));
+        const billingPreflight = await preflightCodeAgentBilling({ params, options, traceId, sendResponse: false });
+        if (billingPreflight?.blocked) {
+          await settleAdmittedCodeAgentFailure({
+            base: initial,
+            params,
+            options,
+            traceId,
+            results,
+            failure: billingPreflight,
+            traceLabel: "billing:preflight:failed"
+          });
+          return;
+        }
+        const dispatchParams = { ...params, userBillingReservation: billingPreflight?.reservation ?? null };
+        initial = withCodeAgentBillingReservation(initial, dispatchParams);
+        results.set(traceId, annotateOwner(initial, dispatchParams));
+        executionOptions = { ...options, ...(await codeAgentChatExecutionOptions(options, dispatchParams)) };
+        const payload = await submitAgentRunChatTurn({ params: dispatchParams, options: executionOptions, traceId, traceStore, results });
         if (isCodeAgentResultCanceled(results.get(traceId))) return;
-        const owned = annotateOwner(withCodeAgentBillingReservation(payload, params), params);
+        const owned = annotateOwner(withCodeAgentBillingReservation(payload, dispatchParams), dispatchParams);
-        await recordCodeAgentSessionOwner({ payload: owned, params, options: executionOptions, status: "running" });
+        await recordCodeAgentSessionOwner({ payload: owned, params: dispatchParams, options: executionOptions, status: "running" });
         results.set(traceId, owned);
-        scheduleAgentRunProjectionSync({ traceId, params, options: executionOptions, traceStore, currentResult: owned });
+        scheduleAgentRunProjectionSync({ traceId, params: dispatchParams, options: executionOptions, traceStore, currentResult: owned });
       } catch (error) {
         if (isCodeAgentResultCanceled(results.get(traceId))) return;
-        const payload = annotateOwner({
+        await settleAdmittedCodeAgentFailure({
-          ...initial,
+          base: initial ?? codeAgentAdmittedFailureBasePayload({ params, options, traceId }),
-          status: "failed",
+          params,
-          error: {
+          options: executionOptions,
-            code: error?.code ?? "agentrun_dispatch_failed",
+          traceId,
-            layer: "agentrun",
+          results,
-            retryable: true,
+          failure: codeAgentDispatchFailure(error),
-            message: error?.message ?? "AgentRun dispatch failed",
+          traceLabel: "agentrun:dispatch:failed"
-            userMessage: "AgentRun 调度失败；trace/result 轮询已保留错误。"
-          },
-          blocker: {
-            code: error?.code ?? "agentrun_dispatch_failed",
-            layer: "agentrun",
-            retryable: true,
-            summary: error?.message ?? "AgentRun dispatch failed"
-          },
-          runnerTrace: traceStore.snapshot(traceId),
-          updatedAt: new Date().toISOString()
-        }, params);
-        recordCodeAgentConversationFact(payload, executionOptions);
-        await finalizeCodeAgentBillingUsage({ payload, params, options: executionOptions });
-        results.set(traceId, payload);
-        traceStore.append(traceId, {
-          type: "result",
-          status: "failed",
-          label: "agentrun:dispatch:failed",
-          errorCode: payload.error.code,
-          message: payload.error.message,
-          terminal: true
         });
-        await recordCodeAgentSessionOwner({ payload, params, options: executionOptions, status: "failed" });
       }
     };
     setImmediate(() => { run(); });
     return;
   }
-  results.set(traceId, acceptedPayload);
   traceStore.ensure(traceId, {
     runnerKind: "codex-app-server-stdio-runner",
     workspace: options.workspace ?? options.env?.HWLAB_CODE_AGENT_CODEX_WORKSPACE ?? options.env?.HWLAB_CODE_AGENT_WORKSPACE ?? null,
@@ -795,11 +821,25 @@ function submitCodeAgentChatTurn({ params, options, traceId }) {
 
   const run = async () => {
     try {
-      const payload = await runCodeAgentChat(params, options);
+      const billingPreflight = await preflightCodeAgentBilling({ params, options, traceId, sendResponse: false });
+      if (billingPreflight?.blocked) {
+        await settleAdmittedCodeAgentFailure({
+          base: codeAgentAdmittedFailureBasePayload({ params, options, traceId }),
+          params,
+          options,
+          traceId,
+          results,
+          failure: billingPreflight,
+          traceLabel: "billing:preflight:failed"
+        });
+        return;
+      }
+      const dispatchParams = { ...params, userBillingReservation: billingPreflight?.reservation ?? null };
+      const payload = await runCodeAgentChat(dispatchParams, options);
       if (isCodeAgentResultCanceled(results.get(traceId))) return;
-      await finalizeCodeAgentBillingUsage({ payload, params, options });
+      await finalizeCodeAgentBillingUsage({ payload, params: dispatchParams, options });
-      await recordCodeAgentSessionOwner({ payload, params, options, status: codeAgentOwnerStatusForResult(payload) });
+      await recordCodeAgentSessionOwner({ payload, params: dispatchParams, options, status: codeAgentOwnerStatusForResult(payload) });
-      results.set(traceId, annotateOwner(payload, params));
+      results.set(traceId, annotateOwner(payload, dispatchParams));
       traceStore.append(traceId, {
         type: "result",
         status: payload.status === "completed" ? "completed" : "failed",
@@ -844,6 +884,178 @@ function submitCodeAgentChatTurn({ params, options, traceId }) {
   });
 }
 
+async function recordCodeAgentTurnAdmission({ payload = {}, params = {}, options = {}, traceId } = {}) {
+  const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
+  let ownerRecord = null;
+  try {
+    ownerRecord = await recordCodeAgentSessionOwner({ payload, params, options, status: "running" });
+  } catch (error) {
+    throw codeAgentAdmissionUnavailableError(error, { params, traceId });
+  }
+  traceStore.append(traceId, {
+    type: "request",
+    status: "admitted",
+    label: "turn:admitted",
+    message: "Code Agent turn was durably admitted before billing or dispatch preflight.",
+    sessionId: safeSessionId(params.sessionId) || null,
+    threadId: safeOpaqueId(params.threadId) || null,
+    waitingFor: "billing-or-dispatch",
+    valuesPrinted: false
+  });
+  return ownerRecord;
+}
+
+function codeAgentAdmissionUnavailableError(error, { params = {}, traceId } = {}) {
+  const payload = codeAgentAdmissionUnavailablePayload({ error, params, traceId });
+  return Object.assign(new Error(payload.error.message), {
+    code: "admission_unavailable",
+    statusCode: 503,
+    payload,
+    alreadyClassified: true
+  });
+}
+
+function isCodeAgentAdmissionUnavailableError(error) {
+  return error?.code === "admission_unavailable" && error?.payload;
+}
+
+function codeAgentAdmissionUnavailablePayload({ error, params = {}, traceId } = {}) {
+  const message = error?.message ?? "Code Agent turn admission is unavailable; the user message was not persisted.";
+  return {
+    ok: false,
+    accepted: false,
+    status: "failed",
+    traceId: safeTraceId(traceId) || null,
+    ...codeAgentTurnLifecycleFields(traceId, params),
+    conversationId: safeConversationId(params.conversationId) || null,
+    sessionId: safeSessionId(params.sessionId) || null,
+    threadId: safeOpaqueId(params.threadId) || null,
+    error: {
+      code: "admission_unavailable",
+      layer: "admission",
+      retryable: true,
+      message,
+      userMessage: "本次输入尚未持久化，请保留草稿后重试。",
+      route: "/v1/agent/chat"
+    },
+    blocker: {
+      code: "admission_unavailable",
+      layer: "admission",
+      retryable: true,
+      summary: message
+    },
+    valuesRedacted: true,
+    secretMaterialStored: false
+  };
+}
+
+function codeAgentDispatchFailure(error) {
+  const message = error?.message ?? "AgentRun dispatch failed";
+  return {
+    statusCode: 503,
+    payload: {
+      error: {
+        code: error?.code ?? "agentrun_dispatch_failed",
+        layer: "agentrun",
+        retryable: true,
+        message,
+        userMessage: "AgentRun 调度失败；trace/result 轮询已保留错误。",
+        route: "/v1/agent/chat"
+      },
+      blocker: {
+        code: error?.code ?? "agentrun_dispatch_failed",
+        layer: "agentrun",
+        retryable: true,
+        summary: message
+      }
+    }
+  };
+}
+
+function codeAgentAdmittedFailureBasePayload({ params = {}, options = {}, traceId } = {}) {
+  const now = new Date().toISOString();
+  return {
+    accepted: true,
+    status: "running",
+    shortConnection: true,
+    traceId,
+    ...codeAgentTurnLifecycleFields(traceId, params),
+    messageId: codeAgentTurnLifecycleFields(traceId, params).assistantMessageId,
+    projectId: params.projectId ?? null,
+    conversationId: safeConversationId(params.conversationId) || null,
+    sessionId: safeSessionId(params.sessionId) || null,
+    threadId: safeOpaqueId(params.threadId) || null,
+    createdAt: now,
+    updatedAt: now,
+    provider: codeAgentAgentRunAdapterEnabled(options.env ?? process.env) ? "agentrun" : "codex-stdio",
+    model: options.env?.HWLAB_CODE_AGENT_MODEL ?? "unknown",
+    backend: codeAgentAgentRunAdapterEnabled(options.env ?? process.env) ? "agentrun-v01" : "hwlab-cloud-api/code-agent-chat",
+    valuesPrinted: false
+  };
+}
+
+async function settleAdmittedCodeAgentFailure({ base = {}, params = {}, options = {}, traceId, results, failure = {}, traceLabel = "code-agent:failed" } = {}) {
+  const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
+  const error = failure.payload?.error ?? failure.error ?? {};
+  const blocker = failure.payload?.blocker ?? failure.blocker ?? null;
+  const message = error.message ?? blocker?.summary ?? "Code Agent request failed after admission.";
+  traceStore.append(traceId, {
+    type: "result",
+    status: "failed",
+    label: traceLabel,
+    errorCode: error.code ?? "code_agent_failed_after_admission",
+    message,
+    terminal: true,
+    valuesPrinted: false
+  });
+  const now = new Date().toISOString();
+  const payload = annotateOwner({
+    ...base,
+    accepted: true,
+    status: "failed",
+    traceId,
+    ...codeAgentTurnLifecycleFields(traceId, base),
+    messageId: base.messageId ?? codeAgentTurnLifecycleFields(traceId, base).assistantMessageId,
+    conversationId: safeConversationId(base.conversationId ?? params.conversationId) || null,
+    sessionId: safeSessionId(base.sessionId ?? params.sessionId) || null,
+    threadId: safeOpaqueId(base.threadId ?? params.threadId) || null,
+    error: {
+      code: error.code ?? "code_agent_failed_after_admission",
+      layer: error.layer ?? blocker?.layer ?? "api",
+      retryable: error.retryable !== false,
+      message,
+      userMessage: error.userMessage ?? message,
+      route: error.route ?? "/v1/agent/chat"
+    },
+    blocker: blocker ? {
+      code: blocker.code ?? error.code ?? "code_agent_failed_after_admission",
+      layer: blocker.layer ?? error.layer ?? "api",
+      retryable: blocker.retryable !== false,
+      summary: blocker.summary ?? message
+    } : null,
+    finalResponse: {
+      text: error.userMessage ?? message,
+      textChars: String(error.userMessage ?? message).length,
+      role: "assistant",
+      status: "failed",
+      traceId,
+      updatedAt: now,
+      source: "code-agent-admitted-failure",
+      valuesPrinted: false
+    },
+    runnerTrace: traceStore.snapshot(traceId),
+    updatedAt: now,
+    valuesPrinted: false,
+    secretMaterialStored: false
+  }, params);
+  await finalizeCodeAgentBillingUsage({ payload, params, options });
+  payload.runnerTrace = traceStore.snapshot(traceId);
+  recordCodeAgentConversationFact(payload, options);
+  results?.set?.(traceId, payload);
+  await recordCodeAgentSessionOwner({ payload, params, options, status: codeAgentOwnerStatusForResult(payload) });
+  return payload;
+}
+
 function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, traceStore = defaultCodeAgentTraceStore, currentResult = null } = {}) {
   if (!safeTraceId(traceId) || !currentResult?.agentRun?.runId || !currentResult?.agentRun?.commandId) return null;
   const env = options.env ?? process.env;
@@ -1281,7 +1493,7 @@ function codeAgentChatShortConnectionRequested(request, params, options = {}) {
     truthyFlag(envValue);
 }
 
-async function preflightCodeAgentBilling({ params = {}, options = {}, traceId, response } = {}) {
+async function preflightCodeAgentBilling({ params = {}, options = {}, traceId, response, sendResponse = true } = {}) {
   const client = options.userBillingClient;
   if (!codeAgentBillingEnabled(options) || !client?.configured || !options.userBillingAuth?.active) return null;
   const estimatedCredits = parsePositiveInteger(options.env?.HWLAB_USER_BILLING_CODE_AGENT_ESTIMATED_CREDITS, 1);
@@ -1295,7 +1507,7 @@ async function preflightCodeAgentBilling({ params = {}, options = {}, traceId, r
   const result = await client.billingPreflight(body);
   if (result.ok && result.body?.allowed !== false) return { reservation: sanitizeBillingReservation(result.body) };
   const status = [402, 403, 429].includes(Number(result.status)) ? Number(result.status) : 503;
-  sendJson(response, status, {
+  const payload = {
     ok: false,
     accepted: false,
     status: status === 402 ? "payment_required" : status === 403 ? "not_entitled" : status === 429 ? "rate_limited" : "billing_unavailable",
@@ -1314,8 +1526,9 @@ async function preflightCodeAgentBilling({ params = {}, options = {}, traceId, r
       summary: result.error?.message ?? "Code Agent billing preflight failed"
     },
     valuesRedacted: true
-  });
+  };
-  return { blocked: true };
+  if (sendResponse && response) sendJson(response, status, payload);
+  return { blocked: true, statusCode: status, payload, error: payload.error, blocker: payload.blocker };
 }
 
 async function recordCodeAgentBillingUsage({ payload = {}, params = {}, options = {} } = {}) {

internal/cloud/server-http-utils.ts

@@ -21,11 +21,13 @@ export function readBody(request, limitBytes = DEFAULT_BODY_LIMIT_BYTES) {
 }
 
 export function sendJson(response, statusCode, body) {
+  if (response.headersSent || response.writableEnded) return false;
   response.writeHead(statusCode, {
     "content-type": "application/json; charset=utf-8",
     "cache-control": "no-store"
   });
   response.end(JSON.stringify(body) + "\n");
+  return true;
 }
 
 export function sendRedirect(response, location, body = {}) {

internal/cloud/server.ts

@@ -129,7 +129,7 @@ export function createCloudApiServer(options = {}) {
       try {
         await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry, hwpodNodeWsRegistry, accessController, sessionRegistry, traceStore, codexStdioManager, codeAgentChatResults, webPerformanceStore, backendPerformanceStore, backendPerformance });
       } catch (error) {
-        if (error?.alreadySent) return;
+        if (error?.alreadySent || response.headersSent || response.writableEnded) return;
         backendPerformance.recordPhase({ phase: "route_exception", durationMs: 0, outcome: "error" });
         sendJson(response, 500, {
           error: {
@@ -1829,11 +1829,13 @@ function readBody(request, limitBytes = DEFAULT_BODY_LIMIT_BYTES) {
 }
 
 function sendJson(response, statusCode, body) {
+  if (response.headersSent || response.writableEnded) return false;
   response.writeHead(statusCode, {
     "content-type": "application/json; charset=utf-8",
     "cache-control": "no-store"
   });
   response.end(`${JSON.stringify(body)}\n`);
+  return true;
 }
 
 function getHeader(request, name) {

web/hwlab-cloud-web/scripts/workbench-e2e-server.ts

@@ -168,6 +168,17 @@ async function handleRequest(request: IncomingMessage, response: ServerResponse)
     const body = await readJson(request);
     state.chatRequests.push(redactRequestBody(body));
     await delay(state.chatDelayMs);
+    if (state.scenarioId === "admission-unavailable") {
+      return json(response, 503, {
+        ok: false,
+        accepted: false,
+        status: "failed",
+        traceId: body.traceId,
+        error: { code: "admission_unavailable", layer: "admission", retryable: true, message: "admission_unavailable: durable turn was not persisted" },
+        blocker: { code: "admission_unavailable", layer: "admission", retryable: true, summary: "durable turn was not persisted" },
+        valuesRedacted: true
+      });
+    }
     return json(response, 200, acceptChatTurn(body));
   }
 

web/hwlab-cloud-web/src/components/workbench/CommandComposer.vue

@@ -31,7 +31,8 @@ async function submit(): Promise<void> {
   }
   if (!value.trim() || workbench.composer.disabled) return;
   draft.value = "";
-  await workbench.submitMessage(value);
+  const accepted = await workbench.submitMessage(value);
+  if (!accepted && !draft.value.trim()) draft.value = value;
 }
 
 async function submitFromInput(): Promise<void> {

web/hwlab-cloud-web/src/stores/workbench.ts

@@ -415,13 +415,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     }));
   }
 
-  async function submitMessage(text: string): Promise<void> {
+  async function submitMessage(text: string): Promise<boolean> {
     const value = text.trim();
-    if (!value) return;
+    if (!value) return false;
     beginSessionSelection();
     if (!composer.value.sessionId) {
       error.value = "session_required：请先显式新建或选择 Code Agent session。";
-      return;
+      return false;
     }
     const steerMode = composer.value.submitMode === "steer";
     const submitEntry = steerMode ? "steer" : "existing";
@@ -444,11 +444,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     const response = await route(payload, codeAgentTimeoutMs.value, () => activityRef.value);
     if (!response.ok || !response.data) {
       markMessage(traceId, { status: "failed", text: response.error ?? "Code Agent 请求失败" });
+      reduceServerState({ type: "turn.forget", traceId });
       failWorkbenchSubmitJourney(traceId, response.status === 0 ? "network" : "error");
-      if (steerMode && response.status === 404) void clearActiveTrace(traceId, "steer-trace-not-found");
+      await clearActiveTrace(traceId, steerMode && response.status === 404 ? "steer-trace-not-found" : "submit-not-admitted");
       chatPending.value = false;
       currentRequest.value = null;
-      return;
+      restartRealtime("submit-not-admitted");
+      return false;
     }
     markWorkbenchSubmitApiAccepted(traceId);
     const canonicalTraceId = alignOptimisticTurnMessages(traceId, response.data);
@@ -468,10 +470,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     void refreshSessions(sessionId);
     if ((response.data as AgentChatResultResponse).terminal === true || isTerminalMessageStatus(response.data.status)) {
       completeTrace(canonicalTraceId, response.data as AgentChatResultResponse);
-      return;
+      return true;
     }
     restartRealtime("submit");
     scheduleRealtimeGapHydration("submit");
+    return true;
   }
 
   async function cancelAgentMessage(message: ChatMessage): Promise<void> {

web/hwlab-cloud-web/tests/workbench-e2e/specs/submit-authority.spec.ts

@@ -35,3 +35,21 @@ test.describe("submit session authority", () => {
     expect(legacyGets).toEqual([]);
   });
 });
+
+test.describe("submit admission failure", () => {
+  test.use({ scenarioId: "admission-unavailable" });
+
+  test("not-admitted chat failure restores draft and does not add preflight routes", async ({ page }) => {
+    await gotoWorkbench(page, "/workbench/sessions/ses_completed");
+    const draft = "admission unavailable draft must survive";
+    await page.locator(selectors.commandInput).fill(draft);
+    await page.locator(selectors.commandSend).click();
+    await expect(page.locator(selectors.commandInput)).toHaveValue(draft);
+    await expect(page.locator(`${selectors.messageCard}[data-role="agent"][data-status="failed"]`).last()).toContainText("admission_unavailable");
+
+    const state = await fakeServerState(page);
+    expect(state.chatRequests).toHaveLength(1);
+    const forbiddenPreflights = (state.requestLedger as Array<{ method?: string; path?: string }>).filter((item) => /billing|readiness|preflight|agentrun/iu.test(item.path ?? ""));
+    expect(forbiddenPreflights).toEqual([]);
+  });
+});