package parser import ( "bufio" "encoding/json" "errors" "fmt" "io" "os" "path/filepath" "regexp" "slices" "sort" "strconv" "strings" "sync" "time" "github.com/tidwall/gjson" ) // Codex JSONL entry types. const ( codexTypeSessionMeta = "session_meta" codexTypeResponseItem = "response_item" codexTypeTurnContext = "turn_context" codexTypeEventMsg = "event_msg" codexOriginatorExec = "codex_exec" ) var errCodexIncrementalNeedsFullParse = errors.New( "codex incremental event requires full parse", ) const codexGoalContextSourceAttr = `source="goal"` var codexGoalContextSourceAttrRe = regexp.MustCompile(`(?:^|\s)` + regexp.QuoteMeta(codexGoalContextSourceAttr) + `(?:\s|/|$)`) var codexSessionIndexCache = struct { mu sync.Mutex entries map[string]codexSessionIndexEntry }{ entries: make(map[string]codexSessionIndexEntry), } type codexSessionIndexEntry struct { mtime int64 size int64 changeTime int64 changeTimeOkay bool titles map[string]string } // codexSessionBuilder accumulates state while scanning a Codex // JSONL session file line by line. type codexSessionBuilder struct { codexCursorState messages []ParsedMessage firstMessage string startedAt time.Time endedAt time.Time sessionID string project string ordinal int callNames map[string]string callRefs map[string]codexToolCallRef agentSpawnCalls map[string]string agentWaitCalls map[string]string pendingAgentEvents map[string][]codexPendingEvent orphanNotificationIx map[string]int unattachedTokenUsage bool } // codexForkGate drops the replayed parent history at the top of a // forked Codex rollout (#643). // // `codex fork` copies the parent's lines — its session_meta, turns, // messages and token_count events — into the new file with re-stamped // envelope timestamps, so the same usage exists in two session files // and gets counted twice. Envelope timestamps cannot locate the // boundary (the replay is re-stamped at fork creation), but turn ids // are UUIDv7 values minted when the turn originally ran: every // replayed turn predates the fork instant, and the first genuine turn // is minted at or after it. The gate stays closed until the first // turn_context whose turn_id timestamp is >= the fork's own creation // time, then everything flows normally. // // Replayed turn_context entries from parents recorded before Codex // stamped turn ids carry no turn_id at all; a CLI new enough to write // forked_from_id always stamps genuine turns, so a missing turn_id // while gated means replayed history. An unparseable turn_id fails // open (pre-#643 behaviour) rather than risk dropping live data. type codexForkGate struct { active bool createdMs int64 } // armFromMeta activates the gate when the session_meta belongs to a // forked session and its creation instant can be anchored: from the // fork's UUIDv7 id, the payload timestamp, or the JSONL envelope // timestamp, in that order. func (g *codexForkGate) armFromMeta(payload gjson.Result, envelopeTS time.Time) { if payload.Get("forked_from_id").Str == "" { return } ms := uuidV7Millis(payload.Get("id").Str) if ms == 0 { if t := parseTimestamp(payload.Get("timestamp").Str); !t.IsZero() { ms = t.UnixMilli() } } if ms == 0 && !envelopeTS.IsZero() { ms = envelopeTS.UnixMilli() } if ms == 0 { return // no anchor for the boundary — fail open } g.active = true g.createdMs = ms } // suppresses reports whether the line is replayed parent history. // turn_context lines open the gate when their turn id was minted at // or after the fork instant. func (g *codexForkGate) suppresses(lineType string, payload gjson.Result) bool { if !g.active { return false } if lineType != codexTypeTurnContext { return true } tid := payload.Get("turn_id").Str if tid == "" { return true // pre-turn_id parent history } if ms := uuidV7Millis(tid); ms != 0 && ms < g.createdMs { return true } g.active = false return false } // uuidV7Millis extracts the millisecond timestamp embedded in a // UUIDv7, returning 0 for anything that is not a v7 UUID. func uuidV7Millis(id string) int64 { hex := strings.ReplaceAll(id, "-", "") if len(hex) != 32 || hex[12] != '7' { return 0 } ms, err := strconv.ParseInt(hex[:12], 16, 64) if err != nil { return 0 } return ms } type codexToolCallRef struct { messageIndex int callIndex int } type codexPendingEvent struct { agentID string source string status string text string timestamp time.Time ordinal int } func newCodexSessionBuilder( _ bool, ) *codexSessionBuilder { return &codexSessionBuilder{ project: "unknown", callNames: make(map[string]string), callRefs: make(map[string]codexToolCallRef), agentSpawnCalls: make(map[string]string), agentWaitCalls: make(map[string]string), pendingAgentEvents: make(map[string][]codexPendingEvent), orphanNotificationIx: make(map[string]int), } } func (b *codexSessionBuilder) incrementalSeed() codexIncrementalSeed { return b.codexCursorState } // processLine handles a single non-empty, valid JSON line. func (b *codexSessionBuilder) processLine( line string, ) (skip bool) { tsStr := gjson.Get(line, "timestamp").Str ts := parseTimestamp(tsStr) if ts.IsZero() { if tsStr != "" { logParseError(tsStr) } } else { if b.startedAt.IsZero() { b.startedAt = ts } b.endedAt = ts } payload := gjson.Get(line, "payload") switch gjson.Get(line, "type").Str { case codexTypeSessionMeta: if b.forkGate.active { // A forked rollout replays the parent's session_meta // too — the fork's own meta came first and wins. return false } return b.handleSessionMeta(payload, ts) case codexTypeTurnContext: if b.forkGate.suppresses(codexTypeTurnContext, payload) { return false } b.model = payload.Get("model").Str case codexTypeResponseItem: if b.forkGate.suppresses(codexTypeResponseItem, payload) { return false } b.handleResponseItem(payload, ts) case codexTypeEventMsg: if b.forkGate.suppresses(codexTypeEventMsg, payload) { return false } b.handleEventMsg(payload) } return false } func (b *codexSessionBuilder) handleSessionMeta( payload gjson.Result, envelopeTS time.Time, ) (skip bool) { b.sessionID = payload.Get("id").Str if cwd := payload.Get("cwd").Str; cwd != "" { b.cwd = cwd branch := payload.Get("git.branch").Str if proj := ExtractProjectFromCwdWithBranch(cwd, branch); proj != "" { b.project = proj } else { b.project = "unknown" } } b.forkGate.armFromMeta(payload, envelopeTS) return false } func (b *codexSessionBuilder) handleResponseItem( payload gjson.Result, ts time.Time, ) { switch payload.Get("type").Str { case "function_call", "custom_tool_call": b.handleFunctionCall(payload, ts) return case "function_call_output", "custom_tool_call_output": b.handleFunctionCallOutput(payload, ts) return } role := payload.Get("role").Str if role != "user" && role != "assistant" { return } content := extractCodexContent(payload) if role == "user" && !b.firstUserSeen { content = extractCodexInitialUserContent(payload) } if strings.TrimSpace(content) == "" { return } if role == "user" && b.handleSubagentNotification(content, ts) { return } if role == "user" { if isCodexTurnAbortedMessage(content) { b.markFirstUserReplayPossible() } if isCodexSystemMessage(content) { return } } if role == "user" { first, replay := b.observeUserPrompt(content) if first { b.firstMessage = truncate( strings.ReplaceAll(content, "\n", " "), 300, ) } if replay { // Codex can re-emit the initial prompt verbatim after a // turn_aborted continuation signal. Drop only that positively // identified replay; otherwise an identical second prompt is // real transcript content. The digest covers the full content, // not the truncated first-message preview. return } } b.messages = append(b.messages, ParsedMessage{ Ordinal: b.ordinal, Role: RoleType(role), Content: content, Timestamp: ts, ContentLength: len(content), Model: b.model, }) b.ordinal++ } func (b *codexSessionBuilder) handleEventMsg(payload gjson.Result) { eventType := payload.Get("type").Str switch eventType { case "task_started", "task_complete", "turn_aborted": b.observeTaskEvent(eventType) case "token_count": b.handleTokenCountEvent(payload) case "collab_agent_spawn_end": b.handleCollabAgentSpawnEnd(payload) } } func (b *codexSessionBuilder) markFirstUserReplayPossible() { b.codexCursorState.markFirstUserReplayPossible() } func (b *codexSessionBuilder) handleTokenCountEvent( payload gjson.Result, ) { raw := payload.Get("info.last_token_usage").Raw if raw == "" || b.observeTokenUsage(raw) { return } // Find last assistant message without usage in the current // turn. Stop at user message boundary so we don't cross // turns. for i, v := range slices.Backward(b.messages) { if v.Role == RoleUser { break } if v.Role == RoleAssistant && v.TokenUsage == nil { b.applyCodexTokenUsage(&b.messages[i], raw) return } } b.unattachedTokenUsage = true } func (b *codexSessionBuilder) handleCollabAgentSpawnEnd( payload gjson.Result, ) { callID := payload.Get("call_id").Str agentID := strings.TrimSpace(payload.Get("new_thread_id").Str) if callID == "" || agentID == "" { return } b.agentSpawnCalls[agentID] = callID b.setCallSubagentSessionID(callID, codexSubagentSessionID(agentID)) } // applyCodexTokenUsage normalizes Codex token usage fields // into the Anthropic-style shape expected by the usage and cost // queries. Codex reports input_tokens as the full input count // (cached portion included), while the downstream cost formula // treats input_tokens as the uncached remainder and bills // cache_read_input_tokens separately. Subtracting cached here // prevents double-counting the cached portion at the full input // rate. // // input_tokens - cached_input_tokens → input_tokens (uncached) // output_tokens → output_tokens // cached_input_tokens → cache_read_input_tokens func (b *codexSessionBuilder) applyCodexTokenUsage( msg *ParsedMessage, raw string, ) { usage := gjson.Parse(raw) totalInput := int(usage.Get("input_tokens").Int()) cached := int(usage.Get("cached_input_tokens").Int()) output := int(usage.Get("output_tokens").Int()) uncached := max(totalInput-cached, 0) normalized := map[string]int{ "input_tokens": uncached, "output_tokens": output, "cache_read_input_tokens": cached, } j, err := json.Marshal(normalized) if err != nil { return } msg.TokenUsage = j msg.OutputTokens = output msg.HasOutputTokens = output > 0 msg.ContextTokens = uncached + cached msg.HasContextTokens = totalInput > 0 || cached > 0 } func (b *codexSessionBuilder) handleFunctionCall( payload gjson.Result, ts time.Time, ) { name := payload.Get("name").Str if name == "" { return } callID := payload.Get("call_id").Str if callID != "" { b.callNames[callID] = name } content := formatCodexFunctionCall(name, payload) inputJSON := extractCodexInputJSON(payload) skillName := inferCodexSkillNameWithBase(name, inputJSON, b.cwd) waitAgentIDs := []string(nil) if isCodexWaitAgentCall(name) && callID != "" { args, _ := parseCodexFunctionArgs(payload) waitAgentIDs = codexWaitAgentIDs(args) } b.messages = append(b.messages, ParsedMessage{ Ordinal: b.ordinal, Role: RoleAssistant, Content: content, Timestamp: ts, HasToolUse: true, ContentLength: len(content), Model: b.model, ToolCalls: []ParsedToolCall{{ ToolUseID: callID, ToolName: name, Category: NormalizeToolCategory(name), InputJSON: inputJSON, SkillName: skillName, }}, }) if callID != "" { b.callRefs[callID] = codexToolCallRef{ messageIndex: len(b.messages) - 1, callIndex: 0, } } b.ordinal++ if isCodexWaitAgentCall(name) && callID != "" { for _, agentID := range waitAgentIDs { b.agentWaitCalls[agentID] = callID b.claimPendingAgentEvents(callID, agentID) } } } func (b *codexSessionBuilder) handleFunctionCallOutput( payload gjson.Result, ts time.Time, ) { callID := payload.Get("call_id").Str if callID == "" { return } output, raw := parseCodexFunctionOutput(payload) if !output.Exists() { if strings.TrimSpace(raw) == "" { return } } switch b.callNames[callID] { case "spawn_agent": agentID := strings.TrimSpace(output.Get("agent_id").Str) if agentID == "" { return } b.agentSpawnCalls[agentID] = callID b.setCallSubagentSessionID(callID, codexSubagentSessionID(agentID)) case "wait", "wait_agent": status := output.Get("status") if !status.Exists() || !status.IsObject() { return } status.ForEach(func(key, entry gjson.Result) bool { agentID := key.Str statusName, text := codexTerminalSubagentEvent(entry) if text == "" { return true } b.appendCallResultEvent(callID, ParsedToolResultEvent{ ToolUseID: callID, AgentID: agentID, SubagentSessionID: codexSubagentSessionID(agentID), Source: "wait_output", Status: statusName, Content: text, Timestamp: ts, }) return true }) default: if text := strings.TrimSpace(raw); text != "" { source := "function_call_output" status := "" if payload.Get("type").Str == "custom_tool_call_output" { source = "custom_tool_call_output" status = payload.Get("status").Str if status == "" { status = "completed" } } b.appendCallResultEvent(callID, ParsedToolResultEvent{ ToolUseID: callID, Source: source, Status: status, Content: text, Timestamp: ts, }) } } } // setCallSubagentSessionID links a tool call to the session of // the subagent it spawned. Callers must invoke this only after // the originating function_call has been processed (which // populates b.callRefs[callID]); otherwise the link is silently // dropped. In real codex session files the spawn function_call // always precedes both its function_call_output and the // collab_agent_spawn_end event_msg. func (b *codexSessionBuilder) setCallSubagentSessionID( callID, sessionID string, ) { if callID == "" || sessionID == "" { return } ref, ok := b.callRefs[callID] if !ok || ref.messageIndex < 0 || ref.messageIndex >= len(b.messages) { return } if ref.callIndex < 0 || ref.callIndex >= len(b.messages[ref.messageIndex].ToolCalls) { return } b.messages[ref.messageIndex].ToolCalls[ref.callIndex].SubagentSessionID = sessionID } func (b *codexSessionBuilder) handleSubagentNotification( content string, ts time.Time, ) bool { agentID, statusName, text := parseCodexSubagentNotification(content) if agentID == "" || text == "" { return false } if callID := b.agentWaitCalls[agentID]; callID != "" { b.appendCallResultEvent(callID, ParsedToolResultEvent{ AgentID: agentID, SubagentSessionID: codexSubagentSessionID(agentID), Source: "subagent_notification", Status: statusName, Content: text, Timestamp: ts, }) return true } b.pendingAgentEvents[agentID] = append( b.pendingAgentEvents[agentID], codexPendingEvent{ agentID: agentID, source: "subagent_notification", status: statusName, text: text, timestamp: ts, ordinal: b.ordinal, }, ) b.ordinal++ return true } func (b *codexSessionBuilder) appendCallResultEvent( callID string, ev ParsedToolResultEvent, ) { if callID == "" { return } ref, ok := b.callRefs[callID] if !ok || ref.messageIndex < 0 || ref.messageIndex >= len(b.messages) { return } if ref.callIndex < 0 || ref.callIndex >= len(b.messages[ref.messageIndex].ToolCalls) { return } tc := &b.messages[ref.messageIndex].ToolCalls[ref.callIndex] if ev.ToolUseID == "" { ev.ToolUseID = tc.ToolUseID } if ev.SubagentSessionID == "" && ev.AgentID != "" { ev.SubagentSessionID = codexSubagentSessionID(ev.AgentID) } if b.hasEquivalentCallResultEvent(tc.ResultEvents, ev) { return } tc.ResultEvents = append(tc.ResultEvents, ev) } func (b *codexSessionBuilder) hasEquivalentCallResultEvent( events []ParsedToolResultEvent, candidate ParsedToolResultEvent, ) bool { for _, existing := range events { if existing.AgentID == candidate.AgentID && existing.Status == candidate.Status && existing.Content == candidate.Content { return true } } return false } func (b *codexSessionBuilder) claimPendingAgentEvents( callID, agentID string, ) { pending := b.pendingAgentEvents[agentID] if len(pending) == 0 { return } for _, ev := range pending { b.appendCallResultEvent(callID, ParsedToolResultEvent{ AgentID: ev.agentID, SubagentSessionID: codexSubagentSessionID(ev.agentID), Source: ev.source, Status: ev.status, Content: ev.text, Timestamp: ev.timestamp, }) } delete(b.pendingAgentEvents, agentID) } func (b *codexSessionBuilder) flushPendingAgentResults() { if len(b.pendingAgentEvents) == 0 { return } agentIDs := make([]string, 0, len(b.pendingAgentEvents)) for agentID := range b.pendingAgentEvents { agentIDs = append(agentIDs, agentID) } sort.Strings(agentIDs) for _, agentID := range agentIDs { pending := b.pendingAgentEvents[agentID] switch { case b.agentWaitCalls[agentID] != "": b.claimPendingAgentEvents(b.agentWaitCalls[agentID], agentID) case b.agentSpawnCalls[agentID] != "": b.claimPendingAgentEvents(b.agentSpawnCalls[agentID], agentID) default: for _, ev := range pending { key := agentID + "\x00" + ev.status + "\x00" + ev.text if _, ok := b.orphanNotificationIx[key]; ok { continue } idx := b.insertMessage(ParsedMessage{ Ordinal: ev.ordinal, Role: RoleUser, Content: ev.text, Timestamp: ev.timestamp, Model: b.model, ContentLength: len(ev.text), }) b.orphanNotificationIx[key] = idx } delete(b.pendingAgentEvents, agentID) } } } func codexSubagentSessionID(agentID string) string { agentID = strings.TrimSpace(agentID) if agentID == "" { return "" } if strings.HasPrefix(agentID, "codex:") { return agentID } return "codex:" + agentID } func (b *codexSessionBuilder) normalizeOrdinals() { sort.SliceStable(b.messages, func(i, j int) bool { if b.messages[i].Ordinal == b.messages[j].Ordinal { return i < j } return b.messages[i].Ordinal < b.messages[j].Ordinal }) for i := range b.messages { b.messages[i].Ordinal = i } } func (b *codexSessionBuilder) insertMessage(msg ParsedMessage) int { idx := len(b.messages) for i, existing := range b.messages { if existing.Ordinal > msg.Ordinal || (existing.Ordinal == msg.Ordinal && !msg.Timestamp.IsZero() && (existing.Timestamp.IsZero() || msg.Timestamp.Before(existing.Timestamp))) { idx = i break } } b.messages = append(b.messages, ParsedMessage{}) copy(b.messages[idx+1:], b.messages[idx:]) b.messages[idx] = msg for callID, ref := range b.callRefs { if ref.messageIndex >= idx { ref.messageIndex++ b.callRefs[callID] = ref } } return idx } func formatCodexFunctionCall( name string, payload gjson.Result, ) string { summary := sanitizeToolLabel(payload.Get("summary").Str) args, rawArgs := parseCodexFunctionArgs(payload) switch name { case "exec_command", "shell_command", "shell": return formatCodexBashCall(summary, args, rawArgs) case "write_stdin": return formatCodexWriteStdinCall(summary, args, rawArgs) case "apply_patch": return formatCodexApplyPatchCall(summary, args, rawArgs) case "spawn_agent": return formatCodexSpawnAgentCall(summary, args, rawArgs) } category := NormalizeToolCategory(name) if category == "Other" { header := formatToolHeader("Tool", name) if summary != "" { return header + "\n" + summary } if preview := codexArgPreview(args, rawArgs); preview != "" { return header + "\n" + preview } return header } detail := firstNonEmpty(summary, codexCategoryDetail(category, args)) header := formatToolHeader(category, detail) if preview := codexArgPreview(args, rawArgs); preview != "" { return header + "\n" + preview } return header } func parseCodexFunctionArgs( payload gjson.Result, ) (gjson.Result, string) { for _, key := range []string{"arguments", "input"} { arg := payload.Get(key) if !arg.Exists() { continue } switch arg.Type { case gjson.String: s := strings.TrimSpace(arg.Str) if s == "" { continue } if gjson.Valid(s) { return gjson.Parse(s), "" } return gjson.Result{}, s default: if arg.IsObject() { if len(arg.Map()) == 0 { continue } return arg, "" } if arg.IsArray() { if len(arg.Array()) == 0 { continue } return arg, "" } raw := strings.TrimSpace(arg.Raw) if raw == "" { continue } if gjson.Valid(raw) { return gjson.Parse(raw), "" } return gjson.Result{}, raw } } return gjson.Result{}, "" } // extractCodexInputJSON returns the raw JSON string of the // function call arguments from the payload. It checks // "arguments" then "input", normalizing string-encoded JSON // to an object string. func extractCodexInputJSON(payload gjson.Result) string { for _, key := range []string{"arguments", "input"} { arg := payload.Get(key) if !arg.Exists() { continue } switch arg.Type { case gjson.String: s := strings.TrimSpace(arg.Str) if s == "" { continue } if gjson.Valid(s) { if s == "{}" || s == "[]" { continue } return s } return arg.Str default: raw := strings.TrimSpace(arg.Raw) if raw == "" || raw == "{}" || raw == "[]" { continue } return arg.Raw } } return "" } func formatCodexBashCall( summary string, args gjson.Result, rawArgs string, ) string { cmd := codexArgValue(args, "cmd", "command") if cmd == "" && rawArgs != "" && !gjson.Valid(rawArgs) { cmd = rawArgs } if cmd == "" && args.Type == gjson.String { cmd = strings.TrimSpace(args.Str) } header := formatToolHeader("Bash", summary) if cmd != "" { firstLine, _, hasMore := strings.Cut(cmd, "\n") if hasMore { return header + "\n$ " + firstLine } return header + "\n$ " + cmd } if preview := codexArgPreview(args, rawArgs); preview != "" { return header + "\n" + preview } return header } func formatCodexWriteStdinCall( summary string, args gjson.Result, rawArgs string, ) string { if summary == "" { if sid := codexArgValue(args, "session_id"); sid != "" { summary = "stdin -> " + sid } else { summary = "stdin" } } header := formatToolHeader("Bash", summary) chars := codexArgString(args, "chars") if chars != "" { quoted := strings.Trim( strconv.QuoteToASCII(chars), "\"", ) return header + "\n" + truncate(quoted, 220) } if preview := codexArgPreview(args, rawArgs); preview != "" { return header + "\n" + preview } return header } func formatCodexApplyPatchCall( summary string, args gjson.Result, rawArgs string, ) string { patch := codexArgString(args, "patch") if patch == "" && strings.Contains(rawArgs, "*** Begin Patch") { patch = rawArgs } files := extractPatchedFiles(patch) if summary == "" { summary = summarizePatchedFiles(files) } header := formatToolHeader("Edit", summary) if len(files) > 1 { limit := min(len(files), 6) body := strings.Join(files[:limit], "\n") if len(files) > limit { body += fmt.Sprintf("\n+%d more files", len(files)-limit) } return header + "\n" + body } if preview := codexArgPreview(args, rawArgs); preview != "" && len(files) == 0 { return header + "\n" + preview } return header } func formatCodexSpawnAgentCall( summary string, args gjson.Result, rawArgs string, ) string { if summary == "" { summary = firstNonEmpty( codexArgValue(args, "agent_type"), codexArgValue(args, "subagent_type"), "spawn_agent", ) } header := formatToolHeader("Task", summary) prompt := firstNonEmpty( codexArgValue(args, "description"), codexArgValue(args, "message"), codexArgValue(args, "prompt"), ) if prompt != "" { firstLine, _, _ := strings.Cut(prompt, "\n") return header + "\n" + truncate(firstLine, 220) } if preview := codexArgPreview(args, rawArgs); preview != "" { return header + "\n" + preview } return header } func extractPatchedFiles(patch string) []string { if patch == "" { return nil } var files []string seen := make(map[string]struct{}) for line := range strings.SplitSeq(patch, "\n") { for _, prefix := range []string{ "*** Add File: ", "*** Update File: ", "*** Delete File: ", "*** Move to: ", } { if !strings.HasPrefix(line, prefix) { continue } file := strings.TrimSpace( strings.TrimPrefix(line, prefix), ) if file == "" { continue } if _, ok := seen[file]; ok { continue } seen[file] = struct{}{} files = append(files, file) break } } return files } func summarizePatchedFiles(files []string) string { switch len(files) { case 0: return "" case 1: return files[0] default: return fmt.Sprintf( "%s (+%d more)", files[0], len(files)-1, ) } } func codexCategoryDetail( category string, args gjson.Result, ) string { switch category { case "Read", "Write", "Edit": return codexArgValue(args, "file_path", "path") case "Grep": return codexArgValue(args, "pattern") case "Glob": pattern := codexArgValue(args, "pattern") path := codexArgValue(args, "path") if pattern != "" && path != "" { return fmt.Sprintf("%s in %s", pattern, path) } return firstNonEmpty(pattern, path) case "Task", "Agent": desc := codexArgValue(args, "description") agent := codexArgValue(args, "subagent_type") if desc != "" && agent != "" { return fmt.Sprintf("%s (%s)", desc, agent) } return firstNonEmpty(desc, agent) default: return "" } } func codexArgString( args gjson.Result, path string, ) string { v := args.Get(path) if !v.Exists() { return "" } if v.Type == gjson.String { return v.Str } raw := strings.TrimSpace(v.Raw) if raw == "" || raw == "null" { return "" } return raw } func codexArgValue( args gjson.Result, paths ...string, ) string { for _, path := range paths { v := strings.TrimSpace(codexArgString(args, path)) if v != "" { return v } } return "" } func codexArgPreview( args gjson.Result, rawArgs string, ) string { if rawArgs != "" { flat := strings.Join( strings.Fields(rawArgs), " ", ) return truncate(flat, 220) } if args.Exists() { flat := strings.Join( strings.Fields(args.Raw), " ", ) if flat != "" { return truncate(flat, 220) } } return "" } func formatToolHeader( label, detail string, ) string { label = sanitizeToolLabel(label) if label == "" { label = "Tool" } detail = sanitizeToolLabel(detail) if detail != "" { return fmt.Sprintf("[%s: %s]", label, detail) } return fmt.Sprintf("[%s]", label) } func sanitizeToolLabel(s string) string { s = strings.TrimSpace(s) if s == "" { return "" } s = strings.ReplaceAll(s, "]", ")") return strings.Join(strings.Fields(s), " ") } func firstNonEmpty(vals ...string) string { for _, v := range vals { v = strings.TrimSpace(v) if v != "" { return v } } return "" } func parseCodexFunctionOutput( payload gjson.Result, ) (gjson.Result, string) { out := payload.Get("output") if !out.Exists() { return gjson.Result{}, "" } switch out.Type { case gjson.String: s := strings.TrimSpace(out.Str) if s == "" { return gjson.Result{}, "" } if gjson.Valid(s) { return gjson.Parse(s), s } return gjson.Result{}, s default: raw := strings.TrimSpace(out.Raw) if raw == "" { return gjson.Result{}, "" } if gjson.Valid(raw) { return gjson.Parse(raw), raw } return gjson.Result{}, raw } } func codexWaitAgentIDs(args gjson.Result) []string { if !args.Exists() { return nil } ids := args.Get("ids") if !ids.Exists() { ids = args.Get("targets") } if !ids.Exists() || !ids.IsArray() { return nil } var out []string for _, item := range ids.Array() { id := strings.TrimSpace(item.Str) if id == "" { continue } out = append(out, id) } return out } func isCodexWaitAgentCall(name string) bool { return name == "wait" || name == "wait_agent" } func parseCodexSubagentNotification( content string, ) (agentID, statusName, text string) { if !isCodexSubagentNotification(content) { return "", "", "" } body := strings.TrimSpace(content) body = strings.TrimPrefix(body, "") body = strings.TrimSuffix(body, "") body = strings.TrimSpace(body) if !gjson.Valid(body) { return "", "", "" } parsed := gjson.Parse(body) agentID = firstNonEmpty( parsed.Get("agent_id").Str, parsed.Get("agent_path").Str, ) status := parsed.Get("status") statusName, text = codexTerminalSubagentEvent(status) return agentID, statusName, text } func codexTerminalSubagentEvent(status gjson.Result) (string, string) { if text := strings.TrimSpace(status.Get("completed").Str); text != "" { return "completed", text } if text := strings.TrimSpace(status.Get("errored").Str); text != "" { return "errored", text } if text := strings.TrimSpace(status.Get("running").Str); text != "" { return "running", text } return "", "" } func codexTerminalSubagentStatus(status gjson.Result) string { _, text := codexTerminalSubagentEvent(status) return text } func isCodexSubagentFunctionOutput(output gjson.Result) bool { if !output.Exists() { return false } if strings.TrimSpace(output.Get("agent_id").Str) != "" { return true } status := output.Get("status") if !status.Exists() || !status.IsObject() { return false } entries := status.Map() if len(entries) == 0 { return false } for agentID, entry := range entries { if strings.TrimSpace(agentID) == "" || !entry.IsObject() { return false } if codexTerminalSubagentStatus(entry) != "" { continue } if strings.TrimSpace(entry.Get("running").Str) != "" { continue } return false } return true } func extractCodexTextBlocks(payload gjson.Result) []string { var texts []string payload.Get("content").ForEach( func(_, block gjson.Result) bool { switch block.Get("type").Str { case "input_text", "output_text", "text": if t := block.Get("text").Str; t != "" { texts = append(texts, t) } } return true }, ) return texts } // extractCodexContent joins all text blocks from a Codex // response item's content array. func extractCodexContent(payload gjson.Result) string { return strings.Join(extractCodexTextBlocks(payload), "\n") } // extractCodexInitialUserContent filters the synthetic blocks bundled with // Codex's recommended-plugins injection while retaining user-authored blocks // from the same response item. func extractCodexInitialUserContent(payload gjson.Result) string { texts := extractCodexTextBlocks(payload) if len(texts) == 0 { return strings.Join(texts, "\n") } stripped := stripCodexRecommendedPlugins(texts[0]) if stripped == texts[0] { return strings.Join(texts, "\n") } texts[0] = stripped kept := texts[:0] for _, text := range texts { text = stripCodexInitialSystemPrefix(text) if strings.TrimSpace(text) == "" || isCodexSystemMessage(text) { continue } kept = append(kept, text) } return strings.Join(kept, "\n") } // IsCodexExecSessionFile reports whether any session_meta // line in a Codex JSONL file has originator=="codex_exec". // The pre-bulk-sync parser called handleSessionMeta on every // session_meta line and flagged the whole session as exec if // any of them carried that originator, so a one-shot check // of only the first session_meta would miss files that were // originally skipped because a later session_meta set the // originator. Scan all session_meta lines to match the old // skip condition exactly. func IsCodexExecSessionFile(path string) bool { f, err := os.Open(path) if err != nil { return false } defer f.Close() s := bufio.NewScanner(f) s.Buffer(make([]byte, 0, 64*1024), maxLineSize) for s.Scan() { line := strings.TrimSpace(s.Text()) if line == "" || !gjson.Valid(line) { continue } if gjson.Get(line, "type").Str != codexTypeSessionMeta { continue } if gjson.Get(line, "payload.originator").Str == codexOriginatorExec { return true } } return false } // parseSession parses a Codex JSONL session file into a session and its // messages. The includeExec parameter is retained for backward compatibility; // exec-originated sessions are now always parsed and imported. This is the // provider-owned parse entrypoint; the package-level free function was folded // onto the provider. func (p *codexProvider) parseSession( path, machine string, includeExec bool, ) (*ParsedSession, []ParsedMessage, error) { f, err := os.Open(path) if err != nil { return nil, nil, fmt.Errorf("open %s: %w", path, err) } defer f.Close() info, err := f.Stat() if err != nil { return nil, nil, fmt.Errorf("stat %s: %w", path, err) } return p.parseSessionSnapshot( path, machine, includeExec, f, info, ) } // parseSessionSnapshot parses exactly the raw-size snapshot captured from f. // Limiting the reader prevents an append racing the scan from being folded into // a cursor keyed by the earlier size. func (p *codexProvider) parseSessionSnapshot( path, machine string, includeExec bool, f *os.File, info os.FileInfo, ) (*ParsedSession, []ParsedMessage, error) { lr := newLineReader(io.LimitReader(f, info.Size()), maxLineSize) defer releaseLineReader(lr) b := newCodexSessionBuilder(includeExec) for { line, ok := lr.next() if !ok { break } if !gjson.Valid(line) { continue } if b.processLine(line) { return nil, nil, nil } } if err := lr.Err(); err != nil { return nil, nil, fmt.Errorf("reading codex %s: %w", path, err) } b.flushPendingAgentResults() b.normalizeOrdinals() inode, device := sourceFileIdentity(info) if safe, safeErr := codexSafeResumeOffsetFile(f, info.Size()); safeErr == nil && safe { p.cursorCache.Put( path, info.Size(), inode, device, b.incrementalSeed(), ) } sessionID := b.sessionID if sessionID == "" { sessionID = strings.TrimSuffix( filepath.Base(path), ".jsonl", ) } sessionID = "codex:" + sessionID userCount := 0 for _, m := range b.messages { if m.Role == RoleUser && m.Content != "" { userCount++ } } mtime := info.ModTime().UnixNano() // Include session_index.jsonl mtime so renames trigger a re-parse. if idxPath := codexSessionIndexPath(path); idxPath != "" { if idxInfo, err := os.Stat(idxPath); err == nil { if idxMtime := idxInfo.ModTime().UnixNano(); idxMtime > mtime { mtime = idxMtime } } } sess := &ParsedSession{ ID: sessionID, Project: b.project, Machine: machine, Agent: AgentCodex, Cwd: b.cwd, FirstMessage: b.firstMessage, SessionName: LookupCodexThreadName(path, b.sessionID), StartedAt: b.startedAt, EndedAt: b.endedAt, MessageCount: len(b.messages), UserMessageCount: userCount, TerminationStatus: classifyCodexTermination(b.lastTaskEvent), File: FileInfo{ Path: path, Size: info.Size(), Mtime: mtime, Inode: int64(inode), Device: int64(device), }, } accumulateMessageTokenUsage(sess, b.messages) return sess, b.messages, nil } // CodexSessionIndexFilename is the name of the Codex index file that maps // session UUIDs to their (renameable) thread titles. It sits next to the // sessions/ and archived_sessions/ directories. const CodexSessionIndexFilename = "session_index.jsonl" // CodexSessionIndexTitles returns the session UUID to thread-title map from // a Codex session_index.jsonl file, or nil when it cannot be read. The // underlying read is cached by path, mtime, and size. func CodexSessionIndexTitles(indexPath string) map[string]string { titles, err := loadCodexSessionIndex(indexPath) if err != nil { return nil } return titles } // EvictCodexSessionIndex removes one cached session_index.jsonl entry. Callers // use it when an explicit change event makes the sidecar stat tuple insufficient // and when transient hydrated indexes should not outlive their parse. func EvictCodexSessionIndex(indexPath string) { codexSessionIndexCache.mu.Lock() delete(codexSessionIndexCache.entries, indexPath) codexSessionIndexCache.mu.Unlock() } // EvictAllCodexSessionIndexes removes every cached session_index.jsonl entry. // Watcher overflow recovery uses this before a force-reverification pass, // because the individual index paths that changed were deliberately coalesced. func EvictAllCodexSessionIndexes() { codexSessionIndexCache.mu.Lock() clear(codexSessionIndexCache.entries) codexSessionIndexCache.mu.Unlock() } // EvictCodexSessionIndexForSession removes the cached sidecar associated with // one Codex transcript. Explicit full-parse callers use this when an external // event says the sidecar changed even if its stat tuple did not. func EvictCodexSessionIndexForSession(sessionPath string) { if indexPath := codexSessionIndexPath(sessionPath); indexPath != "" { EvictCodexSessionIndex(indexPath) } } // LookupCodexThreadName returns the current Codex thread name for a session // from the session_index.jsonl file next to the session root. func LookupCodexThreadName(sessionPath, sessionID string) string { if strings.TrimSpace(sessionID) == "" { return "" } indexPath := codexSessionIndexPath(sessionPath) if indexPath == "" { return "" } titles, err := loadCodexSessionIndex(indexPath) if err != nil { return "" } return strings.TrimSpace(titles[sessionID]) } // CodexEffectiveMtime returns the effective mtime for a Codex session file, // incorporating session_index.jsonl so renames invalidate the cache. func CodexEffectiveMtime(sessionPath string, fileMtime int64) int64 { if idxPath := codexSessionIndexPath(sessionPath); idxPath != "" { if si, err := os.Stat(idxPath); err == nil { if idxMtime := si.ModTime().UnixNano(); idxMtime > fileMtime { return idxMtime } } } return fileMtime } // CodexSessionIndexPath returns the local session_index.jsonl path associated // with a Codex transcript, or an empty string when the transcript is outside a // recognized Codex session directory. func CodexSessionIndexPath(sessionPath string) string { return codexSessionIndexPath(sessionPath) } func codexSessionIndexPath(sessionPath string) string { dir := filepath.Dir(sessionPath) for dir != "" { base := filepath.Base(dir) if base == "sessions" || base == "archived_sessions" { return filepath.Join( filepath.Dir(dir), CodexSessionIndexFilename, ) } parent := filepath.Dir(dir) if parent == dir { break } dir = parent } return "" } func loadCodexSessionIndex(indexPath string) (map[string]string, error) { info, err := os.Stat(indexPath) if err != nil { return nil, err } mtime := info.ModTime().UnixNano() size := info.Size() changeTime, changeTimeOkay := codexIndexChangeTime(indexPath, info) codexSessionIndexCache.mu.Lock() if entry, ok := codexSessionIndexCache.entries[indexPath]; ok && entry.mtime == mtime && entry.size == size && (!changeTimeOkay || (entry.changeTimeOkay && entry.changeTime == changeTime)) { codexSessionIndexCache.mu.Unlock() return entry.titles, nil } codexSessionIndexCache.mu.Unlock() f, err := os.Open(indexPath) if err != nil { return nil, err } defer f.Close() titles, err := ParseCodexSessionIndexTitles(f) if err != nil { return nil, err } codexSessionIndexCache.mu.Lock() codexSessionIndexCache.entries[indexPath] = codexSessionIndexEntry{ mtime: mtime, size: size, changeTime: changeTime, changeTimeOkay: changeTimeOkay, titles: titles, } codexSessionIndexCache.mu.Unlock() return titles, nil } // ParseCodexSessionIndexTitles reads a Codex session_index.jsonl stream and // returns session UUIDs mapped to non-empty thread titles. func ParseCodexSessionIndexTitles(r io.Reader) (map[string]string, error) { titles := make(map[string]string) s := bufio.NewScanner(r) s.Buffer(make([]byte, 0, 64*1024), maxLineSize) for s.Scan() { line := s.Text() if !gjson.Valid(line) { continue } id := gjson.Get(line, "id").Str title := strings.TrimSpace(gjson.Get(line, "thread_name").Str) if id == "" || title == "" { continue } titles[id] = title } if err := s.Err(); err != nil { return nil, err } return titles, nil } // classifyCodexTermination maps the most recent task lifecycle // event seen on a Codex session file to a TerminationStatus. // Codex emits explicit task_started / task_complete / turn_aborted // events, so the classification is unambiguous when any are // present. Returns "" (unknown) for files where no task event // was seen — typically very short or malformed sessions. func classifyCodexTermination(lastTaskEvent string) TerminationStatus { switch lastTaskEvent { case "task_complete": return TerminationAwaitingUser case "task_started", "turn_aborted": // task_started without a matching task_complete after // means the agent was mid-turn when the file last // flushed — treat the same as an orphan tool call. // turn_aborted means the user interrupted; same shape. return TerminationToolCallPending } return "" } // codexIncrementalSeed carries the builder state recovered from the // already-parsed prefix [0, offset) of a Codex JSONL file so an // incremental parse resumes with the same view a full parse would // have at that offset: the current model, the re-emitted-prompt // dedup state, task lifecycle marker, and the fork replay gate (#643). type codexIncrementalSeed = codexCursorState // seedCodexIncrementalState scans a Codex JSONL prefix [0, offset) // and mirrors processLine's dispatch: every turn_context overwrites // the model (including empty strings), user messages feed the // re-emitted-prompt dedup exactly as handleResponseItem would, and // the fork gate arms/opens on the same lines as a full parse. A gate // still active at the end of the scan means the stored offset landed // inside the replayed parent history of a forked rollout, so the // incremental parse must keep suppressing appended replay lines. func seedCodexIncrementalState( path string, offset int64, ) (codexIncrementalSeed, error) { f, err := os.Open(path) if err != nil { return codexIncrementalSeed{}, fmt.Errorf( "open codex prefix %s: %w", path, err, ) } defer f.Close() seed, err := seedCodexIncrementalStateFromReader( io.LimitReader(f, offset), ) if err != nil { return codexIncrementalSeed{}, fmt.Errorf( "read codex prefix %s: %w", path, err, ) } return seed, nil } func seedCodexIncrementalStateFromReader( r io.Reader, ) (codexIncrementalSeed, error) { var seed codexIncrementalSeed lr := newLineReader(r, maxLineSize) defer releaseLineReader(lr) for { line, ok := lr.next() if !ok { break } if !gjson.Valid(line) { continue } lineType := gjson.Get(line, "type").Str payload := gjson.Get(line, "payload") if lineType == codexTypeSessionMeta { // Mirror processLine: the fork's own meta arms the // gate and supplies cwd, and replayed parent metas are // dropped while it is active. if !seed.forkGate.active { if cwd := payload.Get("cwd").Str; cwd != "" { seed.cwd = cwd } seed.forkGate.armFromMeta( payload, parseTimestamp(gjson.Get(line, "timestamp").Str), ) } continue } if seed.forkGate.suppresses(lineType, payload) { continue } switch lineType { case codexTypeTurnContext: seed.model = payload.Get("model").Str case codexTypeEventMsg: eventType := payload.Get("type").Str seed.observeTaskEvent(eventType) if eventType == "token_count" { raw := payload.Get("info.last_token_usage").Raw if raw != "" { seed.observeTokenUsage(raw) } } case codexTypeResponseItem: observeCodexIncrementalUserMessage(&seed, payload) } } if err := lr.Err(); err != nil { return codexIncrementalSeed{}, err } return seed, nil } // observeUserMessage feeds one response_item into the // re-emitted-prompt dedup state, mirroring handleResponseItem's // user-message filtering and full-content matching. func observeCodexIncrementalUserMessage( s *codexIncrementalSeed, payload gjson.Result, ) { if payload.Get("role").Str != "user" { return } content := extractCodexContent(payload) if !s.firstUserSeen { content = extractCodexInitialUserContent(payload) } if strings.TrimSpace(content) == "" { return } if isCodexTurnAbortedMessage(content) { s.markFirstUserReplayPossible() } if isCodexSystemMessage(content) { return } s.observeUserPrompt(content) } // CodexTranscriptConsumedSize returns the byte offset after the last complete, // valid JSON line in a Codex transcript. Bytes after this offset are ignored by // the Codex JSONL parser, so partial trailing writes are not part of the parsed // source snapshot. func CodexTranscriptConsumedSize(path string) (int64, error) { consumed, err := readCodexJSONLFrom(path, 0, func(line string) {}) if errors.Is(err, errCodexIncrementalNeedsFullParse) { return consumed, nil } return consumed, err } // readCodexJSONLFrom is the Codex-specific conservative tail reader. Codex may // expose syntactically valid JSON before the writer appends its record newline; // such an EOF record requires a full-parse fallback and is never staged as a // safe continuation cursor. func readCodexJSONLFrom( path string, offset int64, fn func(line string), ) (consumed int64, err error) { f, err := os.Open(path) if err != nil { return 0, fmt.Errorf("open %s: %w", path, err) } defer f.Close() info, err := f.Stat() if err != nil { return 0, fmt.Errorf("stat %s: %w", path, err) } return readCodexJSONLSection(f, offset, info.Size(), fn) } // readCodexJSONLSection applies the conservative Codex JSONL rules to the // exact byte range [offset, limit). A SectionReader makes the captured limit a // real EOF even if the underlying descriptor has since grown. func readCodexJSONLSection( f *os.File, offset int64, limit int64, fn func(line string), ) (consumed int64, err error) { if offset < 0 || limit < offset { return 0, fmt.Errorf( "invalid codex JSONL section [%d,%d)", offset, limit, ) } section := io.NewSectionReader(f, offset, limit-offset) return readCodexJSONLReader(section, section, fn) } func readCodexJSONLReader( r io.Reader, at io.ReaderAt, fn func(line string), ) (consumed int64, err error) { lr := newLineReader(r, maxLineSize) defer releaseLineReader(lr) for { line, ok := lr.next() if !ok { break } var terminator [1]byte if _, err := at.ReadAt(terminator[:], lr.bytesRead-1); err != nil { return consumed, err } if terminator[0] != '\n' { if gjson.Valid(line) { return consumed, errCodexIncrementalNeedsFullParse } break } if gjson.Valid(line) { fn(line) consumed = lr.bytesRead } } return consumed, lr.Err() } // codexSafeResumeOffset checks in O(1) that offset starts at a physical line // boundary. Offset zero is always safe; every nonzero offset must immediately // follow a newline byte. func codexSafeResumeOffset(path string, offset int64) (bool, error) { f, err := os.Open(path) if err != nil { return false, err } defer f.Close() return codexSafeResumeOffsetFile(f, offset) } func codexSafeResumeOffsetFile(f *os.File, offset int64) (bool, error) { if offset == 0 { return true, nil } if offset < 0 { return false, nil } var previous [1]byte if _, err := f.ReadAt(previous[:], offset-1); err != nil { if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { return false, nil } return false, err } return previous[0] == '\n', nil } type codexIncrementalParseResult struct { messages []ParsedMessage endedAt time.Time consumedBytes int64 initialCursor codexCursorState cursor codexCursorState inode uint64 device uint64 } // parseSessionFromDetailed parses only new lines from a Codex JSONL file. It // resumes from an exact cached cursor when one exists and otherwise reconstructs // the same state by scanning the prefix. A successful result carries the exact // cursor the provider may stage at offset+consumed; the prior offset remains // eligible until the caller persists the new offset. func (p *codexProvider) parseSessionFromDetailed( path string, offset int64, startOrdinal int, includeExec bool, ) (codexIncrementalParseResult, error) { f, err := os.Open(path) if err != nil { return codexIncrementalParseResult{}, fmt.Errorf( "open codex %s: %w", path, err, ) } defer f.Close() info, err := f.Stat() if err != nil { return codexIncrementalParseResult{}, fmt.Errorf( "stat codex %s: %w", path, err, ) } return p.parseSessionFromSnapshot( path, offset, startOrdinal, includeExec, f, info, info.Size(), ) } func (p *codexProvider) parseSessionFromSnapshot( path string, offset int64, startOrdinal int, includeExec bool, f *os.File, info os.FileInfo, limit int64, ) (codexIncrementalParseResult, error) { return p.parseSessionFromWithSources( path, offset, startOrdinal, includeExec, info, func(fn func(string)) (int64, error) { return readCodexJSONLSection(f, offset, limit, fn) }, func() (codexIncrementalSeed, error) { return seedCodexIncrementalStateFromReader( io.NewSectionReader(f, 0, offset), ) }, ) } func (p *codexProvider) parseSessionFromWithReader( path string, offset int64, startOrdinal int, includeExec bool, readLines func(string, int64, func(string)) (int64, error), ) (codexIncrementalParseResult, error) { return p.parseSessionFromWithReaders( path, offset, startOrdinal, includeExec, readLines, seedCodexIncrementalState, ) } func (p *codexProvider) parseSessionFromWithReaders( path string, offset int64, startOrdinal int, includeExec bool, readLines func(string, int64, func(string)) (int64, error), readSeed func(string, int64) (codexIncrementalSeed, error), ) (codexIncrementalParseResult, error) { info, err := os.Stat(path) if err != nil { return codexIncrementalParseResult{}, fmt.Errorf( "stat codex %s: %w", path, err, ) } return p.parseSessionFromWithSources( path, offset, startOrdinal, includeExec, info, func(fn func(string)) (int64, error) { return readLines(path, offset, fn) }, func() (codexIncrementalSeed, error) { return readSeed(path, offset) }, ) } func (p *codexProvider) parseSessionFromWithSources( path string, offset int64, startOrdinal int, includeExec bool, info os.FileInfo, readLines func(func(string)) (int64, error), readSeed func() (codexIncrementalSeed, error), ) (codexIncrementalParseResult, error) { inode, device := sourceFileIdentity(info) seed, cacheHit := p.cursorCache.Get(path, offset, inode, device) if !cacheHit { var err error seed, err = readSeed() if err != nil { return codexIncrementalParseResult{}, fmt.Errorf( "seed codex %s at offset %d: %w", path, offset, err, ) } } b := newCodexSessionBuilder(includeExec) b.ordinal = startOrdinal b.codexCursorState = seed var fallbackErr error consumed, err := readLines( func(line string) { if fallbackErr != nil { return } // Skip session_meta — already processed in // the initial full parse. if gjson.Get(line, "type").Str == codexTypeSessionMeta { return } if codexIncrementalNeedsFullParse(line) { fallbackErr = errCodexIncrementalNeedsFullParse return } b.processLine(line) if b.unattachedTokenUsage { fallbackErr = errCodexIncrementalNeedsFullParse return } }, ) if err != nil { return codexIncrementalParseResult{}, fmt.Errorf( "reading codex %s from offset %d: %w", path, offset, err, ) } if fallbackErr != nil { return codexIncrementalParseResult{}, fallbackErr } b.flushPendingAgentResults() result := codexIncrementalParseResult{ messages: b.messages, endedAt: b.endedAt, consumedBytes: consumed, initialCursor: seed, cursor: b.incrementalSeed(), inode: inode, device: device, } return result, nil } // parseSessionFrom preserves the legacy test-helper and parser signature while // the provider facade consumes the detailed cursor result internally. func (p *codexProvider) parseSessionFrom( path string, offset int64, startOrdinal int, includeExec bool, ) ([]ParsedMessage, time.Time, int64, error) { result, err := p.parseSessionFromWithReader( path, offset, startOrdinal, includeExec, readJSONLFrom, ) if err != nil { return nil, time.Time{}, 0, err } return result.messages, result.endedAt, result.consumedBytes, nil } // IsIncrementalFullParseFallback reports whether an incremental // parse error requires the caller to fall back to a full parse. func IsIncrementalFullParseFallback(err error) bool { return errors.Is(err, errCodexIncrementalNeedsFullParse) || errors.Is(err, ErrClaudeIncrementalNeedsFullParse) } func isCodexSystemMessage(content string) bool { trimmed := strings.TrimSpace(content) return strings.HasPrefix(content, "# AGENTS.md") || strings.HasPrefix(content, "") || strings.HasPrefix(content, "") || isCodexTurnAbortedMessage(content) || strings.HasPrefix(trimmed, "") || isCodexSubagentNotification(content) || isCodexGoalContext(content) } // stripCodexRecommendedPlugins removes the plugin-discovery envelope // that recent Codex versions prepend to the synthetic context item at the // start of a session. It is called only while looking for the first genuine // user turn, so a later user message that quotes the envelope is preserved. func stripCodexRecommendedPlugins(content string) string { const ( openTag = "" closeTag = "" ) start := strings.Index(content, openTag) if start < 0 { return content } relativeEnd := strings.Index(content[start+len(openTag):], closeTag) if relativeEnd < 0 { return content } end := start + len(openTag) + relativeEnd + len(closeTag) prefix := content[:start] suffix := strings.TrimLeft(content[end:], "\r\n") if strings.TrimSpace(prefix) == "" { return suffix } return prefix + suffix } // stripCodexInitialSystemPrefix removes complete synthetic envelopes from the // start of an initial text block. A genuine prompt may follow an injected // envelope in the same block, so only text through the first matching close // tag is removed. func stripCodexInitialSystemPrefix(content string) string { for { trimmed := strings.TrimLeft(content, "\r\n") var closeTag string switch { case strings.HasPrefix(trimmed, "# AGENTS.md"): closeTag = "" case strings.HasPrefix(trimmed, ""): closeTag = "" case strings.HasPrefix(trimmed, ""): closeTag = "" default: return content } _, after, ok := strings.Cut(trimmed, closeTag) if !ok { return content } content = strings.TrimLeft(after, "\r\n") } } // isCodexGoalContext reports whether content is a Codex /goal // continuation envelope. These are harness-injected as role=user // records to keep the model working toward an active thread goal, but // they are not user-authored turns and should be treated as system // content. Current sessions wrap the body in // ; older sessions used // . Detection is scoped to the structured wrapper (and, // for the modern form, the goal source specifically) so that other // internal-context envelopes and real user messages quoting the goal // text are left untouched. func isCodexGoalContext(content string) bool { trimmed := strings.TrimSpace(content) if strings.HasPrefix(trimmed, "") { return true } if strings.HasPrefix(trimmed, "") return ok && codexGoalContextSourceAttrRe.MatchString(openTag) } return false } func isCodexTurnAbortedMessage(content string) bool { return strings.HasPrefix( strings.TrimSpace(content), "", ) } func isCodexSubagentNotification(content string) bool { return strings.HasPrefix( strings.TrimSpace(content), "", ) } func codexIncrementalNeedsFullParse(line string) bool { switch gjson.Get(line, "type").Str { case codexTypeEventMsg: return gjson.Get(line, "payload.type").Str == "collab_agent_spawn_end" case codexTypeResponseItem: default: return false } payload := gjson.Get(line, "payload") switch payload.Get("type").Str { case "function_call", "custom_tool_call": return isCodexWaitAgentCall(payload.Get("name").Str) case "function_call_output", "custom_tool_call_output": output, raw := parseCodexFunctionOutput(payload) return isCodexSubagentFunctionOutput(output) || strings.TrimSpace(raw) != "" default: role := payload.Get("role").Str if role != "user" { return false } agentID, _, text := parseCodexSubagentNotification( extractCodexContent(payload), ) return agentID != "" && text != "" } }