diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 1eb15a2a..1561401e 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -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); diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 5017ab0b..d4679119 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -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 = [ diff --git a/internal/cloud/workbench-turn-projection.ts b/internal/cloud/workbench-turn-projection.ts index 24171421..c514cb46 100644 --- a/internal/cloud/workbench-turn-projection.ts +++ b/internal/cloud/workbench-turn-projection.ts @@ -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))); } diff --git a/web/hwlab-cloud-web/src/stores/workbench-session.ts b/web/hwlab-cloud-web/src/stores/workbench-session.ts index ee3ce3f1..acbe82d2 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-session.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-session.ts @@ -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 { diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index da03b321..f44cba6c 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -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 {