mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: recover AgentRun terminal projection

Browse Source
This commit is contained in:
lyon
2026-06-20 20:03:38 +08:00
parent 49c6fad495
commit 800f674d5c
6 changed files with 314 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.ts
+27 -5
View File
@@ -1,5 +1,5 @@
/*
* SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p0-projector-resume; draft-2026-06-20-p1-zero-split-durable-realtime; draft-2026-06-20-p3-read-side-zero; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
* SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010403 API契约 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010401 Web工作台 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
* 职责: Workbench REST read model。只读投影 session/message/turn/trace facts，不执行 AgentRun sync、billing finalize 或 workspace repair。
*/
import { createHash } from "node:crypto";
@@ -22,7 +22,7 @@ const MAX_PAGE_LIMIT = 100;
const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000;
const WORKBENCH_SESSION_LIST_PAGE_FAMILIES = Object.freeze(["sessions"]);
const WORKBENCH_SESSION_DETAIL_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "checkpoints"]);
const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "messages", "turns", "checkpoints"]);
const WORKBENCH_TRACE_EVENT_PAGE_FAMILIES = Object.freeze(["traceEvents"]);
export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) {
try {
@@ -467,6 +467,26 @@ function canActorReadFactSession(session, actor) {
return session.ownerUserId === actor.id;
}
async function visibleFactSessionForTrace(readModel, facts, actor, traceId) {
const visible = visibleFactSessions(facts, actor);
const byLastTrace = visible.find((item) => item.lastTraceId === traceId) ?? null;
if (byLastTrace) return { session: byLastTrace, facts };
const relatedSessionIds = uniqueText([
factTurnForTrace(facts, traceId)?.sessionId,
factArray(facts.messages).find((message) => message.traceId === traceId)?.sessionId,
factCheckpointForTrace(facts, traceId)?.sessionId
]);
const alreadyLoaded = visible.find((item) => relatedSessionIds.includes(factSessionId(item))) ?? null;
if (alreadyLoaded) return { session: alreadyLoaded, facts };
for (const sessionId of relatedSessionIds) {
const result = await queryFactsByRouteId(readModel, sessionId, { families: ["sessions"], limit: 1 });
if (result.error) return { error: result.error };
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
if (session) return { session, facts: mergeFactSets(facts, result.facts) };
}
return { session: null, facts };
}
function factSessionMatchesRouteId(session, routeId) {
const sessionId = safeSessionId(routeId);
if (sessionId && factSessionId(session) === sessionId) return true;
@@ -1034,8 +1054,10 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
const pageOptions = tracePageOptions(url);
const metadata = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_METADATA_FAMILIES, limit: 1 });
if (metadata.error) return sendJson(response, 503, workbenchProjectionStoreError(metadata.error));
const metadataSessions = visibleFactSessions(metadata.facts, actor);
const session = metadataSessions.find((item) => item.lastTraceId === traceId) ?? metadataSessions[0] ?? null;
const resolved = await visibleFactSessionForTrace(readModel, metadata.facts, actor, traceId);
if (resolved.error) return sendJson(response, 503, workbenchProjectionStoreError(resolved.error));
const session = resolved.session;
const metadataFacts = resolved.facts ?? metadata.facts;
if (!session) {
const context = await readModel.queryFacts({ traceId, families: ["sessions", "turns", "checkpoints"], limit: MAX_PAGE_LIMIT });
if (context.error) return sendJson(response, 503, workbenchProjectionStoreError(context.error));
@@ -1049,7 +1071,7 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
}
return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId }));
}
const projection = factProjectionForTrace(metadata.facts, traceId);
const projection = factProjectionForTrace(metadataFacts, traceId);
const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 });
if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status });