Merge pull request #1600 from pikasTech/test/1596-fake-server-resume
test: cover workbench projector resume
This commit is contained in:
Lyon
@@ -1,4 +1,4 @@
|
||||
// SPEC: PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1.
|
||||
// SPEC: PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-0104010803 唯一投影 draft-2026-06-19-p0-projector-resume; PJ2026-010403 API契约 draft-2026-06-18-r1.
|
||||
// Responsibility: Same-origin fake Workbench API and static server for Playwright browser regression tests.
|
||||
|
||||
import { createReadStream, existsSync, statSync } from "node:fs";
|
||||
@@ -78,6 +78,7 @@ async function handleRequest(request: IncomingMessage, response: ServerResponse)
|
||||
const url = new URL(request.url ?? "/", `http://${request.headers.host ?? `127.0.0.1:${port}`}`);
|
||||
const path = url.pathname;
|
||||
recordLedger(method, url);
|
||||
refreshBackgroundProjection();
|
||||
|
||||
if (path === "/__e2e/health") return json(response, 200, { ok: true, scenarioId: state.scenarioId });
|
||||
if (path === "/__e2e/state") return json(response, 200, stateSummary());
|
||||
@@ -301,6 +302,11 @@ function createScenarioState(scenarioId: string): ScenarioState {
|
||||
sessions.unshift(session);
|
||||
traces.trc_scroll_follow = scrollFollowTrace(90);
|
||||
}
|
||||
if (id === "projector-resume-from-checkpoint") {
|
||||
const session = projectorResumeSession();
|
||||
sessions.unshift(session);
|
||||
traces.trc_f2456023233a485b = projectorResumeEarlyTrace();
|
||||
}
|
||||
if (id === "cross-session-late-events") {
|
||||
sessions.unshift(crossSessionLateB(), crossSessionLateA());
|
||||
traces.trc_late_A = crossSessionLateTraceA();
|
||||
@@ -320,6 +326,8 @@ function createScenarioState(scenarioId: string): ScenarioState {
|
||||
? "ses_terminal_empty"
|
||||
: id === "progress-only-final-response" || id === "terminal-completed-no-final-response" || id === "tool-completed-projection-running"
|
||||
? "ses_running"
|
||||
: id === "projector-resume-from-checkpoint"
|
||||
? "ses_317f78e1-ed91-4b09-a72a-3668d5469f59"
|
||||
: id === "projection-degraded-diagnostics"
|
||||
? "ses_projection_degraded"
|
||||
: id === "projection-sse-error"
|
||||
@@ -346,11 +354,97 @@ function createScenarioState(scenarioId: string): ScenarioState {
|
||||
terminalScript: id === "event-replay" || id === "running-to-terminal" || id === "stale-submit-restore" || id === "progress-only-final-response" || id === "terminal-completed-no-final-response",
|
||||
terminalFailureScript: id === "stale-submit-restore",
|
||||
staleTraceId,
|
||||
liveBackfillTraceId: null,
|
||||
liveBackfillReadyAtMs: null
|
||||
liveBackfillTraceId: id === "projector-resume-from-checkpoint" ? "trc_f2456023233a485b" : null,
|
||||
liveBackfillReadyAtMs: id === "projector-resume-from-checkpoint" ? Date.now() + 5_000 : null
|
||||
};
|
||||
}
|
||||
|
||||
function refreshBackgroundProjection(): void {
|
||||
if (state.scenarioId !== "projector-resume-from-checkpoint") return;
|
||||
if (!state.liveBackfillReadyAtMs || Date.now() < state.liveBackfillReadyAtMs) return;
|
||||
const session = sessionById("ses_317f78e1-ed91-4b09-a72a-3668d5469f59");
|
||||
if (session?.status === "completed") return;
|
||||
commitProjectorResumeProjection();
|
||||
}
|
||||
|
||||
function projectorResumeSession(): SessionRecord {
|
||||
const now = new Date().toISOString();
|
||||
const trace = projectorResumeEarlyTrace();
|
||||
return {
|
||||
sessionId: "ses_317f78e1-ed91-4b09-a72a-3668d5469f59",
|
||||
threadId: "thr_projector_resume_1596",
|
||||
status: "running",
|
||||
lastTraceId: "trc_f2456023233a485b",
|
||||
startedAt: now,
|
||||
updatedAt: now,
|
||||
messageCount: 2,
|
||||
firstUserMessagePreview: "issue 1596 durable projector resume sample",
|
||||
turnSummary: { traceId: "trc_f2456023233a485b", status: "running", running: true, terminal: false, ...projectorResumeProjection("projecting", 4) },
|
||||
messages: [
|
||||
{ id: "msg_projector_resume_user", messageId: "msg_projector_resume_user", role: "user", title: "用户", text: "issue 1596 durable projector resume sample", status: "sent", createdAt: now, sessionId: "ses_317f78e1-ed91-4b09-a72a-3668d5469f59", threadId: "thr_projector_resume_1596", traceId: "trc_f2456023233a485b", turnId: "turn_projector_resume_1596" },
|
||||
{ id: "msg_projector_resume_agent", messageId: "msg_projector_resume_agent", role: "agent", title: "Code Agent", text: "", status: "running", createdAt: now, sessionId: "ses_317f78e1-ed91-4b09-a72a-3668d5469f59", threadId: "thr_projector_resume_1596", traceId: "trc_f2456023233a485b", turnId: "turn_projector_resume_1596", runnerTrace: trace, ...projectorResumeProjection("projecting", 4) }
|
||||
]
|
||||
};
|
||||
}
|
||||
|
||||
function projectorResumeEarlyTrace(): JsonRecord {
|
||||
const createdAt = new Date().toISOString();
|
||||
const projection = projectorResumeProjection("projecting", 4);
|
||||
const events = [
|
||||
{ seq: 1, sourceSeq: 1, createdAt, label: "agentrun:request:accepted", type: "backend_status", status: "running", payload: { commandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704" } },
|
||||
{ seq: 2, sourceSeq: 2, createdAt, label: "agentrun:run:created", type: "backend_status", status: "running", payload: { runId: "run_58ef1811071548539f881515912560d8", runStatus: "claimed" } },
|
||||
{ seq: 3, sourceSeq: 3, createdAt, label: "agentrun:command:created", type: "backend_status", status: "running", payload: { commandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704", commandState: "running" } },
|
||||
{ seq: 4, sourceSeq: 4, createdAt, label: "agentrun:runner-job:created", type: "backend_status", status: "running", payload: { commandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704", namespace: "agentrun-v01" } }
|
||||
];
|
||||
return { traceId: "trc_f2456023233a485b", status: "running", running: true, terminal: false, sessionId: "ses_317f78e1-ed91-4b09-a72a-3668d5469f59", threadId: "thr_projector_resume_1596", turnId: "turn_projector_resume_1596", events, eventCount: events.length, fullTraceLoaded: true, hasMore: false, ...projection };
|
||||
}
|
||||
|
||||
function projectorResumeCompletedTrace(): JsonRecord {
|
||||
const createdAt = new Date().toISOString();
|
||||
const finalText = "fake projector resume completed from durable checkpoint.";
|
||||
const projection = projectorResumeProjection("caught-up", 7);
|
||||
const events = [
|
||||
...(projectorResumeEarlyTrace().events as JsonRecord[]),
|
||||
{ seq: 5, sourceSeq: 2748, createdAt, label: "agentrun:assistant:message", type: "assistant_message", status: "completed", message: finalText, payload: { commandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704" } },
|
||||
{ seq: 6, sourceSeq: 2749, createdAt, label: "item/commandExecution:completed", type: "commandExecution", status: "completed", stdout: "OK\n", payload: { commandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704", exitCode: 0 } },
|
||||
{ seq: 7, sourceSeq: 2750, createdAt, label: "agentrun:backend:turn/completed", type: "backend_status", status: "completed", terminal: true, payload: { commandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704", terminalStatus: "completed" } }
|
||||
];
|
||||
return { traceId: "trc_f2456023233a485b", status: "completed", running: false, terminal: true, sessionId: "ses_317f78e1-ed91-4b09-a72a-3668d5469f59", threadId: "thr_projector_resume_1596", turnId: "turn_projector_resume_1596", events, eventCount: events.length, fullTraceLoaded: true, hasMore: false, finalResponse: { text: finalText, status: "completed" }, traceSummary: { source: "durable-projector-resume", terminalStatus: "completed", valuesPrinted: false }, ...projection };
|
||||
}
|
||||
|
||||
function projectorResumeProjection(status: "projecting" | "caught-up", lastProjectedSeq: number): JsonRecord {
|
||||
const caughtUp = status === "caught-up";
|
||||
const projection = {
|
||||
projectionStatus: status,
|
||||
projectionHealth: status,
|
||||
lastProjectedSeq,
|
||||
sourceLatestSeq: 2750,
|
||||
sourceRunId: "run_58ef1811071548539f881515912560d8",
|
||||
sourceCommandId: "cmd_d4d20aa2bb2c4d19a91143ce87d73704",
|
||||
staleMs: caughtUp ? 0 : 90_000,
|
||||
blocker: caughtUp ? null : { code: "projector_resume_pending", layer: "workbench-projector", retryable: true, message: "后台投影恢复中，已从 checkpoint 记录 lastProjectedSeq=4。", valuesPrinted: false },
|
||||
valuesRedacted: true
|
||||
};
|
||||
return { projection, ...projection };
|
||||
}
|
||||
|
||||
function commitProjectorResumeProjection(): void {
|
||||
const completed = projectorResumeCompletedTrace();
|
||||
state.traces.trc_f2456023233a485b = completed;
|
||||
const session = sessionById("ses_317f78e1-ed91-4b09-a72a-3668d5469f59");
|
||||
if (!session) return;
|
||||
session.status = "completed";
|
||||
session.updatedAt = new Date().toISOString();
|
||||
session.turnSummary = { traceId: "trc_f2456023233a485b", status: "completed", running: false, terminal: true, ...projectorResumeProjection("caught-up", 7) };
|
||||
session.messages = (session.messages ?? []).map((message) => {
|
||||
if (message.role !== "agent" || message.traceId !== "trc_f2456023233a485b") return message;
|
||||
return { ...message, text: "fake projector resume completed from durable checkpoint.", status: "completed", runnerTrace: completed, ...projectorResumeProjection("caught-up", 7) };
|
||||
});
|
||||
const terminalEvent = (completed.events as JsonRecord[]).at(-1) ?? null;
|
||||
if (terminalEvent) broadcastSse("workbench.trace.event", { type: "trace.event", sessionId: session.sessionId, threadId: session.threadId ?? null, traceId: "trc_f2456023233a485b", event: terminalEvent, snapshot: completed });
|
||||
broadcastSse("workbench.turn.snapshot", { type: "turn.snapshot", sessionId: session.sessionId, threadId: session.threadId ?? null, traceId: "trc_f2456023233a485b", turn: turnPayload("trc_f2456023233a485b") });
|
||||
}
|
||||
|
||||
function markSessionText(sessions: SessionRecord[], sessionId: string, userText: string, agentText: string): void {
|
||||
const session = sessions.find((item) => item.sessionId === sessionId);
|
||||
if (!session) return;
|
||||
@@ -1037,22 +1131,29 @@ function workbenchTurnPayload(traceId: string): JsonRecord {
|
||||
function tracePayload(traceId: string, url: URL): JsonRecord {
|
||||
const turn = turnPayload(traceId);
|
||||
const events = Array.isArray(turn.events) ? turn.events as JsonRecord[] : [];
|
||||
const sinceSeq = Number(url.searchParams.get("sinceSeq") ?? 0);
|
||||
const afterSeq = traceAfterSeq(url);
|
||||
const requestedLimit = Number(url.searchParams.get("limit") ?? events.length);
|
||||
const limit = requestedLimit || events.length || 100;
|
||||
const filtered = Number.isFinite(sinceSeq) && sinceSeq > 0 ? events.filter((event) => Number(event.seq ?? 0) > sinceSeq) : events;
|
||||
const filtered = afterSeq > 0 ? events.filter((event) => Number(event.seq ?? 0) > afterSeq) : events;
|
||||
const page = filtered.slice(0, Math.max(1, Number.isFinite(limit) ? limit : 100));
|
||||
const lastSeq = Number(page.at(-1)?.seq ?? 0);
|
||||
const firstSeq = page.length > 0 ? Number(page[0]?.seq ?? 0) : null;
|
||||
const lastSeq = page.length > 0 ? Number(page.at(-1)?.seq ?? 0) : null;
|
||||
const hasMore = filtered.length > page.length;
|
||||
return { ...turn, events: page, eventCount: events.length, hasMore, fullTraceLoaded: !hasMore, nextSinceSeq: hasMore ? lastSeq : null, range: { sinceSeq, returned: page.length, total: events.length } };
|
||||
return { ...turn, events: page, eventCount: events.length, hasMore, fullTraceLoaded: !hasMore, nextSinceSeq: lastSeq, nextSeq: page.length > 0 ? lastSeq : afterSeq, range: { afterSeq, fromSeq: firstSeq, toSeq: lastSeq, limit: Math.max(1, Number.isFinite(limit) ? limit : 100), returned: page.length, total: events.length } };
|
||||
}
|
||||
|
||||
function traceAfterSeq(url: URL): number {
|
||||
const cursor = url.searchParams.get("cursor") ?? "";
|
||||
if (cursor.startsWith("seq:")) return Math.max(0, Number.parseInt(cursor.slice(4), 10) || 0);
|
||||
return Math.max(0, Number.parseInt(url.searchParams.get("sinceSeq") ?? "0", 10) || 0);
|
||||
}
|
||||
|
||||
function workbenchTracePayload(traceId: string, url: URL): JsonRecord {
|
||||
const payload = tracePayload(traceId, url);
|
||||
if (state.scenarioId === "tool-completed-projection-running" && traceId === "trc_running") {
|
||||
return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, sessionId: payload.sessionId ?? null, threadId: payload.threadId ?? null, traceStatus: "completed", events: payload.events, eventCount: payload.eventCount, hasMore: payload.hasMore, nextSeq: payload.nextSinceSeq, range: payload.range, fullTraceLoaded: payload.fullTraceLoaded, projectionStatus: "projecting", terminalEvidence: null, finalResponse: null, traceSummary: payload.traceSummary, retention: payload.retention };
|
||||
return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, sessionId: payload.sessionId ?? null, threadId: payload.threadId ?? null, traceStatus: "completed", events: payload.events, eventCount: payload.eventCount, hasMore: payload.hasMore, nextSeq: payload.nextSeq, range: payload.range, fullTraceLoaded: payload.fullTraceLoaded, projectionStatus: "projecting", terminalEvidence: null, finalResponse: null, traceSummary: payload.traceSummary, retention: payload.retention };
|
||||
}
|
||||
return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, sessionId: payload.sessionId ?? null, threadId: payload.threadId ?? null, traceStatus: payload.status, events: payload.events, eventCount: payload.eventCount, hasMore: payload.hasMore, nextSeq: payload.nextSinceSeq, range: payload.range, fullTraceLoaded: payload.fullTraceLoaded, terminalEvidence: payload.terminalEvidence, finalResponse: payload.finalResponse, traceSummary: payload.traceSummary, retention: payload.retention, ...projectionEnvelope(payload) };
|
||||
return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, sessionId: payload.sessionId ?? null, threadId: payload.threadId ?? null, traceStatus: payload.status, events: payload.events, eventCount: payload.eventCount, hasMore: payload.hasMore, nextSeq: payload.nextSeq, range: payload.range, fullTraceLoaded: payload.fullTraceLoaded, terminalEvidence: payload.terminalEvidence, finalResponse: payload.finalResponse, traceSummary: payload.traceSummary, retention: payload.retention, ...projectionEnvelope(payload) };
|
||||
}
|
||||
|
||||
function projectionEnvelope(payload: JsonRecord): JsonRecord {
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
import { expect, fakeServerState, gotoWorkbench, saveScreenshot, test } from "../fixtures/test";
|
||||
import { selectors, sessionTab } from "../fixtures/selectors";
|
||||
|
||||
test.describe("durable projector resume", () => {
|
||||
test.use({ scenarioId: "projector-resume-from-checkpoint" });
|
||||
|
||||
test("catches up a stale AgentRun trace without read-side repair", async ({ page }, testInfo) => {
|
||||
const sessionId = "ses_317f78e1-ed91-4b09-a72a-3668d5469f59";
|
||||
const traceId = "trc_f2456023233a485b";
|
||||
|
||||
await gotoWorkbench(page, `/workbench/sessions/${sessionId}`);
|
||||
|
||||
const initialTurn = await page.request.get(`/v1/workbench/turns/${traceId}`);
|
||||
expect(initialTurn.status()).toBe(200);
|
||||
const initialTurnBody = await initialTurn.json();
|
||||
expect(initialTurnBody.turn.status).toBe("running");
|
||||
expect(initialTurnBody.turn.projectionStatus).toBe("projecting");
|
||||
expect(initialTurnBody.turn.projectionHealth).toBe("projecting");
|
||||
expect(initialTurnBody.turn.lastProjectedSeq).toBe(4);
|
||||
|
||||
await expect(page.locator(sessionTab(sessionId))).toHaveAttribute("data-active", "true");
|
||||
await expect(page.locator(`${selectors.messageCard}[data-role="agent"]`).last()).toHaveAttribute("data-status", "running");
|
||||
await saveScreenshot(page, testInfo, "projector-resume-before-catchup");
|
||||
|
||||
await expect.poll(async () => {
|
||||
const response = await page.request.get(`/v1/workbench/turns/${traceId}`);
|
||||
if (!response.ok()) return "http-error";
|
||||
const payload = await response.json();
|
||||
return `${payload.turn?.status}/${payload.turn?.projectionHealth}/${payload.turn?.lastProjectedSeq}`;
|
||||
}, { timeout: 12_000 }).toBe("completed/caught-up/7");
|
||||
|
||||
const trace = await page.request.get(`/v1/workbench/traces/${traceId}/events?limit=10`);
|
||||
expect(trace.status()).toBe(200);
|
||||
const traceBody = await trace.json();
|
||||
expect(traceBody.traceStatus).toBe("completed");
|
||||
expect(traceBody.projectionStatus).toBe("caught-up");
|
||||
expect(traceBody.projectionHealth).toBe("caught-up");
|
||||
expect(traceBody.range).toMatchObject({ afterSeq: 0, fromSeq: 1, toSeq: 7, returned: 7, total: 7 });
|
||||
|
||||
const emptyTracePage = await page.request.get(`/v1/workbench/traces/${traceId}/events?cursor=seq:999999&limit=5`);
|
||||
expect(emptyTracePage.status()).toBe(200);
|
||||
const emptyTraceBody = await emptyTracePage.json();
|
||||
expect(emptyTraceBody.events).toHaveLength(0);
|
||||
expect(emptyTraceBody.range).toMatchObject({ afterSeq: 999999, fromSeq: null, toSeq: null, returned: 0, total: 7 });
|
||||
expect(emptyTraceBody.nextSeq).toBe(999999);
|
||||
|
||||
const finalCard = page.locator(`${selectors.messageCard}[data-role="agent"]`).filter({ hasText: "fake projector resume completed from durable checkpoint" });
|
||||
await expect(finalCard).toHaveAttribute("data-status", "completed");
|
||||
await expect(page.locator(`${selectors.traceTimeline}[data-status="completed"]`).last()).toBeVisible();
|
||||
await expect(page.locator(selectors.commandSend)).toHaveAttribute("data-action", "turn");
|
||||
await saveScreenshot(page, testInfo, "projector-resume-after-catchup");
|
||||
|
||||
const state = await fakeServerState(page) as { legacyRequestLedger?: unknown[]; chatRequests?: unknown[] };
|
||||
expect(state.legacyRequestLedger).toEqual([]);
|
||||
expect(state.chatRequests).toEqual([]);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user