From f3309018630c0a078d4ec3fd3615358af4346e63 Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Mon, 22 Jun 2026 18:01:47 +0800 Subject: [PATCH] fix: shorten AgentRun admission and stabilize Workbench projection (#1909) * fix: shorten AgentRun chat admission path * fix: stabilize Workbench turn projection display --- internal/cloud/code-agent-agentrun-adapter.ts | 124 ++++++++++++++---- .../src/components/agent/TraceTimeline.vue | 40 +++++- .../workbench/ConversationPanel.vue | 27 +++- web/hwlab-cloud-web/src/stores/workbench.ts | 19 ++- 4 files changed, 178 insertions(+), 32 deletions(-) diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index c89bd89a..734ef3a5 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -274,8 +274,8 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI type: "request", status: "accepted", label: "agentrun:request:accepted", - message: "HWLAB Code Agent request accepted by the AgentRun v0.1 adapter; hwlab-cloud-api will reuse an active runner when the HWLAB session has an active AgentRun reuse window, otherwise it will create run/command/runner-job over the k3s Service DNS.", - waitingFor: "agentrun-run-reuse-or-create", + message: "HWLAB Code Agent request accepted by the AgentRun v0.1 adapter; hwlab-cloud-api will create a fresh run/command tied to the session-scoped AgentRun PVC, then ensure the runner Job outside the browser admission path.", + waitingFor: "agentrun-run-create", adapter: ADAPTER_ID, managerHost: new URL(managerUrl).hostname, ...codeAgentOtelTraceFields(traceId, env), @@ -382,7 +382,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI let sessionId = baseSessionId; let sessionReset = false; let finalDispatchParams = params; - let run, command, runnerJob, mapping; + let run, command, mapping; for (let attempt = 0; attempt < 2; attempt += 1) { let attemptParams = params; if (sessionReset) { @@ -458,36 +458,93 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI runId, commandId, backendProfile, waitingFor: "agentrun-runner-job-create", valuesPrinted: false, }, backendProfile)); const runnerJobInput = buildAgentRunRunnerJobInput({ env, traceId, commandId, ownerApiKey, toolCapabilities, backendProfile }); - try { - const runnerJobStartedAt = Date.now(); - runnerJob = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs/" + encodeURIComponent(runId) + "/runner-jobs", method: "POST", body: runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, runId, commandId, stage: "runner-job-create" }); - void emitCodeAgentOtelSpan("agentrun_runner_job_create", traceId, env, { startTimeMs: runnerJobStartedAt, attributes: { runId, commandId, backendProfile, managerUrl, jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? null } }); - } catch (error) { - if (attempt === 0 && await shouldResetSessionAfterEviction("session-store-evicted", error?.message)) { - sessionReset = true; - continue; - } - throw error; - } - mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params: dispatchParams }); + mapping = agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob: null, traceId, startedAt, params: dispatchParams }); + traceStore.append(traceId, agentRunTraceEvent({ + type: "backend", + status: "running", + label: "agentrun:runner-job:queued", + message: "AgentRun run/command admission is durable; hwlab-cloud-api will create the runner Job asynchronously so the browser request is not blocked by Kubernetes Job creation.", + runId: mapping.runId, + commandId: mapping.commandId, + backendProfile, + waitingFor: "agentrun-runner-job-create", + valuesPrinted: false + }, mapping)); + scheduleAgentRunRunnerJobCreateAfterAdmission({ fetchImpl, managerUrl, runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, run, command, mapping, options, params: dispatchParams }); finalDispatchParams = dispatchParams; break; } + return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: finalDispatchParams, options, traceId }), mapping, traceStore, traceId }); +} + +function scheduleAgentRunRunnerJobCreateAfterAdmission(args) { + setTimeout(() => { + void ensureAgentRunRunnerJobCreateAfterAdmission(args).catch((error) => { + markAgentRunRunnerJobCreateFailedAfterAdmission(args, error); + }); + }, 0); +} + +async function ensureAgentRunRunnerJobCreateAfterAdmission({ fetchImpl, managerUrl, runnerJobInput, timeoutMs, env = process.env, traceStore = defaultCodeAgentTraceStore, traceId, backendProfile, mapping = {}, options = {}, params = {} }) { + const runId = requiredString(mapping.runId, "mapping.runId"); + const commandId = requiredString(mapping.commandId, "mapping.commandId"); + const runnerJobStartedAt = Date.now(); + const runnerJob = await agentRunDispatchJson({ fetchImpl, managerUrl, path: "/api/v1/runs/" + encodeURIComponent(runId) + "/runner-jobs", method: "POST", body: runnerJobInput, timeoutMs, env, traceStore, traceId, backendProfile, runId, commandId, stage: "runner-job-create-async" }); + const nextMapping = agentRunMappingWithRunnerJob(mapping, runnerJob); + void emitCodeAgentOtelSpan("agentrun_runner_job_create_async", traceId, env, { startTimeMs: runnerJobStartedAt, attributes: { runId, commandId, backendProfile, managerUrl, jobName: nextMapping.jobName ?? null } }); traceStore.append(traceId, agentRunTraceEvent({ type: "backend", status: "running", label: "agentrun:runner-job:created", - message: `AgentRun runner Job ${mapping.jobName ?? "unknown"} created in namespace ${mapping.namespace ?? DEFAULT_RUNNER_NAMESPACE}.`, - runId: mapping.runId, - commandId: mapping.commandId, - attemptId: mapping.attemptId, - runnerId: mapping.runnerId, - jobName: mapping.jobName, - namespace: mapping.namespace, + message: `AgentRun runner Job ${nextMapping.jobName ?? "unknown"} created in namespace ${nextMapping.namespace ?? DEFAULT_RUNNER_NAMESPACE}.`, + runId: nextMapping.runId, + commandId: nextMapping.commandId, + attemptId: nextMapping.attemptId, + runnerId: nextMapping.runnerId, + jobName: nextMapping.jobName, + namespace: nextMapping.namespace, waitingFor: "agentrun-result", valuesPrinted: false - }, mapping)); - return decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params: finalDispatchParams, options, traceId }), mapping, traceStore, traceId }); + }, nextMapping)); + const current = options.codeAgentChatResults?.get?.(traceId); + const base = current && typeof current === "object" ? current : initialAgentRunChatResult({ params, options, traceId }); + options.codeAgentChatResults?.set?.(traceId, decorateAgentRunRunningResult({ base: { ...base, agentRun: nextMapping }, mapping: nextMapping, traceStore, traceId })); +} + +function markAgentRunRunnerJobCreateFailedAfterAdmission({ traceStore = defaultCodeAgentTraceStore, traceId, backendProfile, mapping = {}, options = {}, params = {} }, error) { + const failureKind = agentRunDispatchFailureKind(error); + const retryAttempt = Number(error?.retryAttempt ?? 0); + const retryMax = Number(error?.retryMax ?? agentRunDispatchRetryPolicy().maxRetries); + const message = `AgentRun runner Job creation failed after durable run/command admission; Code Agent stopped this turn instead of silently waiting. failureKind=${failureKind}`; + traceStore.append(traceId, agentRunTraceEvent({ + type: "error", + eventType: "error", + status: "failed", + label: `agentrun:runner-job:create-async-failed:${failureKind}`, + errorCode: failureKind, + failureKind, + retryAttempt: Number.isFinite(retryAttempt) ? retryAttempt : 0, + retryMax: Number.isFinite(retryMax) ? retryMax : null, + retryExhausted: error?.retryExhausted === true, + runId: mapping.runId ?? null, + commandId: mapping.commandId ?? null, + waitingFor: "agentrun-runner-job-failed", + terminal: true, + message, + valuesPrinted: false + }, mapping || backendProfile)); + const current = options.codeAgentChatResults?.get?.(traceId); + const base = current && typeof current === "object" + ? current + : decorateAgentRunRunningResult({ base: initialAgentRunChatResult({ params, options, traceId }), mapping, traceStore, traceId }); + const failed = agentRunResultToCodeAgentPayload({ + base: { ...base, agentRun: { ...mapping, status: "runner-job-failed", commandState: "failed", terminalStatus: "failed", failureKind, valuesPrinted: false } }, + result: { terminalStatus: "failed", failureKind, failureMessage: message, runId: mapping.runId ?? null, commandId: mapping.commandId ?? null, valuesPrinted: false }, + traceStore, + traceId, + appendResultEvent: false + }); + options.codeAgentChatResults?.set?.(traceId, failed); } async function agentRunDispatchJson({ fetchImpl, managerUrl, path, method = "GET", body = undefined, timeoutMs, env = process.env, traceStore = defaultCodeAgentTraceStore, traceId, backendProfile, runId = null, commandId = null, stage = "agentrun-dispatch", terminalOnFailure = true } = {}) { @@ -2810,6 +2867,7 @@ function resolveAgentRunBackendProfile(env = process.env, params = {}) { function agentRunMapping({ env, managerUrl, backendProfile, run, command, runnerJob, traceId, startedAt, params = {} }) { const threadReused = Boolean(safeOpaqueId(params.threadId)); + const runnerJobCreated = Boolean(runnerJob); return { adapter: ADAPTER_ID, managerUrl, @@ -2822,7 +2880,7 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner runnerJobId: runnerJob?.id ?? null, jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? null, namespace: runnerJob?.namespace ?? runnerJob?.jobIdentity?.namespace ?? DEFAULT_RUNNER_NAMESPACE, - status: "runner-job-created", + status: runnerJobCreated ? "runner-job-created" : "runner-job-pending", runStatus: run.status ?? null, commandState: command.state ?? null, terminalStatus: null, @@ -2833,6 +2891,7 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner conversationId: run?.sessionRef?.conversationId ?? null, threadId: run?.sessionRef?.threadId ?? null, runnerReused: false, + runnerJobCount: runnerJobCreated ? 1 : 0, threadReused, persistentResume: threadReused, reused: false, @@ -2843,6 +2902,21 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner }; } +function agentRunMappingWithRunnerJob(mapping = {}, runnerJob = null) { + return { + ...mapping, + attemptId: runnerJob?.attemptId ?? runnerJob?.runner?.attemptId ?? mapping.attemptId ?? null, + runnerId: runnerJob?.runnerId ?? runnerJob?.runner?.runnerId ?? mapping.runnerId ?? null, + runnerJobId: runnerJob?.id ?? mapping.runnerJobId ?? null, + jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? mapping.jobName ?? null, + namespace: runnerJob?.namespace ?? runnerJob?.jobIdentity?.namespace ?? mapping.namespace ?? DEFAULT_RUNNER_NAMESPACE, + status: "runner-job-created", + runnerJobCount: Math.max(1, Number(mapping.runnerJobCount ?? 0) + 1), + updatedAt: nowIso(), + valuesPrinted: false + }; +} + function agentRunReusedMapping({ previous = {}, run = {}, command = {}, runnerJob = null, traceId, startedAt, backendProfile, managerUrl, env }) { const threadReused = Boolean(safeOpaqueId(run?.sessionRef?.threadId ?? previous.threadId)); return { diff --git a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue index 54f46793..882ec4f9 100644 --- a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue +++ b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue @@ -7,7 +7,7 @@ import { useBottomFollowScroll } from "@/composables/useBottomFollowScroll"; import { formatDisplayClock } from "@/config/runtime"; import { traceDisplayRows, type TraceEventRow } from "../../../../../tools/src/hwlab-cli/trace-renderer.ts"; -const props = defineProps<{ trace?: RunnerTrace | null; defaultExpanded?: boolean; storageKey?: string; autoExpanded?: boolean | null }>(); +const props = defineProps<{ trace?: RunnerTrace | null; defaultExpanded?: boolean; storageKey?: string; autoExpanded?: boolean | null; sealedFinalText?: string }>(); const listRef = ref(null); const expanded = ref(false); @@ -15,7 +15,8 @@ const { following, keepBottomAfterUpdate, onScroll, scrollToBottom } = useBottom const events = computed(() => Array.isArray(props.trace?.events) ? props.trace.events : []); const eventCount = computed(() => props.trace?.eventCount ?? events.value.length); -const readableRows = computed(() => traceDisplayRows(traceRecord(props.trace), events.value as Record[], { formatClock: formatDisplayClock })); +const rawReadableRows = computed(() => traceDisplayRows(traceRecord(props.trace), events.value as Record[], { formatClock: formatDisplayClock })); +const readableRows = computed(() => filterReadableRows(rawReadableRows.value, props.sealedFinalText)); const projectionDiagnosticLabel = computed(() => { const projection = props.trace?.projection ?? null; const health = projection?.projectionHealth ?? props.trace?.projectionHealth; @@ -82,6 +83,41 @@ function rowIsTool(row: TraceEventRow): boolean { return row.rowId.startsWith("tool:") || /commandExecution/u.test(row.header); } +function filterReadableRows(rows: TraceEventRow[], sealedFinalText: string | undefined): TraceEventRow[] { + const finalText = normalizeComparableText(sealedFinalText); + if (!finalText || finalText.length < 24) return rows; + return rows.filter((row) => !assistantRowDuplicatesFinal(row, finalText)); +} + +function assistantRowDuplicatesFinal(row: TraceEventRow, finalText: string): boolean { + if (rowIsTool(row)) return false; + const rowTextRaw = [row.header, row.preview, row.body].filter(Boolean).join("\n"); + if (!/(?:助手消息|assistant\s+message|assistant)/iu.test(rowTextRaw)) return false; + const rowText = normalizeComparableText(rowTextRaw); + if (!rowText || rowText.length < 24) return true; + return finalText.includes(rowText) || rowText.includes(finalText) || longestSharedSubstringLength(finalText, rowText) >= Math.min(80, Math.floor(Math.min(finalText.length, rowText.length) * 0.7)); +} + +function normalizeComparableText(value: unknown): string { + return String(value ?? "").replace(/\s+/gu, " ").trim(); +} + +function longestSharedSubstringLength(a: string, b: string): number { + const left = a.length <= b.length ? a : b; + const right = a.length <= b.length ? b : a; + const max = Math.min(left.length, 320); + let best = 0; + for (let start = 0; start < max; start += 1) { + for (let end = Math.min(max, start + 180); end > start + best; end -= 1) { + if (right.includes(left.slice(start, end))) { + best = end - start; + break; + } + } + } + return best; +} + function toolCommandPreview(row: TraceEventRow): string | null { if (!rowIsTool(row)) return null; const preview = singleLinePreview(row.preview); diff --git a/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue b/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue index 4492effe..c94de355 100644 --- a/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue +++ b/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue @@ -294,7 +294,20 @@ function messageTimingForDisplay(message: ChatMessage): ChatMessage["timing"] { } function terminalMessageDurationMs(message: ChatMessage, timing: ChatMessage["timing"]): number | null { - return finiteDurationMs(message.durationMs) ?? finiteDurationMs(timing?.durationMs) ?? finiteDurationMs(message.runnerTrace?.durationMs); + const recorded = firstPositiveDurationMs(message.durationMs, timing?.durationMs, message.runnerTrace?.durationMs, message.runnerTrace?.timing?.durationMs); + if (recorded !== null) return recorded; + const elapsed = elapsedTerminalDurationMs(message, timing); + if (elapsed !== null && elapsed > 0) return elapsed; + return finiteDurationMs(message.durationMs) ?? finiteDurationMs(timing?.durationMs) ?? finiteDurationMs(message.runnerTrace?.durationMs) ?? finiteDurationMs(message.runnerTrace?.timing?.durationMs) ?? elapsed; +} + +function elapsedTerminalDurationMs(message: ChatMessage, timing: ChatMessage["timing"]): number | null { + const trace = message.runnerTrace as Record | null | undefined; + const traceTiming = trace?.timing as Record | null | undefined; + const startedAt = timestampMs(timing?.startedAt ?? message.startedAt ?? traceTiming?.startedAt ?? trace?.startedAt); + const endedAt = timestampMs(timing?.finishedAt ?? message.finishedAt ?? traceTiming?.finishedAt ?? trace?.finishedAt ?? timing?.lastEventAt ?? message.lastEventAt ?? traceTiming?.lastEventAt ?? trace?.lastEventAt); + if (startedAt === null || endedAt === null || endedAt < startedAt) return null; + return Math.trunc(endedAt - startedAt); } function durationSince(timestamp: unknown): number | null { @@ -315,8 +328,16 @@ function finiteDurationMs(value: unknown): number | null { return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; } +function firstPositiveDurationMs(...values: unknown[]): number | null { + for (const value of values) { + const duration = finiteDurationMs(value); + if (duration !== null && duration > 0) return duration; + } + return null; +} + function formatDuration(ms: number): string { - const seconds = Math.max(0, Math.floor(ms / 1000)); + const seconds = ms > 0 && ms < 1000 ? 1 : Math.max(0, Math.floor(ms / 1000)); if (seconds < 60) return `${seconds} 秒`; const minutes = Math.floor(seconds / 60); if (minutes < 60) return `${minutes} 分 ${seconds % 60} 秒`; @@ -353,7 +374,7 @@ function formatDuration(ms: number): string { - + diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index be16f63d..9a4ef5e6 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1619,12 +1619,19 @@ function messageTimingPatchForMerge(message: ChatMessage, value: unknown): Parti if (!patch.timing) return patch; const terminal = timingPatchIsTerminal(value, patch.timing); const durationMs = terminal - ? firstFiniteNumber(patch.timing.durationMs, elapsedTimingDurationMs(patch.timing)) + ? terminalTimingDurationMsForMerge(message, patch.timing) : maxFiniteNumber(patch.timing.durationMs, message.durationMs, message.timing?.durationMs, message.runnerTrace?.durationMs, message.runnerTrace?.timing?.durationMs); const timing = { ...patch.timing, durationMs } as WorkbenchTurnTimingProjection; return { timing, startedAt: timing.startedAt ?? null, lastEventAt: timing.lastEventAt ?? null, finishedAt: timing.finishedAt ?? null, durationMs }; } +function terminalTimingDurationMsForMerge(message: ChatMessage, timing: WorkbenchTurnTimingProjection): number | null { + const elapsed = elapsedTimingDurationMs(timing); + const positive = firstPositiveFiniteNumber(timing.durationMs, elapsed, message.durationMs, message.timing?.durationMs, message.runnerTrace?.durationMs, message.runnerTrace?.timing?.durationMs); + if (positive !== null) return positive; + return firstFiniteNumber(timing.durationMs, elapsed, message.durationMs, message.timing?.durationMs, message.runnerTrace?.durationMs, message.runnerTrace?.timing?.durationMs); +} + function timingPatchIsTerminal(value: unknown, timing: WorkbenchTurnTimingProjection): boolean { const record = recordValue(value); const trace = recordValue(record?.runnerTrace); @@ -1640,6 +1647,14 @@ function elapsedTimingDurationMs(timing: WorkbenchTurnTimingProjection): number return Math.trunc(end - start); } +function firstPositiveFiniteNumber(...values: unknown[]): number | null { + for (const value of values) { + const number = firstFiniteNumber(value); + if (number !== null && number > 0) return number; + } + return null; +} + function maxFiniteNumber(...values: unknown[]): number | null { let max: number | null = null; for (const value of values) { @@ -1824,12 +1839,12 @@ function messageNeedsTerminalDiagnostics(message: ChatMessage): boolean { function messageNeedsTraceHydration(message: ChatMessage): boolean { if (message.role !== "agent") return false; - if (messageHasCompletedFinalResponse(message)) return false; const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); if (!traceId) return false; const trace = message.runnerTrace; const events = Array.isArray(trace?.events) ? trace.events : []; const eventCount = firstFiniteNumber(trace?.eventCount) ?? events.length; + if (messageHasCompletedFinalResponse(message) && events.length === 0) return true; if (trace?.fullTraceLoaded === true && trace.eventsCompacted !== true) return events.length === 0 && (eventCount > 0 || isTraceActiveStatus(trace?.status) || isTraceActiveStatus(message.status)); return events.length === 0 || trace?.eventsCompacted === true || trace?.fullTraceLoaded !== true; }