Merge pull request #1917 from rainyMorning2/fix/codex-large-history

fix(codex): read large session history records reliably
This commit is contained in:
cg33 authored and GitHub committed 2026-10-16 16:57:25 +08:00
commit 2dec7173bd
5 files changed
+278 -34

No files matched your search

+70 -30
View File
@@ -5,6 +5,7 @@ import (
"bytes" "bytes"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io"
"log/slog" "log/slog"
"os" "os"
"path/filepath" "path/filepath"
@@ -15,6 +16,12 @@ import (
"github.com/chenhg5/cc-connect/core" "github.com/chenhg5/cc-connect/core"
) )
// Keep large Codex transcript records readable while bounding scanner memory.
// Records beyond this limit fail explicitly instead of returning partial history.
// ListSessions scans files serially; a future parallel scan must bound its
// concurrency because each scanner can allocate up to this limit.
const maxCodexJSONLLineSize = 64 * 1024 * 1024
// resolveCodexHomeDir returns the effective CODEX_HOME directory. // resolveCodexHomeDir returns the effective CODEX_HOME directory.
// Priority: explicit config value > CODEX_HOME env > ~/.codex // Priority: explicit config value > CODEX_HOME env > ~/.codex
func resolveCodexHomeDir(explicit string) string { func resolveCodexHomeDir(explicit string) string {
@@ -112,7 +119,8 @@ func loadCodexSessionTitles(codexHome string) map[string]string {
} }
// parseCodexSessionFile reads a Codex JSONL transcript. // parseCodexSessionFile reads a Codex JSONL transcript.
// Returns nil if the session's cwd doesn't match filterCwd. // ListSessions skips a damaged file and keeps scanning other sessions; errors
// are logged with the path for diagnosis. Returns nil if the cwd does not match.
func parseCodexSessionFile(path, filterCwd string) *core.AgentSessionInfo { func parseCodexSessionFile(path, filterCwd string) *core.AgentSessionInfo {
f, err := os.Open(path) f, err := os.Open(path)
if err != nil { if err != nil {
@@ -133,11 +141,11 @@ func parseCodexSessionFile(path, filterCwd string) *core.AgentSessionInfo {
userMsgSeen := 0 userMsgSeen := 0
scanner := bufio.NewScanner(f) scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 256*1024), 256*1024) scanner.Buffer(make([]byte, 256*1024), maxCodexJSONLLineSize)
for scanner.Scan() { for scanner.Scan() {
line := scanner.Text() line := scanner.Text()
if line == "" { if strings.TrimSpace(line) == "" {
continue continue
} }
@@ -146,7 +154,8 @@ func parseCodexSessionFile(path, filterCwd string) *core.AgentSessionInfo {
Payload json.RawMessage `json:"payload"` Payload json.RawMessage `json:"payload"`
} }
if err := json.Unmarshal([]byte(line), &entry); err != nil { if err := json.Unmarshal([]byte(line), &entry); err != nil {
continue slog.Warn("codex: decode session file", "path", path, "error", err)
return nil
} }
switch entry.Type { switch entry.Type {
@@ -159,11 +168,13 @@ func parseCodexSessionFile(path, filterCwd string) *core.AgentSessionInfo {
Cwd string `json:"cwd"` Cwd string `json:"cwd"`
Source json.RawMessage `json:"source"` Source json.RawMessage `json:"source"`
} }
if json.Unmarshal(entry.Payload, &meta) == nil { if err := json.Unmarshal(entry.Payload, &meta); err != nil {
sessionID = meta.ID slog.Warn("codex: decode session metadata", "path", path, "error", err)
sessionCwd = meta.Cwd return nil
sessionSource = meta.Source
} }
sessionID = meta.ID
sessionCwd = meta.Cwd
sessionSource = meta.Source
case "response_item": case "response_item":
var item struct { var item struct {
@@ -173,24 +184,31 @@ func parseCodexSessionFile(path, filterCwd string) *core.AgentSessionInfo {
Text string `json:"text"` Text string `json:"text"`
} `json:"content"` } `json:"content"`
} }
if json.Unmarshal(entry.Payload, &item) == nil { if err := json.Unmarshal(entry.Payload, &item); err != nil {
if item.Role == "user" { slog.Warn("codex: decode session response item", "path", path, "error", err)
userMsgSeen++ return nil
msgCount++ }
// The actual user prompt is the last user response_item switch item.Role {
// (earlier ones are system/AGENTS.md instructions). case "user":
// Pick the last content block that looks like a real prompt. userMsgSeen++
for _, c := range item.Content { msgCount++
if c.Type == "input_text" && c.Text != "" && isUserPrompt(c.Text) { // The actual user prompt is the last user response_item
summary = c.Text // (earlier ones are system/AGENTS.md instructions).
} // Pick the last content block that looks like a real prompt.
for _, c := range item.Content {
if c.Type == "input_text" && c.Text != "" && isUserPrompt(c.Text) {
summary = c.Text
} }
} else if item.Role == "assistant" {
msgCount++
} }
case "assistant":
msgCount++
} }
} }
} }
if err := scanner.Err(); err != nil {
slog.Warn("codex: scan session file", "path", path, "error", err)
return nil
}
// Filter by cwd // Filter by cwd
if filterCwd != "" && sessionCwd != "" && sessionCwd != filterCwd { if filterCwd != "" && sessionCwd != "" && sessionCwd != filterCwd {
@@ -244,7 +262,20 @@ func findSessionFile(sessionID, codexHome string) string {
return found return found
} }
func parseCodexTimestamp(value string) time.Time {
if value == "" {
return time.Time{}
}
ts, err := time.Parse(time.RFC3339Nano, value)
if err != nil {
return time.Time{}
}
return ts.In(time.Local)
}
// getSessionHistory reads the JSONL transcript and returns user/assistant messages. // getSessionHistory reads the JSONL transcript and returns user/assistant messages.
// Unlike ListSessions, it returns read errors so /history can report failure
// without presenting incomplete history as a complete conversation.
func getSessionHistory(sessionID, codexHome string, limit int) ([]core.HistoryEntry, error) { func getSessionHistory(sessionID, codexHome string, limit int) ([]core.HistoryEntry, error) {
path := findSessionFile(sessionID, codexHome) path := findSessionFile(sessionID, codexHome)
if path == "" { if path == "" {
@@ -256,15 +287,22 @@ func getSessionHistory(sessionID, codexHome string, limit int) ([]core.HistoryEn
return nil, err return nil, err
} }
defer f.Close() defer f.Close()
entries, err := readCodexSessionHistory(f, limit)
if err != nil {
return nil, fmt.Errorf("codex: scan session history %s: %w", path, err)
}
return entries, nil
}
func readCodexSessionHistory(r io.Reader, limit int) ([]core.HistoryEntry, error) {
var entries []core.HistoryEntry var entries []core.HistoryEntry
scanner := bufio.NewScanner(f) scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 256*1024), 256*1024) scanner.Buffer(make([]byte, 256*1024), maxCodexJSONLLineSize)
for scanner.Scan() { for scanner.Scan() {
line := scanner.Text() line := scanner.Text()
if line == "" { if strings.TrimSpace(line) == "" {
continue continue
} }
@@ -273,8 +311,8 @@ func getSessionHistory(sessionID, codexHome string, limit int) ([]core.HistoryEn
Type string `json:"type"` Type string `json:"type"`
Payload json.RawMessage `json:"payload"` Payload json.RawMessage `json:"payload"`
} }
if json.Unmarshal([]byte(line), &raw) != nil { if err := json.Unmarshal([]byte(line), &raw); err != nil {
continue return nil, fmt.Errorf("decode session record: %w", err)
} }
if raw.Type != "response_item" { if raw.Type != "response_item" {
continue continue
@@ -289,11 +327,11 @@ func getSessionHistory(sessionID, codexHome string, limit int) ([]core.HistoryEn
Text string `json:"text"` Text string `json:"text"`
} `json:"content"` } `json:"content"`
} }
if json.Unmarshal(raw.Payload, &item) != nil { if err := json.Unmarshal(raw.Payload, &item); err != nil {
continue return nil, fmt.Errorf("decode response item: %w", err)
} }
ts, _ := time.Parse(time.RFC3339Nano, raw.Timestamp) ts := parseCodexTimestamp(raw.Timestamp)
switch { switch {
case item.Role == "user" && len(item.Content) > 0: case item.Role == "user" && len(item.Content) > 0:
@@ -316,7 +354,9 @@ func getSessionHistory(sessionID, codexHome string, limit int) ([]core.HistoryEn
// skip reasoning items // skip reasoning items
} }
} }
if err := scanner.Err(); err != nil {
return nil, err
}
if limit > 0 && len(entries) > limit { if limit > 0 && len(entries) > limit {
entries = entries[len(entries)-limit:] entries = entries[len(entries)-limit:]
} }
+153
View File
@@ -3,10 +3,13 @@ package codex
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"io"
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
"testing" "testing"
"time"
) )
func TestAgentListSessions_ExcludesSubagentRollouts(t *testing.T) { func TestAgentListSessions_ExcludesSubagentRollouts(t *testing.T) {
@@ -163,3 +166,153 @@ func TestAgentListSessions_LongThreadNameTruncated(t *testing.T) {
t.Fatalf("ListSessions()[0].Summary = %q, want %q", sessions[0].Summary, want) t.Fatalf("ListSessions()[0].Summary = %q, want %q", sessions[0].Summary, want)
} }
} }
// Regression: Codex JSONL records can exceed the old 256 KiB scanner limit.
// Both /list and /history must read through such a record, and /history must
// still return the last N eligible messages in chronological order.
func TestGetSessionHistory_LargeJSONLLineAndLastEntries(t *testing.T) {
codexHome := t.TempDir()
sessionID := "session-large-line"
sessionsDir := filepath.Join(codexHome, "sessions")
if err := os.MkdirAll(sessionsDir, 0o755); err != nil {
t.Fatal(err)
}
largePrompt := strings.Repeat("x", 300*1024)
data := `{"type":"session_meta","payload":{"id":"` + sessionID + `","cwd":"/project"}}` + "\n" +
`{"type":"response_item","payload":{"role":"user","content":[{"type":"input_text","text":"first"}]}}` + "\n" +
`{"type":"response_item","payload":{"role":"assistant","content":[{"type":"output_text","text":"first reply"}]}}` + "\n" +
`{"type":"response_item","payload":{"role":"user","content":[{"type":"input_text","text":"` + largePrompt + `"}]}}` + "\n" +
`{"type":"response_item","payload":{"role":"assistant","content":[{"type":"output_text","text":"last reply"}]}}` + "\n"
path := filepath.Join(sessionsDir, "rollout-"+sessionID+".jsonl")
if err := os.WriteFile(path, []byte(data), 0o644); err != nil {
t.Fatal(err)
}
info := parseCodexSessionFile(path, "/project")
if info == nil || info.MessageCount != 4 {
t.Fatalf("session info = %+v, want 4 messages", info)
}
entries, err := getSessionHistory(sessionID, codexHome, 2)
if err != nil {
t.Fatal(err)
}
if len(entries) != 2 || entries[0].Content != largePrompt || entries[1].Content != "last reply" {
t.Fatalf("last 2 entries = %d entries; want large prompt and last reply", len(entries))
}
}
type errorReader struct{ err error }
func (r errorReader) Read([]byte) (int, error) { return 0, r.err }
func TestReadCodexSessionHistory_ReportsScanError(t *testing.T) {
want := errors.New("injected read failure")
line := `{"type":"response_item","payload":{"role":"user","content":[{"type":"input_text","text":"hello"}]}}` + "\n"
reader := io.MultiReader(strings.NewReader(line), errorReader{err: want})
entries, err := readCodexSessionHistory(reader, 10)
if !errors.Is(err, want) || entries != nil {
t.Fatalf("entries = %+v, error = %v; want nil entries and injected error", entries, err)
}
}
func TestReadCodexSessionHistory_RejectsCorruptJSONAfterValidPrefix(t *testing.T) {
valid := `{"type":"response_item","payload":{"role":"user","content":[{"type":"input_text","text":"hello"}]}}`
for _, bad := range []string{
`{"type":"response_item",`,
`{"type":"response_item","payload":{"role":`,
} {
entries, err := readCodexSessionHistory(strings.NewReader(valid+"\n"+bad+"\n"), 10)
if err == nil || entries != nil {
t.Fatalf("corrupt record %q: entries = %+v, error = %v; want nil entries and error", bad, entries, err)
}
}
entries, err := readCodexSessionHistory(strings.NewReader(valid+"\n\n \n"+`{"type":"unknown_event","payload":{}}`+"\n"), 10)
if err != nil || len(entries) != 1 {
t.Fatalf("blank and unknown records: entries = %+v, error = %v; want one entry", entries, err)
}
}
func TestAgentListSessions_BadFileDoesNotHideHealthySession(t *testing.T) {
workDir := t.TempDir()
codexHome := t.TempDir()
dir := filepath.Join(codexHome, "sessions")
if err := os.MkdirAll(dir, 0o755); err != nil {
t.Fatal(err)
}
cwd, err := json.Marshal(workDir)
if err != nil {
t.Fatal(err)
}
healthy := `{"type":"session_meta","payload":{"id":"healthy","cwd":` + string(cwd) + `}}` + "\n" +
`{"type":"response_item","payload":{"role":"user","content":[{"type":"input_text","text":"hello"}]}}` + "\n"
if err := os.WriteFile(filepath.Join(dir, "rollout-healthy.jsonl"), []byte(healthy), 0o644); err != nil {
t.Fatal(err)
}
bad := `{"type":"session_meta","payload":{"id":"bad","cwd":` + string(cwd) + `}}` + "\n" + `{"type":"response_item","payload":`
if err := os.WriteFile(filepath.Join(dir, "rollout-bad.jsonl"), []byte(bad), 0o644); err != nil {
t.Fatal(err)
}
a := &Agent{workDir: workDir, codexHome: codexHome}
sessions, err := a.ListSessions(context.Background())
if err != nil || len(sessions) != 1 || sessions[0].ID != "healthy" {
t.Fatalf("ListSessions() = %+v, %v; want only healthy session", sessions, err)
}
}
func TestGetSessionHistory_ConvertsTimestampToLocal(t *testing.T) {
codexHome := t.TempDir()
sessionID := "session-local-time"
sessionsDir := filepath.Join(codexHome, "sessions")
if err := os.MkdirAll(sessionsDir, 0o755); err != nil {
t.Fatalf("mkdir sessions: %v", err)
}
const timestamp = "2026-09-24T12:34:56.123456789Z"
data := `{"type":"response_item","timestamp":"` + timestamp + `","payload":{"role":"user","content":[{"type":"input_text","text":"hello"}]}}` + "\n"
path := filepath.Join(sessionsDir, "rollout-"+sessionID+".jsonl")
if err := os.WriteFile(path, []byte(data), 0o644); err != nil {
t.Fatalf("write session history: %v", err)
}
entries, err := getSessionHistory(sessionID, codexHome, 0)
if err != nil {
t.Fatalf("getSessionHistory: %v", err)
}
if len(entries) != 1 {
t.Fatalf("entries = %d, want 1", len(entries))
}
want, err := time.Parse(time.RFC3339Nano, timestamp)
if err != nil {
t.Fatalf("parse expected timestamp: %v", err)
}
got := entries[0].Timestamp
if !got.Equal(want) {
t.Fatalf("timestamp instant = %v, want %v", got, want)
}
if got.Location() != time.Local {
t.Fatalf("timestamp location = %v, want time.Local (%v)", got.Location(), time.Local)
}
}
func TestParseCodexTimestamp_NonUTCOffsetConvertsToLocal(t *testing.T) {
const timestamp = "2026-09-24T20:34:56.123456789+08:00"
want, err := time.Parse(time.RFC3339Nano, timestamp)
if err != nil {
t.Fatal(err)
}
got := parseCodexTimestamp(timestamp)
if !got.Equal(want) || got.Location() != time.Local {
t.Fatalf("parseCodexTimestamp(%q) = %v (%v), want instant %v in time.Local", timestamp, got, got.Location(), want)
}
}
func TestParseCodexTimestamp_PreservesZeroForInvalidInput(t *testing.T) {
if got := parseCodexTimestamp(""); !got.IsZero() {
t.Fatalf("empty timestamp = %v, want zero", got)
}
if got := parseCodexTimestamp("not-a-timestamp"); !got.IsZero() {
t.Fatalf("invalid timestamp = %v, want zero", got)
}
}
+11 -4
View File
@@ -9688,9 +9688,13 @@ func (e *Engine) cmdHistory(p Platform, msg *Message, args []string) {
agentSID := s.GetAgentSessionID() agentSID := s.GetAgentSessionID()
if len(entries) == 0 && agentSID != "" { if len(entries) == 0 && agentSID != "" {
if hp, ok := agent.(HistoryProvider); ok { if hp, ok := agent.(HistoryProvider); ok {
if agentEntries, err := hp.GetSessionHistory(e.ctx, agentSID, n); err == nil { agentEntries, err := hp.GetSessionHistory(e.ctx, agentSID, n)
entries = agentEntries if err != nil {
slog.Warn("history: agent transcript read failed", "session_id", agentSID, "error", err)
e.reply(p, msg.ReplyCtx, e.i18n.T(MsgHistoryReadFailed))
return
} }
entries = agentEntries
} }
} }
@@ -13861,9 +13865,12 @@ func (e *Engine) renderHistoryCard(sessionKey string) *Card {
agentSID := s.GetAgentSessionID() agentSID := s.GetAgentSessionID()
if len(entries) == 0 && agentSID != "" { if len(entries) == 0 && agentSID != "" {
if hp, ok := agent.(HistoryProvider); ok { if hp, ok := agent.(HistoryProvider); ok {
if agentEntries, err := hp.GetSessionHistory(e.ctx, agentSID, 10); err == nil { agentEntries, err := hp.GetSessionHistory(e.ctx, agentSID, 10)
entries = agentEntries if err != nil {
slog.Warn("history: agent transcript read failed", "session_id", agentSID, "error", err)
return e.simpleCard(e.i18n.T(MsgCardTitleHistory), "turquoise", e.i18n.T(MsgHistoryReadFailed))
} }
entries = agentEntries
} }
} }
+36
View File
@@ -20,6 +20,15 @@ import (
type stubAgent struct{} type stubAgent struct{}
type failingHistoryAgent struct {
stubAgent
err error
}
func (a *failingHistoryAgent) GetSessionHistory(context.Context, string, int) ([]HistoryEntry, error) {
return []HistoryEntry{{Role: "user", Content: "partial history"}}, a.err
}
func (a *stubAgent) Name() string { return "stub" } func (a *stubAgent) Name() string { return "stub" }
func (a *stubAgent) StartSession(_ context.Context, _ string) (AgentSession, error) { func (a *stubAgent) StartSession(_ context.Context, _ string) (AgentSession, error) {
return &stubAgentSession{}, nil return &stubAgentSession{}, nil
@@ -3830,6 +3839,33 @@ func TestCmdCurrent_UsesLegacyTextOnPlatformWithoutCardSupport(t *testing.T) {
} }
} }
func TestCmdHistory_ReadFailureIsVisibleWithoutLeakingPath(t *testing.T) {
const secretPath = "/private/codex-home/sessions/secret.jsonl"
agent := &failingHistoryAgent{err: fmt.Errorf("scan %s: token too long", secretPath)}
msg := &Message{SessionKey: "test:history-failure", ReplyCtx: "ctx"}
plain := &stubPlatformEngine{n: "plain"}
textEngine := NewEngine("test", agent, []Platform{plain}, "", LangEnglish)
textEngine.sessions.GetOrCreateActive(msg.SessionKey).SetAgentSessionID("broken", "test")
textEngine.cmdHistory(plain, msg, []string{"10"})
sent := plain.getSent()
if len(sent) != 1 || sent[0] != textEngine.i18n.T(MsgHistoryReadFailed) || strings.Contains(sent[0], secretPath) {
t.Fatalf("text history response = %q; want generic localized failure without path", sent)
}
cardPlatform := &stubCardPlatform{stubPlatformEngine: stubPlatformEngine{n: "cards"}}
cardEngine := NewEngine("test", agent, []Platform{cardPlatform}, "", LangEnglish)
cardEngine.sessions.GetOrCreateActive(msg.SessionKey).SetAgentSessionID("broken", "test")
cardEngine.cmdHistory(cardPlatform, msg, nil)
if len(cardPlatform.repliedCards) != 1 {
t.Fatalf("card replies = %d, want 1", len(cardPlatform.repliedCards))
}
cardText := cardPlatform.repliedCards[0].RenderText()
if !strings.Contains(cardText, cardEngine.i18n.T(MsgHistoryReadFailed)) || strings.Contains(cardText, secretPath) || strings.Contains(cardText, "partial history") {
t.Fatalf("card history response = %q; want generic localized failure without path or partial history", cardText)
}
}
func TestCmdCurrent_ShowsAgentSummaryWhenNoCustomName(t *testing.T) { func TestCmdCurrent_ShowsAgentSummaryWhenNoCustomName(t *testing.T) {
p := &stubPlatformEngine{n: "plain"} p := &stubPlatformEngine{n: "plain"}
agent := &stubListAgent{sessions: []AgentSessionInfo{ agent := &stubListAgent{sessions: []AgentSessionInfo{
+8
View File
@@ -255,6 +255,7 @@ const (
MsgListSwitchHint MsgKey = "list_switch_hint" MsgListSwitchHint MsgKey = "list_switch_hint"
MsgListError MsgKey = "list_error" MsgListError MsgKey = "list_error"
MsgHistoryEmpty MsgKey = "history_empty" MsgHistoryEmpty MsgKey = "history_empty"
MsgHistoryReadFailed MsgKey = "history_read_failed"
MsgNameUsage MsgKey = "name_usage" MsgNameUsage MsgKey = "name_usage"
MsgNameSet MsgKey = "name_set" MsgNameSet MsgKey = "name_set"
MsgNameNoSession MsgKey = "name_no_session" MsgNameNoSession MsgKey = "name_no_session"
@@ -1510,6 +1511,13 @@ var messages = map[MsgKey]map[Language]string{
LangJapanese: "現在のセッションに履歴がありません。", LangJapanese: "現在のセッションに履歴がありません。",
LangSpanish: "No hay historial en la sesión actual.", LangSpanish: "No hay historial en la sesión actual.",
}, },
MsgHistoryReadFailed: {
LangEnglish: "Failed to read session history.",
LangChinese: "读取会话历史失败。",
LangTraditionalChinese: "讀取會話歷史失敗。",
LangJapanese: "セッション履歴の読み取りに失敗しました。",
LangSpanish: "No se pudo leer el historial de la sesión.",
},
MsgNameUsage: { MsgNameUsage: {
LangEnglish: "Usage:\n`/name <text>` — name the current session\n`/name <number> <text>` — name a session by list number", LangEnglish: "Usage:\n`/name <text>` — name the current session\n`/name <number> <text>` — name a session by list number",
LangChinese: "用法:\n`/name <名称>` — 命名当前会话\n`/name <序号> <名称>` — 按列表序号命名会话", LangChinese: "用法:\n`/name <名称>` — 命名当前会话\n`/name <序号> <名称>` — 按列表序号命名会话",