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
+}

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"))
+}