package db import ( "context" "crypto/sha256" "database/sql" "encoding/hex" "encoding/json" "fmt" "sort" "strings" "time" corerecall "go.kenn.io/agentsview/internal/recall" ) // defaultEvalChunkChars is the rune-window size for splitting a flattened // trajectory into raw-chunk entries (~500 tokens). Finer than extraction // chunk's 12000-char default so keyword recall can pinpoint passages. It is a // server-side tuning knob, deliberately a constant rather than a request field. const defaultEvalChunkChars = 2000 // maxEvalFieldRunes caps the length of the required identifier-like eval // ingest fields (extractor_method, source_version) so a misbehaving caller // cannot stash arbitrarily large text in columns meant for short labels. const maxEvalFieldRunes = 200 const ( evalTrajectoryMachine = "recall-eval-ingest" defaultEvalTrajectoryProject = "eval-harness" defaultEvalTrajectoryAgent = "eval-harness" ) // EvalTrajectoryIngest is the input to IngestEvalTrajectory: one raw eval // trajectory plus its run scope, harness identity, and optional provenance // fields. ExtractorMethod and SourceVersion are caller-supplied so this // generic raw-chunk ingest path carries no knowledge of any specific // benchmark or harness. type EvalTrajectoryIngest struct { RunID string TrajectoryID string Trajectory json.RawMessage ExtractorMethod string SourceVersion string Project string CWD string GitBranch string Agent string } // EvalTrajectoryIngestResult reports how many chunk entries were newly // inserted for the trajectory. An idempotent re-ingest reports 0. type EvalTrajectoryIngestResult struct { RunID string `json:"run_id"` TrajectoryID string `json:"trajectory_id"` CorpusID string `json:"corpus_id,omitempty"` EntriesIndexed int `json:"entries_indexed"` } // IngestEvalTrajectory chunks a raw eval trajectory into FTS-indexed recall // rows scoped by run_id + extractor_method, so the keyword retriever can // recall them. It is lab-only (the HTTP layer guards the // production data dir) and idempotent: deterministic ids mean re-ingesting a // trajectory with the same extractor, source version, and normalized content // inserts only the chunks it is missing. Changing any of those identity inputs // creates a distinct eval corpus instead of silently retaining stale rows. It // mirrors the /import write path — a placeholder session satisfies the // source_session_id FK, then each chunk is inserted only if absent. func (db *DB) IngestEvalTrajectory( ctx context.Context, in EvalTrajectoryIngest, ) (EvalTrajectoryIngestResult, error) { in = normalizeEvalTrajectoryIngest(in) result := EvalTrajectoryIngestResult{ RunID: in.RunID, TrajectoryID: in.TrajectoryID, } if in.RunID == "" || in.TrajectoryID == "" { return result, fmt.Errorf("run_id and trajectory_id are required") } if err := validateEvalRequiredField("extractor_method", in.ExtractorMethod); err != nil { return result, err } if err := validateEvalRequiredField("source_version", in.SourceVersion); err != nil { return result, err } if len(in.Trajectory) == 0 { return result, fmt.Errorf("trajectory is required") } text, err := flattenTrajectoryText(in.Trajectory) if err != nil { return result, err } chunks := chunkText(text, defaultEvalChunkChars) if len(chunks) == 0 { return result, nil } contentDigest := evalTrajectoryContentDigest(text) sessionID, err := evalTrajectorySessionID(in, contentDigest) if err != nil { return result, err } result.CorpusID = sessionID entries := make([]RecallEntry, 0, len(chunks)) for idx, chunk := range chunks { id, err := evalTrajectoryChunkID(in, contentDigest, idx) if err != nil { return result, err } entries = append(entries, newEvalChunkRecallEntry( id, sessionID, in, idx, len(chunks), chunk, )) } indexed, err := db.ingestEvalTrajectoryChunks( ctx, newEvalTrajectorySession(sessionID, in), entries, ) if err != nil { return result, err } result.EntriesIndexed = indexed return result, nil } func normalizeEvalTrajectoryIngest( in EvalTrajectoryIngest, ) EvalTrajectoryIngest { in.RunID = strings.TrimSpace(in.RunID) in.TrajectoryID = strings.TrimSpace(in.TrajectoryID) in.ExtractorMethod = strings.TrimSpace(in.ExtractorMethod) in.SourceVersion = strings.TrimSpace(in.SourceVersion) in.Project = strings.TrimSpace(in.Project) in.CWD = strings.TrimSpace(in.CWD) in.GitBranch = strings.TrimSpace(in.GitBranch) in.Agent = strings.TrimSpace(in.Agent) if in.Project == "" { in.Project = defaultEvalTrajectoryProject } if in.Agent == "" { in.Agent = defaultEvalTrajectoryAgent } return in } // validateEvalRequiredField reports an error if value is empty or exceeds // maxEvalFieldRunes once trimmed. value must already be trimmed (see // normalizeEvalTrajectoryIngest, which runs before this check). func validateEvalRequiredField(field, value string) error { if value == "" { return fmt.Errorf("%s is required", field) } if len([]rune(value)) > maxEvalFieldRunes { return fmt.Errorf("%s exceeds maximum length of %d characters", field, maxEvalFieldRunes) } return nil } // flattenTrajectoryText walks an arbitrary trajectory JSON value and returns // all of its string leaves joined by newlines. Object keys are visited in // sorted order and arrays in their natural order, so the same trajectory always // flattens to byte-identical text — which keeps chunk ids, and thus // idempotency, stable. Keys are structural and skipped; non-string scalars // (numbers, booleans, null) carry no searchable text and are skipped too. func flattenTrajectoryText(raw json.RawMessage) (string, error) { var root any if err := json.Unmarshal(raw, &root); err != nil { return "", fmt.Errorf("parsing trajectory: %w", err) } var b strings.Builder appendTrajectoryStrings(&b, root) return b.String(), nil } func appendTrajectoryStrings(b *strings.Builder, node any) { switch n := node.(type) { case string: s := strings.TrimSpace(n) if s == "" { return } if b.Len() > 0 { b.WriteByte('\n') } b.WriteString(s) case []any: for _, item := range n { appendTrajectoryStrings(b, item) } case map[string]any: keys := make([]string, 0, len(n)) for k := range n { keys = append(keys, k) } sort.Strings(keys) for _, k := range keys { appendTrajectoryStrings(b, n[k]) } } } // chunkText splits s into windows of at most size runes (not bytes, so // multi-byte characters are never split mid-rune). An empty string yields no // chunks. A non-positive size falls back to defaultEvalChunkChars. func chunkText(s string, size int) []string { if size <= 0 { size = defaultEvalChunkChars } if s == "" { return nil } runes := []rune(s) chunks := make([]string, 0, (len(runes)+size-1)/size) for start := 0; start < len(runes); start += size { end := min(start+size, len(runes)) chunks = append(chunks, string(runes[start:end])) } return chunks } // evalIngestID derives a deterministic, collision-safe id as the SHA-256 hex of // the JSON-encoded tuple of parts (e.g. ["eval-trajectory", runID, trajID, // idx]). JSON length-delimits the parts, so ids cannot collide through // delimiter confusion the way raw string concatenation could. func evalIngestID(parts ...any) (string, error) { encoded, err := json.Marshal(parts) if err != nil { return "", fmt.Errorf("encoding eval id parts: %w", err) } sum := sha256.Sum256(encoded) return hex.EncodeToString(sum[:]), nil } func evalTrajectoryContentDigest(text string) string { sum := sha256.Sum256([]byte(text)) return hex.EncodeToString(sum[:]) } func evalTrajectorySessionID( in EvalTrajectoryIngest, contentDigest string, ) (string, error) { return evalIngestID( "eval-trajectory-session", in.RunID, in.TrajectoryID, in.ExtractorMethod, in.SourceVersion, contentDigest, ) } func evalTrajectoryChunkID( in EvalTrajectoryIngest, contentDigest string, idx int, ) (string, error) { return evalIngestID( "eval-trajectory", in.RunID, in.TrajectoryID, in.ExtractorMethod, in.SourceVersion, contentDigest, idx, ) } // newEvalTrajectorySession builds the placeholder session required by the // chunk entries' source_session_id foreign key. func newEvalTrajectorySession( sessionID string, in EvalTrajectoryIngest, ) Session { now := time.Now().UTC().Format(time.RFC3339Nano) firstMessage := "Eval trajectory " + in.TrajectoryID displayName := firstMessage return Session{ ID: sessionID, Project: in.Project, Machine: evalTrajectoryMachine, Agent: in.Agent, FirstMessage: &firstMessage, DisplayName: &displayName, StartedAt: &now, EndedAt: &now, MessageCount: 0, UserMessageCount: 0, Outcome: "unknown", OutcomeConfidence: "low", Cwd: in.CWD, GitBranch: in.GitBranch, SourceVersion: in.SourceVersion, } } func (db *DB) ingestEvalTrajectoryChunks( ctx context.Context, session Session, entries []RecallEntry, ) (int, error) { db.mu.Lock() defer db.mu.Unlock() tx, err := db.getWriter().BeginTx(ctx, nil) if err != nil { return 0, fmt.Errorf("begin eval trajectory ingest: %w", err) } defer func() { _ = tx.Rollback() }() if err := validateRecallImportPlaceholderSessionStateWithQueryer( ctx, tx, session.ID, ); err != nil { return 0, fmt.Errorf("preparing eval session: %w", err) } if _, err := tx.ExecContext( ctx, insertSessionIfAbsentSQL, upsertSessionArgs(session)..., ); err != nil { return 0, fmt.Errorf("preparing eval session: %w", err) } indexed := 0 for idx, entry := range entries { inserted, err := insertEvalRecallEntryIfAbsentTx(ctx, tx, entry) if err != nil { return 0, fmt.Errorf("inserting chunk %d: %w", idx, err) } if inserted { indexed++ } } if err := tx.Commit(); err != nil { return 0, fmt.Errorf("commit eval trajectory ingest: %w", err) } return indexed, nil } func insertEvalRecallEntryIfAbsentTx( ctx context.Context, tx *sql.Tx, entry RecallEntry, ) (bool, error) { result, err := tx.ExecContext(ctx, ` INSERT OR IGNORE INTO recall_entries ( id, type, scope, status, review_state, title, body, trigger, confidence, uncertainty, project, cwd, git_branch, agent, source_session_id, source_episode_id, source_run_id, extractor_method, model, transferable, provenance_ok, supersedes_entry_id, superseded_by_entry_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, entry.ID, entry.Type, entry.Scope, entry.Status, entry.ReviewState, entry.Title, entry.Body, entry.Trigger, sqlFloat(entry.Confidence), entry.Uncertainty, entry.Project, entry.CWD, entry.GitBranch, entry.Agent, entry.SourceSessionID, entry.SourceEpisodeID, entry.SourceRunID, entry.ExtractorMethod, entry.Model, entry.Transferable, entry.ProvenanceOK, entry.SupersedesEntryID, entry.SupersededByEntryID, ) if err != nil { return false, fmt.Errorf("inserting recall entry: %w", err) } rows, err := result.RowsAffected() if err != nil { return false, fmt.Errorf("counting inserted recall entry: %w", err) } return rows == 1, nil } // newEvalChunkRecallEntry builds the raw-chunk recall row for one chunk. // Confidence is left nil (raw chunks earn no confidence ranking bonus). // Transferability and provenance remain measurable, while eval_raw keeps these // rows quarantined from trusted-only recall. func newEvalChunkRecallEntry( id, sessionID string, in EvalTrajectoryIngest, idx, total int, body string, ) RecallEntry { return RecallEntry{ ID: id, Type: corerecall.TypeFact, Scope: corerecall.ScopeRepository, Status: corerecall.StatusAccepted, ReviewState: corerecall.ReviewStateEvalRaw, Title: fmt.Sprintf("%s [chunk %d/%d]", in.TrajectoryID, idx+1, total), Body: body, Project: in.Project, CWD: in.CWD, GitBranch: in.GitBranch, Agent: in.Agent, SourceSessionID: sessionID, SourceEpisodeID: fmt.Sprintf("%s:chunk:%d", in.TrajectoryID, idx), SourceRunID: in.RunID, ExtractorMethod: in.ExtractorMethod, Transferable: true, ProvenanceOK: true, } }