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

fix: project workbench session current turn authority

Browse Source
This commit is contained in:
UniDesk Codex
2026-07-02 08:43:03 +08:00
parent 8b6c2eaa60
commit 683042900f
4 changed files with 302 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workbenchruntime/service.go
+158 -31
View File
@@ -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 {
internal/workbenchruntime/service_test.go
+38
View File
@@ -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)
}
}