fix: resume workbench agentrun projection
This commit is contained in:
lyon
@@ -1,4 +1,4 @@
|
||||
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
|
||||
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
|
||||
// Responsibility: Code Agent HTTP turn admission, lifecycle compatibility wrappers, and Workbench projection writer/finalizer integration.
|
||||
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
@@ -40,6 +40,10 @@ const MAX_CODE_AGENT_TRACE_PAGE_LIMIT = 100;
|
||||
const DEFAULT_CODE_AGENT_TURN_STATUS_REFRESH_TIMEOUT_MS = 2500;
|
||||
const DEFAULT_CODE_AGENT_PROJECT_ID = "prj_hwpod_workbench";
|
||||
const DEFAULT_CODE_AGENT_PROVIDER_PROFILE = "deepseek";
|
||||
const DEFAULT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS = 5000;
|
||||
const DEFAULT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS = 60000;
|
||||
const DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS = 30000;
|
||||
const DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE = 100;
|
||||
const CODE_AGENT_TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled"]);
|
||||
const CODE_AGENT_PROVIDER_PROFILE_ALIASES = Object.freeze({ codex: "codex-api" });
|
||||
const CODE_AGENT_PROVIDER_PROFILE_ID_PATTERN = /^[a-z0-9][a-z0-9-]{0,63}$/u;
|
||||
@@ -866,6 +870,242 @@ function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, tr
|
||||
});
|
||||
}
|
||||
|
||||
export function startAgentRunProjectionResume({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console } = {}) {
|
||||
const env = options.env ?? process.env;
|
||||
if (!truthyFlag(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_ENABLED)) {
|
||||
return { started: false, reason: "disabled", stop() {}, valuesPrinted: false };
|
||||
}
|
||||
const initialDelayMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_INITIAL_DELAY_MS);
|
||||
const intervalMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_INTERVAL_MS);
|
||||
const minAgeMs = parseNonNegativeInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS);
|
||||
const batchSize = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE);
|
||||
let stopped = false;
|
||||
let active = false;
|
||||
let timer = null;
|
||||
|
||||
const schedule = (delayMs, reason) => {
|
||||
if (stopped) return;
|
||||
timer = setTimeout(() => {
|
||||
timer = null;
|
||||
void runAgentRunProjectionResumePass({ options, traceStore, logger, batchSize, minAgeMs, reason })
|
||||
.catch((error) => logAgentRunProjectionResume(logger, {
|
||||
event: "workbench_projection_resume",
|
||||
ok: false,
|
||||
status: "failed",
|
||||
reason,
|
||||
errorCode: error?.code ?? "projection_resume_pass_failed",
|
||||
message: error?.message ?? "AgentRun projection resume pass failed.",
|
||||
valuesRedacted: true
|
||||
}))
|
||||
.finally(() => {
|
||||
active = false;
|
||||
if (!stopped && intervalMs > 0) schedule(intervalMs, "interval");
|
||||
});
|
||||
active = true;
|
||||
}, Math.max(0, Number(delayMs) || 0));
|
||||
timer.unref?.();
|
||||
};
|
||||
|
||||
schedule(initialDelayMs, "startup");
|
||||
return {
|
||||
started: true,
|
||||
initialDelayMs,
|
||||
intervalMs,
|
||||
minAgeMs,
|
||||
batchSize,
|
||||
valuesPrinted: false,
|
||||
stop() {
|
||||
stopped = true;
|
||||
if (timer) clearTimeout(timer);
|
||||
timer = null;
|
||||
},
|
||||
active: () => active
|
||||
};
|
||||
}
|
||||
|
||||
async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, reason = "manual" } = {}) {
|
||||
const sessions = await listAgentRunProjectionResumeSessions(options, batchSize);
|
||||
const candidates = sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs }));
|
||||
let resumed = 0;
|
||||
let terminal = 0;
|
||||
let failed = 0;
|
||||
for (const candidate of candidates) {
|
||||
try {
|
||||
const result = await resumeAgentRunProjectionCandidate({ candidate, options, traceStore });
|
||||
resumed += 1;
|
||||
if (isTraceCommandTerminalStatus(result?.status)) terminal += 1;
|
||||
} catch (error) {
|
||||
failed += 1;
|
||||
appendProjectionResumeError(traceStore, candidate, error);
|
||||
}
|
||||
}
|
||||
logAgentRunProjectionResume(logger, {
|
||||
event: "workbench_projection_resume",
|
||||
ok: failed === 0,
|
||||
status: failed === 0 ? "completed" : "degraded",
|
||||
reason,
|
||||
scannedSessions: sessions.length,
|
||||
candidates: candidates.length,
|
||||
resumed,
|
||||
terminal,
|
||||
failed,
|
||||
valuesRedacted: true
|
||||
});
|
||||
return { scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed };
|
||||
}
|
||||
|
||||
async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE) {
|
||||
const accessStore = options.accessController?.store ?? options.accessController ?? null;
|
||||
if (typeof accessStore?.listAgentSessionsForUser !== "function") return [];
|
||||
return await accessStore.listAgentSessionsForUser({
|
||||
actorRole: "admin",
|
||||
ownerScoped: false,
|
||||
includeArchived: false,
|
||||
limit: batchSize,
|
||||
offset: 0
|
||||
}) ?? [];
|
||||
}
|
||||
|
||||
function agentRunProjectionResumeCandidates(session = {}, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, nowMs = Date.now() } = {}) {
|
||||
const snapshot = session?.session && typeof session.session === "object" ? session.session : {};
|
||||
const records = new Map();
|
||||
const traceResults = snapshot.traceResults && typeof snapshot.traceResults === "object" ? snapshot.traceResults : null;
|
||||
if (traceResults) {
|
||||
for (const [traceId, record] of Object.entries(traceResults)) {
|
||||
const safeId = safeTraceId(traceId);
|
||||
if (safeId && record && typeof record === "object") records.set(safeId, { ...record, traceId: safeId });
|
||||
}
|
||||
}
|
||||
const directTraceId = safeTraceId(snapshot.traceId ?? session.lastTraceId);
|
||||
if (directTraceId && snapshot.agentRun && typeof snapshot.agentRun === "object") {
|
||||
records.set(directTraceId, { ...snapshot, traceId: directTraceId });
|
||||
}
|
||||
return [...records.values()]
|
||||
.map((record) => agentRunProjectionResumeCandidate(session, record, { minAgeMs, nowMs }))
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
function agentRunProjectionResumeCandidate(session = {}, record = {}, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, nowMs = Date.now() } = {}) {
|
||||
const traceId = safeTraceId(record.traceId ?? session.lastTraceId);
|
||||
const agentRun = record.agentRun && typeof record.agentRun === "object" ? record.agentRun : null;
|
||||
if (!traceId || !agentRun?.runId || !agentRun?.commandId) return null;
|
||||
if (isTraceCommandTerminalStatus(record.status ?? agentRun.terminalStatus ?? agentRun.commandState ?? agentRun.status)) return null;
|
||||
const updatedAtMs = Date.parse(record.updatedAt ?? session.updatedAt ?? "");
|
||||
if (minAgeMs > 0 && Number.isFinite(updatedAtMs) && nowMs - updatedAtMs < minAgeMs) return null;
|
||||
const sessionId = safeSessionId(record.sessionId ?? session.id) || null;
|
||||
const conversationId = safeConversationId(record.conversationId ?? session.conversationId) || null;
|
||||
const threadId = safeOpaqueId(record.threadId ?? session.threadId) || null;
|
||||
const ownerUserId = textValue(record.ownerUserId ?? session.ownerUserId);
|
||||
const ownerRole = textValue(record.ownerRole ?? session.ownerRole) || "user";
|
||||
return {
|
||||
traceId,
|
||||
params: {
|
||||
traceId,
|
||||
sessionId,
|
||||
conversationId,
|
||||
threadId,
|
||||
projectId: record.projectId ?? session.projectId ?? null,
|
||||
ownerUserId,
|
||||
ownerRole
|
||||
},
|
||||
result: {
|
||||
...record,
|
||||
traceId,
|
||||
sessionId,
|
||||
conversationId,
|
||||
threadId,
|
||||
ownerUserId,
|
||||
ownerRole,
|
||||
status: record.status || session.status || "running",
|
||||
finalResponse: null,
|
||||
traceSummary: null,
|
||||
agentRun: {
|
||||
...agentRun,
|
||||
traceId,
|
||||
providerTrace: agentRun.providerTrace ?? { traceId, runId: agentRun.runId, commandId: agentRun.commandId, valuesPrinted: false },
|
||||
lastSeq: Number.isFinite(Number(agentRun.lastSeq)) ? Number(agentRun.lastSeq) : 0,
|
||||
valuesPrinted: false
|
||||
},
|
||||
valuesRedacted: true
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) {
|
||||
const currentResult = await agentRunProjectionResumeResultWithCursor(candidate, options);
|
||||
const synced = await syncAgentRunChatResult({
|
||||
traceId: candidate.traceId,
|
||||
currentResult,
|
||||
options,
|
||||
traceStore,
|
||||
forceResultSync: true,
|
||||
refreshEvents: true
|
||||
});
|
||||
const payload = synced?.result ?? currentResult;
|
||||
if (isTraceCommandTerminalStatus(payload?.status)) {
|
||||
await recordCodeAgentTerminalTurnStatusEffects({ payload, params: candidate.params, options, preserveLastTraceId: true });
|
||||
}
|
||||
return payload;
|
||||
}
|
||||
|
||||
async function agentRunProjectionResumeResultWithCursor(candidate, options = {}) {
|
||||
const readModel = createWorkbenchReadModel(options, { id: candidate.params.ownerUserId || "projection-resume", role: "admin" });
|
||||
const trace = await readModel.traceSnapshot(candidate.traceId);
|
||||
const durableSourceSeq = agentRunProjectionLastSourceSeq(trace, candidate.result.agentRun.commandId);
|
||||
const existingLastSeq = Number(candidate.result.agentRun.lastSeq ?? 0);
|
||||
const lastSeq = Math.max(Number.isFinite(existingLastSeq) ? existingLastSeq : 0, durableSourceSeq);
|
||||
return {
|
||||
...candidate.result,
|
||||
traceSummary: null,
|
||||
agentRun: {
|
||||
...candidate.result.agentRun,
|
||||
lastSeq,
|
||||
valuesPrinted: false
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
function agentRunProjectionLastSourceSeq(trace = {}, commandId = null) {
|
||||
const events = Array.isArray(trace?.events) ? trace.events : [];
|
||||
return events.reduce((max, event) => {
|
||||
if (commandId && event?.commandId && event.commandId !== commandId) return max;
|
||||
const sourceSeq = Number(event?.sourceSeq ?? 0);
|
||||
return Number.isFinite(sourceSeq) && sourceSeq > max ? Math.trunc(sourceSeq) : max;
|
||||
}, 0);
|
||||
}
|
||||
|
||||
function appendProjectionResumeError(traceStore, candidate, error) {
|
||||
const traceId = safeTraceId(candidate?.traceId);
|
||||
if (!traceId) return;
|
||||
traceStore.append(traceId, {
|
||||
type: "turn-status",
|
||||
status: "degraded",
|
||||
label: "projection-resume:sync-failed",
|
||||
errorCode: error?.code ?? "projection_resume_sync_failed",
|
||||
message: error?.message ?? "AgentRun projection resume failed.",
|
||||
runId: candidate?.result?.agentRun?.runId ?? null,
|
||||
commandId: candidate?.result?.agentRun?.commandId ?? null,
|
||||
waitingFor: "agentrun-result-resume",
|
||||
terminal: false,
|
||||
valuesPrinted: false
|
||||
});
|
||||
}
|
||||
|
||||
function logAgentRunProjectionResume(logger, payload) {
|
||||
try {
|
||||
const line = JSON.stringify(payload);
|
||||
if (payload.ok === false && typeof logger?.warn === "function") logger.warn(line);
|
||||
else if (typeof logger?.log === "function") logger.log(line);
|
||||
} catch {
|
||||
// Projection resume logging must not affect the projector.
|
||||
}
|
||||
}
|
||||
|
||||
function parseNonNegativeInteger(value, fallback) {
|
||||
const parsed = Number.parseInt(value ?? "", 10);
|
||||
return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback;
|
||||
}
|
||||
|
||||
function agentRunProjectionActivitySignature(result, runnerTrace) {
|
||||
const agentRun = result?.agentRun && typeof result.agentRun === "object" ? result.agentRun : {};
|
||||
const trace = runnerTrace && typeof runnerTrace === "object" ? runnerTrace : result?.runnerTrace;
|
||||
@@ -2617,7 +2857,7 @@ function paginateTraceSnapshot(snapshot, { sinceSeq, limit }) {
|
||||
const pageItems = indexedEvents.slice(startIndex, startIndex + limit);
|
||||
const events = pageItems.map((item) => item.event);
|
||||
const fromSeq = pageItems.length > 0 ? pageItems[0].seq : null;
|
||||
const toSeq = pageItems.length > 0 ? pageItems[pageItems.length - 1].seq : sinceSeq;
|
||||
const toSeq = pageItems.length > 0 ? pageItems[pageItems.length - 1].seq : null;
|
||||
const hasMore = startIndex + pageItems.length < indexedEvents.length;
|
||||
const totalEvents = snapshot?.eventCount ?? indexedEvents.length;
|
||||
return {
|
||||
@@ -2634,7 +2874,7 @@ function paginateTraceSnapshot(snapshot, { sinceSeq, limit }) {
|
||||
},
|
||||
truncated: hasMore,
|
||||
hasMore,
|
||||
nextSinceSeq: toSeq,
|
||||
nextSinceSeq: pageItems.length > 0 ? toSeq : sinceSeq,
|
||||
fullTraceLoaded: !hasMore
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user