From 09327f1ded2d114f65ba1a9672dcbd5ae4e177e9 Mon Sep 17 00:00:00 2001 From: lyon1998 Date: Sun, 28 Jun 2026 12:52:58 +0800 Subject: [PATCH] fix: keep workbench projection checkpoint behind events --- AGENTS.md | 1 + .../cloud/workbench-projection-writer.test.ts | 83 +++++++++++++++++++ internal/cloud/workbench-projection-writer.ts | 23 +++-- 3 files changed, 101 insertions(+), 6 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 85697a65..d97f2611 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,6 +6,7 @@ HWLAB 是硬件实验室运行面和控制面项目。本文件只作为低噪 - 先按 issue、PR、CLI 参数或受控 lane 配置解析目标 `node + lane`；没有明确目标时才读取受控配置，禁止把 G14、v0.2、D601 legacy 或任意历史路径写成全局默认。 - D601 v0.3 的固定 workspace 是 `/home/ubuntu/workspace/hwlab-v03`，固定跟踪 `origin/v0.3`；它是当前 D601/v03 source truth，不是 legacy 对照面。 +- D518 v0.3 的固定主 workspace 是 `D518:/home/ubuntu/workspace/hwlab-v03`，固定跟踪 `github/v0.3`；D518 源码、配置、部署和 issue 修复必须从这里创建 `.worktree/`，不要使用 master server 或 `/root/HWLAB` 作为 D518 source truth。 - k3s 操作必须通过 UniDesk route，例如 `D601:k3s` 或 `G14:k3s`；不要使用另一节点 kubeconfig、Docker Desktop Kubernetes、master server 本地构建或旧 JS CD 结果作为当前 lane 证据。 - `hyueapi.com` / `.hyueapi.com` 必须保留在 `NO_PROXY` / `no_proxy`，Codex API 通道不得被 HTTP/SOCKS proxy 劫持。 diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index 45eb96bc..d93a1b77 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -357,6 +357,89 @@ test("workbench projection writer does not synthesize terminal trace events from assert.equal(loaded.facts.messages.some((message) => message.role !== "user" && message.text === "BACKFILL_DONE"), true); }); +test("workbench projection terminal facts do not advance checkpoint past durable trace events", async () => { + const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:45:00.000Z" }); + const traceId = "trc_writer_terminal_checkpoint_gap"; + const sessionId = "ses_writer_terminal_checkpoint_gap"; + const finalText = "CHECKPOINT_GAP_DONE"; + const accessController = { + async recordAgentSessionOwner(input) { + return { + id: input.sessionId, + projectId: input.projectId, + ownerUserId: input.ownerUserId, + conversationId: input.conversationId, + threadId: input.threadId, + lastTraceId: input.traceId, + status: input.status, + session: input.session, + updatedAt: "2026-06-20T11:45:00.000Z" + }; + } + }; + + for (let sourceSeq = 1; sourceSeq <= 33; sourceSeq += 1) { + await writeWorkbenchProjectionEvent({ + runtimeStore, + event: { + traceId, + sessionId, + source: "agentrun", + sourceSeq, + type: sourceSeq === 33 ? "terminal" : "backend", + eventType: sourceSeq === 33 ? "terminal" : "backend", + status: sourceSeq === 33 ? "completed" : "running", + label: sourceSeq === 33 ? "agentrun:result:completed" : `agentrun:backend:${sourceSeq}`, + terminal: sourceSeq === 33, + runId: "run_writer_terminal_checkpoint_gap", + commandId: "cmd_writer_terminal_checkpoint_gap", + createdAt: `2026-06-20T11:44:${String(sourceSeq).padStart(2, "0")}.000Z` + } + }); + } + + await writeWorkbenchProjectionSession({ + accessController, + runtimeStore, + traceId, + ownerUserId: "usr_writer", + ownerRole: "user", + sessionId, + projectId: "prj_writer", + conversationId: "cnv_writer_terminal_checkpoint_gap", + threadId: "thread-writer-terminal-checkpoint-gap", + status: "completed", + payload: { + traceId, + status: "completed", + finalResponse: { text: finalText, status: "completed", traceId }, + runnerTrace: { + traceId, + status: "completed", + eventCount: 34, + events: Array.from({ length: 34 }, (_, index) => ({ seq: index + 1, type: "backend", status: index === 33 ? "completed" : "running", terminal: index === 33 })) + }, + agentRun: { runId: "run_writer_terminal_checkpoint_gap", commandId: "cmd_writer_terminal_checkpoint_gap", status: "completed", terminalStatus: "completed", lastSeq: 30 }, + updatedAt: "2026-06-20T11:45:00.000Z" + }, + session: { + sessionStatus: "completed", + messages: [ + { messageId: "msg_writer_checkpoint_gap_user", role: "user", text: "question", status: "sent", turnId: traceId, traceId }, + { messageId: "msg_writer_checkpoint_gap_agent", role: "agent", text: finalText, status: "completed", turnId: traceId, traceId } + ], + finalResponse: { text: finalText, status: "completed", traceId } + } + }); + + const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["traceEvents", "checkpoints", "turns"], limit: 80 }); + assert.equal(loaded.facts.traceEvents.length, 33); + assert.equal(Math.max(...loaded.facts.traceEvents.map((event) => event.projectedSeq)), 33); + assert.equal(loaded.facts.checkpoints[0].projectedSeq, 33); + assert.equal(loaded.facts.checkpoints[0].sourceSeq, 30); + assert.equal(loaded.facts.turns[0].terminal, true); +}); + test("workbench projection writer does not duplicate terminal trace events that already exist", async () => { const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:25:00.000Z" }); const traceId = "trc_writer_terminal_backfill_existing"; diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index d37038b8..0fcb4f22 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -534,6 +534,7 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own const diagnostic = timingAuthorityIssue ? { ...baseDiagnostic, projectionHealth: baseDiagnostic.projectionHealth === "unavailable" ? "unavailable" : "degraded", blocker: baseDiagnostic.blocker ?? timingAuthorityIssue, timingAuthorityIssue, valuesRedacted: true } : baseDiagnostic; + const factSeq = workbenchProjectionFactSequence({ projection, previousCheckpoint, agentRun }); const previousFinalText = finalResponseTextValue(previousCheckpoint?.finalResponse, previousCheckpoint?.assistantText, previousCheckpoint?.finalText); const finalText = terminal ? finalResponseTextValue(projection.finalResponse, payload?.finalResponse, session?.finalResponse, payload?.assistantText, previousFinalText) : null; const projectedFinalResponse = terminal && finalText ? { text: finalText, status: terminalStatus, traceId: safeId, valuesPrinted: false } : null; @@ -558,8 +559,8 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own threadId: resolvedThreadId, status: terminal ? terminalStatus : "running", lastTraceId: safeId, - projectedSeq: nonNegativeInteger(projection.lastProjectedSeq), - sourceSeq: nonNegativeInteger(projection.lastProjectedSeq), + projectedSeq: factSeq.projectedSeq, + sourceSeq: factSeq.sourceSeq, sourceEventId: safeId, terminal, sealed: terminal, @@ -580,8 +581,8 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own traceId: safeId, messageId: messages.find((message) => message.role !== "user")?.messageId ?? null, status: projection.status ?? normalizedStatus, - projectedSeq: nonNegativeInteger(projection.lastProjectedSeq), - sourceSeq: nonNegativeInteger(projection.lastProjectedSeq), + projectedSeq: factSeq.projectedSeq, + sourceSeq: factSeq.sourceSeq, sourceEventId: safeId, terminal: projection.terminal, sealed: projection.terminal, @@ -602,8 +603,8 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own turnId: projection.turnId ?? safeId, runId: textValue(agentRun?.runId ?? projection.sourceRunId) || null, commandId: textValue(agentRun?.commandId ?? projection.sourceCommandId) || null, - projectedSeq: nonNegativeInteger(projection.lastProjectedSeq), - sourceSeq: nonNegativeInteger(agentRun?.lastSeq ?? projection.lastProjectedSeq), + projectedSeq: factSeq.projectedSeq, + sourceSeq: factSeq.sourceSeq, sourceEventId: safeId, projectionStatus: projection.terminal ? "caught_up" : "projecting", projectionHealth: diagnostic.projectionHealth === "unavailable" ? "unavailable" : diagnostic.projectionHealth === "stalled" ? "stalled" : "healthy", @@ -623,6 +624,16 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own }; } +function workbenchProjectionFactSequence({ projection = {}, previousCheckpoint = null, agentRun = null } = {}) { + const projectionSeq = nonNegativeInteger(projection?.lastProjectedSeq); + const previousSeq = nonNegativeInteger(previousCheckpoint?.projectedSeq ?? previousCheckpoint?.lastProjectedSeq); + const sourceSeq = nonNegativeInteger(agentRun?.lastSeq ?? projection?.sourceSeq ?? projectionSeq); + return { + projectedSeq: previousSeq > 0 ? Math.min(projectionSeq || previousSeq, previousSeq) : projectionSeq, + sourceSeq + }; +} + function workbenchProjectionError(code, message) { const error = new Error(message); error.code = code;