fix(evidence): skip oversized JSONL lines instead of aborting the import

A codex rollout with a single line past bufio.Scanner's 10MB limit killed
the whole import-codex step with 'token too long' (miseledger-refresh has
failed on it daily since 2026-07-14). Replace the scanners in the shared
JSONL walker and the adapter importer with a bounded line reader that
drains oversized lines, records a per-line warning, and keeps importing
the rest of the file.

The two interrupted-import tests used an 11MB line as their interruption
mechanism, which relied on the old aborting behavior; they now interrupt
with a mid-stream read error and keep the same assertions.

Author: solomonneas

engines/evidence-ledger/internal/ingest/importer.go

@@ -1,7 +1,6 @@
 package ingest
 
 import (
-	"bufio"
 	"compress/gzip"
 	"crypto/sha256"
 	"database/sql"
@@ -113,8 +112,6 @@ func importAdapterReaderProgress(db *sql.DB, r io.Reader, sourcePath, sourceOver
 
 	result := AdapterResult{SourceKind: sourceKind, SourcePath: sourcePath}
 	h := sha256.New()
-	scanner := bufio.NewScanner(r)
-	scanner.Buffer(make([]byte, 0, 64*1024), 10*1024*1024)
 	ordinal := int64(0)
 	warnedOverrideKinds := map[string]bool{}
 
@@ -145,38 +142,39 @@ func importAdapterReaderProgress(db *sql.DB, r io.Reader, sourcePath, sourceOver
 		return nil
 	}
 
-	for scanner.Scan() {
-		line := append([]byte(nil), scanner.Bytes()...)
+	scanErr := sources.EachLine(r, sources.MaxLineBytes, func(raw []byte, tooLong bool, size int64) error {
 		ordinal++
+		if tooLong {
+			result.Warnings = append(result.Warnings, fmt.Sprintf("line %d: line too long (%d bytes > %d limit), skipped", ordinal, size, sources.MaxLineBytes))
+			return nil
+		}
+		line := append([]byte(nil), raw...)
 		if len(strings.TrimSpace(string(line))) == 0 {
-			continue
+			return nil
 		}
 		if recordScan != nil {
 			file, ok, err := parseSourceScanSentinel(line)
 			if err != nil {
-				return AdapterResult{}, err
+				return err
 			}
 			if ok {
 				if sourceKind == "" {
 					sourceKind = "adapter"
 					result.SourceKind = sourceKind
 				}
 				if err := flush(batchCount > 0); err != nil {
-					return AdapterResult{}, err
+					return err
 				}
 				generatedHash := "sha256:" + hex.EncodeToString(h.Sum(nil))
-				if err := recordScan(sourceKind, generatedHash, file); err != nil {
-					return AdapterResult{}, err
-				}
-				continue
+				return recordScan(sourceKind, generatedHash, file)
 			}
 		}
 		_, _ = h.Write(line)
 		_, _ = h.Write([]byte("\n"))
 		rec, err := adapter.Parse(line)
 		if err != nil {
 			result.Warnings = append(result.Warnings, fmt.Sprintf("line %d: %s", ordinal, err))
-			continue
+			return nil
 		}
 		embeddedKind := rec.Source.Kind
 		if sourceOverride != "" && embeddedKind != "" && embeddedKind != sourceOverride && !warnedOverrideKinds[embeddedKind] {
@@ -193,20 +191,21 @@ func importAdapterReaderProgress(db *sql.DB, r io.Reader, sourcePath, sourceOver
 		inserted, err := upsertRecord(tx, rec, sourcePath, ordinal, line)
 		if err != nil {
 			result.Warnings = append(result.Warnings, fmt.Sprintf("line %d: %s", ordinal, err))
-			continue
+			return nil
 		}
 		if inserted {
 			result.Inserted++
 		}
 		batchCount++
 		if batchCount >= importBatchSize {
 			if err := flush(true); err != nil {
-				return AdapterResult{}, err
+				return err
 			}
 		}
-	}
-	if err := scanner.Err(); err != nil {
-		return AdapterResult{}, err
+		return nil
+	})
+	if scanErr != nil {
+		return AdapterResult{}, scanErr
 	}
 	sourceHash := "sha256:" + hex.EncodeToString(h.Sum(nil))
 	if sourceKind == "" {

engines/evidence-ledger/internal/ingest/importer_test.go

@@ -3,10 +3,13 @@ package ingest
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"strconv"
 	"strings"
 	"testing"
+	"testing/iotest"
 
 	"github.com/escoffier-labs/miseledger/internal/archive"
 	"github.com/escoffier-labs/miseledger/internal/sources"
@@ -46,6 +49,42 @@ func TestImportAdapterReaderIdempotent(t *testing.T) {
 	}
 }
 
+// Regression: an adapter line beyond the 10MB scanner limit used to abort the
+// whole import with bufio.Scanner: token too long. It must be skipped with a
+// warning while surrounding records still import.
+func TestImportAdapterReaderSkipsOversizedLine(t *testing.T) {
+	db, err := archive.Open(t.TempDir() + "/miseledger.db")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if err := archive.Migrate(db); err != nil {
+		t.Fatal(err)
+	}
+	record := func(id, text string) string {
+		return `{"schema":"miseledger.adapter.v1","source":{"kind":"oversize-test","name":"Oversize Test"},"collection":{"external_id":"oversize:collection","kind":"agent_session","name":"oversize"},"item":{"external_id":"oversize:item:` + id + `","kind":"message","created_at":"2026-07-14T00:00:00Z","text":"` + text + `","tags":["oversize"]},"actor":{"external_id":"oversize:actor","type":"human","name":"reader"},"artifacts":[],"links":[],"relations":[],"raw":{"format":"json","path":"oversize.jsonl","ordinal":1}}`
+	}
+	jsonl := record("1", "before") + "\n" +
+		record("huge", strings.Repeat("a", sources.MaxLineBytes+1024)) + "\n" +
+		record("2", "after") + "\n"
+	res, err := ImportAdapterReader(db, strings.NewReader(jsonl), "oversize://fixture", "oversize-test")
+	if err != nil {
+		t.Fatalf("import must not abort on an oversized line: %v", err)
+	}
+	if res.Inserted != 2 {
+		t.Fatalf("inserted = %d, want 2", res.Inserted)
+	}
+	var warned bool
+	for _, w := range res.Warnings {
+		if strings.Contains(w, "line too long") {
+			warned = true
+		}
+	}
+	if !warned {
+		t.Fatalf("expected a line-too-long warning, got %d warnings", len(res.Warnings))
+	}
+}
+
 func TestImportAdapterReaderPreservesSchemaConformantCodeReferencesInItemMetadata(t *testing.T) {
 	db, err := archive.Open(t.TempDir() + "/miseledger.db")
 	if err != nil {
@@ -282,9 +321,12 @@ func TestNativeReaderRecordsCommittedFileScanBeforeInterruptedImport(t *testing.
 	}); err != nil {
 		t.Fatal(err)
 	}
-	b.WriteString(strings.Repeat("x", 11*1024*1024))
+	// Interrupt the stream with a read error after the sentinel. (This used to
+	// be an 11MB line tripping bufio.Scanner's limit; oversized lines are now
+	// skipped with a warning instead of aborting, so they no longer interrupt.)
+	interrupted := io.MultiReader(strings.NewReader(b.String()), iotest.ErrReader(errors.New("simulated interruption")))
 
-	_, err = ImportNativeReaderProgress(db, strings.NewReader(b.String()), "native://fixture", "native-scan", nil, func(sourceKind, generatedHash string, file sources.FileScan) error {
+	_, err = ImportNativeReaderProgress(db, interrupted, "native://fixture", "native-scan", nil, func(sourceKind, generatedHash string, file sources.FileScan) error {
 		return RecordSourceScans(db, sourceKind, generatedHash, []sources.FileScan{file}, true)
 	})
 	if err == nil {
@@ -315,8 +357,13 @@ func TestNativeReaderDoesNotRecordScanForUncommittedFile(t *testing.T) {
 	if err := archive.Migrate(db); err != nil {
 		t.Fatal(err)
 	}
-	stream := adapterRecord("native-uncommitted", "file1:item:1", "uncommitted file record", "file1.jsonl", 1) + strings.Repeat("x", 11*1024*1024)
-	_, err = ImportNativeReaderProgress(db, strings.NewReader(stream), "native://fixture", "native-uncommitted", nil, func(sourceKind, generatedHash string, file sources.FileScan) error {
+	// Interrupt with a read error before any file-complete sentinel arrives
+	// (formerly an 11MB line tripping bufio.Scanner's limit).
+	stream := io.MultiReader(
+		strings.NewReader(adapterRecord("native-uncommitted", "file1:item:1", "uncommitted file record", "file1.jsonl", 1)),
+		iotest.ErrReader(errors.New("simulated interruption")),
+	)
+	_, err = ImportNativeReaderProgress(db, stream, "native://fixture", "native-uncommitted", nil, func(sourceKind, generatedHash string, file sources.FileScan) error {
 		return RecordSourceScans(db, sourceKind, generatedHash, []sources.FileScan{file}, true)
 	})
 	if err == nil {

engines/evidence-ledger/internal/sources/codex/codex_test.go

@@ -107,6 +107,38 @@ func TestGenerateMalformedAndUnknownInput(t *testing.T) {
 	}
 }
 
+// Regression: a rollout file with a single line beyond the 10MB scanner limit
+// used to abort the whole codex import with bufio.Scanner: token too long.
+// The oversized line must be skipped with a warning while every other line in
+// the same file (and the rest of the tree) still imports.
+func TestGenerateSkipsOversizedLine(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "rollout-oversized.jsonl")
+	huge := `{"type":"event_msg","timestamp":"2026-07-14T19:29:49Z","payload":{"session_id":"s","role":"assistant","message":"` +
+		strings.Repeat("a", sources.MaxLineBytes+1024) + `"}}`
+	content := strings.Join([]string{
+		`{"type":"event_msg","timestamp":"2026-07-14T19:29:48Z","payload":{"session_id":"s","role":"user","message":"before the oversized line"}}`,
+		huge,
+		`{"type":"event_msg","timestamp":"2026-07-14T19:29:50Z","payload":{"session_id":"s","role":"assistant","message":"after the oversized line"}}`,
+	}, "\n") + "\n"
+	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	recs, res := parseRecords(t, path, sources.Options{})
+	if len(recs) != 2 {
+		t.Fatalf("records = %d, want the two normal lines to import", len(recs))
+	}
+	var warned bool
+	for _, w := range res.Warnings {
+		if strings.Contains(w, "line too long") {
+			warned = true
+		}
+	}
+	if !warned {
+		t.Fatalf("expected a line-too-long warning, got %v", res.Warnings)
+	}
+}
+
 func TestGenerateMissingPathErrors(t *testing.T) {
 	var buf bytes.Buffer
 	if _, err := Generate(filepath.Join(t.TempDir(), "nope.jsonl"), sources.Options{}, &buf); err == nil {

engines/evidence-ledger/internal/sources/lines_test.go

@@ -0,0 +1,87 @@
+package sources
+
+import (
+	"strings"
+	"testing"
+)
+
+type collectedLine struct {
+	line    string
+	tooLong bool
+	size    int64
+}
+
+func collectLines(t *testing.T, input string, max int) []collectedLine {
+	t.Helper()
+	var out []collectedLine
+	err := EachLine(strings.NewReader(input), max, func(line []byte, tooLong bool, size int64) error {
+		out = append(out, collectedLine{line: string(line), tooLong: tooLong, size: size})
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("EachLine: %v", err)
+	}
+	return out
+}
+
+func TestEachLineSkipsOversizedLineAndContinues(t *testing.T) {
+	long := strings.Repeat("x", 100)
+	got := collectLines(t, "before\n"+long+"\nafter\n", 10)
+	if len(got) != 3 {
+		t.Fatalf("lines = %d, want 3: %+v", len(got), got)
+	}
+	if got[0].line != "before" || got[0].tooLong {
+		t.Fatalf("line 1 = %+v", got[0])
+	}
+	if !got[1].tooLong || got[1].line != "" || got[1].size != 100 {
+		t.Fatalf("oversized line = %+v, want tooLong with size 100", got[1])
+	}
+	if got[2].line != "after" || got[2].tooLong {
+		t.Fatalf("line 3 = %+v, oversized line must not poison later lines", got[2])
+	}
+}
+
+// A line far larger than the read buffer must be drained chunk by chunk
+// without buffering it whole.
+func TestEachLineOversizedSpansManyReadChunks(t *testing.T) {
+	long := strings.Repeat("y", 300*1024)
+	got := collectLines(t, long+"\nok\n", 1024)
+	if len(got) != 2 {
+		t.Fatalf("lines = %d, want 2", len(got))
+	}
+	if !got[0].tooLong || got[0].size != int64(len(long)) {
+		t.Fatalf("oversized line = %+v", got[0])
+	}
+	if got[1].line != "ok" {
+		t.Fatalf("line after oversized = %+v", got[1])
+	}
+}
+
+func TestEachLineFinalLineWithoutNewline(t *testing.T) {
+	got := collectLines(t, "a\nb", 10)
+	if len(got) != 2 || got[1].line != "b" || got[1].tooLong {
+		t.Fatalf("lines = %+v", got)
+	}
+	// Oversized final line without a terminator is still reported.
+	got = collectLines(t, "a\n"+strings.Repeat("z", 20), 10)
+	if len(got) != 2 || !got[1].tooLong || got[1].size != 20 {
+		t.Fatalf("lines = %+v", got)
+	}
+}
+
+func TestEachLineStripsCarriageReturns(t *testing.T) {
+	got := collectLines(t, "a\r\nb\r", 10)
+	if len(got) != 2 || got[0].line != "a" || got[1].line != "b" {
+		t.Fatalf("lines = %+v", got)
+	}
+}
+
+func TestEachLineBlankAndEmptyInput(t *testing.T) {
+	if got := collectLines(t, "", 10); len(got) != 0 {
+		t.Fatalf("empty input produced %+v", got)
+	}
+	got := collectLines(t, "\n\n", 10)
+	if len(got) != 2 || got[0].line != "" || got[1].line != "" {
+		t.Fatalf("blank lines = %+v", got)
+	}
+}