fix: persist workbench turn admission before preflight
This commit is contained in:
lyon
@@ -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 = {} } = {}) {
|
||||
|
||||
Reference in New Issue
Block a user