mirror of
https://github.com/boshu2/agentops.git
synced 2026-09-14 15:08:13 +08:00
66323acc64
Session mining could overwrite its source when the checkpoint referred to the same file, consume pending events during `--dry-run`, and miss native Codex custom-tool input edits when checking for transcript rewrites. The CLI now rejects source/checkpoint aliases before output or writes, previews from the current watermark without persisting it, and preserves native custom input so edits invalidate the checkpoint. Regression controls retain normal incremental mining and legacy argument handling. Postmortem now accepts real goal/session evidence without requiring a fabricated verdict file, uses current Memory ownership and protected external storage, and separates structural checks from causal support. Implement guidance emphasizes early representative checks and exact repair verification. The Go standard and its local rule now point to the actual CI complexity gate and explain its committed-file scope. Normal projections are regenerated; no skill roots, framework or gate thresholds were added or weakened. The bounded evaluation used three useful coding tasks with fixed behavior acceptance. Trial two exposed a real workflow defect: lint and fast checks passed while CI rejected complexity 25. Checkpoint construction was extracted, reducing MineSession to 20, and the Go guidance was corrected before the third task. The failed attempt stays in the evidence; this is an observational result, not proof of token savings or superiority over native prompting. Final local Go build/vet/test, lint, strict whole-campaign complexity and 37 selected gates pass. [CI run 34628874411](https://github.com/boshu2/agentops/actions/runs/34628874411) is green on `62205367e91946f84723147763662ca39ad4b580`, including Linux race/shuffle tests, Windows, Bats and security. The independent review checked all 27 changed paths and returned PASS with no unresolved findings or unchecked acceptance; its earlier complexity FAIL remains preserved. The local aggregate retains its existing skip for the absent legacy OL test directory.
809 lines
22 KiB
Go
809 lines
22 KiB
Go
// Package parser provides streaming JSONL parsing for Claude and Codex transcripts.
|
|
package parser
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/boshu2/agentops/cli/internal/types"
|
|
)
|
|
|
|
// DefaultMaxContentLength is the default truncation limit for content fields.
|
|
const DefaultMaxContentLength = 500
|
|
|
|
// Message type constants for transcript entries.
|
|
const (
|
|
msgTypeUser = "user"
|
|
msgTypeAssistant = "assistant"
|
|
msgTypeToolUse = "tool_use"
|
|
msgTypeToolResult = "tool_result"
|
|
// msgTypeCodexTokenCount is an internal sentinel: a Codex token_count event
|
|
// folded into ParseResult.FinalUsage rather than the message list.
|
|
msgTypeCodexTokenCount = "codex_token_count" // #nosec G101 nosemgrep -- message-type sentinel string, not a credential.
|
|
)
|
|
|
|
// Error classification constants for parse errors.
|
|
const (
|
|
errClassJSON = "json"
|
|
errClassSchema = "schema"
|
|
errClassEncoding = "encoding"
|
|
)
|
|
|
|
// Parser handles streaming JSONL parsing with configurable options.
|
|
type Parser struct {
|
|
// MaxContentLength is the maximum characters before truncation.
|
|
MaxContentLength int
|
|
|
|
// SkipMalformed skips malformed lines instead of erroring.
|
|
SkipMalformed bool
|
|
|
|
// OnProgress is called with progress updates for large files.
|
|
OnProgress func(linesProcessed, totalLines int)
|
|
}
|
|
|
|
// NewParser creates a parser with default settings.
|
|
func NewParser() *Parser {
|
|
return &Parser{
|
|
MaxContentLength: DefaultMaxContentLength,
|
|
SkipMalformed: true,
|
|
}
|
|
}
|
|
|
|
// rawMessage represents the raw JSON structure from supported transcript surfaces.
|
|
type rawMessage struct {
|
|
Type string `json:"type"`
|
|
SessionID string `json:"sessionId"`
|
|
Timestamp string `json:"timestamp"`
|
|
UUID string `json:"uuid"`
|
|
ParentUUID string `json:"parentUuid,omitempty"`
|
|
RequestID string `json:"requestId,omitempty"`
|
|
Role string `json:"role,omitempty"`
|
|
Content any `json:"content,omitempty"`
|
|
ToolName string `json:"tool_name,omitempty"`
|
|
ToolInput map[string]any `json:"tool_input,omitempty"`
|
|
ToolOutput any `json:"tool_output,omitempty"`
|
|
Payload json.RawMessage `json:"payload,omitempty"`
|
|
Message *struct {
|
|
ID string `json:"id,omitempty"`
|
|
Type string `json:"type,omitempty"`
|
|
Role string `json:"role"`
|
|
Content any `json:"content"` // Can be string or array
|
|
Tools []struct {
|
|
Name string `json:"name"`
|
|
Input map[string]any `json:"input"`
|
|
Output any `json:"output"`
|
|
} `json:"tools,omitempty"`
|
|
Usage *types.TokenUsage `json:"usage,omitempty"`
|
|
} `json:"message,omitempty"`
|
|
// ToolUseResult contains structured tool output (e.g., for TodoWrite)
|
|
ToolUseResult any `json:"toolUseResult,omitempty"`
|
|
}
|
|
|
|
type codexSessionMeta struct {
|
|
ID string `json:"id"`
|
|
Timestamp string `json:"timestamp"`
|
|
}
|
|
|
|
type codexEventPayload struct {
|
|
Type string `json:"type"`
|
|
Message string `json:"message"`
|
|
// Info carries the cumulative token accounting on a "token_count" event.
|
|
Info *struct {
|
|
TotalTokenUsage *codexTokenUsage `json:"total_token_usage"`
|
|
} `json:"info,omitempty"`
|
|
}
|
|
|
|
// codexTokenUsage is the Codex transcript's running token total. Unlike Claude
|
|
// (per-message usage), Codex emits a CUMULATIVE total on each token_count event,
|
|
// so the LAST one is the session total. input_tokens already includes cached
|
|
// input, and total_tokens == input_tokens + output_tokens.
|
|
type codexTokenUsage struct {
|
|
InputTokens int `json:"input_tokens"`
|
|
CachedInputTokens int `json:"cached_input_tokens"`
|
|
OutputTokens int `json:"output_tokens"`
|
|
}
|
|
|
|
type codexResponseItem struct {
|
|
Type string `json:"type"`
|
|
Role string `json:"role"`
|
|
Name string `json:"name"`
|
|
Arguments string `json:"arguments"`
|
|
Input *string `json:"input"`
|
|
Output string `json:"output"`
|
|
Content []struct {
|
|
Type string `json:"type"`
|
|
Text string `json:"text"`
|
|
} `json:"content"`
|
|
}
|
|
|
|
// ParseResult contains the result of parsing a JSONL stream.
|
|
type ParseResult struct {
|
|
Messages []types.TranscriptMessage
|
|
TotalLines int
|
|
MalformedLines int
|
|
Errors []error
|
|
|
|
// Checksum is SHA256 hash of the parsed content (first 16 hex chars).
|
|
// Used for detecting changes and validating re-parsing.
|
|
Checksum string
|
|
|
|
// FilePath is the source file path (if parsed from file).
|
|
FilePath string
|
|
|
|
// ParsedAt is when parsing completed.
|
|
ParsedAt time.Time
|
|
|
|
// FinalUsage is the Codex cumulative session token total (the last
|
|
// token_count event). When set, it is authoritative and supersedes the
|
|
// per-message Claude usage sum — see TokenTotals.
|
|
FinalUsage *types.TokenUsage
|
|
}
|
|
|
|
// TokenTotals returns the session's real token footprint, choosing the right
|
|
// aggregation per runtime: Codex emits a cumulative total (FinalUsage, last
|
|
// wins), while Claude emits per-message usage that SumUsage de-dups by response
|
|
// id. Producer truth (age-membrane-memory-arch-tz2s.3.1/.3.2).
|
|
func (r *ParseResult) TokenTotals() (tokensIn, tokensOut int) {
|
|
if r.FinalUsage != nil {
|
|
return r.FinalUsage.TotalInputTokens(), r.FinalUsage.OutputTokens
|
|
}
|
|
return types.SumUsage(r.Messages)
|
|
}
|
|
|
|
// ParseError provides structured error information for transcript parsing failures.
|
|
type ParseError struct {
|
|
Line int `json:"line"`
|
|
Column int `json:"column,omitempty"`
|
|
Message string `json:"message"`
|
|
RawContent string `json:"raw_content,omitempty"`
|
|
ErrorType string `json:"error_type"` // "json", "schema", "encoding"
|
|
}
|
|
|
|
func (e *ParseError) Error() string {
|
|
if e.Column > 0 {
|
|
return fmt.Sprintf("line %d, col %d: %s (%s)", e.Line, e.Column, e.Message, e.ErrorType)
|
|
}
|
|
return fmt.Sprintf("line %d: %s (%s)", e.Line, e.Message, e.ErrorType)
|
|
}
|
|
|
|
// Parse reads JSONL from the reader and returns parsed messages.
|
|
func (p *Parser) Parse(r io.Reader) (*ParseResult, error) {
|
|
result := &ParseResult{
|
|
Messages: make([]types.TranscriptMessage, 0),
|
|
}
|
|
|
|
hasher := sha256.New()
|
|
if err := readJSONLLines(r, func(line []byte, lineNum int) error {
|
|
result.TotalLines = lineNum
|
|
|
|
if len(line) == 0 {
|
|
return nil
|
|
}
|
|
|
|
_, _ = hasher.Write(line)
|
|
_, _ = hasher.Write([]byte("\n"))
|
|
|
|
p.processLine(line, lineNum, result)
|
|
|
|
if p.OnProgress != nil && lineNum%100 == 0 {
|
|
p.OnProgress(lineNum, 0)
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
return result, fmt.Errorf("read jsonl: %w", err)
|
|
}
|
|
|
|
hash := hasher.Sum(nil)
|
|
result.Checksum = hex.EncodeToString(hash[:8])
|
|
result.ParsedAt = time.Now()
|
|
|
|
return result, nil
|
|
}
|
|
|
|
// processLine parses a single JSONL line and appends the result or error.
|
|
func (p *Parser) processLine(line []byte, lineNum int, result *ParseResult) {
|
|
msg, err := p.parseLine(line, lineNum)
|
|
if err != nil {
|
|
result.MalformedLines++
|
|
if !p.SkipMalformed {
|
|
result.Errors = append(result.Errors, &ParseError{
|
|
Line: lineNum,
|
|
Message: err.Error(),
|
|
ErrorType: classifyError(err),
|
|
RawContent: truncateForError(string(line), 100),
|
|
})
|
|
}
|
|
return
|
|
}
|
|
if msg != nil {
|
|
// Codex token_count events carry the CUMULATIVE session total; keep the
|
|
// last one as FinalUsage and never add it to the summed message list.
|
|
if msg.Type == msgTypeCodexTokenCount {
|
|
result.FinalUsage = msg.Usage
|
|
return
|
|
}
|
|
result.Messages = append(result.Messages, *msg)
|
|
}
|
|
}
|
|
|
|
// classifyError determines the error type for structured reporting.
|
|
func classifyError(err error) string {
|
|
errStr := err.Error()
|
|
switch {
|
|
case strings.Contains(errStr, "invalid character"):
|
|
return errClassJSON
|
|
case strings.Contains(errStr, "unexpected end"):
|
|
return errClassJSON
|
|
case strings.Contains(errStr, "cannot unmarshal"):
|
|
return errClassSchema
|
|
case strings.Contains(errStr, "invalid UTF-8"):
|
|
return errClassEncoding
|
|
default:
|
|
return errClassJSON
|
|
}
|
|
}
|
|
|
|
// truncateForError limits error context to a reasonable size.
|
|
func truncateForError(s string, maxLen int) string {
|
|
// Fast path: len(s) is an upper bound on rune count, so any string whose
|
|
// byte length already fits needs no conversion. This avoids a []rune
|
|
// allocation on the common short-ASCII case.
|
|
if len(s) <= maxLen {
|
|
return s
|
|
}
|
|
runes := []rune(s)
|
|
if len(runes) <= maxLen {
|
|
return s
|
|
}
|
|
return string(runes[:maxLen]) + "..."
|
|
}
|
|
|
|
// openFileFunc is the function used to open files for parsing.
|
|
// It can be overridden in tests to inject close errors.
|
|
var openFileFunc = func(path string) (io.ReadCloser, error) {
|
|
return os.Open(path)
|
|
}
|
|
|
|
// ParseFile parses a JSONL file by path.
|
|
func (p *Parser) ParseFile(path string) (result *ParseResult, err error) {
|
|
f, err := openFileFunc(path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open file: %w", err)
|
|
}
|
|
defer func() {
|
|
if cerr := f.Close(); cerr != nil && err == nil {
|
|
err = cerr
|
|
}
|
|
}()
|
|
|
|
result, err = p.Parse(f)
|
|
if result != nil {
|
|
result.FilePath = path
|
|
}
|
|
return result, err
|
|
}
|
|
|
|
// timestampFormats lists the formats to try when parsing timestamps.
|
|
var timestampFormats = []string{
|
|
time.RFC3339,
|
|
"2006-01-02T15:04:05.000Z",
|
|
}
|
|
|
|
// parseTimestamp parses a timestamp string, trying multiple formats.
|
|
// Returns zero time if all formats fail.
|
|
func parseTimestamp(s string) time.Time {
|
|
for _, format := range timestampFormats {
|
|
if ts, err := time.Parse(format, s); err == nil {
|
|
return ts
|
|
}
|
|
}
|
|
return time.Time{}
|
|
}
|
|
|
|
// isValidMessageType checks if the type is one we should parse.
|
|
func isValidMessageType(msgType string) bool {
|
|
switch msgType {
|
|
case msgTypeUser, msgTypeAssistant, msgTypeToolUse, msgTypeToolResult:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// parseContentBlocks extracts text and tool calls from content block array.
|
|
func (p *Parser) parseContentBlocks(blocks []any) (string, []types.ToolCall) {
|
|
var content string
|
|
var tools []types.ToolCall
|
|
|
|
for _, block := range blocks {
|
|
blockMap, ok := block.(map[string]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
text, tool := p.classifyBlock(blockMap)
|
|
content += text
|
|
if tool != nil {
|
|
tools = append(tools, *tool)
|
|
}
|
|
}
|
|
|
|
return content, tools
|
|
}
|
|
|
|
// classifyBlock dispatches a single content block into text or tool call.
|
|
func (p *Parser) classifyBlock(blockMap map[string]any) (string, *types.ToolCall) {
|
|
blockType, _ := blockMap["type"].(string)
|
|
switch blockType {
|
|
case "text":
|
|
if text, ok := blockMap["text"].(string); ok {
|
|
return p.truncate(text), nil
|
|
}
|
|
case msgTypeToolUse:
|
|
return "", p.parseToolUse(blockMap)
|
|
case msgTypeToolResult:
|
|
return "", p.parseToolResult(blockMap)
|
|
}
|
|
return "", nil
|
|
}
|
|
|
|
// parseLine parses a single JSON line.
|
|
func (p *Parser) parseLine(line []byte, lineNum int) (*types.TranscriptMessage, error) {
|
|
var raw rawMessage
|
|
if err := json.Unmarshal(line, &raw); err != nil {
|
|
return nil, fmt.Errorf("invalid JSON: %w", err)
|
|
}
|
|
|
|
switch raw.Type {
|
|
case msgTypeUser, msgTypeAssistant:
|
|
return p.parseClaudeMessage(raw, lineNum), nil
|
|
case msgTypeToolUse:
|
|
return p.parseClaudeToolUse(raw, lineNum), nil
|
|
case msgTypeToolResult:
|
|
return p.parseClaudeToolResult(raw, lineNum), nil
|
|
case "session_meta":
|
|
return p.parseCodexSessionMeta(raw, lineNum)
|
|
case "event_msg":
|
|
return p.parseCodexEvent(raw, lineNum)
|
|
case "response_item":
|
|
return p.parseCodexResponseItem(raw, lineNum)
|
|
default:
|
|
return nil, nil
|
|
}
|
|
}
|
|
|
|
func (p *Parser) parseClaudeMessage(raw rawMessage, lineNum int) *types.TranscriptMessage {
|
|
msg := &types.TranscriptMessage{
|
|
Type: raw.Type,
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
Role: raw.Role,
|
|
}
|
|
|
|
if raw.Message != nil {
|
|
msg.Role = raw.Message.Role
|
|
p.extractMessageContent(raw.Message.Content, msg)
|
|
if raw.Message.Usage != nil {
|
|
msg.Usage = raw.Message.Usage
|
|
}
|
|
// Response identity for usage de-dup: prefer message.id, fall back to
|
|
// the top-level requestId. One response spans several rows that repeat
|
|
// the same usage; SumUsage counts each id once.
|
|
if raw.Message.ID != "" {
|
|
msg.MessageID = raw.Message.ID
|
|
} else if raw.RequestID != "" {
|
|
msg.MessageID = raw.RequestID
|
|
}
|
|
}
|
|
if raw.Content != nil {
|
|
p.extractMessageContent(raw.Content, msg)
|
|
}
|
|
if msg.Role == "" {
|
|
msg.Role = raw.Type
|
|
}
|
|
|
|
return msg
|
|
}
|
|
|
|
func (p *Parser) parseClaudeToolUse(raw rawMessage, lineNum int) *types.TranscriptMessage {
|
|
var tools []types.ToolCall
|
|
if raw.ToolName != "" {
|
|
tools = append(tools, types.ToolCall{
|
|
Name: raw.ToolName,
|
|
Input: raw.ToolInput,
|
|
})
|
|
}
|
|
if raw.Message != nil {
|
|
for _, tool := range raw.Message.Tools {
|
|
if tool.Name == "" {
|
|
continue
|
|
}
|
|
tools = append(tools, types.ToolCall{
|
|
Name: tool.Name,
|
|
Input: tool.Input,
|
|
Output: p.extractTopLevelToolOutput(tool.Output),
|
|
})
|
|
}
|
|
}
|
|
if len(tools) == 0 && raw.Message != nil {
|
|
msg := &types.TranscriptMessage{
|
|
Type: msgTypeToolUse,
|
|
Role: raw.Message.Role,
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
}
|
|
p.extractMessageContent(raw.Message.Content, msg)
|
|
return msg
|
|
}
|
|
if len(tools) == 0 {
|
|
return nil
|
|
}
|
|
return &types.TranscriptMessage{
|
|
Type: msgTypeToolUse,
|
|
Role: msgTypeAssistant,
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
Tools: tools,
|
|
}
|
|
}
|
|
|
|
func (p *Parser) parseClaudeToolResult(raw rawMessage, lineNum int) *types.TranscriptMessage {
|
|
output := p.extractTopLevelToolOutput(raw.ToolOutput)
|
|
if output == "" && raw.ToolUseResult != nil {
|
|
output = p.extractTopLevelToolOutput(raw.ToolUseResult)
|
|
}
|
|
if output == "" && raw.Message != nil && raw.Message.Content != nil {
|
|
output = p.extractTopLevelToolOutput(raw.Message.Content)
|
|
}
|
|
role := raw.Role
|
|
if role == "" && raw.Message != nil {
|
|
role = raw.Message.Role
|
|
}
|
|
return &types.TranscriptMessage{
|
|
Type: msgTypeToolResult,
|
|
Role: coalesce(role, msgTypeAssistant),
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
Tools: []types.ToolCall{{
|
|
Name: coalesce(raw.ToolName, msgTypeToolResult),
|
|
Input: raw.ToolInput,
|
|
Output: output,
|
|
}},
|
|
}
|
|
}
|
|
|
|
func (p *Parser) parseCodexSessionMeta(raw rawMessage, lineNum int) (*types.TranscriptMessage, error) {
|
|
var meta codexSessionMeta
|
|
if err := json.Unmarshal(raw.Payload, &meta); err != nil {
|
|
return nil, fmt.Errorf("invalid session_meta payload: %w", err)
|
|
}
|
|
return &types.TranscriptMessage{
|
|
Type: "session_meta",
|
|
Timestamp: parseTimestamp(coalesce(meta.Timestamp, raw.Timestamp)),
|
|
SessionID: coalesce(meta.ID, raw.SessionID),
|
|
MessageIndex: lineNum,
|
|
}, nil
|
|
}
|
|
|
|
func (p *Parser) parseCodexEvent(raw rawMessage, lineNum int) (*types.TranscriptMessage, error) {
|
|
var payload codexEventPayload
|
|
if err := json.Unmarshal(raw.Payload, &payload); err != nil {
|
|
return nil, fmt.Errorf("invalid event payload: %w", err)
|
|
}
|
|
|
|
var (
|
|
msgType string
|
|
role string
|
|
)
|
|
switch payload.Type {
|
|
case "user_message":
|
|
msgType = msgTypeUser
|
|
role = msgTypeUser
|
|
case "agent_message":
|
|
msgType = msgTypeAssistant
|
|
role = msgTypeAssistant
|
|
case "token_count":
|
|
// Codex's cumulative session total. Surfaced as a sentinel message that
|
|
// processLine folds into ParseResult.FinalUsage (last wins), never into
|
|
// the summed message list.
|
|
if payload.Info == nil || payload.Info.TotalTokenUsage == nil {
|
|
return nil, nil
|
|
}
|
|
usage := payload.Info.TotalTokenUsage
|
|
if usage.InputTokens < 0 || usage.OutputTokens < 0 || usage.CachedInputTokens < 0 || usage.CachedInputTokens > usage.InputTokens {
|
|
return nil, fmt.Errorf("invalid Codex cumulative token counters")
|
|
}
|
|
return &types.TranscriptMessage{
|
|
Type: msgTypeCodexTokenCount,
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
Usage: &types.TokenUsage{
|
|
InputTokens: usage.InputTokens - usage.CachedInputTokens,
|
|
CacheReadInputTokens: usage.CachedInputTokens,
|
|
OutputTokens: usage.OutputTokens,
|
|
},
|
|
}, nil
|
|
default:
|
|
return nil, nil
|
|
}
|
|
|
|
return &types.TranscriptMessage{
|
|
Type: msgType,
|
|
Role: role,
|
|
Content: p.truncate(payload.Message),
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
}, nil
|
|
}
|
|
|
|
func (p *Parser) parseCodexResponseItem(raw rawMessage, lineNum int) (*types.TranscriptMessage, error) {
|
|
var item codexResponseItem
|
|
if err := json.Unmarshal(raw.Payload, &item); err != nil {
|
|
return nil, fmt.Errorf("invalid response_item payload: %w", err)
|
|
}
|
|
|
|
switch item.Type {
|
|
case "message":
|
|
if item.Role != msgTypeUser && item.Role != msgTypeAssistant {
|
|
return nil, nil
|
|
}
|
|
var content strings.Builder
|
|
for _, block := range item.Content {
|
|
switch block.Type {
|
|
case "input_text", "output_text", "text":
|
|
content.WriteString(block.Text)
|
|
}
|
|
}
|
|
return &types.TranscriptMessage{
|
|
Type: item.Role,
|
|
Role: item.Role,
|
|
Content: p.truncate(content.String()),
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
}, nil
|
|
case "function_call", "custom_tool_call":
|
|
input := parseCodexToolInput(item.Arguments)
|
|
if item.Type == "custom_tool_call" && item.Input != nil {
|
|
// Native custom input is literal text, including empty or JSON-like
|
|
// strings. Only calls without it use the legacy arguments decoder.
|
|
input = map[string]any{"raw": *item.Input}
|
|
}
|
|
return &types.TranscriptMessage{
|
|
Type: msgTypeToolUse,
|
|
Role: msgTypeAssistant,
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
Tools: []types.ToolCall{{
|
|
Name: coalesce(item.Name, item.Type),
|
|
Input: input,
|
|
}},
|
|
}, nil
|
|
case "function_call_output", "custom_tool_call_output":
|
|
return &types.TranscriptMessage{
|
|
Type: msgTypeToolResult,
|
|
Role: msgTypeAssistant,
|
|
Timestamp: parseTimestamp(raw.Timestamp),
|
|
SessionID: raw.SessionID,
|
|
MessageIndex: lineNum,
|
|
Tools: []types.ToolCall{{
|
|
Name: msgTypeToolResult,
|
|
Output: p.truncate(item.Output),
|
|
}},
|
|
}, nil
|
|
default:
|
|
return nil, nil
|
|
}
|
|
}
|
|
|
|
// extractMessageContent populates msg.Content and msg.Tools from raw content.
|
|
func (p *Parser) extractMessageContent(rawContent any, msg *types.TranscriptMessage) {
|
|
switch content := rawContent.(type) {
|
|
case string:
|
|
msg.Content = p.truncate(content)
|
|
case []any:
|
|
msg.Content, msg.Tools = p.parseContentBlocks(content)
|
|
}
|
|
}
|
|
|
|
// parseToolUse extracts tool call information from a tool_use block.
|
|
func (p *Parser) parseToolUse(block map[string]any) *types.ToolCall {
|
|
name, _ := block["name"].(string)
|
|
if name == "" {
|
|
return nil
|
|
}
|
|
|
|
toolCall := &types.ToolCall{
|
|
Name: name,
|
|
}
|
|
|
|
// Extract input parameters
|
|
if input, ok := block["input"].(map[string]any); ok {
|
|
toolCall.Input = input
|
|
}
|
|
|
|
return toolCall
|
|
}
|
|
|
|
// parseToolResult extracts tool result information from a tool_result block.
|
|
func (p *Parser) parseToolResult(block map[string]any) *types.ToolCall {
|
|
toolCall := &types.ToolCall{
|
|
Name: msgTypeToolResult,
|
|
}
|
|
|
|
// Check if it's an error result
|
|
if isError, ok := block["is_error"].(bool); ok && isError {
|
|
toolCall.Error = "tool error"
|
|
}
|
|
|
|
toolCall.Output = p.extractToolResultContent(block["content"])
|
|
return toolCall
|
|
}
|
|
|
|
// extractToolResultContent extracts text from a tool_result content field,
|
|
// which may be a plain string or an array of text blocks.
|
|
func (p *Parser) extractToolResultContent(content any) string {
|
|
switch c := content.(type) {
|
|
case string:
|
|
return p.truncate(c)
|
|
case []any:
|
|
var out string
|
|
for _, item := range c {
|
|
if itemMap, ok := item.(map[string]any); ok {
|
|
if text, ok := itemMap["text"].(string); ok {
|
|
out += p.truncate(text)
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
|
|
func (p *Parser) extractTopLevelToolOutput(content any) string {
|
|
switch c := content.(type) {
|
|
case nil:
|
|
return ""
|
|
case string:
|
|
return p.truncate(c)
|
|
case []any:
|
|
return p.extractToolResultContent(c)
|
|
default:
|
|
data, err := json.Marshal(c)
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
return p.truncate(string(data))
|
|
}
|
|
}
|
|
|
|
// truncate limits content to MaxContentLength characters.
|
|
// Slices at rune boundaries to avoid splitting multi-byte UTF-8 sequences.
|
|
func (p *Parser) truncate(s string) string {
|
|
if p.MaxContentLength <= 0 {
|
|
return s
|
|
}
|
|
// Fast path: byte length is an upper bound on rune count.
|
|
if len(s) <= p.MaxContentLength {
|
|
return s
|
|
}
|
|
runes := []rune(s)
|
|
if len(runes) <= p.MaxContentLength {
|
|
return s
|
|
}
|
|
return string(runes[:p.MaxContentLength]) + "... [truncated]"
|
|
}
|
|
|
|
// ParseChannel returns a channel that emits messages as they're parsed.
|
|
// Useful for streaming large files without loading all into memory.
|
|
func (p *Parser) ParseChannel(r io.Reader) (<-chan types.TranscriptMessage, <-chan error) {
|
|
msgCh := make(chan types.TranscriptMessage, 100)
|
|
errCh := make(chan error, 1)
|
|
|
|
go func() {
|
|
defer close(msgCh)
|
|
defer close(errCh)
|
|
p.channelScanner(r, msgCh, errCh)
|
|
}()
|
|
|
|
return msgCh, errCh
|
|
}
|
|
|
|
// processChannelLine parses one scanner line and forwards the result to msgCh/errCh.
|
|
// Returns false if scanning should stop (fatal parse error).
|
|
func (p *Parser) processChannelLine(line []byte, lineNum int, msgCh chan<- types.TranscriptMessage, errCh chan<- error) bool {
|
|
if len(line) == 0 {
|
|
return true
|
|
}
|
|
msg, err := p.parseLine(line, lineNum)
|
|
if err != nil {
|
|
if !p.SkipMalformed {
|
|
errCh <- fmt.Errorf("line %d: %w", lineNum, err)
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
if msg != nil {
|
|
msgCh <- *msg
|
|
}
|
|
return true
|
|
}
|
|
|
|
// channelScanner scans r line by line, sending parsed messages to msgCh and errors to errCh.
|
|
func (p *Parser) channelScanner(r io.Reader, msgCh chan<- types.TranscriptMessage, errCh chan<- error) {
|
|
if err := readJSONLLines(r, func(line []byte, lineNum int) error {
|
|
if !p.processChannelLine(line, lineNum, msgCh, errCh) {
|
|
return errStopChannelScan
|
|
}
|
|
return nil
|
|
}); err != nil && !errors.Is(err, errStopChannelScan) {
|
|
errCh <- fmt.Errorf("read jsonl: %w", err)
|
|
}
|
|
}
|
|
|
|
var errStopChannelScan = errors.New("stop channel scan")
|
|
|
|
func readJSONLLines(r io.Reader, fn func([]byte, int) error) error {
|
|
reader := bufio.NewReader(r)
|
|
lineNum := 0
|
|
|
|
for {
|
|
line, err := reader.ReadBytes('\n')
|
|
if len(line) > 0 {
|
|
lineNum++
|
|
line = bytes.TrimSuffix(line, []byte("\n"))
|
|
line = bytes.TrimSuffix(line, []byte("\r"))
|
|
if callErr := fn(line, lineNum); callErr != nil {
|
|
return callErr
|
|
}
|
|
}
|
|
|
|
if err != nil {
|
|
if errors.Is(err, io.EOF) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
func parseCodexToolInput(raw string) map[string]any {
|
|
if strings.TrimSpace(raw) == "" {
|
|
return nil
|
|
}
|
|
|
|
var data any
|
|
if err := json.Unmarshal([]byte(raw), &data); err != nil {
|
|
return map[string]any{"raw": raw}
|
|
}
|
|
if obj, ok := data.(map[string]any); ok {
|
|
return obj
|
|
}
|
|
return map[string]any{"value": data}
|
|
}
|
|
|
|
func coalesce(values ...string) string {
|
|
for _, value := range values {
|
|
if strings.TrimSpace(value) != "" {
|
|
return value
|
|
}
|
|
}
|
|
return ""
|
|
}
|