diff --git a/docs/architecture/cli.md b/docs/architecture/cli.md index 8a66f45f..90b89688 100644 --- a/docs/architecture/cli.md +++ b/docs/architecture/cli.md @@ -104,8 +104,15 @@ bytes. This keeps them testable and keeps each command file small. spinner, no in-place updates. The scheduler invokes prosa this way, so this is the default for the production install. -The fallback decision lives in `internal/cli/sync.go` and looks at whether -`stderr` is a TTY (not stdout — sync writes its summary to stdout). +The fallback decision lives in `internal/cli/sync.go` and requires both +`stdout` and `stderr` to be TTYs (`IsInteractive` in `internal/cli/term.go`). +Sync writes its summary to `stderr`, keeping `stdout` free for piped output. + +The interactive branch owns the terminal for the duration of the frame, so +it swaps the *scoped* loggers of the pusher and of every importer +(`opts.Logger`) for one that tallies warn-level records; the count is +reported in the summary afterwards. The process-global `slog` default is +never touched. ## Store access diff --git a/docs/architecture/importers.md b/docs/architecture/importers.md index f63a3a81..9835a944 100644 --- a/docs/architecture/importers.md +++ b/docs/architecture/importers.md @@ -49,6 +49,11 @@ type SkipCache interface { decide whether to no-op, parse it, classify its usage, and write into the sink. `opts.Overwrite` bypasses the idempotency short-circuit and the no_usage skip cache (used by `prosa sync --overwrite`). + `opts.Logger` is where this import's diagnostics go; resolve it with + `importerutil.Logger(opts)`, which falls back to `slog.Default()` when + the caller left it nil. The interactive sync path passes a counting + logger so warnings are tallied instead of written into its progress + frame. The `Sink` is implemented by `internal/store` (locally) and by an in-memory fake for tests. Importers never know about SQLite, Postgres, or the server. @@ -227,7 +232,9 @@ The short version: 2. Create `internal/importers//`. Mirror the shape of `internal/importers/claudecode/`. 3. Implement `Walk` and `Import`. Be paranoid about malformed records and - partial files. + partial files: skip blank lines silently, and report malformed records + once per file with a count and the first offending line rather than + once per record. 4. Write parser tests (`parse_test.go`) covering representative records, missing fields, malformed JSON, truncated files, sessions with no turns. @@ -265,3 +272,5 @@ Skill backing the reviewer: - Special-case a single user. If a behavior is needed, document it in `docs/sources/.md` and put it under a flag if it's optional. - Mutate the agent's raw files in any way. +- Call package-level `slog`. Diagnostics go through + `importerutil.Logger(opts)` so the caller decides where they land. diff --git a/docs/cli/motion.md b/docs/cli/motion.md index ffd1672f..35a21c95 100644 --- a/docs/cli/motion.md +++ b/docs/cli/motion.md @@ -130,6 +130,11 @@ The command should finish with a non-zero exit status only when the requested operation cannot complete. Per-file importer errors can be summarized at the end while allowing the rest of the import to proceed. +Warn-level diagnostics are counted rather than printed while the frame is +live — an out-of-band write desyncs the in-place repaint — and the tally +lands in the final summary. Plain and `--json` mode stream them to `stderr` +as they happen. + ## Cancellation `ctrl+c` cancels the progress program and releases the terminal. After diff --git a/docs/cli/rendering-contract.md b/docs/cli/rendering-contract.md index 96380a00..8d7b8d62 100644 --- a/docs/cli/rendering-contract.md +++ b/docs/cli/rendering-contract.md @@ -93,6 +93,12 @@ Use `stderr` for operational context: - cancellation notices; - logs. +Between the first and last frame of an in-place progress render, nothing +writes to `stdout` or `stderr` outside the renderer: the frame repaints by +moving the cursor over its own previous output, so an out-of-band line +desyncs that accounting. Warn-level diagnostics raised in that window are +counted and reported in the summary instead. + For non-interactive automation, progress may stream to `stderr` while the final result remains pipeable on `stdout`. @@ -341,10 +347,14 @@ Live: imported N · skipped N · errors N Legacy: imported N · skipped N · errors N (of N catalog rows) Push: sent N · skipped N · errors N Catch-up: sent N · skipped N · errors N (local L · remote R) +Warnings: N diagnostic logs suppressed in TTY; use `--verbose` to see them Remote: server unavailable at ; local import is saved. Run `prosa sync` again when it is back. ``` -`Legacy` only appears when `--legacy-bundle` was passed. `Push` and +`Warnings` appears in the TTY summary when the run raised warn-level +diagnostics — from the importers or from the catch-up phase — that the +in-place frame could not display. `Legacy` only appears when +`--legacy-bundle` was passed. `Push` and `Catch-up` only appear when the device is logged in to a prosa-server (i.e. when `~/.config/prosa/auth.json` exists). `Catch-up` is the manifest-driven reconcile that makes the remote converge to the local diff --git a/docs/contributing.md b/docs/contributing.md index df736f75..4cdcc313 100644 --- a/docs/contributing.md +++ b/docs/contributing.md @@ -89,6 +89,9 @@ they are enforced by reviewers (human or agent). - **Error wrapping**: `fmt.Errorf("doing X: %w", err)`. No `pkg/errors`-style ladders. - **Logging**: `log/slog` with the default text handler in CLI commands. + Importers and the pusher write through a caller-supplied `*slog.Logger` + so the interactive sync path can scope them; `slog.SetDefault` is never + called at runtime. - **Comments** are sparse, local, and surgical. Reach for one only to document a behaviour that is not immediately obvious; otherwise let the code speak for itself. Docstrings are allowed but stay short and avoid diff --git a/docs/sources/claude-code.md b/docs/sources/claude-code.md index 41b21777..6660abea 100644 --- a/docs/sources/claude-code.md +++ b/docs/sources/claude-code.md @@ -230,6 +230,10 @@ metadata-rich version with tool-call summaries. ## Notes for prosa importers +- Claude Code appends to the transcript while the session runs, so a torn + write occasionally leaves a line cut mid-token. Such lines are skipped, + as are blank lines; the file earns one warning carrying the count, the + first offending line, and the first parse error. - Walk `*.jsonl` directly. `sessions-index.json` is a hint, never the source of truth. - Subagents are the majority of files in active workspaces. Include diff --git a/docs/sources/codex.md b/docs/sources/codex.md index 43406378..bbaeb292 100644 --- a/docs/sources/codex.md +++ b/docs/sources/codex.md @@ -213,6 +213,10 @@ metadata and tool-call summaries; the recipe above is for raw inspection. ## Notes for prosa importers +- Codex appends to the rollout file while the session runs, so a torn + write occasionally leaves a line cut mid-token. Such lines are skipped, + as are blank lines; the file earns one warning carrying the count, the + first offending line, and the first parse error. - Recent files use the envelope; older files may emit top-level `message`, `reasoning`, `function_call`, `function_call_output`. Both shapes must be projected. diff --git a/internal/cli/sync_run.go b/internal/cli/sync_run.go index 53c68bb0..7d64cc22 100644 --- a/internal/cli/sync_run.go +++ b/internal/cli/sync_run.go @@ -38,12 +38,8 @@ func runSyncInteractive( } updates := make(chan spinner.Update, len(work)*2+16) - // Scope log suppression to the pusher's logger, not the process-global slog - // default, so only reconcile chatter is silenced during Bubble Tea repaints. var suppressedWarnings atomic.Int64 - if push != nil { - push.logger = slog.New(warningCounterHandler{count: &suppressedWarnings}) - } + opts = quietSyncLogging(opts, push, &suppressedWarnings) go func() { defer close(updates) @@ -242,6 +238,22 @@ func runSyncPlain(ctx context.Context, work []syncJob, sink importer.Sink, push return nil } +// quietSyncLogging points the pusher and every importer at a logger that +// tallies warn-level records instead of writing them. Bubble Tea repaints +// by cursor-up over its own frame on stdout, so an unrelated stderr write +// mid-frame desyncs that line accounting and smears the output. The tally +// surfaces afterwards in the summary; the plain and JSON paths keep +// slog.Default(). Scoped per call — the process-global default is never +// swapped. +func quietSyncLogging(opts importer.ImportOptions, push *pusher, count *atomic.Int64) importer.ImportOptions { + quiet := slog.New(warningCounterHandler{count: count}) + if push != nil { + push.logger = quiet + } + opts.Logger = quiet + return opts +} + // localItemErr returns what the spinner shows on a local-phase row. // Import error wins; a pushFailed outcome surfaces the push error so a healthy // import with a failed push isn't rendered as a clean check mark (issue #74). diff --git a/internal/cli/sync_slog_test.go b/internal/cli/sync_slog_test.go index b58fed64..bd74dd7e 100644 --- a/internal/cli/sync_slog_test.go +++ b/internal/cli/sync_slog_test.go @@ -1,11 +1,13 @@ package cli import ( + "bytes" "log/slog" "strings" "sync/atomic" "testing" + "github.com/c3-oss/prosa/pkg/importer" "github.com/stretchr/testify/require" ) @@ -37,3 +39,41 @@ func TestSuppressedWarningsTextSingular(t *testing.T) { require.True(t, strings.Contains(got, "1 diagnostic log suppressed")) require.True(t, strings.Contains(got, "see it")) } + +// TestQuietSyncLoggingScopesImporterWarnings asserts the interactive path +// routes importer diagnostics into the same counter the pusher uses, so +// the summary's Warnings row covers both phases — and that nothing leaks +// onto the process-global slog default, which issue #154 took out of this +// code path for good. +func TestQuietSyncLoggingScopesImporterWarnings(t *testing.T) { + var global bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&global, nil))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + var count atomic.Int64 + push := &pusher{} + got := quietSyncLogging(importer.ImportOptions{Overwrite: true, Profile: "mz"}, push, &count) + + require.NotNil(t, got.Logger) + require.Same(t, got.Logger, push.logger, "one counter must feed both phases") + require.True(t, got.Overwrite, "the value copy must preserve the rest of opts") + require.Equal(t, "mz", got.Profile) + + got.Logger.Warn("codex: malformed JSONL lines skipped", "count", 2) + push.log().Warn("reconcile: catching up") + + require.Equal(t, int64(2), count.Load()) + require.Empty(t, global.String(), "sync diagnostics must not reach the global slog default") +} + +// TestQuietSyncLoggingWithoutPusher covers the server-less sync (no +// auth.json): importers still get the scoped logger. +func TestQuietSyncLoggingWithoutPusher(t *testing.T) { + var count atomic.Int64 + got := quietSyncLogging(importer.ImportOptions{}, nil, &count) + + require.NotNil(t, got.Logger) + got.Logger.Warn("counted") + require.Equal(t, int64(1), count.Load()) +} diff --git a/internal/importers/claudecode/importer.go b/internal/importers/claudecode/importer.go index 951321c7..03b2e16d 100644 --- a/internal/importers/claudecode/importer.go +++ b/internal/importers/claudecode/importer.go @@ -12,6 +12,7 @@ import ( "github.com/c3-oss/prosa/internal/importers/importerutil" "github.com/c3-oss/prosa/internal/paths" "github.com/c3-oss/prosa/pkg/importer" + "github.com/c3-oss/prosa/pkg/session" ) // Name is the agent identifier used in session rows and CLI output. @@ -46,7 +47,9 @@ func (i *Importer) Import(ctx context.Context, jsonlPath string, sink importer.S Opts: opts, Hash: importerutil.HashAndSize, PeekID: peekSessionID, - Parse: parseSession, + Parse: func(ctx context.Context, p string) (session.Session, []session.Turn, []session.ToolUsage, session.UsageState, error) { + return parseSession(ctx, p, importerutil.Logger(opts)) + }, PreserveRaw: func(srcPath, sessionID string, startedAt time.Time) (string, error) { return importerutil.PreserveRaw(Name, sessionID, ".jsonl", startedAt, srcPath) }, diff --git a/internal/importers/claudecode/importer_test.go b/internal/importers/claudecode/importer_test.go index 7790eb8e..c6cf8d4a 100644 --- a/internal/importers/claudecode/importer_test.go +++ b/internal/importers/claudecode/importer_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "log/slog" "os" "path/filepath" "strings" @@ -812,3 +813,38 @@ func TestImportStripsStdoutWrapperAndANSI(t *testing.T) { require.Equal(t, "user", turns[0].Role) require.Equal(t, "now refactor the sync logic", turns[0].Content) } + +// TestImportAggregatesTornAndBlankLines mirrors the codex case: a torn +// append leaves a line cut mid-token, and the session must survive it +// with one warning for the whole file. +func TestImportAggregatesTornAndBlankLines(t *testing.T) { + ctx := context.Background() + t.Setenv("PROSA_HOME", filepath.Join(t.TempDir(), "prosa-home")) + src := writeFixtureSmall(t, t.TempDir()) + + clean, err := os.ReadFile(src) + require.NoError(t, err) + + torn := []byte(`{"type":"user","sessionId":"` + fixtureSessionID + `","message":{"role":"use` + "\n" + + `{"type":"assistant","sessionId":"` + fixtureSessionID + "\n" + + "\n \t\n") + require.NoError(t, os.WriteFile(src, append(clean, torn...), 0o644)) + + var logs bytes.Buffer + sink := newSink() + res, err := New().Import(ctx, src, sink, importer.ImportOptions{ + Logger: slog.New(slog.NewTextHandler(&logs, nil)), + }) + require.NoError(t, err) + require.False(t, res.Skipped) + + s := sink.Sessions[fixtureSessionID] + require.Equal(t, fixtureSessionID, s.ID) + require.NotNil(t, s.Usage) + require.Equal(t, int64(135), s.Usage.TotalTokens) + require.Len(t, sink.Turns[fixtureSessionID], 3) + + require.Equal(t, 1, strings.Count(logs.String(), "level=WARN")) + require.Contains(t, logs.String(), "claude-code: malformed JSONL lines skipped") + require.Contains(t, logs.String(), "count=2") +} diff --git a/internal/importers/claudecode/parse.go b/internal/importers/claudecode/parse.go index 6d405959..33ba313a 100644 --- a/internal/importers/claudecode/parse.go +++ b/internal/importers/claudecode/parse.go @@ -2,6 +2,7 @@ package claudecode import ( "bufio" + "bytes" "context" "encoding/json" "errors" @@ -93,7 +94,7 @@ func peekSessionID(path string) (string, error) { // parseSession streams the JSONL once and returns the projected metadata // plus a UsageState classifying whether the transcript carried any usage // event (and if so, whether totals were positive). -func parseSession(ctx context.Context, path string) (session.Session, []session.Turn, []session.ToolUsage, session.UsageState, error) { +func parseSession(ctx context.Context, path string, log *slog.Logger) (session.Session, []session.Turn, []session.ToolUsage, session.UsageState, error) { f, err := os.Open(path) if err != nil { return session.Session{}, nil, nil, session.UsageStateUnknown, err @@ -114,6 +115,10 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. modelSet bool firstPromptSet bool line int + + malformed int + malformedAt int + malformedErr error ) // Subagent transcripts live under `/subagents/` and @@ -135,10 +140,17 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. return session.Session{}, nil, nil, session.UsageStateUnknown, err } + raw := sc.Bytes() + if len(bytes.TrimSpace(raw)) == 0 { + continue + } + var r rawRecord - if err := json.Unmarshal(sc.Bytes(), &r); err != nil { - slog.Warn("claude-code: malformed JSONL line skipped", - "path", path, "line", line, "err", err) + if err := json.Unmarshal(raw, &r); err != nil { + malformed++ + if malformedAt == 0 { + malformedAt, malformedErr = line, err + } continue } @@ -228,13 +240,18 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. if err := sc.Err(); err != nil { if errors.Is(err, bufio.ErrTooLong) { - slog.Warn("claude-code: JSONL line exceeded 16 MiB scan buffer; partial session", + log.Warn("claude-code: JSONL line exceeded 16 MiB scan buffer; partial session", "path", path, "line", line+1) } else { return session.Session{}, nil, nil, session.UsageStateUnknown, fmt.Errorf("scan %s: %w", path, err) } } + if malformed > 0 { + log.Warn("claude-code: malformed JSONL lines skipped", + "path", path, "count", malformed, "first_line", malformedAt, "err", malformedErr) + } + tools := make([]session.ToolUsage, 0, len(toolCounts)) for name, count := range toolCounts { tools = append(tools, session.ToolUsage{Name: name, Count: count}) diff --git a/internal/importers/codex/importer.go b/internal/importers/codex/importer.go index d7c2884a..01b238a5 100644 --- a/internal/importers/codex/importer.go +++ b/internal/importers/codex/importer.go @@ -14,6 +14,7 @@ import ( "github.com/c3-oss/prosa/internal/importers/importerutil" "github.com/c3-oss/prosa/internal/paths" "github.com/c3-oss/prosa/pkg/importer" + "github.com/c3-oss/prosa/pkg/session" ) // Name is the agent identifier used in session rows and CLI output. @@ -48,7 +49,9 @@ func (i *Importer) Import(ctx context.Context, jsonlPath string, sink importer.S Opts: opts, Hash: importerutil.HashAndSize, PeekID: peekSessionID, - Parse: parseSession, + Parse: func(ctx context.Context, p string) (session.Session, []session.Turn, []session.ToolUsage, session.UsageState, error) { + return parseSession(ctx, p, importerutil.Logger(opts)) + }, PreserveRaw: func(srcPath, sessionID string, startedAt time.Time) (string, error) { return importerutil.PreserveRaw(Name, sessionID, ".jsonl", startedAt, srcPath) }, diff --git a/internal/importers/codex/importer_test.go b/internal/importers/codex/importer_test.go index dddb848e..250a8d20 100644 --- a/internal/importers/codex/importer_test.go +++ b/internal/importers/codex/importer_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "log/slog" "os" "path/filepath" "strings" @@ -725,3 +726,48 @@ func TestTruncatePreviewKeepsUTF8Valid(t *testing.T) { require.True(t, utf8.ValidString(got)) require.Contains(t, got, "…") } + +// TestImportAggregatesTornAndBlankLines models what Codex actually writes: +// concurrent appends occasionally tear a line mid-token. Those lines are +// skipped without touching the rest of the session, blank lines are not +// diagnostics at all, and the file earns exactly one warning. +func TestImportAggregatesTornAndBlankLines(t *testing.T) { + ctx := context.Background() + t.Setenv("PROSA_HOME", filepath.Join(t.TempDir(), "prosa-home")) + + root := filepath.Join(t.TempDir(), "codex-root") + src := writeFixtureEnvelope(t, root) + + clean, err := os.ReadFile(src) + require.NoError(t, err) + + torn := []byte(`{"timestamp":"2026-05-30T12:00:40Z","ordinal":520,"type":"inter_agent_communicatio` + "\n" + + `{"timestamp":"2026-05-30T12:00:41Z","type":"response_ite` + "\n" + + "\n \t\n") + require.NoError(t, os.WriteFile(src, append(clean, torn...), 0o644)) + + var logs bytes.Buffer + sink := newSink() + res, err := New().Import(ctx, src, sink, importer.ImportOptions{ + Logger: slog.New(slog.NewTextHandler(&logs, nil)), + }) + require.NoError(t, err) + require.False(t, res.Skipped) + + // The good records still project exactly as they do without the damage. + s := sink.Sessions[fixtureSessionID] + require.Equal(t, fixtureSessionID, s.ID) + require.NotNil(t, s.Usage) + require.Equal(t, int64(1120), s.Usage.TotalTokens) + turns := sink.Turns[fixtureSessionID] + require.Len(t, turns, 2) + require.Equal(t, "user", turns[0].Role) + require.Equal(t, "assistant", turns[1].Role) + require.Len(t, sink.Tools[fixtureSessionID], 1) + + // One record for the file, carrying the tally — not one per bad line, + // and blank/whitespace-only lines are not counted as malformed. + require.Equal(t, 1, strings.Count(logs.String(), "level=WARN")) + require.Contains(t, logs.String(), "codex: malformed JSONL lines skipped") + require.Contains(t, logs.String(), "count=2") +} diff --git a/internal/importers/codex/parse.go b/internal/importers/codex/parse.go index a37316e0..d2b8d521 100644 --- a/internal/importers/codex/parse.go +++ b/internal/importers/codex/parse.go @@ -2,6 +2,7 @@ package codex import ( "bufio" + "bytes" "context" "encoding/json" "errors" @@ -147,7 +148,7 @@ func peekSessionID(path string) (string, error) { // parseSession streams the JSONL once and returns the projected session, // turns, tool usages, and a UsageState. Hash + size are computed by Import. -func parseSession(ctx context.Context, path string) (session.Session, []session.Turn, []session.ToolUsage, session.UsageState, error) { +func parseSession(ctx context.Context, path string, log *slog.Logger) (session.Session, []session.Turn, []session.ToolUsage, session.UsageState, error) { f, err := os.Open(path) if err != nil { return session.Session{}, nil, nil, session.UsageStateUnknown, err @@ -169,6 +170,10 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. firstPromptSet bool seenUsageEvent bool line int + + malformed int + malformedAt int + malformedErr error ) for sc.Scan() { @@ -177,10 +182,17 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. return session.Session{}, nil, nil, session.UsageStateUnknown, err } + raw := sc.Bytes() + if len(bytes.TrimSpace(raw)) == 0 { + continue + } + var r rawRecord - if err := json.Unmarshal(sc.Bytes(), &r); err != nil { - slog.Warn("codex: malformed JSONL line skipped", - "path", path, "line", line, "err", err) + if err := json.Unmarshal(raw, &r); err != nil { + malformed++ + if malformedAt == 0 { + malformedAt, malformedErr = line, err + } continue } @@ -266,7 +278,7 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. // Legacy records carry the call_id alongside; round-trip // it via the raw payload so function_call_output can find // the tool name later. - if id := legacyCallID(sc.Bytes()); id != "" { + if id := legacyCallID(raw); id != "" { callIDToName[id] = r.Name } } @@ -276,7 +288,7 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. if text == "" { break } - id := legacyCallID(sc.Bytes()) + id := legacyCallID(raw) name := callIDToName[id] ts, _ := importerutil.ParseRFC3339(r.Timestamp) turns = append(turns, session.Turn{ @@ -291,13 +303,20 @@ func parseSession(ctx context.Context, path string) (session.Session, []session. if err := sc.Err(); err != nil { if errors.Is(err, bufio.ErrTooLong) { - slog.Warn("codex: JSONL line exceeded 16 MiB scan buffer; partial session", + log.Warn("codex: JSONL line exceeded 16 MiB scan buffer; partial session", "path", path, "line", line+1) } else { return session.Session{}, nil, nil, session.UsageStateUnknown, fmt.Errorf("scan %s: %w", path, err) } } + // One record per file, not per line: Codex appends concurrently and a + // torn write leaves a line cut mid-token every few thousand records. + if malformed > 0 { + log.Warn("codex: malformed JSONL lines skipped", + "path", path, "count", malformed, "first_line", malformedAt, "err", malformedErr) + } + tools := make([]session.ToolUsage, 0, len(toolCounts)) for name, count := range toolCounts { tools = append(tools, session.ToolUsage{Name: name, Count: count}) diff --git a/internal/importers/grokbuild/importer.go b/internal/importers/grokbuild/importer.go index fe8a07fd..ec22ae31 100644 --- a/internal/importers/grokbuild/importer.go +++ b/internal/importers/grokbuild/importer.go @@ -66,6 +66,7 @@ type projectedLine struct { // re-import. func (i *Importer) Import(ctx context.Context, summaryPath string, sink importer.Sink, opts importer.ImportOptions) (importer.ImportResult, error) { dir := filepath.Dir(summaryPath) + log := importerutil.Logger(opts) summaryRaw, err := os.ReadFile(summaryPath) if err != nil { @@ -80,7 +81,7 @@ func (i *Importer) Import(ctx context.Context, summaryPath string, sink importer if id == "" { id = filepath.Base(dir) } else if id != filepath.Base(dir) { - slog.Warn("grok-build: summary id differs from directory name", + log.Warn("grok-build: summary id differs from directory name", "summary_id", id, "dir", filepath.Base(dir)) } if err := session.ValidateID(id); err != nil { @@ -92,14 +93,14 @@ func (i *Importer) Import(ctx context.Context, summaryPath string, sink importer parentMetaRaw json.RawMessage ) if sum.SessionKind == "subagent" { - parentID, parentMetaRaw = resolveParentMeta(dir, id) + parentID, parentMetaRaw = resolveParentMeta(dir, id, log) } - chatLines, err := readJSONLines(filepath.Join(dir, "chat_history.jsonl")) + chatLines, err := readJSONLines(filepath.Join(dir, "chat_history.jsonl"), log) if err != nil { return importer.ImportResult{}, fmt.Errorf("read chat history %s: %w", dir, err) } - updateLines, err := readJSONLines(filepath.Join(dir, "updates.jsonl")) + updateLines, err := readJSONLines(filepath.Join(dir, "updates.jsonl"), log) if err != nil { return importer.ImportResult{}, fmt.Errorf("read updates %s: %w", dir, err) } @@ -136,7 +137,7 @@ func (i *Importer) Import(ctx context.Context, summaryPath string, sink importer if parentID != "" { sess.ParentSessionID = &parentID } - turns, toolCounts, firstPrompt := projectChat(chatLines, sess.StartedAt) + turns, toolCounts, firstPrompt := projectChat(chatLines, sess.StartedAt, log) sess.FirstPrompt = firstPrompt usage, seenUsage := sumUsage(usageRecs) sess.Usage = usage @@ -256,7 +257,7 @@ func marshalProjectedLine(typ string, data []byte) (json.RawMessage, error) { // returns the parent session id plus the raw meta bytes for the // projection. Resolution happens before hashing so a late-appearing or // changed meta.json changes the hash and triggers re-import. -func resolveParentMeta(dir, childID string) (string, json.RawMessage) { +func resolveParentMeta(dir, childID string, log *slog.Logger) (string, json.RawMessage) { projectDir := filepath.Dir(dir) patterns := []string{ filepath.Join(projectDir, "*", "subagents", childID, "meta.json"), @@ -277,6 +278,6 @@ func resolveParentMeta(dir, childID string) (string, json.RawMessage) { } return meta.ParentSessionID, raw } - slog.Warn("grok-build: subagent session has no resolvable parent meta", "session", childID) + log.Warn("grok-build: subagent session has no resolvable parent meta", "session", childID) return "", nil } diff --git a/internal/importers/grokbuild/importer_test.go b/internal/importers/grokbuild/importer_test.go index 4ed4fc2b..a12d92c2 100644 --- a/internal/importers/grokbuild/importer_test.go +++ b/internal/importers/grokbuild/importer_test.go @@ -7,6 +7,7 @@ import ( "encoding/hex" "encoding/json" "errors" + "log/slog" "os" "path/filepath" "strings" @@ -570,8 +571,11 @@ func TestImportDropsMalformedAndTornLines(t *testing.T) { chat = append(chat, []byte(`{"type":"assistant","content":"torn lin`)...) // torn tail, no newline require.NoError(t, os.WriteFile(filepath.Join(filepath.Dir(summaryPath), "chat_history.jsonl"), chat, 0o644)) + var logs bytes.Buffer sink := newSink() - res := importSession(t, summaryPath, sink, importer.ImportOptions{}) + res := importSession(t, summaryPath, sink, importer.ImportOptions{ + Logger: slog.New(slog.NewTextHandler(&logs, nil)), + }) require.False(t, res.Skipped) turns := sink.Turns[uuidA] @@ -582,6 +586,36 @@ func TestImportDropsMalformedAndTornLines(t *testing.T) { require.NoError(t, err) require.NotContains(t, string(raw), "broken json") require.NotContains(t, string(raw), "torn lin") + + // Both bad lines are one warning for the file, not one apiece. + require.Equal(t, 1, strings.Count(logs.String(), "level=WARN")) + require.Contains(t, logs.String(), "grok-build: malformed JSONL lines dropped") + require.Contains(t, logs.String(), "count=2") +} + +// TestImportHashUnchangedByBlankLines locks the projection bytes. The +// lines readJSONLines collects are hashed verbatim, and that hash is the +// dedup key, the sync hash, and RawHash at once — so trimming a collected +// line would silently re-import and re-push every touched session. +func TestImportHashUnchangedByBlankLines(t *testing.T) { + t.Setenv("PROSA_HOME", t.TempDir()) + root := t.TempDir() + + summaryPath := writeSessionDir(t, root, encodedCwd, uuidA, defaultSummary(uuidA), + []map[string]any{humanUser(0, "hello")}, + []map[string]any{turnCompleted(uuidA, "p", usageDelta(10, 5, 0, 0))}) + + first := importSession(t, summaryPath, newSink(), importer.ImportOptions{}) + require.False(t, first.Skipped) + + chatPath := filepath.Join(filepath.Dir(summaryPath), "chat_history.jsonl") + chat, err := os.ReadFile(chatPath) + require.NoError(t, err) + require.NoError(t, os.WriteFile(chatPath, append(chat, []byte("\n")...), 0o644)) + + second := importSession(t, summaryPath, newSink(), importer.ImportOptions{}) + require.Equal(t, first.RawHash, second.RawHash, "a blank line must not move the projection hash") + require.Equal(t, first.RawSize, second.RawSize) } // failingLastHashSink wraps the in-memory sink so LastHash fails, diff --git a/internal/importers/grokbuild/parse.go b/internal/importers/grokbuild/parse.go index 38d6755a..bb5c2692 100644 --- a/internal/importers/grokbuild/parse.go +++ b/internal/importers/grokbuild/parse.go @@ -90,7 +90,7 @@ const subagentCompletedPrefix = "subagent-completed-" // file. Malformed or torn lines are warned and dropped so they never // reach the raw projection (a torn tail would change on the next append // anyway). A missing file returns nil with no error. -func readJSONLines(path string) ([]json.RawMessage, error) { +func readJSONLines(path string, log *slog.Logger) ([]json.RawMessage, error) { f, err := os.Open(path) if err != nil { if os.IsNotExist(err) { @@ -105,6 +105,8 @@ func readJSONLines(path string) ([]json.RawMessage, error) { var out []json.RawMessage line := 0 + malformed := 0 + malformedAt := 0 for sc.Scan() { line++ b := sc.Bytes() @@ -112,14 +114,23 @@ func readJSONLines(path string) ([]json.RawMessage, error) { continue } if !json.Valid(b) { - slog.Warn("grok-build: malformed JSONL line dropped", "path", path, "line", line) + malformed++ + if malformedAt == 0 { + malformedAt = line + } continue } out = append(out, json.RawMessage(append([]byte(nil), b...))) } + // One record per file, not per line: a torn write can cut a line + // mid-token, and a truncated file can cut many. + if malformed > 0 { + log.Warn("grok-build: malformed JSONL lines dropped", + "path", path, "count", malformed, "first_line", malformedAt) + } if err := sc.Err(); err != nil { if errors.Is(err, bufio.ErrTooLong) { - slog.Warn("grok-build: JSONL line exceeded 16 MiB scan buffer; partial file", + log.Warn("grok-build: JSONL line exceeded 16 MiB scan buffer; partial file", "path", path, "line", line+1) return out, nil } @@ -187,14 +198,14 @@ func sumUsage(recs []updateRecord) (usage *session.TokenUsage, seenUsage bool) { // counters, and the first human prompt. Chat records carry no // timestamps, so every turn is stamped with the session's created_at — // the store orders by (ts, id) so insertion order is preserved. -func projectChat(lines []json.RawMessage, createdAt time.Time) (turns []session.Turn, toolCounts map[string]int, firstPrompt *string) { +func projectChat(lines []json.RawMessage, createdAt time.Time, log *slog.Logger) (turns []session.Turn, toolCounts map[string]int, firstPrompt *string) { toolCounts = map[string]int{} callIDToName := map[string]string{} for n, line := range lines { var r chatRecord if err := json.Unmarshal(line, &r); err != nil { - slog.Warn("grok-build: undecodable chat record skipped", "line", n+1, "err", err) + log.Warn("grok-build: undecodable chat record skipped", "line", n+1, "err", err) continue } switch r.Type { @@ -269,7 +280,7 @@ func projectChat(lines []json.RawMessage, createdAt time.Time) (turns []session. } default: - slog.Warn("grok-build: unknown chat record type skipped", "type", r.Type, "line", n+1) + log.Warn("grok-build: unknown chat record type skipped", "type", r.Type, "line", n+1) } } return turns, toolCounts, firstPrompt diff --git a/internal/importers/importerutil/logger.go b/internal/importers/importerutil/logger.go new file mode 100644 index 00000000..e9af682f --- /dev/null +++ b/internal/importers/importerutil/logger.go @@ -0,0 +1,18 @@ +package importerutil + +import ( + "log/slog" + + "github.com/c3-oss/prosa/pkg/importer" +) + +// Logger returns the logger an import's diagnostics belong on. Importers +// never reach for package-level slog: the caller decides where warnings +// land, so the interactive sync path can tally them instead of writing +// into its own progress frame. +func Logger(opts importer.ImportOptions) *slog.Logger { + if opts.Logger != nil { + return opts.Logger + } + return slog.Default() +} diff --git a/internal/importers/importerutil/logger_test.go b/internal/importers/importerutil/logger_test.go new file mode 100644 index 00000000..71edff4c --- /dev/null +++ b/internal/importers/importerutil/logger_test.go @@ -0,0 +1,25 @@ +package importerutil + +import ( + "bytes" + "log/slog" + "testing" + + "github.com/c3-oss/prosa/pkg/importer" + "github.com/stretchr/testify/require" +) + +func TestLoggerFallsBackToDefault(t *testing.T) { + require.Same(t, slog.Default(), Logger(importer.ImportOptions{})) +} + +func TestLoggerPrefersOptsLogger(t *testing.T) { + var buf bytes.Buffer + scoped := slog.New(slog.NewTextHandler(&buf, nil)) + + got := Logger(importer.ImportOptions{Logger: scoped}) + require.Same(t, scoped, got) + + got.Warn("scoped") + require.Contains(t, buf.String(), "scoped") +} diff --git a/pkg/importer/importer.go b/pkg/importer/importer.go index 5d5b5f92..92ac6216 100644 --- a/pkg/importer/importer.go +++ b/pkg/importer/importer.go @@ -8,6 +8,7 @@ package importer import ( "context" + "log/slog" "github.com/c3-oss/prosa/pkg/session" ) @@ -49,6 +50,11 @@ type ImportOptions struct { // Profile names the profile the scanned file belongs to; empty means "default". Profile string + + // Logger receives this import's diagnostics; nil means slog.Default(). + // Scoped per call so the interactive sync path can tally warnings + // instead of interleaving them with its in-place progress frame. + Logger *slog.Logger } // Importer is the plugin contract every per-agent connector implements.