diff --git a/cmd/lard-client/main.go b/cmd/lard-client/main.go index 1f08f27..c19bd87 100644 --- a/cmd/lard-client/main.go +++ b/cmd/lard-client/main.go @@ -17,6 +17,7 @@ import ( "charm.land/fang/v2" "github.com/spf13/cobra" + "golang.org/x/term" "github.com/taciturnaxolotl/lard/internal/client" "github.com/taciturnaxolotl/lard/internal/dotenv" @@ -106,6 +107,26 @@ func printConnected(cfg *client.Config) { fmt.Printf("Connected to %s via %s.\n", ui.Link(cfg.URL, "lard-server"), cfg.AuthMode()) } +// plural renders a count with its singular/plural noun: "1 session", "3 sessions". +func plural(n int, noun string) string { + if n == 1 { + return fmt.Sprintf("%d %s", n, noun) + } + return fmt.Sprintf("%d %ss", n, noun) +} + +// shortID trims a session id to its first segment so progress lines stay +// readable: "3f1ab2c9-4d0e-…" not the whole uuid. +func shortID(id string) string { + if i := strings.IndexAny(id, "-:"); i > 0 { + return id[:i] + } + if len(id) > 12 { + return id[:12] + } + return id +} + // --- collection --- func backfillCmd() *cobra.Command { @@ -205,7 +226,43 @@ only needed to skip the wait.`, if err != nil { return err } - return up.Consolidate(cmd.Context()) + fmt.Println("Consolidating on the server; a backfill can take a while...") + interactive := term.IsTerminal(int(os.Stdout.Fd())) + updating := false + progress := func(phase, name string, done, total int) { + verb, subject := "extracting", "session "+shortID(name) + if phase == "synthesize" { + verb, subject = "synthesizing", name + } + line := fmt.Sprintf("%s %s (%d/%d)", verb, subject, done, total) + if interactive { + fmt.Printf("\r\033[2K%s", line) + updating = true + } else { + fmt.Println(line) + } + } + res, err := up.Consolidate(cmd.Context(), progress) + if updating { + fmt.Println() + } + if err != nil { + return err + } + switch { + case res.Extracted > 0 && res.Synthesized > 0: + fmt.Printf("Done: extracted facts from %s, rewrote %s.\n", + plural(res.Extracted, "session"), plural(res.Synthesized, "subject file")) + case res.Extracted > 0: + fmt.Printf("Done: extracted facts from %s; no subjects needed rewriting.\n", + plural(res.Extracted, "session")) + case res.Synthesized > 0: + fmt.Printf("Done: rewrote %s; nothing new to extract.\n", + plural(res.Synthesized, "subject file")) + default: + fmt.Println("Done: nothing to do, memory is up to date.") + } + return nil }, } } diff --git a/internal/client/client.go b/internal/client/client.go index 726b0b2..c0b1b59 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -71,9 +72,67 @@ func (u *Uploader) Ingest(ctx context.Context, req types.IngestRequest) error { return u.post(ctx, "/ingest", req, nil) } -// Consolidate triggers a server-side consolidation pass. -func (u *Uploader) Consolidate(ctx context.Context) error { - return u.post(ctx, "/consolidate", nil, nil) +// ConsolidateResult reports what the server's consolidation pass did. +type ConsolidateResult struct { + Extracted int `json:"extracted"` // sessions that yielded facts + Synthesized int `json:"synthesized"` // subject files rewritten +} + +// ProgressFn gets one call per completed step of a consolidation pass. +type ProgressFn func(phase, name string, done, total int) + +// Consolidate triggers a server-side consolidation pass and waits for it, +// streaming progress as steps finish and reporting what the pass did. A +// backfill can take a long time (one LLM call per session and per dirty +// subject), so the request itself carries no timeout; cancellation comes +// from ctx. The pass survives a client going away: it is detached on the +// server side. +func (u *Uploader) Consolidate(ctx context.Context, progress ProgressFn) (*ConsolidateResult, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.BaseURL+"/consolidate", nil) + if err != nil { + return nil, err + } + req.Header.Set("accept", "application/x-ndjson") + if u.Token != "" { + req.Header.Set("authorization", "Bearer "+u.Token) + } + cl := *u.HTTP + cl.Timeout = 0 // let ctx own cancellation for a long-running pass + resp, err := cl.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("/consolidate %d: %s", resp.StatusCode, string(b)) + } + + dec := json.NewDecoder(resp.Body) + for { + var ev struct { + Phase string `json:"phase"` + Name string `json:"name"` + Done int `json:"done"` + Total int `json:"total"` + Finished bool `json:"finished"` + Extracted int `json:"extracted"` + Synthesized int `json:"synthesized"` + Error string `json:"error"` + } + if err := dec.Decode(&ev); err != nil { + return nil, fmt.Errorf("consolidate stream ended early: %w", err) + } + if ev.Finished { + if ev.Error != "" { + return nil, errors.New(ev.Error) + } + return &ConsolidateResult{Extracted: ev.Extracted, Synthesized: ev.Synthesized}, nil + } + if progress != nil { + progress(ev.Phase, ev.Name, ev.Done, ev.Total) + } + } } // ResolveProject asks the service to canonicalize hints and returns the id. diff --git a/internal/client/consolidate_test.go b/internal/client/consolidate_test.go new file mode 100644 index 0000000..0f7c932 --- /dev/null +++ b/internal/client/consolidate_test.go @@ -0,0 +1,84 @@ +package client + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" +) + +// TestConsolidateStream exercises the NDJSON contract /consolidate speaks: +// progress events as they arrive, then the summary line. +func TestConsolidateStream(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/consolidate" || r.Method != http.MethodPost { + t.Errorf("unexpected request: %s %s", r.Method, r.URL.Path) + } + if r.Header.Get("authorization") != "Bearer tok" { + t.Errorf("bearer token not forwarded: %q", r.Header.Get("authorization")) + } + w.Header().Set("content-type", "application/x-ndjson") + fmt.Fprintln(w, `{"phase":"extract","name":"sess-1","done":1,"total":2}`) + fmt.Fprintln(w, `{"phase":"extract","name":"sess-2","done":2,"total":2}`) + fmt.Fprintln(w, `{"phase":"synthesize","name":"areas/crush","done":1,"total":1}`) + fmt.Fprintln(w, `{"finished":true,"extracted":2,"synthesized":1}`) + })) + defer srv.Close() + + up := NewUploader(srv.URL, "tok") + var seen []string + res, err := up.Consolidate(context.Background(), func(phase, name string, done, total int) { + seen = append(seen, fmt.Sprintf("%s %s %d/%d", phase, name, done, total)) + }) + if err != nil { + t.Fatal(err) + } + if res.Extracted != 2 || res.Synthesized != 1 { + t.Fatalf("bad result: %+v", res) + } + want := []string{ + "extract sess-1 1/2", + "extract sess-2 2/2", + "synthesize areas/crush 1/1", + } + if len(seen) != len(want) { + t.Fatalf("progress calls: %v", seen) + } + for i := range want { + if seen[i] != want[i] { + t.Fatalf("progress %d: got %q, want %q", i, seen[i], want[i]) + } + } +} + +// TestConsolidateError covers the failure line: the pass ended with an +// error, which surfaces instead of a result. +func TestConsolidateError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("content-type", "application/x-ndjson") + fmt.Fprintln(w, `{"finished":true,"error":"llm: rate limited"}`) + })) + defer srv.Close() + + up := NewUploader(srv.URL, "tok") + _, err := up.Consolidate(context.Background(), nil) + if err == nil || err.Error() != "llm: rate limited" { + t.Fatalf("want the pass error, got %v", err) + } +} + +// TestConsolidateRefused covers a non-2xx body (e.g. 503 without an LLM): +// the server's message comes back verbatim. +func TestConsolidateRefused(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + fmt.Fprintln(w, `{"error":"consolidation unavailable: no LLM client configured"}`) + })) + defer srv.Close() + + up := NewUploader(srv.URL, "tok") + if _, err := up.Consolidate(context.Background(), nil); err == nil { + t.Fatal("want an error for 503") + } +} diff --git a/internal/httpapi/httpapi.go b/internal/httpapi/httpapi.go index 0e09321..a6d2943 100644 --- a/internal/httpapi/httpapi.go +++ b/internal/httpapi/httpapi.go @@ -10,6 +10,7 @@ import ( "errors" "log/slog" "net/http" + "sync" "time" "github.com/taciturnaxolotl/lard/internal/auth" @@ -26,6 +27,58 @@ type Server struct { llm *llm.Client mux *http.ServeMux auto *autoConsolidator + + // Single-flight consolidation. A manual /consolidate and the quiet + // timer both funnel through one job, so concurrent callers never start + // competing passes, and a caller going away doesn't kill the pass. + consolMu sync.Mutex + consolJob *consolidationJob +} + +// consolidationJob is one in-flight consolidation pass. +type consolidationJob struct { + done chan struct{} // closed when the pass finishes + res pipeline.Result + err error + + // Progress fan-out. Each listener gets events as steps finish; a slow + // listener is dropped rather than stalling the pass. + mu sync.Mutex + subs []chan pipeline.ProgressEvent +} + +// publish fans one event out to every listener. Buffered sends mean a +// consumer going away can never block consolidation. +func (j *consolidationJob) publish(ev pipeline.ProgressEvent) { + j.mu.Lock() + defer j.mu.Unlock() + for _, ch := range j.subs { + select { + case ch <- ev: + default: + } + } +} + +// subscribe registers a progress listener. The returned channel is +// unregistered when done is closed or unsubscribe is called. +func (j *consolidationJob) subscribe() chan pipeline.ProgressEvent { + ch := make(chan pipeline.ProgressEvent, 64) + j.mu.Lock() + j.subs = append(j.subs, ch) + j.mu.Unlock() + return ch +} + +func (j *consolidationJob) unsubscribe(ch chan pipeline.ProgressEvent) { + j.mu.Lock() + defer j.mu.Unlock() + for i, s := range j.subs { + if s == ch { + j.subs = append(j.subs[:i], j.subs[i+1:]...) + break + } + } } // New builds the HTTP server. llmClient may be nil if consolidation is never @@ -45,7 +98,7 @@ func (s *Server) EnableAutoConsolidate(after, maxWait time.Duration) { return } s.auto = newAutoConsolidator(after, maxWait, func(ctx context.Context) error { - _, err := s.Consolidator().Run(ctx, 0) + _, err := s.consolidate(ctx) return err }) slog.Info("auto-consolidate enabled", "quiet_period", after, "max_wait", maxWait) @@ -304,17 +357,106 @@ func (s *Server) handleConsolidate(w http.ResponseWriter, r *http.Request) { writeErr(w, 503, errors.New("consolidation unavailable: no LLM client configured")) return } + job := s.startConsolidation() + events := job.subscribe() + defer job.unsubscribe(events) + + // One JSON object per line: a progress event as each step finishes, then + // a final line carrying the result (or error). A client that goes away + // just stops reading; the pass keeps running server-side. + w.Header().Set("content-type", "application/x-ndjson") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + writeLine := func(v any) bool { + if err := json.NewEncoder(w).Encode(v); err != nil { + return false + } + if flusher != nil { + flusher.Flush() + } + return true + } + progressLine := func(ev pipeline.ProgressEvent) bool { + return writeLine(map[string]any{"phase": ev.Phase, "name": ev.Name, "done": ev.Done, "total": ev.Total}) + } + + for { + select { + case ev := <-events: + if !progressLine(ev) { + return + } + case <-job.done: + // Events published before the pass ended may still be buffered; + // drain them so the summary is always the last line. + for { + select { + case ev := <-events: + if !progressLine(ev) { + return + } + default: + goto finished + } + } + finished: + out := map[string]any{ + "finished": true, + "extracted": job.res.Extracted, + "synthesized": job.res.Synthesized, + } + if job.err != nil { + out["error"] = job.err.Error() + } + writeLine(out) + return + case <-r.Context().Done(): + return + } + } +} + +// consolidate waits for one consolidation pass and returns its result. +// Used by the quiet-timer auto-pass, which wants the outcome but not the +// step-by-step stream. +func (s *Server) consolidate(ctx context.Context) (pipeline.Result, error) { + job := s.startConsolidation() + select { + case <-job.done: + return job.res, job.err + case <-ctx.Done(): + return pipeline.Result{}, ctx.Err() + } +} + +// startConsolidation starts a consolidation pass, or joins the one already +// running. The pass is single-flight and detached from the caller: a manual +// call, a second caller mid-pass, and the quiet timer all share one job, and +// a client going away never kills work already in flight. Both phases are +// checkpointed, so even a server restart resumes where the pass stopped. +func (s *Server) startConsolidation() *consolidationJob { + s.consolMu.Lock() + defer s.consolMu.Unlock() + if s.consolJob != nil { + return s.consolJob + } + job := &consolidationJob{done: make(chan struct{})} + s.consolJob = job go func() { - ctx, cancel := context.WithTimeout(context.Background(), 6*time.Hour) + runCtx, cancel := context.WithTimeout(context.Background(), 6*time.Hour) defer cancel() - n, err := s.Consolidator().Run(ctx, 0) - if err != nil { - slog.Error("consolidate", "error", err) + job.res, job.err = s.Consolidator().Run(runCtx, 0, job.publish) + close(job.done) + s.consolMu.Lock() + s.consolJob = nil + s.consolMu.Unlock() + if job.err != nil { + slog.Error("consolidate", "error", job.err) } else { - slog.Info("consolidate done", "extracted", n) + slog.Info("consolidate done", "extracted", job.res.Extracted, "synthesized", job.res.Synthesized) } }() - writeJSON(w, 202, map[string]string{"status": "started"}) + return job } // --- projects --- diff --git a/internal/httpapi/httpapi_test.go b/internal/httpapi/httpapi_test.go new file mode 100644 index 0000000..fcbdb50 --- /dev/null +++ b/internal/httpapi/httpapi_test.go @@ -0,0 +1,60 @@ +package httpapi + +import ( + "testing" + "time" + + "github.com/taciturnaxolotl/lard/internal/pipeline" +) + +// TestConsolidationJobProgress covers the pub-sub contract the /consolidate +// handler streams from: listeners get every event, buffered delivery means a +// slow listener never blocks the pass, and unsubscribe stops delivery. +func TestConsolidationJobProgress(t *testing.T) { + job := &consolidationJob{done: make(chan struct{})} + + events := job.subscribe() + job.publish(pipeline.ProgressEvent{Phase: "extract", Name: "sess-1", Done: 1, Total: 2}) + job.publish(pipeline.ProgressEvent{Phase: "extract", Name: "sess-2", Done: 2, Total: 2}) + + for i, want := range []string{"sess-1", "sess-2"} { + select { + case ev := <-events: + if ev.Name != want { + t.Fatalf("event %d: got %q, want %q", i, ev.Name, want) + } + case <-time.After(time.Second): + t.Fatalf("event %d: timed out", i) + } + } + + // Unsubscribing stops delivery. + job.unsubscribe(events) + job.publish(pipeline.ProgressEvent{Phase: "synthesize", Name: "areas/x", Done: 1, Total: 1}) + select { + case ev := <-events: + t.Fatalf("got event after unsubscribe: %+v", ev) + default: + } +} + +// TestConsolidationJobDrain covers the finish drain: events published just +// before the pass ends are still readable from the buffered channel after +// done closes, so the summary line is always last. +func TestConsolidationJobDrain(t *testing.T) { + job := &consolidationJob{done: make(chan struct{})} + events := job.subscribe() + + job.publish(pipeline.ProgressEvent{Phase: "synthesize", Name: "areas/x", Done: 1, Total: 1}) + job.res = pipeline.Result{Extracted: 1, Synthesized: 1} + close(job.done) + + select { + case ev := <-events: + if ev.Name != "areas/x" { + t.Fatalf("drained the wrong event: %+v", ev) + } + default: + t.Fatal("buffered event lost on finish") + } +} diff --git a/internal/pipeline/consolidate.go b/internal/pipeline/consolidate.go index 77c453a..d6dc0aa 100644 --- a/internal/pipeline/consolidate.go +++ b/internal/pipeline/consolidate.go @@ -30,23 +30,42 @@ func New(st *store.Store, client *llm.Client, resolve func(*types.ProjectHints) return &Consolidator{store: st, llm: client, Resolve: resolve} } +// Result summarizes one consolidation pass. +type Result struct { + Extracted int // sessions that yielded facts this pass + Synthesized int // subject files rewritten +} + +// ProgressEvent reports one completed step of a pass, so callers can show +// live feedback instead of a single summary at the end. Total is an estimate: +// sessions uploaded mid-pass push it up as later batches are discovered. +type ProgressEvent struct { + Phase string // "extract" or "synthesize" + Name string // the session or subject path that finished + Done int // steps completed so far in this phase + Total int // steps expected in this phase +} + // Run drains all pending work: extract facts from every unextracted session, // then synthesize every dirty subject. Both phases are checkpointed, so a // crash or re-run resumes cleanly. limit>0 caps sessions extracted this run. -func (c *Consolidator) Run(ctx context.Context, limit int) (extracted int, err error) { - extracted, err = c.extractPhase(ctx, limit) +// progress, when non-nil, is called as each step finishes. +func (c *Consolidator) Run(ctx context.Context, limit int, progress func(ProgressEvent)) (Result, error) { + var res Result + extracted, err := c.extractPhase(ctx, limit, progress) + res.Extracted = extracted if err != nil { - return extracted, err - } - if err := c.synthesizePhase(ctx); err != nil { - return extracted, err + return res, err } - return extracted, nil + synthesized, err := c.synthesizePhase(ctx, progress) + res.Synthesized = synthesized + return res, err } // extractPhase runs extraction across unextracted sessions in parallel, // persisting facts per session. Each session is an independent checkpoint. -func (c *Consolidator) extractPhase(ctx context.Context, limit int) (int, error) { +// progress, when non-nil, is called as each session finishes. +func (c *Consolidator) extractPhase(ctx context.Context, limit int, progress func(ProgressEvent)) (int, error) { total := 0 for { if err := ctx.Err(); err != nil { @@ -76,6 +95,8 @@ func (c *Consolidator) extractPhase(ctx context.Context, limit int) (int, error) sem := make(chan struct{}, extractConcurrency) var mu sync.Mutex done := 0 + batchStart := total + batchTotal := total + len(sessions) // best estimate; later batches may follow for _, sess := range sessions { wg.Add(1) sem <- struct{}{} @@ -83,15 +104,22 @@ func (c *Consolidator) extractPhase(ctx context.Context, limit int) (int, error) defer wg.Done() defer func() { <-sem }() if err := c.extractSession(ctx, s, listing); err != nil { + if ctx.Err() != nil { + // The pass is going away; leave the session queued + // so a later pass still gets its facts. + return + } slog.Error("extract: session failed", "session", s.SessionID, "error", err) // Mark extracted-with-no-facts so a poison session doesn't // wedge the queue; it simply contributes nothing. _ = c.store.SaveFacts(s.Source, s.SessionID, time.Unix(s.EndedAt, 0).UTC(), nil) - return } mu.Lock() done++ mu.Unlock() + if progress != nil { + progress(ProgressEvent{Phase: "extract", Name: s.SessionID, Done: batchStart + done, Total: batchTotal}) + } }(sess) } wg.Wait() @@ -190,15 +218,17 @@ func (c *Consolidator) ensureSubject(kind types.SubjectKind, name, desc string, } // synthesizePhase rewrites every subject that has facts newer than its last -// synthesis, in parallel across subjects (each writes a distinct file). -func (c *Consolidator) synthesizePhase(ctx context.Context) error { +// synthesis, in parallel across subjects (each writes a distinct file). It +// returns how many subjects it rewrote. progress, when non-nil, is called as +// each subject finishes. +func (c *Consolidator) synthesizePhase(ctx context.Context, progress func(ProgressEvent)) (int, error) { dirty, err := c.store.DirtySubjects() if err != nil { - return err + return 0, err } if len(dirty) == 0 { slog.Info("synthesize: nothing dirty") - return nil + return 0, nil } var wg sync.WaitGroup sem := make(chan struct{}, synthConcurrency) @@ -217,15 +247,18 @@ func (c *Consolidator) synthesizePhase(ctx context.Context) error { if err != nil { failed++ slog.Error("synthesize: subject failed", "kind", kind, "name", name, "error", err) - return + } else { + ok++ + slog.Info("synthesize: subject done", "kind", kind, "name", name, "progress", ok+failed, "total", len(dirty)) + } + if progress != nil { + progress(ProgressEvent{Phase: "synthesize", Name: types.SubjectPath(kind, name), Done: ok + failed, Total: len(dirty)}) } - ok++ - slog.Info("synthesize: subject done", "kind", kind, "name", name, "progress", ok+failed, "total", len(dirty)) }(kind, name) } wg.Wait() slog.Info("synthesize: complete", "subjects", len(dirty), "written", ok, "failed", failed) - return nil + return ok, nil } func (c *Consolidator) synthesizeSubject(ctx context.Context, kind types.SubjectKind, name string) error {