From 683042900fb15afe7f8478498b959d73deb550ca Mon Sep 17 00:00:00 2001 From: UniDesk Codex Date: Thu, 2 Jul 2026 08:43:03 +0800 Subject: [PATCH] fix: project workbench session current turn authority --- internal/cloud/server-workbench-http.test.ts | 54 ++++++ internal/cloud/server-workbench-http.ts | 56 +++++- internal/workbenchruntime/service.go | 189 ++++++++++++++++--- internal/workbenchruntime/service_test.go | 38 ++++ 4 files changed, 302 insertions(+), 35 deletions(-) diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 3204b253..242b0c8f 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1661,6 +1661,60 @@ test("workbench read model projects current turn running state for idle session } }); +test("workbench read model projects newest running turn over stale completed session lastTraceId", async () => { + const oldTraceId = "trc_workbench_stale_completed_old"; + const newTraceId = "trc_workbench_new_running_current"; + const session = { + id: "ses_workbench_stale_completed_new_running", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "completed", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_stale_completed_new_running", + threadId: "thread-workbench-stale-completed-new-running", + lastTraceId: oldTraceId, + updatedAt: "2026-07-01T00:00:02.000Z", + session: { + sessionStatus: "completed", + lastTraceId: oldTraceId, + messages: [ + { messageId: "msg_workbench_old_user", role: "user", text: "first", traceId: oldTraceId, status: "sent", projectedSeq: 10 }, + { messageId: "msg_workbench_old_agent", role: "agent", text: "first done", traceId: oldTraceId, status: "completed", projectedSeq: 20 } + ] + } + }; + const facts = emptyFacts(); + mergeFacts(facts, buildDurableFactsForSession({ session, status: "completed", finalText: "first done", lastProjectedSeq: 20 })); + facts.messages.push( + { messageId: "msg_workbench_new_user", sessionId: session.id, turnId: newTraceId, traceId: newTraceId, role: "user", status: "sent", text: "second", projectedSeq: 30, sourceSeq: 30, sourceEventId: "evt_new_user", terminal: false, sealed: false, updatedAt: "2026-07-01T00:00:03.000Z", valuesRedacted: true }, + { messageId: "msg_workbench_new_agent", sessionId: session.id, turnId: newTraceId, traceId: newTraceId, role: "agent", status: "running", text: "", projectedSeq: 40, sourceSeq: 40, sourceEventId: "evt_new_agent", terminal: false, sealed: false, updatedAt: "2026-07-01T00:00:04.000Z", valuesRedacted: true } + ); + facts.turns.push({ turnId: newTraceId, sessionId: session.id, traceId: newTraceId, messageId: "msg_workbench_new_agent", status: "running", projectedSeq: 40, sourceSeq: 40, sourceEventId: "evt_new_turn", terminal: false, sealed: false, diagnostic: { projectionStatus: "projecting", projectionHealth: "healthy", valuesRedacted: true }, timing: { startedAt: "2026-07-01T00:00:03.000Z", lastEventAt: "2026-07-01T00:00:04.000Z", finishedAt: null, durationMs: null, valuesRedacted: true }, startedAt: "2026-07-01T00:00:03.000Z", lastEventAt: "2026-07-01T00:00:04.000Z", finishedAt: null, durationMs: null, updatedAt: "2026-07-01T00:00:04.000Z", valuesRedacted: true }); + facts.checkpoints.push({ traceId: newTraceId, sessionId: session.id, turnId: newTraceId, projectedSeq: 40, sourceSeq: 40, sourceEventId: "evt_new_checkpoint", projectionStatus: "projecting", projectionHealth: "healthy", terminal: false, sealed: false, diagnostic: { projectionStatus: "projecting", projectionHealth: "healthy", valuesRedacted: true }, updatedAt: "2026-07-01T00:00:04.000Z", valuesRedacted: true }); + const runtimeStore = createRuntimeStoreFromFacts(facts); + const accessController = { + store: { async listAgentSessionsForUser() { return [session]; } }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const server = createCloudApiServer({ accessController, traceStore: createCodeAgentTraceStore(), runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`); + assert.equal(sessions.status, 200); + assert.equal(sessions.body.sessions[0].lastTraceId, newTraceId); + assert.equal(sessions.body.sessions[0].status, "running"); + assert.equal(sessions.body.sessions[0].running, true); + assert.equal(sessions.body.sessions[0].terminal, false); + assert.equal(sessions.body.sessions[0].turnSummary.traceId, newTraceId); + assert.equal(sessions.body.sessions[0].turnSummary.status, "running"); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench read model projects completed current turn for idle session summary", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 0d8bfeae..4eb621f7 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -872,8 +872,7 @@ function queryRequestsOnlySessionFacts(query = {}) { async function queryFactsForSessionSummaries(readModel, sessions = []) { const sessionIds = uniqueText(sessions.map(factSessionId)); - const traceIds = uniqueText(sessions.map(factLastTraceId)); - if (sessionIds.length === 0 && traceIds.length === 0) return { facts: emptyFactSet(), count: 0, durable: false, persistence: null }; + if (sessionIds.length === 0 && uniqueText(sessions.map(factLastTraceId)).length === 0) return { facts: emptyFactSet(), count: 0, durable: false, persistence: null }; // Session list summary is hot and is also sampled by the two-page web-probe // observer. Avoid parallel fan-out here so one list request cannot consume // multiple durable-read connections while another page is refreshing. @@ -881,6 +880,7 @@ async function queryFactsForSessionSummaries(readModel, sessions = []) { ? await readModel.queryFacts({ sessionIds, families: ["messages", "parts", "turns"] }) : { facts: emptyFactSet(), count: 0, durable: false, persistence: null }; if (bySession.error) return bySession; + const traceIds = summaryTraceIds(sessions, bySession.facts); const byTrace = traceIds.length > 0 ? await readModel.queryFacts({ traceIds, families: ["checkpoints"] }) : { facts: emptyFactSet(), count: 0, durable: false, persistence: null }; @@ -889,6 +889,13 @@ async function queryFactsForSessionSummaries(readModel, sessions = []) { return { facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0), durable: bySession.durable ?? byTrace.durable, persistence: bySession.persistence ?? byTrace.persistence ?? null }; } +function summaryTraceIds(sessions = [], facts = {}) { + return uniqueText([ + ...sessions.map(factLastTraceId), + ...sessions.map((session) => factCurrentTraceForSession(facts, factSessionId(session), session)?.traceId) + ]); +} + function visibleFactSessions(facts, actor) { return hideShadowedAgentRunAliasSessions(factArray(facts?.sessions) .filter((session) => canActorReadFactSession(session, actor)) @@ -945,7 +952,8 @@ function factSessionMatchesRouteId(session, routeId) { function factSessionSummary(session, facts = {}) { const sessionId = factSessionId(session); if (!sessionId) return null; - const traceId = factLastTraceId(session) ?? factLatestTraceIdForSession(facts, sessionId); + const currentTrace = factCurrentTraceForSession(facts, sessionId, session); + const traceId = currentTrace?.traceId ?? factLastTraceId(session) ?? factLatestTraceIdForSession(facts, sessionId); const projection = traceId ? factProjectionForTrace(facts, traceId) : null; const turn = traceId ? factTurnForTrace(facts, traceId) : null; const checkpoint = traceId ? factCheckpointForTrace(facts, traceId) : null; @@ -957,7 +965,7 @@ function factSessionSummary(session, facts = {}) { const messageTerminalStatus = factTerminalStatusFromMessageProjection(terminalMessage); const turnStatus = normalizeStatus(turn?.status); const launchContext = compactLaunchContext(session?.sessionJson?.launchContext); - const status = normalizeStatus(checkpointStatus ?? traceStatus ?? messageTerminalStatus ?? turnStatus ?? session?.status); + const status = normalizeStatus(checkpointStatus ?? traceStatus ?? messageTerminalStatus ?? nonUnknownStatus(turnStatus) ?? nonUnknownStatus(currentTrace?.status) ?? session?.status); const timing = traceId ? factCombinedTimingProjection(status, checkpoint, messageTerminalStatus ? terminalMessage : null, turn, trace) : factTimingProjection(session, status); @@ -1613,6 +1621,46 @@ function factLatestTraceIdForSession(facts = {}, sessionId) { return safeTraceId(turn?.traceId) ?? null; } +function factCurrentTraceForSession(facts = {}, sessionId, session = null) { + const candidates = []; + const sessionTraceId = factLastTraceId(session); + if (sessionTraceId) candidates.push(factTraceCandidate(session, sessionTraceId, normalizeStatus(session?.status), 0)); + for (const message of factArray(facts.messages)) { + if (message?.sessionId !== sessionId || !safeTraceId(message?.traceId)) continue; + candidates.push(factMessageTraceCandidate(message)); + } + for (const turn of factArray(facts.turns)) { + if (turn?.sessionId !== sessionId || !safeTraceId(turn?.traceId)) continue; + candidates.push(factTraceCandidate(turn, safeTraceId(turn.traceId), normalizeStatus(turn?.status), 3)); + } + return candidates.sort(compareTraceCandidatesDesc)[0] ?? null; +} + +function factMessageTraceCandidate(message) { + const role = textValue(message?.role); + let status = normalizeStatus(message?.status); + let priority = 1; + if (isAssistantLikeRole(role)) { + priority = 2; + if (status === "unknown") status = "running"; + } else if (safeTraceId(message?.traceId)) { + status = "running"; + } + return factTraceCandidate(message, safeTraceId(message?.traceId), status, priority); +} + +function factTraceCandidate(record, traceId, status, priority) { + return { traceId: safeTraceId(traceId) ?? null, status: normalizeStatus(status), seq: factSeq(record), updatedAt: factUpdatedAt(record), priority }; +} + +function compareTraceCandidatesDesc(left, right) { + return compareNumberDesc(left?.seq, right?.seq) || compareTimestampDesc(left?.updatedAt, right?.updatedAt) || compareNumberDesc(left?.priority, right?.priority); +} + +function nonUnknownStatus(status) { + return normalizeStatus(status) === "unknown" ? null : status; +} + function factTurnId(turn, fallbackTraceId = null) { return safeTurnId(turn?.turnId) ?? safeTraceId(turn?.turnId) ?? safeTraceId(fallbackTraceId) ?? null; } diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index 168fa86f..b81912a8 100644 --- a/internal/workbenchruntime/service.go +++ b/internal/workbenchruntime/service.go @@ -1346,44 +1346,55 @@ func (s *Server) queryIncludedSession(ctx context.Context, routeID string) (map[ func (s *Server) querySummaryFacts(ctx context.Context, sessions []map[string]any) (map[string][]any, error) { sessionIDs := uniqueStrings(mapSessions(sessions, sessionID)) - traceIDs := uniqueStrings(mapSessions(sessions, lastTraceID)) facts := emptyFacts() - type summaryFactsResult struct { - kind string - facts map[string][]any - err error - } - resultCount := 0 - results := make(chan summaryFactsResult, 2) if len(sessionIDs) > 0 { - resultCount++ - go func() { - messageSummaries, err := s.queryMessageSummaries(ctx, sessionIDs) - results <- summaryFactsResult{kind: "messageSummaries", facts: map[string][]any{"messageSummaries": messageSummaries}, err: err} - }() + messageSummaries, err := s.queryMessageSummaries(ctx, sessionIDs) + if err != nil { + return nil, err + } + facts = mergeFacts(facts, map[string][]any{"messageSummaries": messageSummaries}) + + bySession, err := s.queryFacts(ctx, factQuery{SessionIDs: sessionIDs, Families: []string{"messages", "turns"}, Limit: summarySessionFactLimit(len(sessionIDs)), MessagesOrder: "updated_desc", TurnsOrder: "updated_desc"}) + if err != nil { + return nil, err + } + facts = mergeFacts(facts, bySession) } + traceIDs := summaryTraceIDs(sessions, facts) if len(traceIDs) > 0 { - resultCount++ - go func() { - byTrace, err := s.queryFacts(ctx, factQuery{TraceIDs: traceIDs, Families: []string{"turns", "checkpoints"}, Limit: len(traceIDs), TurnsOrder: "updated_desc", CheckpointsOrder: "updated_desc"}) - results <- summaryFactsResult{kind: "traceFacts", facts: byTrace, err: err} - }() - } - for i := 0; i < resultCount; i++ { - result := <-results - if result.err != nil { - return nil, result.err - } - switch result.kind { - case "messageSummaries": - facts = mergeFacts(facts, result.facts) - case "traceFacts": - facts = mergeFacts(facts, result.facts) + byTrace, err := s.queryFacts(ctx, factQuery{TraceIDs: traceIDs, Families: []string{"turns", "checkpoints"}, Limit: len(traceIDs), TurnsOrder: "updated_desc", CheckpointsOrder: "updated_desc"}) + if err != nil { + return nil, err } + facts = mergeFacts(facts, byTrace) } return facts, nil } +func summarySessionFactLimit(sessionCount int) int { + if sessionCount <= 0 { + return 0 + } + limit := sessionCount * 8 + if limit < sessionCount { + return sessionCount + } + if limit > 1000 { + return 1000 + } + return limit +} + +func summaryTraceIDs(sessions []map[string]any, facts map[string][]any) []string { + values := mapSessions(sessions, lastTraceID) + for _, session := range sessions { + if candidate := currentTraceForSession(session, facts, sessionID(session)); candidate.traceID != "" { + values = append(values, candidate.traceID) + } + } + return uniqueStrings(values) +} + func (s *Server) queryFacts(ctx context.Context, query factQuery) (map[string][]any, error) { families := factFamilySet(query) facts := emptyFacts() @@ -2047,7 +2058,8 @@ func sessionSummary(session map[string]any, facts map[string][]any) map[string]a if sid == "" { return nil } - traceID := first(lastTraceID(session), latestTraceIDForSession(facts, sid)) + currentTrace := currentTraceForSession(session, facts, sid) + traceID := currentTrace.traceID var projection map[string]any if traceID != "" { projection = projectionForTrace(facts, traceID) @@ -2066,7 +2078,7 @@ func sessionSummary(session map[string]any, facts map[string][]any) map[string]a } checkpointStatus := terminalStatus(text(checkpoint["status"])) turnStatus := normalizeStatus(turn["status"]) - status := normalizeStatus(first(checkpointStatus, turnStatus, text(session["status"]))) + status := normalizeStatus(first(checkpointStatus, nonUnknownStatus(turnStatus), nonUnknownStatus(currentTrace.status), text(session["status"]))) timingSource := turn if checkpointStatus != "" { timingSource = checkpoint @@ -2080,6 +2092,93 @@ func sessionSummary(session map[string]any, facts map[string][]any) map[string]a return map[string]any{"sessionId": sid, "threadId": nullableText(session["threadId"]), "agentId": first(text(session["agentId"]), "hwlab-code-agent"), "title": title, "name": title, "status": status, "running": isRunning(status), "terminal": isTerminal(status), "lastTraceId": nullableText(traceID), "projection": projection, "projectionStatus": projection["projectionStatus"], "projectionHealth": projection["projectionHealth"], "staleMs": projection["staleMs"], "blocker": projection["blocker"], "providerProfile": nullableText(first(text(session["providerProfile"]), text(objectMap(session["sessionJson"])["providerProfile"]))), "messageCount": messageCount, "firstUserMessagePreview": firstUserMessagePreview, "updatedAt": updatedAt(session), "turnSummary": turnSummary, "valuesRedacted": true} } +type traceCandidate struct { + traceID string + status string + seq int64 + hasSeq bool + updatedAt string + priority int +} + +func currentTraceForSession(session map[string]any, facts map[string][]any, sid string) traceCandidate { + best := traceCandidateFromRecord(session, lastTraceID(session), normalizeStatus(session["status"]), 0) + for _, row := range facts["messages"] { + item := objectMap(row) + if text(item["sessionId"]) != sid || text(item["traceId"]) == "" { + continue + } + if candidate := messageTraceCandidate(item); traceCandidateNewer(candidate, best) { + best = candidate + } + } + for _, row := range facts["turns"] { + item := objectMap(row) + if text(item["sessionId"]) != sid || text(item["traceId"]) == "" { + continue + } + candidate := traceCandidateFromRecord(item, text(item["traceId"]), normalizeStatus(item["status"]), 3) + if traceCandidateNewer(candidate, best) { + best = candidate + } + } + if best.traceID == "" { + best.traceID = latestTraceIDForSession(facts, sid) + } + return best +} + +func messageTraceCandidate(message map[string]any) traceCandidate { + status := normalizeStatus(message["status"]) + priority := 1 + if text(message["role"]) == "agent" { + priority = 2 + if status == "unknown" { + status = "running" + } + } else if text(message["traceId"]) != "" { + status = "running" + } + return traceCandidateFromRecord(message, text(message["traceId"]), status, priority) +} + +func traceCandidateFromRecord(record map[string]any, traceID string, status string, priority int) traceCandidate { + seq, hasSeq := seqInt64(record) + return traceCandidate{traceID: traceID, status: status, seq: seq, hasSeq: hasSeq, updatedAt: text(updatedAt(record)), priority: priority} +} + +func traceCandidateNewer(candidate traceCandidate, current traceCandidate) bool { + if candidate.traceID == "" { + return false + } + if current.traceID == "" { + return true + } + if candidate.hasSeq && current.hasSeq && candidate.seq != current.seq { + return candidate.seq > current.seq + } + if candidate.hasSeq != current.hasSeq { + return candidate.hasSeq + } + if candidate.updatedAt != "" && current.updatedAt != "" && candidate.updatedAt != current.updatedAt { + return candidate.updatedAt > current.updatedAt + } + if candidate.updatedAt != "" && current.updatedAt == "" { + return true + } + if candidate.priority != current.priority { + return candidate.priority > current.priority + } + return false +} + +func nonUnknownStatus(status string) string { + if normalizeStatus(status) == "unknown" { + return "" + } + return status +} + func sessionDisplayTitle(session map[string]any, firstUserMessagePreview any) string { sessionJSON := objectMap(session["sessionJson"]) if title := first(text(session["title"]), text(session["name"]), text(session["displayName"]), text(sessionJSON["title"]), text(sessionJSON["name"]), text(sessionJSON["displayName"])); title != "" { @@ -2357,6 +2456,34 @@ func seq(record map[string]any) any { return nil } +func seqInt64(record map[string]any) (int64, bool) { + for _, key := range []string{"projectedSeq", "sourceSeq", "seq"} { + switch value := record[key].(type) { + case int64: + if value >= 0 { + return value, true + } + case int: + if value >= 0 { + return int64(value), true + } + case float64: + if value >= 0 { + return int64(value), true + } + case json.Number: + if parsed, err := value.Int64(); err == nil && parsed >= 0 { + return parsed, true + } + case string: + if parsed, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64); err == nil && parsed >= 0 { + return parsed, true + } + } + } + return 0, false +} + func timestamp(value any) string { parsed, err := time.Parse(time.RFC3339Nano, text(value)) if err == nil { diff --git a/internal/workbenchruntime/service_test.go b/internal/workbenchruntime/service_test.go index 681972c4..a6fba8f2 100644 --- a/internal/workbenchruntime/service_test.go +++ b/internal/workbenchruntime/service_test.go @@ -18,3 +18,41 @@ func TestWorkbenchRuntimeReadIndexesCoverTraceFanout(t *testing.T) { } } } + +func TestSessionSummaryUsesNewestTurnFactsOverStaleSessionLastTrace(t *testing.T) { + sessionID := "ses_triad_current_trace" + oldTraceID := "trc_triad_completed" + newTraceID := "trc_triad_running" + facts := emptyFacts() + facts["messages"] = []any{ + map[string]any{"messageId": "msg_old_user", "sessionId": sessionID, "traceId": oldTraceID, "role": "user", "status": "sent", "updatedAt": "2026-07-01T00:00:00Z", "projectedSeq": int64(10)}, + map[string]any{"messageId": "msg_old_agent", "sessionId": sessionID, "traceId": oldTraceID, "role": "agent", "status": "completed", "text": "first done", "updatedAt": "2026-07-01T00:00:02Z", "projectedSeq": int64(20)}, + map[string]any{"messageId": "msg_new_user", "sessionId": sessionID, "traceId": newTraceID, "role": "user", "status": "sent", "updatedAt": "2026-07-01T00:00:03Z", "projectedSeq": int64(30)}, + map[string]any{"messageId": "msg_new_agent", "sessionId": sessionID, "traceId": newTraceID, "role": "agent", "status": "running", "updatedAt": "2026-07-01T00:00:04Z", "projectedSeq": int64(40)}, + } + facts["turns"] = []any{ + map[string]any{"turnId": oldTraceID, "sessionId": sessionID, "traceId": oldTraceID, "status": "completed", "terminal": true, "updatedAt": "2026-07-01T00:00:02Z", "projectedSeq": int64(20)}, + map[string]any{"turnId": newTraceID, "sessionId": sessionID, "traceId": newTraceID, "status": "running", "terminal": false, "updatedAt": "2026-07-01T00:00:04Z", "projectedSeq": int64(40)}, + } + summary := sessionSummary(map[string]any{ + "sessionId": sessionID, + "ownerUserId": "usr_triad", + "status": "completed", + "lastTraceId": oldTraceID, + "updatedAt": "2026-07-01T00:00:02Z", + }, facts) + + if got := text(summary["lastTraceId"]); got != newTraceID { + t.Fatalf("lastTraceId=%q, want %q; summary=%#v", got, newTraceID, summary) + } + if got := text(summary["status"]); got != "running" { + t.Fatalf("status=%q, want running; summary=%#v", got, summary) + } + turnSummary := objectMap(summary["turnSummary"]) + if got := text(turnSummary["traceId"]); got != newTraceID { + t.Fatalf("turnSummary.traceId=%q, want %q; turnSummary=%#v", got, newTraceID, turnSummary) + } + if got := text(turnSummary["status"]); got != "running" { + t.Fatalf("turnSummary.status=%q, want running; turnSummary=%#v", got, turnSummary) + } +}