fix: index historical AgentRun traces

2026-06-06 11:20:48 +08:00
parent 4a4aa1bc29
commit 7c6df8d9b4
3 changed files with 369 additions and 17 deletions
@@ -606,22 +606,35 @@ export async function steerAgentRunChatTurn({ traceId, currentResult = null, par
 }

 export async function loadPersistedAgentRunResult(traceId, options = {}) {
-  if (!safeTraceId(traceId) || typeof options.accessController?.getAgentSessionByTraceId !== "function") return null;
-  const session = await options.accessController.getAgentSessionByTraceId(traceId);
-  const agentRun = session?.session?.agentRun;
+  const safeId = safeTraceId(traceId);
+  if (!safeId || typeof options.accessController?.getAgentSessionByTraceId !== "function") return null;
+  const session = await options.accessController.getAgentSessionByTraceId(safeId);
+  const traceEvidence = agentSessionTraceEvidence(session, safeId);
+  const latestTraceId = safeTraceId(session?.lastTraceId ?? session?.session?.lastTraceId ?? session?.session?.traceId);
+  const topLevelAgentRun = session?.session?.agentRun && typeof session.session.agentRun === "object" ? session.session.agentRun : null;
+  const agentRunTraceId = safeTraceId(topLevelAgentRun?.traceId);
+  const mayUseTopLevelAgentRun = latestTraceId === safeId || agentRunTraceId === safeId || (!latestTraceId && !agentRunTraceId);
+  const agentRun = traceEvidence?.agentRun ?? (mayUseTopLevelAgentRun ? topLevelAgentRun : null);
+  const restored = persistedAgentRunResultFromSession({ traceId: safeId, session, agentRun, traceEvidence, options });
+  if (restored) return restored;
+  return await repairPersistedAgentRunResultFromRunCommands({ traceId: safeId, session, options });
+}
+
+function persistedAgentRunResultFromSession({ traceId, session, agentRun, traceEvidence, options = {} }) {
  if (!agentRun || typeof agentRun !== "object" || !agentRun.runId || !agentRun.commandId) return null;
+  const status = firstNonEmpty(traceEvidence?.status, session?.status === "canceled" ? "canceled" : null, agentRun.terminalStatus, agentRun.commandState, "running");
  return {
    accepted: true,
-    status: session.status === "canceled" ? "canceled" : "running",
+    status,
    shortConnection: true,
    traceId,
-    conversationId: safeConversationId(session.conversationId) || agentRun.conversationId || null,
-    sessionId: safeSessionId(session.id) || agentRun.sessionId || null,
-    threadId: safeOpaqueId(session.threadId) || agentRun.threadId || null,
-    messageId: session.session?.messageId ?? `msg_${traceId.slice(4)}`,
-    createdAt: session.startedAt ?? session.updatedAt ?? nowIso(options.now),
-    updatedAt: session.updatedAt ?? nowIso(options.now),
-    ownerUserId: session.ownerUserId,
+    conversationId: safeConversationId(traceEvidence?.conversationId ?? session?.conversationId) || agentRun.conversationId || null,
+    sessionId: safeSessionId(traceEvidence?.sessionId ?? session?.id) || agentRun.sessionId || null,
+    threadId: safeOpaqueId(traceEvidence?.threadId ?? session?.threadId) || agentRun.threadId || null,
+    messageId: traceEvidence?.messageId ?? session?.session?.messageId ?? `msg_${traceId.slice(4)}`,
+    createdAt: traceEvidence?.createdAt ?? session?.startedAt ?? session?.updatedAt ?? nowIso(options.now),
+    updatedAt: traceEvidence?.updatedAt ?? session?.updatedAt ?? nowIso(options.now),
+    ownerUserId: session?.ownerUserId,
    provider: providerForBackendProfile(agentRun.backendProfile ?? "deepseek"),
    model: modelForBackendProfile(agentRun.backendProfile, options.env ?? process.env),
    backend: backendForBackendProfile(agentRun.backendProfile ?? "deepseek"),
@@ -629,13 +642,129 @@ export async function loadPersistedAgentRunResult(traceId, options = {}) {
    capabilityLevel: AGENTRUN_CAPABILITY_LEVEL,
    sessionMode: AGENTRUN_SESSION_MODE,
    implementationType: AGENTRUN_IMPLEMENTATION_TYPE,
-    finalResponse: session.session?.finalResponse ?? null,
-    traceSummary: session.session?.traceSummary ?? null,
+    finalResponse: traceEvidence?.finalResponse ?? session?.session?.finalResponse ?? null,
+    traceSummary: traceEvidence?.traceSummary ?? session?.session?.traceSummary ?? null,
    agentRun: { ...agentRun, adapter: ADAPTER_ID, valuesPrinted: false },
    valuesPrinted: false
  };
 }

+async function repairPersistedAgentRunResultFromRunCommands({ traceId, session, options = {} }) {
+  const seedAgentRun = session?.session?.agentRun && typeof session.session.agentRun === "object" ? session.session.agentRun : null;
+  if (!seedAgentRun?.runId) return null;
+  const env = options.env ?? process.env;
+  const fetchImpl = options.fetchImpl ?? globalThis.fetch;
+  const managerUrl = resolveAgentRunManagerUrl(env, seedAgentRun.managerUrl);
+  const timeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS, 20_000);
+  const commands = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(seedAgentRun.runId)}/commands?afterSeq=0&limit=100`, { method: "GET", timeoutMs });
+  const command = (Array.isArray(commands?.items) ? commands.items : []).find((item) => agentRunCommandMatchesTrace(item, traceId));
+  if (!command?.id) return null;
+  const [result, eventsResponse] = await Promise.all([
+    agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(seedAgentRun.runId)}/commands/${encodeURIComponent(command.id)}/result`, { method: "GET", timeoutMs }),
+    agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(seedAgentRun.runId)}/events?afterSeq=0&limit=500`, { method: "GET", timeoutMs }).catch(() => ({ items: [] }))
+  ]);
+  const rawEvents = Array.isArray(eventsResponse?.items) ? eventsResponse.items : [];
+  const commandSeqs = rawEvents
+    .filter((event) => agentRunEventCommandId(event) === command.id)
+    .map((event) => Number(event?.seq ?? 0))
+    .filter(Number.isFinite);
+  const firstSeq = commandSeqs.length ? Math.min(...commandSeqs) : 0;
+  const afterSeq = firstSeq > 0 ? firstSeq - 1 : 0;
+  const endSeq = Math.max(0, ...commandSeqs);
+  const traceEvents = rawEvents.filter((event) => agentRunEventBelongsToTrace(event, { currentCommandId: command.id, afterSeq, endSeq }));
+  const now = nowIso(options.now);
+  const backendProfile = firstNonEmpty(seedAgentRun.backendProfile, result?.backendProfile, command?.payload?.providerProfile, "deepseek");
+  const agentRun = {
+    ...seedAgentRun,
+    adapter: ADAPTER_ID,
+    managerUrl,
+    backendProfile,
+    runId: result?.runId ?? seedAgentRun.runId,
+    commandId: command.id,
+    attemptId: result?.attemptId ?? seedAgentRun.attemptId ?? null,
+    runnerId: result?.runnerId ?? seedAgentRun.runnerId ?? null,
+    jobName: result?.jobName ?? seedAgentRun.jobName ?? null,
+    namespace: result?.namespace ?? seedAgentRun.namespace ?? DEFAULT_RUNNER_NAMESPACE,
+    status: result?.status ?? command.state ?? seedAgentRun.status ?? null,
+    runStatus: result?.runStatus ?? seedAgentRun.runStatus ?? null,
+    commandState: result?.commandState ?? command.state ?? null,
+    terminalStatus: result?.terminalStatus ?? null,
+    eventStartSeq: afterSeq > 0 ? afterSeq + 1 : null,
+    lastSeq: endSeq || 0,
+    traceId,
+    sessionId: result?.sessionRef?.sessionId ?? command?.payload?.sessionId ?? seedAgentRun.sessionId ?? null,
+    conversationId: result?.sessionRef?.conversationId ?? command?.payload?.conversationId ?? seedAgentRun.conversationId ?? session?.conversationId ?? null,
+    threadId: result?.sessionRef?.threadId ?? command?.payload?.threadId ?? seedAgentRun.threadId ?? session?.threadId ?? null,
+    valuesPrinted: false
+  };
+  const finalText = firstNonEmpty(result?.reply);
+  const finalResponse = finalText ? {
+    text: finalText,
+    textChars: finalText.length,
+    messageId: `msg_${traceId.slice(4)}`,
+    role: "assistant",
+    status: result?.status ?? result?.terminalStatus ?? "completed",
+    traceId,
+    createdAt: command.updatedAt ?? result?.updatedAt ?? now,
+    updatedAt: command.updatedAt ?? result?.updatedAt ?? now,
+    source: "agentrun-command-result-repair",
+    valuesPrinted: false
+  } : null;
+  const traceSummary = {
+    traceId,
+    source: "agent-session-agentrun-command-repair",
+    sourceEventCount: traceEvents.length,
+    terminalStatus: result?.terminalStatus ?? command.state ?? null,
+    agentRun: { runId: agentRun.runId, commandId: agentRun.commandId, lastSeq: agentRun.lastSeq, eventStartSeq: agentRun.eventStartSeq, valuesPrinted: false },
+    updatedAt: now,
+    valuesPrinted: false
+  };
+  const traceEvidence = {
+    traceId,
+    status: result?.status ?? result?.terminalStatus ?? command.state ?? "completed",
+    conversationId: agentRun.conversationId,
+    sessionId: safeSessionId(command?.payload?.hwlabSessionId) || safeSessionId(session?.id) || agentRun.sessionId,
+    threadId: agentRun.threadId,
+    messageId: `msg_${traceId.slice(4)}`,
+    createdAt: command.createdAt ?? session?.startedAt ?? now,
+    updatedAt: command.updatedAt ?? now,
+    finalResponse,
+    traceSummary,
+    agentRun,
+    valuesRedacted: true,
+    secretMaterialStored: false
+  };
+  await persistTraceRepairEvidence({ session, traceId, traceEvidence, options });
+  return persistedAgentRunResultFromSession({ traceId, session, agentRun, traceEvidence, options });
+}
+
+function agentRunCommandMatchesTrace(command, traceId) {
+  const payload = command?.payload && typeof command.payload === "object" ? command.payload : {};
+  return command?.idempotencyKey === traceId || payload.traceId === traceId;
+}
+
+async function persistTraceRepairEvidence({ session, traceId, traceEvidence, options = {} }) {
+  if (!session?.id || !session.ownerUserId || typeof options.accessController?.recordAgentSessionOwner !== "function") return;
+  await options.accessController.recordAgentSessionOwner({
+    ownerUserId: session.ownerUserId,
+    sessionId: session.id,
+    projectId: session.projectId,
+    agentId: session.agentId,
+    status: session.status,
+    conversationId: session.conversationId,
+    threadId: session.threadId,
+    lastTraceId: session.lastTraceId,
+    session: { traceResults: { [traceId]: traceEvidence }, source: "agent-session-trace-repair", valuesRedacted: true, secretMaterialStored: false }
+  });
+}
+
+function agentSessionTraceEvidence(session, traceId) {
+  const id = safeTraceId(traceId);
+  const traceResults = session?.session?.traceResults && typeof session.session.traceResults === "object" ? session.session.traceResults : null;
+  const evidence = id && traceResults?.[id] && typeof traceResults[id] === "object" ? traceResults[id] : null;
+  return evidence ? { ...evidence, traceId: id } : null;
+}
+
 export function agentRunSessionEvidence(payload = {}) {
  if (!payload?.agentRun) return {};
  return {