package eval import ( "context" "database/sql" "errors" "sync" ) // DefaultConcurrency bounds parallel Claude CLI subprocesses. Claude's API // rate limits and local CPU/memory make 3 a safe default; power users can // raise it via Batch.Concurrency. const DefaultConcurrency = 3 // Event is emitted on the channel returned by Batch.Run. Exactly one Start // and one Done (or Error) event is emitted per submitted job. type Event struct { Kind EventKind Job JobContext Saved Saved // populated on EventDone Err error // populated on EventError Eval Evaluation // populated on EventDone (lightweight fields only — no Raw) } // EventKind enumerates the lifecycle events a caller can observe. type EventKind int const ( EventStart EventKind = iota EventDone EventError // EventSkipped is emitted when MinScore filtered an otherwise-successful // eval. Saved is populated (the report is still on disk), but the // application row has been marked Discarded. EventSkipped ) // Batch fans evaluations out to N worker goroutines, capped by Concurrency. // The Client and Writer are shared across workers; both are safe to reuse. // // MinScore (0 = disabled) post-filters low-fit offers: after a successful // eval, if the parsed Score is below MinScore, the persisted application row // is demoted to status='Discarded' and the event is emitted as EventSkipped. // This keeps weak matches out of the review queue without losing the report. type Batch struct { Client *Client Writer *ReportWriter DB *sql.DB CVPath string Concurrency int MinScore float64 } // Run evaluates every job in the input slice in parallel, up to Concurrency // at a time. Returns a channel that emits Event values and closes when all // workers finish or the context is canceled. // // Cancellation: closing ctx terminates in-flight subprocesses and stops the // worker pool; no new jobs are started. Events for already-running jobs are // still emitted (typically as EventError with context.Canceled). func (b *Batch) Run(ctx context.Context, jobs []JobContext) <-chan Event { events := make(chan Event, len(jobs)) if len(jobs) == 0 { close(events) return events } concurrency := b.Concurrency if concurrency <= 0 { concurrency = DefaultConcurrency } if concurrency > len(jobs) { concurrency = len(jobs) } sem := make(chan struct{}, concurrency) var wg sync.WaitGroup go func() { defer close(events) for _, j := range jobs { select { case <-ctx.Done(): // Remaining jobs never started — emit a synthetic error so // the caller's progress counter reconciles. events <- Event{Kind: EventError, Job: j, Err: ctx.Err()} continue case sem <- struct{}{}: } wg.Add(1) go func(job JobContext) { defer wg.Done() defer func() { <-sem }() b.runOne(ctx, job, events) }(j) } wg.Wait() }() return events } // runOne evaluates a single job and emits its lifecycle events. Errors are // surfaced as EventError; successful completion yields EventDone carrying // the persisted Saved record and a lightweight copy of the Evaluation // (Raw is stripped to keep event payloads small). func (b *Batch) runOne(ctx context.Context, job JobContext, events chan<- Event) { events <- Event{Kind: EventStart, Job: job} if b.Client == nil { events <- Event{Kind: EventError, Job: job, Err: errors.New("batch: nil client")} return } prompt, err := BuildPrompt(b.DB, job, b.CVPath) if err != nil { events <- Event{Kind: EventError, Job: job, Err: err} return } eval, err := b.Client.Evaluate(ctx, prompt, nil) if err != nil { events <- Event{Kind: EventError, Job: job, Err: err} return } if b.Writer == nil { // Caller asked us to evaluate but skip persistence (unusual but valid). events <- Event{Kind: EventDone, Job: job, Eval: slimEval(eval)} return } saved, err := b.Writer.Save(ctx, job, eval) if err != nil { events <- Event{Kind: EventError, Job: job, Err: err} return } // Post-filter: if a MinScore threshold is set and the eval fell below it, // demote the application to Discarded so the pipeline view doesn't queue // it. The on-disk report is preserved for the user's reference. if b.MinScore > 0 && eval.Score > 0 && eval.Score < b.MinScore { if err := b.demoteBelowScore(ctx, saved.ApplicationID); err != nil { // Surface the demotion failure but don't lose the Save success. events <- Event{Kind: EventError, Job: job, Err: err} return } events <- Event{Kind: EventSkipped, Job: job, Saved: saved, Eval: slimEval(eval)} return } events <- Event{Kind: EventDone, Job: job, Saved: saved, Eval: slimEval(eval)} } // demoteBelowScore marks an application as 'Discarded' after a MinScore miss. // Only demotes rows currently in 'Evaluated' — user-advanced states // (Applied, Interview, etc.) are preserved in case of re-evaluation. func (b *Batch) demoteBelowScore(ctx context.Context, appID int64) error { if b.DB == nil { return errors.New("demote: nil DB") } _, err := b.DB.ExecContext(ctx, ` UPDATE applications SET status = 'Discarded', updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'Evaluated' `, appID) return err } // slimEval returns a copy of eval without Raw so Event payloads stay small // when emitted into a buffered channel watched by the TUI. func slimEval(e Evaluation) Evaluation { e.Raw = "" return e }