fix: persist workbench projection cursor
This commit is contained in:
lyon
@@ -1,4 +1,4 @@
|
||||
// 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.
|
||||
// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor; 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";
|
||||
@@ -19,6 +19,7 @@ import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
|
||||
import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts";
|
||||
import { scheduleWorkbenchProjectionFinalizer } from "./workbench-projection-finalizer.ts";
|
||||
import { writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts";
|
||||
import { buildAgentRunProjectionEventsFetchPlan } from "./workbench-projection-cursor.ts";
|
||||
import { createWorkbenchReadModel } from "./workbench-read-model.ts";
|
||||
import {
|
||||
firstHeaderValue,
|
||||
@@ -942,8 +943,9 @@ export function startAgentRunProjectionResume({ options = {}, traceStore = defau
|
||||
}
|
||||
|
||||
async function runAgentRunProjectionResumePass({ options = {}, traceStore = defaultCodeAgentTraceStore, logger = console, batchSize, minAgeMs, offset = 0, reason = "manual" } = {}) {
|
||||
const sessions = await listAgentRunProjectionResumeSessions(options, batchSize, offset);
|
||||
const candidates = sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs }));
|
||||
const stateCandidates = await listAgentRunProjectionResumeStateCandidates(options, batchSize, offset, { minAgeMs });
|
||||
const sessions = stateCandidates.length > 0 ? [] : await listAgentRunProjectionResumeSessions(options, batchSize, offset);
|
||||
const candidates = stateCandidates.length > 0 ? stateCandidates : sessions.flatMap((session) => agentRunProjectionResumeCandidates(session, { minAgeMs }));
|
||||
let resumed = 0;
|
||||
let terminal = 0;
|
||||
let failed = 0;
|
||||
@@ -963,6 +965,7 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa
|
||||
status: failed === 0 ? "completed" : "degraded",
|
||||
reason,
|
||||
offset,
|
||||
scannedProjectionStates: stateCandidates.length,
|
||||
scannedSessions: sessions.length,
|
||||
candidates: candidates.length,
|
||||
resumed,
|
||||
@@ -973,6 +976,25 @@ async function runAgentRunProjectionResumePass({ options = {}, traceStore = defa
|
||||
return { offset, scannedSessions: sessions.length, candidates: candidates.length, resumed, terminal, failed };
|
||||
}
|
||||
|
||||
async function listAgentRunProjectionResumeStateCandidates(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, offset = 0, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS } = {}) {
|
||||
const runtimeStore = options.runtimeStore;
|
||||
if (typeof runtimeStore?.queryWorkbenchProjectionStates !== "function") return [];
|
||||
let result;
|
||||
try {
|
||||
result = await runtimeStore.queryWorkbenchProjectionStates({
|
||||
projectionStatuses: ["projecting", "degraded"],
|
||||
dueAt: new Date().toISOString(),
|
||||
limit: batchSize,
|
||||
offset
|
||||
});
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
return (Array.isArray(result?.states) ? result.states : [])
|
||||
.map((state) => agentRunProjectionResumeCandidateFromState(state, { minAgeMs }))
|
||||
.filter(Boolean);
|
||||
}
|
||||
|
||||
async function listAgentRunProjectionResumeSessions(options = {}, batchSize = DEFAULT_AGENTRUN_PROJECTION_RESUME_BATCH_SIZE, offset = 0) {
|
||||
const accessStore = options.accessController?.store ?? options.accessController ?? null;
|
||||
if (typeof accessStore?.listAgentSessionsForUser !== "function") return [];
|
||||
@@ -1050,14 +1072,70 @@ function agentRunProjectionResumeCandidate(session = {}, record = {}, { minAgeMs
|
||||
};
|
||||
}
|
||||
|
||||
function agentRunProjectionResumeCandidateFromState(state = {}, { minAgeMs = DEFAULT_AGENTRUN_PROJECTION_RESUME_MIN_AGE_MS, nowMs = Date.now() } = {}) {
|
||||
const traceId = safeTraceId(state.traceId);
|
||||
const runId = textValue(state.runId ?? state.sourceRunId);
|
||||
const commandId = textValue(state.commandId ?? state.sourceCommandId);
|
||||
if (!traceId || !runId || !commandId) return null;
|
||||
if (isTraceCommandTerminalStatus(state.projectionStatus)) return null;
|
||||
const updatedAtMs = Date.parse(state.updatedAt ?? "");
|
||||
if (minAgeMs > 0 && Number.isFinite(updatedAtMs) && nowMs - updatedAtMs < minAgeMs) return null;
|
||||
const sessionId = safeSessionId(state.sessionId) || null;
|
||||
const conversationId = safeConversationId(state.conversationId) || null;
|
||||
const threadId = safeOpaqueId(state.threadId) || null;
|
||||
const ownerUserId = textValue(state.ownerUserId);
|
||||
const ownerRole = textValue(state.ownerRole) || "user";
|
||||
const lastSeq = Number.isFinite(Number(state.lastSourceSeq ?? state.lastAgentRunSeq)) ? Number(state.lastSourceSeq ?? state.lastAgentRunSeq) : 0;
|
||||
return {
|
||||
traceId,
|
||||
params: {
|
||||
traceId,
|
||||
sessionId,
|
||||
conversationId,
|
||||
threadId,
|
||||
ownerUserId,
|
||||
ownerRole
|
||||
},
|
||||
result: {
|
||||
accepted: true,
|
||||
status: "running",
|
||||
shortConnection: true,
|
||||
traceId,
|
||||
sessionId,
|
||||
conversationId,
|
||||
threadId,
|
||||
ownerUserId,
|
||||
ownerRole,
|
||||
finalResponse: null,
|
||||
traceSummary: null,
|
||||
agentRun: {
|
||||
traceId,
|
||||
runId,
|
||||
commandId,
|
||||
lastSeq,
|
||||
managerUrl: state.managerUrl ?? null,
|
||||
backendProfile: state.backendProfile ?? null,
|
||||
providerId: state.providerId ?? null,
|
||||
status: "running",
|
||||
commandState: "running",
|
||||
resultSyncState: state.resultSyncState ?? null,
|
||||
providerTrace: { traceId, runId, commandId, valuesPrinted: false },
|
||||
valuesPrinted: false
|
||||
},
|
||||
valuesRedacted: true
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
async function resumeAgentRunProjectionCandidate({ candidate, options = {}, traceStore = defaultCodeAgentTraceStore } = {}) {
|
||||
const currentResult = await agentRunProjectionResumeResultWithCursor(candidate, options);
|
||||
const resumeOptions = { ...options, deferAgentRunResultSync: true };
|
||||
const synced = await syncAgentRunChatResult({
|
||||
traceId: candidate.traceId,
|
||||
currentResult,
|
||||
options,
|
||||
options: resumeOptions,
|
||||
traceStore,
|
||||
forceResultSync: true,
|
||||
forceResultSync: false,
|
||||
refreshEvents: true
|
||||
});
|
||||
const payload = synced?.result ?? currentResult;
|
||||
@@ -1068,29 +1146,35 @@ async function resumeAgentRunProjectionCandidate({ candidate, options = {}, trac
|
||||
}
|
||||
|
||||
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);
|
||||
const projectionState = await loadAgentRunProjectionState(options.runtimeStore, candidate);
|
||||
const plan = buildAgentRunProjectionEventsFetchPlan({
|
||||
projectionState,
|
||||
agentRun: candidate.result.agentRun,
|
||||
pageLimit: options.env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT
|
||||
});
|
||||
return {
|
||||
...candidate.result,
|
||||
traceSummary: null,
|
||||
agentRun: {
|
||||
...candidate.result.agentRun,
|
||||
lastSeq,
|
||||
lastSeq: plan.afterSeq,
|
||||
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);
|
||||
async function loadAgentRunProjectionState(runtimeStore, candidate) {
|
||||
if (!runtimeStore || typeof runtimeStore.getWorkbenchProjectionState !== "function") return null;
|
||||
try {
|
||||
const result = await runtimeStore.getWorkbenchProjectionState({ traceId: candidate.traceId });
|
||||
const state = result?.projectionState ?? null;
|
||||
if (!state) return null;
|
||||
if (String(state.runId ?? state.sourceRunId ?? "") !== String(candidate.result.agentRun.runId ?? "")) return null;
|
||||
if (String(state.commandId ?? state.sourceCommandId ?? "") !== String(candidate.result.agentRun.commandId ?? "")) return null;
|
||||
return state;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function appendProjectionResumeError(traceStore, candidate, error) {
|
||||
|
||||
Reference in New Issue
Block a user