From 5f1763aeac9af299be08d90f04843ae134bbdd30 Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 19 Jun 2026 21:53:53 +0800 Subject: [PATCH] fix: persist workbench turn admission before preflight --- internal/cloud/server-agent-chat.test.ts | 102 ++++++ internal/cloud/server-code-agent-http.ts | 333 ++++++++++++++---- internal/cloud/server-http-utils.ts | 2 + internal/cloud/server.ts | 4 +- .../scripts/workbench-e2e-server.ts | 11 + .../components/workbench/CommandComposer.vue | 3 +- web/hwlab-cloud-web/src/stores/workbench.ts | 15 +- .../specs/submit-authority.spec.ts | 18 + 8 files changed, 420 insertions(+), 68 deletions(-) diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index cef68326..d9d9fe29 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/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"; diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 82e8545d..6b9421fd 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/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 }); - if (billingPreflight?.blocked) return; - const nativeSessionChatParams = stripSyntheticConversationContext({ ...chatParams, userBillingReservation: billingPreflight?.reservation ?? null }, traceId, options); + const nativeSessionChatParams = stripSyntheticConversationContext(chatParams, traceId, options); if (codeAgentChatShortConnectionRequested(request, params, options)) { if (perf) perf.recordPhase({ phase: "short_connection_accept", durationMs: 0, outcome: "ok" }); - submitCodeAgentChatTurn({ - params: nativeSessionChatParams, - options, - traceId - }); + try { + await submitCodeAgentChatTurn({ + params: nativeSessionChatParams, + 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); - await (perf ? perf.measure("billing_finalize", () => finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options })) : finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options })); - await (perf ? perf.measure("session_owner_record", () => recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: codeAgentOwnerStatusForResult(payload) })) : recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: codeAgentOwnerStatusForResult(payload) })); - const responsePayload = annotateOwner(payload, nativeSessionChatParams); + const admittedBase = codeAgentAdmittedFailureBasePayload({ params: nativeSessionChatParams, options, traceId }); + try { + await recordCodeAgentTurnAdmission({ payload: admittedBase, params: nativeSessionChatParams, options, traceId }); + } 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)) }; - const payload = await submitAgentRunChatTurn({ params, options: executionOptions, traceId, traceStore, results }); + initial = initialAgentRunChatResult({ params, options, traceId }); + 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); - await recordCodeAgentSessionOwner({ payload: owned, params, options: executionOptions, status: "running" }); + const owned = annotateOwner(withCodeAgentBillingReservation(payload, dispatchParams), dispatchParams); + 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({ - ...initial, - status: "failed", - error: { - code: error?.code ?? "agentrun_dispatch_failed", - layer: "agentrun", - retryable: true, - message: error?.message ?? "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 settleAdmittedCodeAgentFailure({ + base: initial ?? codeAgentAdmittedFailureBasePayload({ params, options, traceId }), + params, + options: executionOptions, + traceId, + results, + failure: codeAgentDispatchFailure(error), + traceLabel: "agentrun:dispatch:failed" }); - 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 recordCodeAgentSessionOwner({ payload, params, options, status: codeAgentOwnerStatusForResult(payload) }); - results.set(traceId, annotateOwner(payload, params)); + await finalizeCodeAgentBillingUsage({ payload, params: dispatchParams, options }); + await recordCodeAgentSessionOwner({ payload, params: dispatchParams, options, status: codeAgentOwnerStatusForResult(payload) }); + 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 = {} } = {}) { diff --git a/internal/cloud/server-http-utils.ts b/internal/cloud/server-http-utils.ts index 56da0e71..33236514 100644 --- a/internal/cloud/server-http-utils.ts +++ b/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 = {}) { diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 5c356a46..cde71de5 100644 --- a/internal/cloud/server.ts +++ b/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) { diff --git a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts index a1a8ae35..0ad86f0f 100644 --- a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts +++ b/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)); } diff --git a/web/hwlab-cloud-web/src/components/workbench/CommandComposer.vue b/web/hwlab-cloud-web/src/components/workbench/CommandComposer.vue index abdb7a3a..5c3a51d2 100644 --- a/web/hwlab-cloud-web/src/components/workbench/CommandComposer.vue +++ b/web/hwlab-cloud-web/src/components/workbench/CommandComposer.vue @@ -31,7 +31,8 @@ async function submit(): Promise { } 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 { diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index e402942c..272033ac 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -415,13 +415,13 @@ export const useWorkbenchStore = defineStore("workbench", () => { })); } - async function submitMessage(text: string): Promise { + async function submitMessage(text: string): Promise { 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 { diff --git a/web/hwlab-cloud-web/tests/workbench-e2e/specs/submit-authority.spec.ts b/web/hwlab-cloud-web/tests/workbench-e2e/specs/submit-authority.spec.ts index 73df9c3c..261d5239 100644 --- a/web/hwlab-cloud-web/tests/workbench-e2e/specs/submit-authority.spec.ts +++ b/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([]); + }); +});