This is an automated email from the ASF dual-hosted git repository. hsluoyz pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
commit ce2028bd9d5600af4edda9b603de3a11516c8f87 Author: Yang Luo <[email protected]> AuthorDate: Fri Sep 4 22:16:27 2026 +0800 fix(sessions): read every agent's transcript and key it to its own file --- agenthistory/cursor.go | 159 +++++++++++++++++++++++++++++++++++++++++++++ agenthistory/dsh.go | 58 +++++++++++++++-- agenthistory/history.go | 74 ++++++++++++++++++--- agenthistory/opencode.go | 118 +++++++++++++++++++++++++++++++++ agenthistory/transcript.go | 16 ++++- 5 files changed, 410 insertions(+), 15 deletions(-) diff --git a/agenthistory/cursor.go b/agenthistory/cursor.go index 4062a9b..f3c1149 100644 --- a/agenthistory/cursor.go +++ b/agenthistory/cursor.go @@ -19,6 +19,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "time" _ "modernc.org/sqlite" // read another agent's session database @@ -78,6 +79,7 @@ type cursorComposer struct { LastUpdatedAt int64 `json:"lastUpdatedAt"` Headers []struct { BubbleId string `json:"bubbleId"` + Type int `json:"type"` } `json:"fullConversationHeadersOnly"` } @@ -142,3 +144,160 @@ func cursorTime(value int64) string { } return time.UnixMilli(value).Local().Format(time.RFC3339) } + +// readCursorTranscript reads one chat out of the state database. The chat row +// holds only the order of the messages; each one is a row of its own. +func readCursorTranscript(session Session) (Transcript, error) { + transcript := Transcript{Session: session, Messages: []Message{}} + + db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(session.Path)+"?mode=ro") + if err != nil { + return Transcript{}, err + } + defer db.Close() + + // A row the editor emptied rather than deleted holds a NULL, which is not a + // string and would end the read. + var value sql.NullString + if err := db.QueryRow(`SELECT value FROM cursorDiskKV WHERE key = ?`, + "composerData:"+session.SessionKey).Scan(&value); err != nil { + return Transcript{}, err + } + var composer cursorComposer + if err := json.Unmarshal([]byte(value.String), &composer); err != nil { + return Transcript{}, err + } + + bubbles, err := readCursorBubbles(db, session.SessionKey) + if err != nil { + return Transcript{}, err + } + + for _, header := range composer.Headers { + bubble, found := bubbles[header.BubbleId] + if !found { + continue + } + if len(transcript.Messages) >= maxMessages { + transcript.Truncated = true + break + } + + kind := header.Type + if kind == 0 { + kind = bubble.Type + } + role := "assistant" + if kind == cursorUserBubble { + role = "user" + } + transcript.Messages = append(transcript.Messages, Message{Role: role, Blocks: cursorBlocks(bubble)}) + } + return transcript, nil +} + +// cursorUserBubble is the message kind the editor gives what a person typed. +const cursorUserBubble = 1 + +// cursorBubble is one message: the prose, what the model thought before it, and +// the tool call it made, all optional. +type cursorBubble struct { + Type int `json:"type"` + Text string `json:"text"` + RichText string `json:"richText"` + Thinking struct { + Text string `json:"text"` + } `json:"thinking"` + Tool *struct { + Name string `json:"name"` + Params string `json:"params"` + RawArgs string `json:"rawArgs"` + Result string `json:"result"` + Error string `json:"error"` + Status string `json:"status"` + } `json:"toolFormerData"` +} + +// readCursorBubbles reads every message of one chat at once: they are keyed by +// the chat they belong to, and there is no other way to find them. +func readCursorBubbles(db *sql.DB, composerId string) (map[string]cursorBubble, error) { + rows, err := db.Query(`SELECT key, value FROM cursorDiskKV WHERE key LIKE ?`, "bubbleId:"+composerId+":%") + if err != nil { + return nil, err + } + defer rows.Close() + + bubbles := map[string]cursorBubble{} + for rows.Next() { + var key string + var value sql.NullString + if err := rows.Scan(&key, &value); err != nil { + return nil, err + } + var bubble cursorBubble + if err := json.Unmarshal([]byte(value.String), &bubble); err != nil { + continue + } + bubbles[key[strings.LastIndex(key, ":")+1:]] = bubble + } + return bubbles, nil +} + +func cursorBlocks(bubble cursorBubble) []Block { + blocks := []Block{} + if thinking := strings.TrimSpace(bubble.Thinking.Text); thinking != "" { + blocks = append(blocks, Block{Kind: blockThinking, Text: clip(thinking, maxTextRunes)}) + } + // What a person typed is kept as an editor document, and only the older + // messages carry the same words as plain text. + if text := firstNonEmpty(bubble.Text, cursorRichText(bubble.RichText)); strings.TrimSpace(text) != "" { + blocks = append(blocks, Block{Kind: blockText, Text: clip(text, maxTextRunes)}) + } + if call := bubble.Tool; call != nil { + blocks = append(blocks, Block{ + Kind: blockToolUse, + Tool: call.Name, + Text: clip(firstNonEmpty(call.Params, call.RawArgs), maxToolRunes), + }) + if output := firstNonEmpty(call.Error, call.Result); output != "" { + blocks = append(blocks, Block{ + Kind: blockToolResult, + Text: clip(output, maxToolRunes), + IsError: call.Status == "error", + }) + } + } + return blocks +} + +// cursorRichText reads the words out of the editor document a prompt is stored +// as: a tree of nodes, of which only the text ones are the message. +func cursorRichText(value string) string { + if strings.TrimSpace(value) == "" { + return "" + } + var document any + if err := json.Unmarshal([]byte(value), &document); err != nil { + return "" + } + + pieces := []string{} + var walk func(node any) + walk = func(node any) { + switch value := node.(type) { + case map[string]any: + if text, ok := value["text"].(string); ok && text != "" { + pieces = append(pieces, text) + } + if children, ok := value["content"]; ok { + walk(children) + } + case []any: + for _, child := range value { + walk(child) + } + } + } + walk(document) + return strings.Join(pieces, "\n") +} diff --git a/agenthistory/dsh.go b/agenthistory/dsh.go index 54f3ce6..a887c4a 100644 --- a/agenthistory/dsh.go +++ b/agenthistory/dsh.go @@ -151,16 +151,22 @@ func openDshTranscript(path string) (io.Reader, func(), error) { }, nil } -// dshContent is the model-facing blocks of one message, which is what the event -// data is for a user turn. +// dshContent is the model-facing blocks of one message. A turn spells them on +// the event itself, or under the message the event carries. func dshContent(data json.RawMessage) json.RawMessage { - var message struct { + var event struct { Content json.RawMessage `json:"content"` + Message struct { + Content json.RawMessage `json:"content"` + } `json:"message"` } - if err := json.Unmarshal(data, &message); err != nil { + if err := json.Unmarshal(data, &event); err != nil { return nil } - return message.Content + if len(event.Content) > 0 { + return event.Content + } + return event.Message.Content } // addDshUsage reads what one model call spent. The counts ride the assistant @@ -229,3 +235,45 @@ func dshHeaderTime(value json.RawMessage) string { } return "" } + +// readDshTranscript reads the conversation out of a session log. The scan reads +// the same lines to count them, and only the two message events carry a turn. +func readDshTranscript(session Session) (Transcript, error) { + transcript := Transcript{Session: session, Messages: []Message{}} + + source, closer, err := openDshTranscript(session.Path) + if err != nil { + return Transcript{}, err + } + defer closer() + + skipped, _ := eachLineIn(source, func(data []byte) { + var entry dshLine + if err := json.Unmarshal(data, &entry); err != nil { + return + } + role := "" + switch entry.Type { + case "user/message": + role = "user" + case "assistant/message": + role = "assistant" + default: + return + } + if len(transcript.Messages) >= maxMessages { + transcript.Truncated = true + return + } + + transcript.Messages = append(transcript.Messages, Message{ + Role: role, + Time: dshEventTime(entry.Time), + Blocks: blocks(dshContent(entry.Data)), + }) + }) + if skipped > 0 { + transcript.Truncated = true + } + return transcript, nil +} diff --git a/agenthistory/history.go b/agenthistory/history.go index c46b432..80157a0 100644 --- a/agenthistory/history.go +++ b/agenthistory/history.go @@ -64,6 +64,15 @@ const ( maxTitleRunes = 120 ) +// claudeSubagentDir holds the transcripts of the subagents one session ran. +const claudeSubagentDir = "subagents" + +// The two Codex front ends, which write their sessions into the same directory. +const ( + codexCliAgent = "codex-cli" + codexDesktopAgent = "codex" +) + // transcriptDirs are where each agent keeps its own transcripts, relative to a // home directory. Both write one JSONL file per session. var transcriptDirs = []struct { @@ -74,7 +83,7 @@ var transcriptDirs = []struct { keyOf func(path string) string }{ {agent: "claude-code", parts: []string{".claude", "projects"}}, - {agent: "codex-cli", parts: []string{".codex", "sessions"}}, + {agent: codexCliAgent, parts: []string{".codex", "sessions"}}, // The Gemini CLI and its Qwen Code fork keep one directory per project hash, // and OpenClaw one per agent id; all hold the sessions a level or two // further down, which is where the walk finds them. @@ -250,6 +259,10 @@ type line struct { // counts of the turn that just ended. Model string `json:"model"` Info json.RawMessage `json:"info"` + // Originator is the Codex front end that opened the session. They all + // write into one sessions directory, so it is the only thing that says + // which one a transcript belongs to. + Originator string `json:"originator"` } `json:"payload"` } @@ -314,6 +327,12 @@ func parse(agent string, file transcript, keyOf func(string) string) (Session, b LastTime: file.info.ModTime().Local().Format(time.RFC3339), Historical: true, } + // A Claude Code subagent writes a transcript of its own under the session + // that spawned it, and stamps every line with that session's id. Taken from + // the lines they would all be that one session, and opening it would open + // whichever of the files was read first. + subagent := filepath.Base(filepath.Dir(file.path)) == claudeSubagentDir + // The file name is the session id for Claude Code, and carries it for Codex; // the transcript itself overrides this when it names one. session.SessionKey = sessionKeyFromName(file.info.Name()) @@ -331,11 +350,26 @@ func parse(agent string, file transcript, keyOf func(string) string) (Session, b return } - if entry.SessionId != "" { + if entry.SessionId != "" && !subagent { session.SessionKey = entry.SessionId } - if entry.Type == "session_meta" && entry.Payload.Id != "" { - session.SessionKey = entry.Payload.Id + if entry.Type == "session_meta" { + if entry.Payload.Id != "" { + // Resuming a thread writes a new file, named after the thread and + // then after the continuation, while the header still names the + // thread alone. Keying on the header would list every file of a + // resumed thread as the same session, and open the first of them + // for all of them. + session.SessionKey = firstNonEmpty(trailingUuid(strings.TrimSuffix(file.info.Name(), ".jsonl")), entry.Payload.Id) + } + // Codex Desktop and the CLI share ~/.codex/sessions, and monitoring + // reports them apart. A session listed under the front end that did + // not write it is a second row for one conversation. + if agent == codexCliAgent { + if front := codexAgentForOriginator(entry.Payload.Originator); front != "" { + session.Agent = front + } + } } if session.Cwd == "" { session.Cwd = firstNonEmpty(entry.Cwd, entry.Payload.Cwd) @@ -372,6 +406,18 @@ func parse(agent string, file transcript, keyOf func(string) string) (Session, b return session, true } +// codexAgentForOriginator is the agent id one Codex front end reports under, +// spelled the way agentmonitor spells it for the same rollout files. +func codexAgentForOriginator(originator string) string { + switch strings.NewReplacer("_", "-", " ", "-").Replace(strings.ToLower(strings.TrimSpace(originator))) { + case "codex-desktop": + return codexDesktopAgent + case "codex-tui", "codex-exec", "codex-vscode": + return codexCliAgent + } + return "" +} + // roleAndContent picks the messages out of a transcript, in either format. Only // the conversation counts: the tool calls and events around it are what // monitoring is for. @@ -489,15 +535,25 @@ func trimTitle(text string) string { // "<uuid>.jsonl" for Claude Code, "rollout-<date>-<uuid>.jsonl" for Codex. func sessionKeyFromName(name string) string { base := strings.TrimSuffix(name, ".jsonl") - const uuidLength = 36 - if len(base) >= uuidLength { - if candidate := base[len(base)-uuidLength:]; strings.Count(candidate, "-") == 4 { - return candidate - } + if id := trailingUuid(base); id != "" { + return id } return base } +// trailingUuid is the id a transcript's name ends in, or "" when it ends in +// something else. +func trailingUuid(base string) string { + const uuidLength = 36 + if len(base) < uuidLength { + return "" + } + if candidate := base[len(base)-uuidLength:]; strings.Count(candidate, "-") == 4 { + return candidate + } + return "" +} + // parentDirName is the name of the directory a transcript sits in, which is the // session id for an agent that writes every session's file under a fixed name. func parentDirName(path string) string { diff --git a/agenthistory/opencode.go b/agenthistory/opencode.go index 478e032..30124f4 100644 --- a/agenthistory/opencode.go +++ b/agenthistory/opencode.go @@ -193,3 +193,121 @@ func opencodeTime(value int64) string { } return time.UnixMilli(value).Local().Format(time.RFC3339) } + +// readOpencodeTranscript reads one session out of the database. A message is a +// row of its own, and what it holds is the parts filed under it. +func readOpencodeTranscript(session Session) (Transcript, error) { + transcript := Transcript{Session: session, Messages: []Message{}} + + db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(session.Path)+"?mode=ro") + if err != nil { + return Transcript{}, err + } + defer db.Close() + + parts, err := readOpencodeParts(db, session.SessionKey) + if err != nil { + return Transcript{}, err + } + + rows, err := db.Query(`SELECT id, time_created, data FROM message + WHERE session_id = ? ORDER BY time_created, id LIMIT ?`, session.SessionKey, maxMessages+1) + if err != nil { + return Transcript{}, err + } + defer rows.Close() + + for rows.Next() { + var id, data string + var created int64 + if err := rows.Scan(&id, &created, &data); err != nil { + return Transcript{}, err + } + if len(transcript.Messages) >= maxMessages { + transcript.Truncated = true + break + } + + var message struct { + Role string `json:"role"` + } + json.Unmarshal([]byte(data), &message) + transcript.Messages = append(transcript.Messages, Message{ + Role: message.Role, + Time: opencodeTime(created), + Blocks: parts[id], + }) + } + return transcript, nil +} + +// opencodePart is one part of a message: the prose, the model thinking aloud, +// or a tool call with what it returned. +type opencodePart struct { + Type string `json:"type"` + Text string `json:"text"` + Tool string `json:"tool"` + State struct { + Status string `json:"status"` + Input json.RawMessage `json:"input"` + Output string `json:"output"` + Error string `json:"error"` + } `json:"state"` + Filename string `json:"filename"` + Mime string `json:"mime"` +} + +// readOpencodeParts reads every part of a session at once, keyed by the message +// it belongs to: one query rather than one per message. +func readOpencodeParts(db *sql.DB, sessionKey string) (map[string][]Block, error) { + rows, err := db.Query(`SELECT message_id, data FROM part + WHERE session_id = ? ORDER BY time_created, id`, sessionKey) + if err != nil { + return nil, err + } + defer rows.Close() + + blocks := map[string][]Block{} + for rows.Next() { + var messageId, data string + if err := rows.Scan(&messageId, &data); err != nil { + return nil, err + } + var part opencodePart + if err := json.Unmarshal([]byte(data), &part); err != nil { + continue + } + blocks[messageId] = append(blocks[messageId], opencodeBlocks(part)...) + } + return blocks, nil +} + +func opencodeBlocks(part opencodePart) []Block { + switch part.Type { + case "text": + if strings.TrimSpace(part.Text) == "" { + return nil + } + return []Block{{Kind: blockText, Text: clip(part.Text, maxTextRunes)}} + case "reasoning": + return []Block{{Kind: blockThinking, Text: clip(part.Text, maxTextRunes)}} + case "file": + if strings.HasPrefix(part.Mime, "image/") { + return []Block{{Kind: blockImage}} + } + return []Block{{Kind: blockText, Text: part.Filename}} + case "tool": + blocks := []Block{{Kind: blockToolUse, Tool: part.Tool, Text: clip(string(part.State.Input), maxToolRunes)}} + // A call still running has neither, and stands as the call alone. + if output := firstNonEmpty(part.State.Error, part.State.Output); output != "" { + blocks = append(blocks, Block{ + Kind: blockToolResult, + Text: clip(output, maxToolRunes), + IsError: part.State.Status == "error", + }) + } + return blocks + } + // step-start, step-finish and the snapshots around them are bookkeeping. + return nil +} diff --git a/agenthistory/transcript.go b/agenthistory/transcript.go index 8ab1e96..146a3b8 100644 --- a/agenthistory/transcript.go +++ b/agenthistory/transcript.go @@ -67,8 +67,22 @@ const ( // pointed at an arbitrary file. // // The messages are the same ones Scan counts, so the number the Sessions page -// shows is the number of messages this returns. +// shows is the number of messages this returns. An agent that keeps its +// sessions in something other than a plain JSONL file is read by the same +// reader the scan used, rather than by the one below. func ReadTranscript(session Session) (Transcript, error) { + switch session.Agent { + case opencodeAgent: + return readOpencodeTranscript(session) + case cursorAgent: + return readCursorTranscript(session) + case dshAgent: + return readDshTranscript(session) + } + return readJsonlTranscript(session) +} + +func readJsonlTranscript(session Session) (Transcript, error) { transcript := Transcript{Session: session, Messages: []Message{}} // A line eachLine passed over is a message this cannot show, so it counts --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
