fix code queue trace summary sync contract
This commit is contained in:
Codex
@@ -249,6 +249,25 @@ interface OaTraceStepRow {
|
||||
updated_at: Date | string;
|
||||
}
|
||||
|
||||
interface OaTraceStatsRow {
|
||||
scope_id: string;
|
||||
service_id: string;
|
||||
subject_kind: string;
|
||||
subject_id: string;
|
||||
stats_revision: number | string;
|
||||
step_count: number | string;
|
||||
llm_step_count: number | string;
|
||||
read_count: number | string;
|
||||
edit_count: number | string;
|
||||
run_count: number | string;
|
||||
error_count: number | string;
|
||||
trace_line_count: number | string;
|
||||
output_max_seq: number | string;
|
||||
attempt_stats_json: unknown;
|
||||
last_event_sequence: number | string | null;
|
||||
updated_at: Date | string;
|
||||
}
|
||||
|
||||
interface TraceStepLine {
|
||||
seq: number;
|
||||
at: string;
|
||||
@@ -446,6 +465,11 @@ function timestampMs(value: string | Date | null | undefined): number | null {
|
||||
return Number.isFinite(ms) ? ms : null;
|
||||
}
|
||||
|
||||
function positiveNumber(value: unknown): number | null {
|
||||
const parsed = Number(value);
|
||||
return Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : null;
|
||||
}
|
||||
|
||||
function prefixPreview(value: unknown, maxChars: number): string {
|
||||
const text = String(value ?? "");
|
||||
return text.length <= maxChars ? text : `${text.slice(0, Math.max(0, maxChars - 1))}…`;
|
||||
@@ -814,6 +838,41 @@ function activeTask(task: QueueTask): boolean {
|
||||
return task.status === "running" || task.status === "judging";
|
||||
}
|
||||
|
||||
function visibleTraceOutputs(task: QueueTask): LiveOutput[] {
|
||||
return task.output.filter((item) => item.channel !== "system" || item.method !== "enqueue");
|
||||
}
|
||||
|
||||
function rawTraceStats(task: QueueTask): JsonRecord {
|
||||
const steps = visibleTraceOutputs(task);
|
||||
const count = (channel: string) => steps.filter((item) => item.channel === channel).length;
|
||||
const stepCount = positiveNumber(task.stepCount ?? task.llmStepCount) ?? steps.length;
|
||||
return {
|
||||
stepCount,
|
||||
llmStepCount: positiveNumber(task.llmStepCount ?? task.stepCount) ?? stepCount,
|
||||
traceLineCount: steps.length,
|
||||
outputMaxSeq: outputMaxSeq(task),
|
||||
readCount: count("tool"),
|
||||
editCount: count("diff"),
|
||||
runCount: count("command"),
|
||||
errorCount: count("error"),
|
||||
};
|
||||
}
|
||||
|
||||
function traceStatsFallbackPatch(task: QueueTask): JsonRecord {
|
||||
const fallback = rawTraceStats(task);
|
||||
const rawTracePresent = Number(fallback.stepCount ?? 0) > 0 || Number(fallback.traceLineCount ?? 0) > 0 || Number(fallback.outputMaxSeq ?? 0) > 0;
|
||||
return {
|
||||
...fallback,
|
||||
traceStats: null,
|
||||
statsSource: rawTracePresent ? "raw-trace-fallback" : "raw-trace-empty",
|
||||
traceStatsState: rawTracePresent ? "degraded" : "empty",
|
||||
traceStatsReason: rawTracePresent ? "oa-event-flow-stats-unavailable-retained-trace-present" : "no-trace-steps-yet",
|
||||
statsUnavailable: true,
|
||||
statsSyncing: rawTracePresent,
|
||||
rawTraceStepCount: fallback.stepCount,
|
||||
};
|
||||
}
|
||||
|
||||
function schedulerHeartbeat(task: QueueTask): JsonRecord | null {
|
||||
const heartbeat = asRecord(task.schedulerHeartbeat);
|
||||
return heartbeat !== null && heartbeat.taskId === task.id ? heartbeat as JsonRecord : null;
|
||||
@@ -1917,10 +1976,7 @@ function taskListResponse(task: QueueTask, lite = true): JsonRecord {
|
||||
...promptFields,
|
||||
promptEditable: queuedTaskPromptEditable(task),
|
||||
finalResponseChars: task.finalResponse.length,
|
||||
stepCount: numberField(task.stepCount ?? task.llmStepCount, 0),
|
||||
llmStepCount: numberField(task.llmStepCount ?? task.stepCount, 0),
|
||||
traceStats: null,
|
||||
statsSource: "code-queue-mgr",
|
||||
...traceStatsFallbackPatch(task),
|
||||
summaryOnly: true,
|
||||
referenceTaskIds: task.referenceTaskIds,
|
||||
referenceInjection: task.referenceInjection,
|
||||
@@ -2299,6 +2355,66 @@ function traceScopeId(taskId: string, url: URL): string {
|
||||
return attempt > 0 ? `task:${taskId}:attempt:${attempt}` : `task:${taskId}`;
|
||||
}
|
||||
|
||||
function taskScopeId(taskId: string): string {
|
||||
return `task:${taskId}`;
|
||||
}
|
||||
|
||||
function taskAttemptScopeId(taskId: string, attemptIndex: number): string {
|
||||
return `${taskScopeId(taskId)}:attempt:${Math.floor(attemptIndex)}`;
|
||||
}
|
||||
|
||||
function traceStatsRowToRecord(row: OaTraceStatsRow): JsonRecord {
|
||||
const taskAttempt = row.scope_id.match(/^task:([^:]+):attempt:(\d+)$/u);
|
||||
const taskOnly = row.scope_id.match(/^task:([^:]+)$/u);
|
||||
const taskId = taskAttempt?.[1] ?? taskOnly?.[1] ?? "";
|
||||
const attemptIndex = taskAttempt === null ? null : Number(taskAttempt[2]);
|
||||
return {
|
||||
scopeId: row.scope_id,
|
||||
serviceId: row.service_id,
|
||||
subjectKind: row.subject_kind,
|
||||
subjectId: row.subject_id,
|
||||
statsRevision: numberField(row.stats_revision, 0),
|
||||
stepCount: numberField(row.step_count, 0),
|
||||
llmStepCount: numberField(row.llm_step_count, 0),
|
||||
readCount: numberField(row.read_count, 0),
|
||||
editCount: numberField(row.edit_count, 0),
|
||||
runCount: numberField(row.run_count, 0),
|
||||
errorCount: numberField(row.error_count, 0),
|
||||
traceLineCount: numberField(row.trace_line_count, 0),
|
||||
outputMaxSeq: numberField(row.output_max_seq, 0),
|
||||
attemptStats: toJsonValue(row.attempt_stats_json ?? null),
|
||||
lastEventSequence: row.last_event_sequence === null ? null : numberField(row.last_event_sequence, 0),
|
||||
updatedAt: timestampToIso(row.updated_at),
|
||||
taskId,
|
||||
attemptIndex,
|
||||
source: "oa-event-flow",
|
||||
};
|
||||
}
|
||||
|
||||
async function loadOaTraceStats(scopeIds: string[]): Promise<Map<string, JsonRecord>> {
|
||||
const uniqueScopeIds = Array.from(new Set(scopeIds.map((scopeId) => scopeId.trim()).filter(Boolean)));
|
||||
const result = new Map<string, JsonRecord>();
|
||||
if (uniqueScopeIds.length === 0) return result;
|
||||
try {
|
||||
const rows = await traceSql<OaTraceStatsRow[]>`
|
||||
SELECT scope_id, service_id, subject_kind, subject_id, stats_revision,
|
||||
step_count, llm_step_count, read_count, edit_count, run_count, error_count,
|
||||
trace_line_count, output_max_seq, attempt_stats_json, last_event_sequence, updated_at
|
||||
FROM oa_trace_stats
|
||||
WHERE scope_id IN ${traceSql(uniqueScopeIds)}
|
||||
ORDER BY updated_at DESC
|
||||
LIMIT ${Math.max(100, uniqueScopeIds.length)}
|
||||
`;
|
||||
for (const row of rows) {
|
||||
const record = traceStatsRowToRecord(row);
|
||||
result.set(String(record.scopeId || row.scope_id), record);
|
||||
}
|
||||
} catch (error) {
|
||||
log("warn", "oa_trace_stats_load_failed", { scopeIds: uniqueScopeIds, error: errorToJson(error) });
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function oaTraceStepToLine(row: OaTraceStepRow): TraceStepLine | null {
|
||||
const seq = numberField(row.step_seq, 0);
|
||||
if (seq <= 0) return null;
|
||||
@@ -2336,6 +2452,17 @@ async function loadOaTraceStepLines(taskId: string, url: URL): Promise<TraceStep
|
||||
return coalesceCodexToolLifecycleTraceSteps(rows.map(oaTraceStepToLine).filter((line): line is TraceStepLine => line !== null));
|
||||
}
|
||||
|
||||
async function loadOaTraceStepLinesForScope(scopeId: string): Promise<TraceStepLine[]> {
|
||||
const rows = await traceSql<OaTraceStepRow[]>`
|
||||
SELECT scope_id, step_seq, kind, title, status, summary_lines, raw_seqs, created_at, updated_at
|
||||
FROM oa_trace_steps
|
||||
WHERE scope_id = ${scopeId}
|
||||
ORDER BY step_seq ASC
|
||||
LIMIT 5000
|
||||
`;
|
||||
return coalesceCodexToolLifecycleTraceSteps(rows.map(oaTraceStepToLine).filter((line): line is TraceStepLine => line !== null));
|
||||
}
|
||||
|
||||
function fallbackTraceStepLines(task: QueueTask): TraceStepLine[] {
|
||||
return task.output
|
||||
.filter((item) => item.channel !== "system" || item.method !== "enqueue")
|
||||
@@ -2397,21 +2524,147 @@ async function traceStepDetail(task: QueueTask, url: URL): Promise<JsonRecord |
|
||||
return { ok: true, taskId: task.id, source: "code-queue-mgr-postgres", step: item as unknown as JsonValue, rawOutput: item as unknown as JsonValue, line: payload };
|
||||
}
|
||||
|
||||
function traceSummary(task: QueueTask): JsonRecord {
|
||||
const steps = task.output.filter((item) => item.channel !== "system" || item.method !== "enqueue");
|
||||
function attemptIndexesForTrace(task: QueueTask): number[] {
|
||||
const indexes = new Set<number>();
|
||||
for (const attempt of task.attempts) {
|
||||
const index = Number(attempt.index);
|
||||
if (Number.isInteger(index) && index > 0) indexes.add(index);
|
||||
}
|
||||
const current = Number(task.currentAttempt || 0);
|
||||
const max = Math.max(current, ...Array.from(indexes), 0);
|
||||
for (let index = 1; index <= max; index += 1) indexes.add(index);
|
||||
return Array.from(indexes).sort((left, right) => left - right);
|
||||
}
|
||||
|
||||
function traceStepExecutionStats(steps: TraceStepLine[]): JsonRecord {
|
||||
const countKind = (kind: string) => steps.filter((step) => step.kind === kind).length;
|
||||
return {
|
||||
stepCount: steps.length,
|
||||
llmStepCount: steps.length,
|
||||
traceLineCount: steps.length,
|
||||
outputMaxSeq: steps.at(-1)?.seq ?? 0,
|
||||
readCount: countKind("explored"),
|
||||
editCount: countKind("edited"),
|
||||
runCount: countKind("ran"),
|
||||
errorCount: countKind("error"),
|
||||
};
|
||||
}
|
||||
|
||||
function fallbackTraceStatsRecord(scopeId: string, fallback: JsonRecord): JsonRecord {
|
||||
const stepCount = positiveNumber(fallback.stepCount) ?? 0;
|
||||
return {
|
||||
scopeId,
|
||||
source: "oa-event-flow",
|
||||
stepCount,
|
||||
llmStepCount: positiveNumber(fallback.llmStepCount) ?? stepCount,
|
||||
traceLineCount: positiveNumber(fallback.traceLineCount) ?? stepCount,
|
||||
outputMaxSeq: positiveNumber(fallback.outputMaxSeq) ?? 0,
|
||||
readCount: positiveNumber(fallback.readCount) ?? 0,
|
||||
editCount: positiveNumber(fallback.editCount) ?? 0,
|
||||
runCount: positiveNumber(fallback.runCount) ?? 0,
|
||||
errorCount: positiveNumber(fallback.errorCount) ?? 0,
|
||||
sourceHint: "raw-trace-fallback",
|
||||
};
|
||||
}
|
||||
|
||||
function applyStatsOrFallback(fallback: JsonRecord, stats: JsonRecord | null, scopeId: string): JsonRecord {
|
||||
if (stats !== null) {
|
||||
return {
|
||||
...fallback,
|
||||
...stats,
|
||||
traceStats: stats,
|
||||
statsSource: "oa-event-flow",
|
||||
traceStatsState: "ready",
|
||||
traceStatsReason: null,
|
||||
statsUnavailable: false,
|
||||
statsSyncing: false,
|
||||
};
|
||||
}
|
||||
const rawTracePresent = Number(fallback.stepCount ?? 0) > 0 || Number(fallback.traceLineCount ?? 0) > 0 || Number(fallback.outputMaxSeq ?? 0) > 0;
|
||||
const traceStats = rawTracePresent ? fallbackTraceStatsRecord(scopeId, fallback) : null;
|
||||
return {
|
||||
...fallback,
|
||||
traceStats,
|
||||
statsSource: rawTracePresent ? "raw-trace-fallback" : "raw-trace-empty",
|
||||
traceStatsState: rawTracePresent ? "degraded" : "empty",
|
||||
traceStatsReason: rawTracePresent ? "oa-event-flow-stats-unavailable-raw-trace-present" : "no-trace-steps-yet",
|
||||
statsUnavailable: true,
|
||||
statsSyncing: rawTracePresent,
|
||||
rawTraceStepCount: fallback.stepCount ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
async function traceSummary(task: QueueTask): Promise<JsonRecord> {
|
||||
const fallbackSteps = fallbackTraceStepLines(task);
|
||||
const attempts = attemptIndexesForTrace(task);
|
||||
const scopeIds = [taskScopeId(task.id), ...attempts.map((index) => taskAttemptScopeId(task.id, index))];
|
||||
const [stats, allOaSteps, attemptStepEntries] = await Promise.all([
|
||||
loadOaTraceStats(scopeIds),
|
||||
loadOaTraceStepLinesForScope(taskScopeId(task.id)).catch((error) => {
|
||||
log("warn", "oa_trace_summary_steps_load_failed", { taskId: task.id, error: errorToJson(error) });
|
||||
return [] as TraceStepLine[];
|
||||
}),
|
||||
Promise.all(attempts.map(async (index) => {
|
||||
const steps = await loadOaTraceStepLinesForScope(taskAttemptScopeId(task.id, index)).catch((error) => {
|
||||
log("warn", "oa_trace_summary_attempt_steps_load_failed", { taskId: task.id, attemptIndex: index, error: errorToJson(error) });
|
||||
return [] as TraceStepLine[];
|
||||
});
|
||||
return [index, steps] as const;
|
||||
})),
|
||||
]);
|
||||
const visibleSteps = allOaSteps.length > 0 ? allOaSteps : fallbackSteps;
|
||||
const taskStats = applyStatsOrFallback(traceStepExecutionStats(visibleSteps), stats.get(taskScopeId(task.id)) ?? null, taskScopeId(task.id));
|
||||
const attemptSteps = new Map(attemptStepEntries);
|
||||
const attemptRows = attempts.length > 0 ? attempts : task.attempts.map((attempt) => attempt.index);
|
||||
return {
|
||||
id: task.id,
|
||||
taskId: task.id,
|
||||
queueId: queueIdOf(task),
|
||||
status: task.status,
|
||||
providerId: task.providerId,
|
||||
executionMode: task.executionMode,
|
||||
executionModeInfo: executionModeInfo(task.executionMode),
|
||||
model: task.model,
|
||||
stepCount: numberField(task.stepCount ?? task.llmStepCount, steps.length),
|
||||
retainedStepCount: steps.length,
|
||||
agentPort: codeAgentPortForModel(task.model),
|
||||
agentPortInfo: codeAgentPortInfo(codeAgentPortForModel(task.model)),
|
||||
cwd: task.cwd,
|
||||
reasoningEffort: task.reasoningEffort,
|
||||
createdAt: task.createdAt,
|
||||
startedAt: task.startedAt,
|
||||
finishedAt: task.finishedAt,
|
||||
updatedAt: task.updatedAt,
|
||||
currentAttempt: task.currentAttempt,
|
||||
currentMode: task.currentMode,
|
||||
maxAttempts: task.maxAttempts,
|
||||
...taskStats,
|
||||
retainedStepCount: fallbackSteps.length,
|
||||
outputMaxSeq: outputMaxSeq(task),
|
||||
schedulerHeartbeat: task.schedulerHeartbeat ?? null,
|
||||
statsSource: "code-queue-mgr-postgres",
|
||||
attempts: task.attempts.map((attempt) => ({
|
||||
execution: taskStats,
|
||||
attempts: attemptRows.map((index) => {
|
||||
const attempt = task.attempts.find((item) => Number(item.index) === index) ?? task.attempts[index - 1] ?? null;
|
||||
const steps = attemptSteps.get(index) ?? fallbackSteps.filter((step) => step.seq >= Number(attempt?.outputStartSeq ?? -Infinity) && step.seq <= Number(attempt?.outputEndSeq ?? Infinity));
|
||||
const attemptFallback = traceStepExecutionStats(steps);
|
||||
const attemptStats = applyStatsOrFallback(attemptFallback, stats.get(taskAttemptScopeId(task.id, index)) ?? null, taskAttemptScopeId(task.id, index));
|
||||
return {
|
||||
...(attempt ?? {}),
|
||||
index,
|
||||
mode: attempt?.mode ?? (index <= 1 ? "initial" : "retry"),
|
||||
startedAt: attempt?.startedAt ?? steps[0]?.at ?? task.startedAt,
|
||||
finishedAt: attempt?.finishedAt ?? null,
|
||||
terminalStatus: attempt?.terminalStatus ?? null,
|
||||
startSeq: steps[0]?.seq ?? attempt?.outputStartSeq ?? null,
|
||||
endSeq: steps.at(-1)?.seq ?? attempt?.outputEndSeq ?? null,
|
||||
finalResponsePreview: attempt?.finalResponsePreview ?? "",
|
||||
judge: attempt?.judge ?? (index === task.currentAttempt ? task.lastJudge : null),
|
||||
feedbackPromptPreview: attempt?.feedbackPromptPreview ?? null,
|
||||
feedbackPromptChars: attempt?.feedbackPromptChars ?? null,
|
||||
attemptScopeId: taskAttemptScopeId(task.id, index),
|
||||
...attemptStats,
|
||||
execution: attemptStats,
|
||||
};
|
||||
}),
|
||||
storedAttempts: task.attempts.map((attempt) => ({
|
||||
index: attempt.index,
|
||||
mode: attempt.mode,
|
||||
startedAt: attempt.startedAt,
|
||||
@@ -2429,6 +2682,12 @@ function traceSummary(task: QueueTask): JsonRecord {
|
||||
lines: task.prompt.split(/\r?\n/u).length,
|
||||
},
|
||||
lastAssistantMessage: lastAssistantMessage(task),
|
||||
finalResponse: task.finalResponse,
|
||||
finalResponseChars: task.finalResponse.length,
|
||||
lastJudge: task.lastJudge,
|
||||
lastError: task.lastError,
|
||||
errorCount: taskStats.errorCount,
|
||||
timing: taskTiming(task),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -2976,7 +3235,7 @@ async function route(req: Request): Promise<Response> {
|
||||
const traceSummaryMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/trace-summary$/u);
|
||||
if (traceSummaryMatch !== null && req.method === "GET") {
|
||||
const task = await loadTask(decodeURIComponent(traceSummaryMatch[1] ?? ""));
|
||||
return task === null ? jsonResponse({ ok: false, error: "task not found" }, 404) : jsonResponse({ ok: true, summary: traceSummary(task) });
|
||||
return task === null ? jsonResponse({ ok: false, error: "task not found" }, 404) : jsonResponse({ ok: true, summary: await traceSummary(task) });
|
||||
}
|
||||
const traceStepsMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/trace-steps$/u);
|
||||
if (traceStepsMatch !== null && req.method === "GET") {
|
||||
|
||||
Reference in New Issue
Block a user