diff --git a/ingestion/internal/watcher/watcher.go b/ingestion/internal/watcher/watcher.go index 2a951a3..512ee49 100644 --- a/ingestion/internal/watcher/watcher.go +++ b/ingestion/internal/watcher/watcher.go @@ -15,6 +15,7 @@ import ( "github.com/mathiasbq/hyperguild/ingestion/internal/extract" "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" + "github.com/mathiasbq/hyperguild/ingestion/internal/wing" ) // Config holds watcher configuration. @@ -27,6 +28,11 @@ type Config struct { // from raw/ nor marked (e.g. a read-only raw/), so later polls skip them // instead of ingesting them again. In memory only: a restart retries once. stuck *sync.Map + + // Resolver, when set, decides each note's wing (hyperguild#205): explicit + // subfolder/frontmatter, else classifier + scorer. A note it cannot place + // is parked in raw/needs-review/ for a human. Nil keeps the retired layout. + Resolver *wing.Resolver } // Start launches the watcher in a background goroutine. @@ -94,7 +100,7 @@ func processDir(ctx context.Context, cfg Config, date string) []error { // path that treats raw/ as "awaiting processing" — this is that path. if d.IsDir() { name := d.Name() - if name == "processed" || name == "failed" || name == "rejected" { + if name == "processed" || name == "failed" || name == "rejected" || name == "needs-review" { return filepath.SkipDir } return nil @@ -165,7 +171,16 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d return fmt.Errorf("extract text: %w", err) } - _, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false) + pcfg := cfg.Pipeline + if cfg.Resolver != nil { + d := cfg.Resolver.Resolve(ctx, wing.Note{Dir: topDir(rel), FrontmatterWing: frontmatterWing(content), Text: content}) + if d.Source == wing.Exception { + return parkForReview(cfg, root, rawDir, rel, path, filename, content, d, date) + } + pcfg.Wing = d.Wing + slog.Info("watcher: wing resolved", "file", filename, "wing", d.Wing, "how", d.Source, "reason", d.Reason) + } + _, runErr := pipeline.Run(ctx, pcfg, cfg.BrainDir, content, source, false) if runErr != nil { // Copy to failed/ and leave a .failed marker so we don't retry. failedDir := filepath.Join(cfg.BrainDir, "raw", "failed") @@ -308,3 +323,65 @@ func copyFromRoot(root *os.Root, rel, dst string) error { } return out.Close() } + +// topDir is the first directory of a raw/-relative path, "" for a top-level file. +func topDir(rel string) string { + if i := strings.IndexByte(filepath.ToSlash(rel), '/'); i > 0 { + return filepath.ToSlash(rel)[:i] + } + return "" +} + +// frontmatterWing reads `wing:` from a note's leading frontmatter, "" if none. +func frontmatterWing(content string) string { + if !strings.HasPrefix(content, "---\n") { + return "" + } + block, _, _ := strings.Cut(content[4:], "\n---") + for _, line := range strings.Split(block, "\n") { + if key, val, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(key) == "wing" { + return strings.Trim(strings.TrimSpace(val), `"'`) + } + } + return "" +} + +// parkForReview moves a note the resolver could not place into +// raw/needs-review/, recording the proposal in its frontmatter so a human can +// approve it (brain_promote). Nothing is extracted or written to the wiki. The +// original is removed only after the parked copy is written. +func parkForReview(cfg Config, root *os.Root, rawDir, rel, path, filename, content string, d wing.Decision, date string) error { + dir := filepath.Join(rawDir, "needs-review") + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("mkdir needs-review: %w", err) + } + dest := filepath.Join(dir, filename) + if _, err := os.Stat(dest); err == nil { + ext := filepath.Ext(filename) + dest = filepath.Join(dir, strings.TrimSuffix(filename, ext)+"-"+time.Now().UTC().Format("150405")+ext) + } + if strings.ToLower(filepath.Ext(filename)) == ".pdf" { + // Binary: park the bytes untouched; the proposal is in log.md only. + if err := copyFromRoot(root, rel, dest); err != nil { + return fmt.Errorf("park pdf: %w", err) + } + } else if err := os.WriteFile(dest, []byte(withProposal(content, d)), 0o644); err != nil { + return fmt.Errorf("park note: %w", err) + } + if err := appendWatcherLog(cfg.BrainDir, filename, fmt.Errorf("needs review (proposed wing %q): %s", d.Wing, d.Reason), date); err != nil { + slog.Error("watcher: failed to write log entry", "error", err) + } + slog.Info("watcher: parked for review", "file", filename, "proposed_wing", d.Wing, "reason", d.Reason) + retire(cfg, root, rel, path, ".processed", date) + return nil +} + +// withProposal adds the resolver's proposal to a note's frontmatter, creating +// the block when the note has none. +func withProposal(content string, d wing.Decision) string { + kv := fmt.Sprintf("proposed_wing: %s\nproposal_confidence: %.2f\nproposal_reason: %q\n", d.Wing, d.Confidence, d.Reason) + if strings.HasPrefix(content, "---\n") { + return "---\n" + kv + content[4:] + } + return "---\n" + kv + "---\n" + content +} diff --git a/ingestion/internal/watcher/watcher_wing_test.go b/ingestion/internal/watcher/watcher_wing_test.go new file mode 100644 index 0000000..46cea24 --- /dev/null +++ b/ingestion/internal/watcher/watcher_wing_test.go @@ -0,0 +1,146 @@ +package watcher + +import ( + "context" + "errors" + "os" + "path/filepath" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/mathiasbq/hyperguild/ingestion/internal/pipeline" + "github.com/mathiasbq/hyperguild/ingestion/internal/wing" +) + +type stubClassifier struct { + s wing.Suggestion + err error + calls atomic.Int32 +} + +func (c *stubClassifier) Classify(context.Context, string, []string) (wing.Suggestion, error) { + c.calls.Add(1) + return c.s, c.err +} + +type stubScorer map[string]float64 + +func (s stubScorer) Score(string) map[string]float64 { return s } + +func wingResolver(c wing.Classifier, s wing.Scorer) *wing.Resolver { + return &wing.Resolver{ + Known: []string{"radar", "homelab", "client-x"}, + Eligible: func(w string) bool { return w != "client-x" }, + Classifier: c, + Scorer: s, + Threshold: 0.8, + } +} + +func wingTestCfg(t *testing.T, r *wing.Resolver, llmCalls *atomic.Int32) (Config, string) { + t.Helper() + brainDir := setupBrainDir(t) + return Config{ + BrainDir: brainDir, + Resolver: r, + Pipeline: pipeline.Config{ + Schema: "# Schema", + Complete: func(ctx context.Context, s, u string) (string, error) { + llmCalls.Add(1) + return successComplete(pipeline.RawPage{Title: "Talk", Type: "source", Subtype: "video", Content: "## Summary\n\nx\n"})(ctx, s, u) + }, + }, + }, brainDir +} + +func put(t *testing.T, brainDir, rel, body string) { + t.Helper() + full := filepath.Join(brainDir, "raw", rel) + require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755)) + require.NoError(t, os.WriteFile(full, []byte(body), 0o644)) +} + +func has(brainDir, rel string) bool { + _, err := os.Stat(filepath.Join(brainDir, rel)) + return err == nil +} + +func TestProcessDir_ASubfolderNamesTheWing(t *testing.T) { + var llmCalls atomic.Int32 + clf := &stubClassifier{} + cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{}), &llmCalls) + put(t, brainDir, "radar/dark-factory.md", "A talk about factories.") + + errs := processDir(context.Background(), cfg, "2026-10-20") + + require.Empty(t, errs) + assert.True(t, has(brainDir, "wiki/radar/sources/dark-factory.md"), "lands in the named wing") + assert.False(t, has(brainDir, "raw/radar/dark-factory.md"), "inbox zero holds for subfolders too") + assert.Zero(t, clf.calls.Load(), "an explicit wing costs no classification") +} + +func TestProcessDir_ANoteTheResolverCannotPlaceIsParkedNotIngested(t *testing.T) { + var llmCalls atomic.Int32 + clf := &stubClassifier{s: wing.Suggestion{Wing: "radar", Confidence: 0.5, Reason: "maybe"}} + cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{"radar": 3}), &llmCalls) + put(t, brainDir, "unsure.md", "Some note.") + + errs := processDir(context.Background(), cfg, "2026-10-20") + + require.Empty(t, errs) + assert.Zero(t, llmCalls.Load(), "no extraction runs for an unplaced note") + assert.False(t, has(brainDir, "raw/unsure.md"), "leaves the top of raw/") + parked, err := os.ReadFile(filepath.Join(brainDir, "raw", "needs-review", "unsure.md")) + require.NoError(t, err) + assert.Contains(t, string(parked), "proposed_wing: radar") + assert.Contains(t, string(parked), "below threshold") + assert.Contains(t, string(parked), "Some note.", "the note body is preserved") +} + +func TestProcessDir_ANoteInAClientWingFolderIsParked(t *testing.T) { + var llmCalls atomic.Int32 + cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{}, stubScorer{}), &llmCalls) + put(t, brainDir, "client-x/secret.md", "Client material.") + + require.Empty(t, processDir(context.Background(), cfg, "2026-10-20")) + + assert.Zero(t, llmCalls.Load()) + assert.False(t, has(brainDir, "wiki/client-x"), "nothing is ever written into an ineligible wing") + assert.True(t, has(brainDir, "raw/needs-review/secret.md")) +} + +func TestProcessDir_ClassifierFailureParksTheNote(t *testing.T) { + var llmCalls atomic.Int32 + cfg, brainDir := wingTestCfg(t, wingResolver(&stubClassifier{err: errors.New("llm down")}, stubScorer{}), &llmCalls) + put(t, brainDir, "n.md", "Note.") + + require.Empty(t, processDir(context.Background(), cfg, "2026-10-20")) + + assert.Zero(t, llmCalls.Load()) + assert.True(t, has(brainDir, "raw/needs-review/n.md")) +} + +func TestProcessDir_ParkedNotesAreNotReprocessed(t *testing.T) { + var llmCalls atomic.Int32 + clf := &stubClassifier{s: wing.Suggestion{Wing: "radar", Confidence: 0.99}} + cfg, brainDir := wingTestCfg(t, wingResolver(clf, stubScorer{"radar": 3}), &llmCalls) + put(t, brainDir, "needs-review/waiting.md", "---\nproposed_wing: radar\n---\nA parked note.") + + require.Empty(t, processDir(context.Background(), cfg, "2026-10-20")) + + assert.Zero(t, clf.calls.Load(), "needs-review/ is a human queue, like rejected/") + assert.True(t, has(brainDir, "raw/needs-review/waiting.md")) +} + +func TestProcessDir_WithoutAResolverTheRetiredLayoutIsUnchanged(t *testing.T) { + var llmCalls atomic.Int32 + cfg, brainDir := wingTestCfg(t, nil, &llmCalls) + put(t, brainDir, "plain.md", "Plain note.") + + require.Empty(t, processDir(context.Background(), cfg, "2026-10-20")) + + assert.True(t, has(brainDir, "wiki/sources/plain.md")) +}