mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1675 from pikasTech/fix/issue-1673-provider-retry

Browse Source 
fix: 区分 provider retry 与终态失败
This commit is contained in:
Lyon
2026-06-20 09:52:21 +08:00
committed by GitHub
parent 40a54eef20 c51dd122e2
commit 4adbfd3988
5 changed files with 151 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/code-agent-agentrun-adapter.ts
+57 -5
View File
@@ -1612,6 +1612,7 @@ function agentRunProviderTrace({ base = {}, result = {}, terminalStatus = null }
terminalStatus: terminalStatus ?? result?.terminalStatus ?? null,
failureKind: result?.failureKind ?? null,
failureMessage: result?.failureMessage ?? result?.blocker?.message ?? null,
willRetry: result?.willRetry === true ? true : undefined,
valuesPrinted: false
};
}
@@ -1692,16 +1693,36 @@ function agentRunReplyText(result = {}) {
function agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId, appendResultEvent = true }) {
const terminalStatus = String(result?.terminalStatus ?? "");
const terminal = terminalStatus === "completed" || terminalStatus === "failed" || terminalStatus === "blocked" || terminalStatus === "cancelled";
const normalizedTerminalStatus = terminalStatus.trim().toLowerCase().replace(/_/gu, "-");
const retryableInterruption = normalizedTerminalStatus !== "completed" && isRetryableProviderInterruptionPayload(result);
const terminal = !retryableInterruption && ["completed", "failed", "blocked", "cancelled", "canceled"].includes(normalizedTerminalStatus);
if (!terminal) {
return decorateAgentRunRunningResult({ base, mapping: base.agentRun, traceStore, traceId });
if (retryableInterruption && appendResultEvent) {
traceStore.append(traceId, agentRunProviderRetryEvent({ base, payload: result, status: "retrying", label: "agentrun:provider-retry:provider-stream-disconnected" }), base.agentRun);
}
const runningMapping = retryableInterruption ? {
...base.agentRun,
status: "retrying",
commandState: "retrying",
failureKind: normalizedFailureKind(result?.failureKind),
willRetry: true
} : base.agentRun;
const running = decorateAgentRunRunningResult({ base, mapping: runningMapping, traceStore, traceId });
if (!retryableInterruption) return running;
return {
...running,
status: "retrying",
providerTrace: agentRunProviderTrace({ base, result, terminalStatus }),
transient: { failureKind: normalizedFailureKind(result?.failureKind), willRetry: true, valuesRedacted: true },
valuesPrinted: false
};
}
const now = nowIso();
const runnerTrace = traceStore.snapshot(traceId, agentRunTraceMeta({}, {}));
const terminalEventCreatedAt = agentRunResultTraceCreatedAt(runnerTrace, now);
const providerTrace = agentRunProviderTrace({ base, result, terminalStatus });
const replyText = agentRunReplyText(result);
if (terminalStatus === "completed") {
if (normalizedTerminalStatus === "completed") {
const traceReplyText = agentRunTraceTerminalAssistantText(runnerTrace);
const completedReplyText = replyText || traceReplyText;
const finalResponse = completedReplyText ? agentRunCompletedFinalResponse({ base, result, traceId, now, replyText: completedReplyText }) : null;
@@ -2222,15 +2243,46 @@ function mapAgentRunEvent(event, mapping = {}) {
return { ...base, type: "diff", eventType: "status", status: "running", label: "agentrun:diff", message: textPayload(payload, "diff") };
}
if (event.type === "error") {
return { ...base, type: "error", eventType: "error", status: "failed", label: `agentrun:error:${String(payload.failureKind ?? "backend")}`, errorCode: payload.failureKind ?? "agentrun_error", message: textPayload(payload, "AgentRun error") };
if (isRetryableProviderInterruptionPayload(payload)) return agentRunProviderRetryEvent({ base, payload, status: "retrying", label: `agentrun:provider-retry:${String(payload.failureKind ?? "provider-stream-disconnected")}` });
return { ...base, type: "error", eventType: "error", status: "failed", label: `agentrun:error:${String(payload.failureKind ?? "backend")}`, errorCode: payload.failureKind ?? "agentrun_error", failureKind: payload.failureKind ?? null, willRetry: payload.willRetry === true ? true : undefined, message: textPayload(payload, "AgentRun error") };
}
if (event.type === "terminal_status") {
const terminalStatus = String(payload.terminalStatus ?? "failed");
return { ...base, type: "result", eventType: "terminal", status: terminalStatus === "cancelled" ? "canceled" : terminalStatus, label: `agentrun:terminal:${terminalStatus}`, errorCode: payload.failureKind ?? null, message: textPayload(payload, `AgentRun terminal status ${terminalStatus}`), terminal: true };
if (isRetryableProviderInterruptionPayload(payload)) return agentRunProviderRetryEvent({ base, payload, status: "retrying", label: `agentrun:provider-retry:${terminalStatus}` });
return { ...base, type: "result", eventType: "terminal", status: terminalStatus === "cancelled" ? "canceled" : terminalStatus, label: `agentrun:terminal:${terminalStatus}`, errorCode: payload.failureKind ?? null, failureKind: payload.failureKind ?? null, willRetry: payload.willRetry === true ? true : undefined, message: textPayload(payload, `AgentRun terminal status ${terminalStatus}`), terminal: true };
}
return { ...base, type: "backend", eventType: "backend", status: "running", label: `agentrun:event:${String(event.type ?? "unknown")}`, message: textPayload(payload, "AgentRun event") };
}
function agentRunProviderRetryEvent({ base = {}, payload = {}, status = "retrying", label = "agentrun:provider-retry" } = {}) {
const failureKind = normalizedFailureKind(payload.failureKind ?? payload.errorCode ?? "provider-stream-disconnected");
return {
...base,
type: "provider_retry",
eventType: "backend",
status,
label,
errorCode: failureKind,
failureKind,
willRetry: payload.willRetry === true,
retryable: true,
transient: true,
terminal: false,
retryOf: payload.retryOf ?? null,
terminalStatus: payload.terminalStatus ?? null,
message: textPayload(payload, "Provider stream disconnected; AgentRun will retry."),
valuesPrinted: false
};
}
function isRetryableProviderInterruptionPayload(payload = {}) {
return normalizedFailureKind(payload?.failureKind ?? payload?.errorCode ?? payload?.code) === "provider-stream-disconnected" && payload?.willRetry === true;
}
function normalizedFailureKind(value) {
return String(value ?? "").trim().toLowerCase().replace(/_/gu, "-") || null;
}
function agentRunTraceEvent(event = {}, backendSource = {}) {
const source = backendSource && typeof backendSource === "object" ? backendSource : { backendProfile: backendSource };
const backendProfile = firstNonEmpty(event.backendProfile, source.backendProfile, source.providerProfile, source.agentRun?.backendProfile);
internal/cloud/server-workbench-http.test.ts
+58 -1
View File
@@ -8,7 +8,7 @@ import { createCloudApiServer } from "./server.ts";
import { createBackendPerformanceStore } from "./backend-performance.ts";
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts";
import { createWorkbenchTurnProjection, projectionDiagnostics } from "./workbench-turn-projection.ts";
import { createWorkbenchTurnProjection, durableTraceStatus, projectionDiagnostics, traceTerminalEvidence } from "./workbench-turn-projection.ts";
const ACTOR = { id: "usr_workbench_reader", username: "reader", displayName: "Reader", role: "user", status: "active" };
@@ -58,6 +58,63 @@ test("workbench projection diagnostics keeps projecting health distinct from cau
assert.equal(diagnostics.projectionHealth, "projecting");
});
test("workbench turn projection treats retryable provider stream disconnect as active", () => {
const traceId = "trc_retryable_provider_stream_disconnect";
const result = {
traceId,
status: "canceled",
failureKind: "provider-stream-disconnected",
willRetry: true,
agentRun: {
runId: "run_retryable_provider_stream_disconnect",
commandId: "cmd_retryable_provider_stream_disconnect",
terminalStatus: "cancelled",
failureKind: "provider-stream-disconnected",
willRetry: true
},
providerTrace: { failureKind: "provider-stream-disconnected", willRetry: true }
};
const trace = {
traceId,
status: "canceled",
events: [
{ seq: 1, type: "error", status: "retrying", errorCode: "provider-stream-disconnected", willRetry: true, terminal: false },
{ seq: 2, type: "result", status: "canceled", errorCode: "provider-stream-disconnected", willRetry: true, terminal: true }
],
eventCount: 2
};
const projection = createWorkbenchTurnProjection({ traceId, result, trace });
assert.equal(projection.status, "retrying");
assert.equal(projection.running, true);
assert.equal(projection.terminal, false);
assert.equal(projection.finalResponse, null);
assert.equal(durableTraceStatus(trace.events), "retrying");
assert.equal(traceTerminalEvidence(trace), null);
const completedProjection = createWorkbenchTurnProjection({
traceId,
result: {
...result,
status: "completed",
finalResponse: "retry recovered",
agentRun: { ...result.agentRun, terminalStatus: "completed" }
},
trace: {
...trace,
status: "completed",
events: [
...trace.events,
{ seq: 3, type: "assistant", status: "completed", message: "retry recovered", terminal: true, replyAuthority: true }
],
eventCount: 3
}
});
assert.equal(completedProjection.status, "completed");
assert.equal(completedProjection.running, false);
assert.equal(completedProjection.terminal, true);
assert.equal(completedProjection.finalResponse?.text, "retry recovered");
});
test("workbench turn projection seals final response from authoritative terminal assistant trace event (#1629)", () => {
const traceId = "trc_workbench_terminal_assistant_final";
const finalText = [
internal/cloud/workbench-turn-projection.ts
+33 -2
View File
@@ -4,7 +4,7 @@
*/
export const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]);
export const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active", "processing", "busy", "creating"]);
export const RUNNING_STATUSES = new Set(["running", "retrying", "pending", "queued", "accepted", "dispatching", "streaming", "active", "processing", "busy", "creating"]);
export function createWorkbenchTurnProjection({ turnId = null, traceId = null, result = null, session = null, trace = null } = {}) {
const projectionTraceId = textValue(traceId ?? trace?.traceId ?? result?.traceId ?? session?.lastTraceId) || null;
@@ -72,6 +72,7 @@ export function traceTerminalEvidence(trace = null) {
const direct = objectValue(trace?.terminalEvidence);
if (direct) {
const status = terminalStatusFromValue(direct.status ?? direct.terminalStatus ?? trace?.status) ?? "completed";
if (status !== "completed" && retryableProviderInterruptionEvidence(direct, trace)) return null;
return { source: "trace-terminal-evidence", status, evidence: direct, valuesRedacted: true };
}
const finalResponse = objectValue(trace?.finalResponse);
@@ -91,11 +92,17 @@ export function normalizeWorkbenchStatus(value) {
function terminalTurnEvidence({ result = null, traceTerminal = null } = {}) {
const resultStatus = terminalStatusFromValue(result?.status ?? result?.terminalStatus ?? result?.agentRun?.terminalStatus);
if (resultStatus) return { source: "result", status: resultStatus, finalResponse: traceTerminal?.finalResponse ?? null, valuesRedacted: true };
if (resultStatus) {
if (resultStatus !== "completed" && retryableProviderInterruptionEvidence(result, result?.agentRun, result?.providerTrace, traceTerminal?.evidence)) return null;
return { source: "result", status: resultStatus, finalResponse: traceTerminal?.finalResponse ?? null, valuesRedacted: true };
}
if (traceTerminal && retryableProviderInterruptionEvidence(traceTerminal.evidence)) return null;
return traceTerminal;
}
function activeTurnEvidence({ result = null, session = null, trace = null } = {}) {
const retryEvidence = retryableProviderInterruptionEvidence(result, result?.agentRun, result?.providerTrace, trace);
if (retryEvidence) return { source: "provider-retry", status: "retrying", evidence: retryEvidence, valuesRedacted: true };
const resultStatus = firstStatus([
result?.status,
result?.agentRun?.status,
@@ -133,6 +140,7 @@ function terminalTraceEventEvidence(events = []) {
for (let index = events.length - 1; index >= 0; index -= 1) {
const event = events[index];
if (!event || typeof event !== "object") continue;
if (retryableProviderInterruptionEvidence(event)) continue;
const terminal = event.terminal === true || event.final === true || event.replyAuthority === true;
if (!terminal) continue;
const status = terminalStatusFromValue(event.status ?? event.terminalStatus) ?? (event.error || event.errorCode ? "failed" : "completed");
@@ -178,12 +186,35 @@ function isAssistantTraceEvent(event = {}) {
function activeTraceEventStatus(events = []) {
for (const event of [...events].reverse()) {
if (retryableProviderInterruptionEvidence(event)) return "retrying";
const status = normalizeWorkbenchStatus(event?.status ?? event?.type);
if (RUNNING_STATUSES.has(status)) return normalizeActiveStatus(status);
}
return null;
}
function retryableProviderInterruptionEvidence(...values) {
for (const value of values) {
const direct = retryableProviderInterruptionRecord(value);
if (direct) return direct;
}
return null;
}
function retryableProviderInterruptionRecord(value) {
const record = objectValue(value);
if (!record) return null;
const kind = normalizeWorkbenchStatus(record.failureKind ?? record.errorCode ?? record.code ?? record.name);
if (kind === "provider-stream-disconnected" && record.willRetry === true) {
return { failureKind: kind, willRetry: true, runId: textValue(record.runId) || null, commandId: textValue(record.commandId) || null, valuesRedacted: true };
}
for (const nested of [record.payload, record.providerTrace, record.agentRun, record.error, record.blocker, record.terminalEvidence]) {
const found = retryableProviderInterruptionRecord(nested);
if (found) return found;
}
return null;
}
function hasTraceProjection(trace) {
return Boolean(trace && trace.status !== "missing" && (normalizedEventCount(trace) > 0 || traceLastEvent(trace)));
}
web/hwlab-cloud-web/src/stores/workbench-session.ts
+2 -2
View File
@@ -326,12 +326,12 @@ function latestSessionMessage(messages: ChatMessage[], activeSessionId: string |
function isActiveStatus(value: unknown): boolean {
const status = normalizeSessionStatus(value);
return ["active", "running", "pending", "accepted", "processing", "busy", "creating"].includes(status ?? "");
return ["active", "running", "retrying", "pending", "queued", "accepted", "dispatching", "streaming", "processing", "busy", "creating"].includes(status ?? "");
}
function isRunningIndicatorStatus(value: unknown): boolean {
const status = normalizeSessionStatus(value);
return ["running", "pending", "accepted", "processing", "busy"].includes(status ?? "");
return ["running", "retrying", "pending", "queued", "accepted", "dispatching", "streaming", "processing", "busy"].includes(status ?? "");
}
function isTerminalStatus(value: unknown): boolean {
web/hwlab-cloud-web/src/stores/workbench.ts
+1 -1
View File
@@ -1543,7 +1543,7 @@ function statusFromResult(status: string | undefined): ChatMessage["status"] {
}
function isTraceActiveStatus(status: unknown): boolean {
return ["accepted", "pending", "processing", "running", "busy", "creating"].includes(String(status ?? "").trim().toLowerCase().replace(/_/gu, "-"));
return ["accepted", "pending", "queued", "dispatching", "streaming", "processing", "running", "retrying", "busy", "creating"].includes(String(status ?? "").trim().toLowerCase().replace(/_/gu, "-"));
}
function readString(key: string, fallback: string): string {