Merge pull request #972 from pikasTech/fix/issue955-historical-trace-index
fix: index historical AgentRun traces
This commit is contained in:
Lyon
@@ -1555,7 +1555,7 @@ class MemoryAccessStore {
|
||||
async getAgentSession(id) { return this.agentSessions.get(id) ?? null; }
|
||||
async getAgentSessionByTraceId(traceId) {
|
||||
return [...this.agentSessions.values()]
|
||||
.filter((session) => session.lastTraceId === traceId)
|
||||
.filter((session) => session.lastTraceId === traceId || Boolean(agentSessionTraceEvidence(session, traceId)) || agentSessionContainsTraceId(session, traceId))
|
||||
.sort((a, b) => String(b.updatedAt ?? "").localeCompare(String(a.updatedAt ?? "")))[0] ?? null;
|
||||
}
|
||||
async listAgentSessionsForUser(input = {}) {
|
||||
@@ -1694,7 +1694,7 @@ class PostgresAccessStore extends MemoryAccessStore {
|
||||
return pgAgentSession(result.rows?.[0]);
|
||||
}
|
||||
async getAgentSession(id) { await this.ensureSchema(); const result = await this.query("SELECT id, project_id, agent_id, status, started_at, ended_at, owner_user_id, conversation_id, thread_id, last_trace_id, session_json, updated_at FROM agent_sessions WHERE id = $1 LIMIT 1", [id]); return pgAgentSession(result.rows?.[0]); }
|
||||
async getAgentSessionByTraceId(traceId) { await this.ensureSchema(); const result = await this.query("SELECT id, project_id, agent_id, status, started_at, ended_at, owner_user_id, conversation_id, thread_id, last_trace_id, session_json, updated_at FROM agent_sessions WHERE last_trace_id = $1 ORDER BY updated_at DESC NULLS LAST LIMIT 1", [traceId]); return pgAgentSession(result.rows?.[0]); }
|
||||
async getAgentSessionByTraceId(traceId) { await this.ensureSchema(); const result = await this.query("SELECT id, project_id, agent_id, status, started_at, ended_at, owner_user_id, conversation_id, thread_id, last_trace_id, session_json, updated_at FROM agent_sessions WHERE last_trace_id = $1 OR session_json LIKE $2 ORDER BY updated_at DESC NULLS LAST LIMIT 1", [traceId, `%${traceId}%`]); return pgAgentSession(result.rows?.[0]); }
|
||||
async listAgentSessionsForUser(input = {}) {
|
||||
await this.ensureSchema();
|
||||
const role = textOr(input.actorRole, "user");
|
||||
@@ -1852,6 +1852,7 @@ function mergeAgentSessionOwnerEvidence(nextValue, existingValue) {
|
||||
const existing = normalizeObject(existingValue);
|
||||
const next = normalizeObject(nextValue);
|
||||
const merged = { ...existing, ...next };
|
||||
merged.traceResults = mergeAgentSessionTraceResults(existing.traceResults, next.traceResults ?? next.traceResult ?? next);
|
||||
const existingMessages = Array.isArray(existing.messages) ? existing.messages : null;
|
||||
const nextMessages = Array.isArray(next.messages) ? next.messages : null;
|
||||
if (nextMessages?.length) {
|
||||
@@ -1871,6 +1872,11 @@ function mergeAgentSessionOwnerEvidence(nextValue, existingValue) {
|
||||
const nextChatMessages = Array.isArray(next.chatMessages) ? next.chatMessages : null;
|
||||
if (existingChatMessages?.length && (!nextChatMessages || nextChatMessages.length === 0)) merged.chatMessages = existingChatMessages;
|
||||
if (!next.agentRun && existing.agentRun) merged.agentRun = existing.agentRun;
|
||||
const activeTraceEvidence = latestAgentSessionTraceEvidence(merged);
|
||||
if (activeTraceEvidence) {
|
||||
if (!merged.finalResponse && activeTraceEvidence.finalResponse) merged.finalResponse = activeTraceEvidence.finalResponse;
|
||||
if (!merged.traceSummary && activeTraceEvidence.traceSummary) merged.traceSummary = activeTraceEvidence.traceSummary;
|
||||
}
|
||||
const nextCount = numberOrNull(next.messageCount);
|
||||
const existingCount = numberOrNull(existing.messageCount);
|
||||
if (Array.isArray(merged.messages)) {
|
||||
@@ -1885,6 +1891,71 @@ function mergeAgentSessionOwnerEvidence(nextValue, existingValue) {
|
||||
}
|
||||
return merged;
|
||||
}
|
||||
function mergeAgentSessionTraceResults(existingValue, nextValue) {
|
||||
const records = new Map();
|
||||
for (const source of [existingValue, nextValue]) {
|
||||
for (const entry of traceResultEntries(source)) {
|
||||
const normalized = normalizeAgentSessionTraceResult(entry);
|
||||
if (!normalized) continue;
|
||||
records.set(normalized.traceId, { ...(records.get(normalized.traceId) ?? {}), ...normalized });
|
||||
}
|
||||
}
|
||||
const items = [...records.values()]
|
||||
.sort((a, b) => String(b.updatedAt ?? b.createdAt ?? "").localeCompare(String(a.updatedAt ?? a.createdAt ?? "")))
|
||||
.slice(0, 50);
|
||||
return Object.fromEntries(items.reverse().map((item) => [item.traceId, item]));
|
||||
}
|
||||
function traceResultEntries(value) {
|
||||
if (!value) return [];
|
||||
if (Array.isArray(value)) return value;
|
||||
if (typeof value === "object") {
|
||||
if (safeTraceIdLocal(value.traceId)) return [value];
|
||||
return Object.entries(value)
|
||||
.filter(([key]) => safeTraceIdLocal(key))
|
||||
.map(([traceId, entry]) => ({ ...normalizeObject(entry), traceId }));
|
||||
}
|
||||
return [];
|
||||
}
|
||||
function normalizeAgentSessionTraceResult(value) {
|
||||
const record = normalizeObject(value);
|
||||
const traceId = safeTraceIdLocal(record.traceId);
|
||||
if (!traceId) return null;
|
||||
const agentRun = normalizeObject(record.agentRun);
|
||||
const result = pruneEmpty({
|
||||
traceId,
|
||||
status: textOr(record.status, ""),
|
||||
conversationId: textOr(record.conversationId, ""),
|
||||
sessionId: textOr(record.sessionId, ""),
|
||||
threadId: textOr(record.threadId, ""),
|
||||
messageId: textOr(record.messageId, ""),
|
||||
createdAt: textOr(record.createdAt, ""),
|
||||
updatedAt: textOr(record.updatedAt, ""),
|
||||
finalResponse: redactTraceFinalResponse(record.finalResponse),
|
||||
traceSummary: redactTraceSummary(record.traceSummary),
|
||||
agentRun: redactAgentRunSummary(agentRun),
|
||||
valuesRedacted: true,
|
||||
secretMaterialStored: false
|
||||
});
|
||||
return result.agentRun?.runId || result.finalResponse || result.traceSummary ? result : null;
|
||||
}
|
||||
function agentSessionTraceEvidence(session, traceId) {
|
||||
const id = safeTraceIdLocal(traceId);
|
||||
if (!id) return null;
|
||||
const snapshot = normalizeObject(session?.session);
|
||||
return normalizeAgentSessionTraceResult(snapshot.traceResults?.[id] ?? snapshot.traceResult?.[id] ?? null);
|
||||
}
|
||||
function latestAgentSessionTraceEvidence(snapshot) {
|
||||
const id = safeTraceIdLocal(snapshot?.lastTraceId ?? snapshot?.traceId);
|
||||
if (!id) return null;
|
||||
return normalizeAgentSessionTraceResult(snapshot.traceResults?.[id] ?? null);
|
||||
}
|
||||
function agentSessionContainsTraceId(session, traceId) {
|
||||
const id = safeTraceIdLocal(traceId);
|
||||
if (!id) return false;
|
||||
const snapshot = normalizeObject(session?.session);
|
||||
if (safeTraceIdLocal(snapshot.lastTraceId ?? snapshot.traceId) === id) return true;
|
||||
return [snapshot.messages, snapshot.chatMessages].some((messages) => Array.isArray(messages) && messages.some((message) => safeTraceIdLocal(message?.traceId) === id));
|
||||
}
|
||||
function normalizeFinalResponseConversationMessages(messages, finalResponse) {
|
||||
const finalText = textOr(finalResponse?.text, "");
|
||||
const traceId = safeTraceIdLocal(finalResponse?.traceId);
|
||||
@@ -2357,11 +2428,16 @@ function redactTraceSummary(value) {
|
||||
const summary = normalizeObject(value);
|
||||
if (Object.keys(summary).length === 0) return undefined;
|
||||
return pruneEmpty({
|
||||
traceId: textOr(summary.traceId, ""),
|
||||
source: textOr(summary.source, ""),
|
||||
terminalStatus: textOr(summary.terminalStatus, ""),
|
||||
sourceEventCount: numberOrNull(summary.sourceEventCount ?? summary.eventCount),
|
||||
noiseEventCount: numberOrNull(summary.noiseEventCount),
|
||||
omittedNoiseCount: numberOrNull(summary.omittedNoiseCount),
|
||||
lastEventLabel: textOr(summary.lastEventLabel, ""),
|
||||
finalAssistantRow: redactTraceFinalResponse(summary.finalAssistantRow)
|
||||
updatedAt: textOr(summary.updatedAt, ""),
|
||||
finalAssistantRow: redactTraceFinalResponse(summary.finalAssistantRow),
|
||||
agentRun: redactAgentRunSummary(summary.agentRun)
|
||||
});
|
||||
}
|
||||
function redactAgentRunSummary(value) {
|
||||
@@ -2369,12 +2445,27 @@ function redactAgentRunSummary(value) {
|
||||
if (Object.keys(agentRun).length === 0) return undefined;
|
||||
return pruneEmpty({
|
||||
adapter: textOr(agentRun.adapter, ""),
|
||||
managerUrl: textOr(agentRun.managerUrl, ""),
|
||||
backendProfile: textOr(agentRun.backendProfile, ""),
|
||||
providerId: textOr(agentRun.providerId, ""),
|
||||
runId: textOr(agentRun.runId, ""),
|
||||
commandId: textOr(agentRun.commandId, ""),
|
||||
attemptId: textOr(agentRun.attemptId, ""),
|
||||
runnerId: textOr(agentRun.runnerId, ""),
|
||||
runnerJobId: textOr(agentRun.runnerJobId, ""),
|
||||
jobName: textOr(agentRun.jobName, ""),
|
||||
namespace: textOr(agentRun.namespace, ""),
|
||||
terminalStatus: textOr(agentRun.terminalStatus, "")
|
||||
status: textOr(agentRun.status, ""),
|
||||
runStatus: textOr(agentRun.runStatus, ""),
|
||||
commandState: textOr(agentRun.commandState, ""),
|
||||
terminalStatus: textOr(agentRun.terminalStatus, ""),
|
||||
lastSeq: numberOrNull(agentRun.lastSeq),
|
||||
eventStartSeq: numberOrNull(agentRun.eventStartSeq ?? agentRun.commandStartSeq ?? agentRun.startSeq),
|
||||
sessionId: textOr(agentRun.sessionId, ""),
|
||||
conversationId: textOr(agentRun.conversationId, ""),
|
||||
threadId: textOr(agentRun.threadId, ""),
|
||||
traceId: textOr(agentRun.traceId, ""),
|
||||
reuseEligible: typeof agentRun.reuseEligible === "boolean" ? agentRun.reuseEligible : undefined
|
||||
});
|
||||
}
|
||||
function conversationsFromAgentSessions(sessions = []) {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -1758,6 +1758,138 @@ test("cloud api trace replays an earlier AgentRun command after same-run lastSeq
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api repairs historical same-session AgentRun trace after lastTraceId advances (#955)", async () => {
|
||||
const calls = [];
|
||||
const persistedOwnerInputs = [];
|
||||
const ownerSessions = new Map();
|
||||
const runId = "run_issue955_historical";
|
||||
const firstTraceId = "trc_issue955_historical_first";
|
||||
const secondTraceId = "trc_issue955_historical_second";
|
||||
const firstCommandId = "cmd_issue955_historical_first";
|
||||
const secondCommandId = "cmd_issue955_historical_second";
|
||||
const firstFinalText = "目前只有一个 HWPOD 可用：\n\n- d601-f103-v2 (board: D601-F103-V2 / STM32F103)\n\nAPI 返回 count=1, availableCount=1。";
|
||||
const secondFinalText = "编译成功！hwpod build completed。";
|
||||
const events = [
|
||||
{ id: "evt_issue955_hist_first_created", runId, seq: 1, type: "command_created", payload: { commandId: firstCommandId }, createdAt: "2026-06-06T01:43:23.000Z" },
|
||||
{ id: "evt_issue955_hist_first_tool", runId, seq: 25, type: "tool_call", payload: { method: "item/completed", type: "commandExecution", toolName: "commandExecution", itemId: "call_issue955_hwpod_list", command: "/bin/sh -lc 'hwpod list --available'", status: "completed", exitCode: 0, outputSummary: '{\"count\":1,\"availableCount\":1}', commandId: firstCommandId }, createdAt: "2026-06-06T01:43:46.000Z" },
|
||||
{ id: "evt_issue955_hist_first_message", runId, seq: 34, type: "assistant_message", payload: { commandId: firstCommandId, itemId: "msg_issue955_first", text: firstFinalText }, createdAt: "2026-06-06T01:46:19.000Z" },
|
||||
{ id: "evt_issue955_hist_first_done", runId, seq: 35, type: "terminal_status", payload: { commandId: firstCommandId, terminalStatus: "completed" }, createdAt: "2026-06-06T01:46:19.000Z" },
|
||||
{ id: "evt_issue955_hist_second_created", runId, seq: 36, type: "command_created", payload: { commandId: secondCommandId }, createdAt: "2026-06-06T01:46:11.000Z" },
|
||||
{ id: "evt_issue955_hist_second_message", runId, seq: 58, type: "assistant_message", payload: { commandId: secondCommandId, itemId: "msg_issue955_second", text: secondFinalText }, createdAt: "2026-06-06T01:46:28.000Z" },
|
||||
{ id: "evt_issue955_hist_second_done", runId, seq: 67, type: "terminal_status", payload: { commandId: secondCommandId, terminalStatus: "completed" }, createdAt: "2026-06-06T01:46:28.000Z" }
|
||||
];
|
||||
const agentRunServer = createHttpServer(async (request, response) => {
|
||||
const url = new URL(request.url || "/", "http://127.0.0.1");
|
||||
calls.push({ method: request.method, path: url.pathname, search: url.search });
|
||||
const send = (data) => {
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end(`${JSON.stringify({ ok: true, data, traceId: "trc_fake_issue955_historical" })}\n`);
|
||||
};
|
||||
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands`) {
|
||||
return send({ items: [
|
||||
{ id: firstCommandId, runId, state: "completed", idempotencyKey: firstTraceId, createdAt: "2026-06-06T01:43:30.040Z", updatedAt: "2026-06-06T01:43:46.360Z", payload: { traceId: firstTraceId, conversationId: "cnv_issue955_historical", hwlabSessionId: "ses_issue955_historical", threadId: "thread-issue955-historical", providerProfile: "deepseek" } },
|
||||
{ id: secondCommandId, runId, state: "completed", idempotencyKey: secondTraceId, createdAt: "2026-06-06T01:46:11.000Z", updatedAt: "2026-06-06T01:46:28.000Z", payload: { traceId: secondTraceId, conversationId: "cnv_issue955_historical", hwlabSessionId: "ses_issue955_historical", threadId: "thread-issue955-historical", providerProfile: "deepseek" } }
|
||||
] });
|
||||
}
|
||||
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${firstCommandId}/result`) {
|
||||
return send({ runId, commandId: firstCommandId, status: "completed", runStatus: "completed", commandState: "completed", terminalStatus: "completed", completed: true, reply: firstFinalText, lastSeq: 35, eventCount: 35, sessionRef: { sessionId: "ses_agentrun_deepseek_issue955", conversationId: "cnv_issue955_historical", threadId: "thread-issue955-historical" } });
|
||||
}
|
||||
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/commands/${secondCommandId}/result`) {
|
||||
return send({ runId, commandId: secondCommandId, status: "completed", runStatus: "completed", commandState: "completed", terminalStatus: "completed", completed: true, reply: secondFinalText, lastSeq: 67, eventCount: 67, sessionRef: { sessionId: "ses_agentrun_deepseek_issue955", conversationId: "cnv_issue955_historical", threadId: "thread-issue955-historical" } });
|
||||
}
|
||||
if (request.method === "GET" && url.pathname === `/api/v1/runs/${runId}/events`) {
|
||||
return send({ items: events });
|
||||
}
|
||||
response.writeHead(404, { "content-type": "application/json" });
|
||||
response.end(`${JSON.stringify({ ok: false, message: `unexpected ${request.method} ${url.pathname}` })}\n`);
|
||||
});
|
||||
await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
const agentRunPort = agentRunServer.address().port;
|
||||
ownerSessions.set("ses_issue955_historical", testAgentSessionRecord({
|
||||
sessionId: "ses_issue955_historical",
|
||||
projectId: "prj_v02_code_agent",
|
||||
conversationId: "cnv_issue955_historical",
|
||||
threadId: "thread-issue955-historical",
|
||||
lastTraceId: secondTraceId,
|
||||
status: "active",
|
||||
session: {
|
||||
messages: [
|
||||
{ id: "msg_issue955_user_first", role: "user", text: "看看有几个hwpod可用", traceId: firstTraceId, status: "submitted" },
|
||||
{ id: "msg_issue955_user_second", role: "user", text: "试一下编译 d601-f103-v2", traceId: secondTraceId, status: "submitted" },
|
||||
{ id: "msg_issue955_second", role: "agent", text: secondFinalText, traceId: secondTraceId, status: "idle" }
|
||||
],
|
||||
agentRun: { adapter: "agentrun-v01", managerUrl: `http://127.0.0.1:${agentRunPort}`, runId, commandId: secondCommandId, traceId: secondTraceId, backendProfile: "deepseek", terminalStatus: "completed", lastSeq: 67, valuesPrinted: false },
|
||||
finalResponse: { text: secondFinalText, textChars: secondFinalText.length, role: "assistant", status: "completed", traceId: secondTraceId, valuesPrinted: false },
|
||||
traceSummary: { traceId: secondTraceId, source: "agent-session-snapshot", sourceEventCount: 31, terminalStatus: "completed", agentRun: { runId, commandId: secondCommandId, lastSeq: 67, valuesPrinted: false }, valuesPrinted: false },
|
||||
valuesRedacted: true
|
||||
}
|
||||
}));
|
||||
|
||||
const server = createCloudApiServer({
|
||||
traceStore: createCodeAgentTraceStore(),
|
||||
env: {
|
||||
HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01",
|
||||
AGENTRUN_MGR_URL: `http://127.0.0.1:${agentRunPort}`,
|
||||
HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1",
|
||||
HWLAB_ENVIRONMENT: "v02",
|
||||
HWLAB_GITOPS_PROFILE: "v02"
|
||||
},
|
||||
accessController: {
|
||||
required: false,
|
||||
async authenticate() {
|
||||
return { ok: true, actor: TEST_AGENT_ACTOR, session: TEST_AUTH_SESSION };
|
||||
},
|
||||
async getAgentSessionByTraceId(traceId) {
|
||||
return [...ownerSessions.values()].find((session) => session.lastTraceId === traceId || session.session?.messages?.some((message) => message.traceId === traceId)) ?? null;
|
||||
},
|
||||
async recordAgentSessionOwner(input) {
|
||||
persistedOwnerInputs.push(input);
|
||||
const existing = ownerSessions.get(input.sessionId) ?? testAgentSessionRecord(input);
|
||||
const nextSession = { ...(existing.session ?? {}), ...(input.session ?? {}) };
|
||||
nextSession.traceResults = { ...(existing.session?.traceResults ?? {}), ...(input.session?.traceResults ?? {}) };
|
||||
const record = testAgentSessionRecord({ ...existing, ...input, session: nextSession });
|
||||
ownerSessions.set(record.id, record);
|
||||
return record;
|
||||
}
|
||||
}
|
||||
});
|
||||
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
|
||||
try {
|
||||
const { port } = server.address();
|
||||
const response = await fetch(`http://127.0.0.1:${port}/v1/agent/chat/trace/${firstTraceId}`, {
|
||||
headers: { cookie: "hwlab_session=test-stub-session" }
|
||||
});
|
||||
assert.equal(response.status, 200);
|
||||
const body = await response.json();
|
||||
const text = JSON.stringify(body);
|
||||
assert.equal(body.traceId, firstTraceId);
|
||||
assert.equal(body.status, "completed");
|
||||
assert.equal(body.finalResponse.text, firstFinalText);
|
||||
assert.equal(body.agentRun.commandId, firstCommandId);
|
||||
assert.equal(body.traceSummary.sourceEventCount, 4);
|
||||
assert.ok(body.events.some((event) => event.commandId === firstCommandId));
|
||||
assert.equal(body.events.some((event) => event.commandId === secondCommandId), false);
|
||||
assert.match(text, /d601-f103-v2/u);
|
||||
assert.doesNotMatch(text, /编译成功/u);
|
||||
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands`));
|
||||
assert.ok(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${firstCommandId}/result`));
|
||||
assert.equal(calls.some((call) => call.path === `/api/v1/runs/${runId}/commands/${secondCommandId}/result`), false);
|
||||
const repair = persistedOwnerInputs.find((input) => input.session?.traceResults?.[firstTraceId]);
|
||||
assert.ok(repair);
|
||||
assert.equal(repair.session.traceResults[firstTraceId].agentRun.commandId, firstCommandId);
|
||||
assert.equal(ownerSessions.get("ses_issue955_historical").session.traceResults[firstTraceId].finalResponse.text, firstFinalText);
|
||||
} finally {
|
||||
await new Promise((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
await new Promise((resolve, reject) => {
|
||||
agentRunServer.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api result polling compacts large runnerTrace while preserving providerTrace", async () => {
|
||||
const workspace = await mkdtemp(path.join(os.tmpdir(), "hwlab-agent-result-compact-"));
|
||||
const codexHome = await mkdtemp(path.join(os.tmpdir(), "hwlab-agent-result-compact-codex-home-"));
|
||||
|
||||
Reference in New Issue
Block a user