hyperguild/ingestion/internal/watcher/watcher_test.go

// ingestion/internal/watcher/watcher_test.go
package watcher

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
)

// successComplete returns a valid JSON-encoded RawPage array for any call.
func successComplete(raw pipeline.RawPage) pipeline.CompleteFunc {
	return func(ctx context.Context, system, user string) (string, error) {
		b, err := json.Marshal([]pipeline.RawPage{raw})
		if err != nil {
			return "", err
		}
		return string(b), nil
	}
}

// errorComplete always returns an error simulating an LLM failure.
func errorComplete(_ context.Context, _, _ string) (string, error) {
	return "", fmt.Errorf("LLM unavailable")
}

func setupBrainDir(t *testing.T) string {
	t.Helper()
	brainDir := t.TempDir()
	for _, sub := range []string{"wiki/concepts", "wiki/entities", "wiki/sources", "raw"} {
		require.NoError(t, os.MkdirAll(filepath.Join(brainDir, sub), 0o755))
	}
	return brainDir
}

func TestStart_ProcessesFile(t *testing.T) {
	brainDir := setupBrainDir(t)

	// Place a .md file in raw/.
	rawFile := filepath.Join(brainDir, "raw", "shape-up-book.md")
	require.NoError(t, os.WriteFile(rawFile, []byte("Content about Shape Up."), 0o644))

	date := time.Now().UTC().Format("2006-01-02")
	rawPage := pipeline.RawPage{
		Title:   "Shape Up Book",
		Type:    "source",
		Subtype: "article",
		Domain:  "product-management",
		Content: "## Summary\n\nA book about Shape Up.\n",
	}

	cfg := Config{
		BrainDir: brainDir,
		Interval: 50 * time.Millisecond,
		Pipeline: pipeline.Config{
			Complete:  successComplete(rawPage),
			ChunkSize: 0,
			Schema:    "# Schema\nThree page types.",
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	Start(ctx, cfg)

	// Poll until the file is moved to processed/.
	processedPath := filepath.Join(brainDir, "raw", "processed", date, "shape-up-book.md")
	var found bool
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(processedPath); err == nil {
			found = true
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	require.True(t, found, "file should be copied to processed/")

	// Original file should still exist (copy, not move — keeps Obsidian vault intact).
	_, err := os.Stat(rawFile)
	assert.NoError(t, err, "original file should remain in raw/")

	// A .processed marker should exist next to the original.
	_, err = os.Stat(rawFile + ".processed")
	assert.NoError(t, err, ".processed marker should be written")

	// Wiki page should exist.
	wikiPath := filepath.Join(brainDir, "wiki", "sources", "shape-up-book.md")
	_, err = os.Stat(wikiPath)
	assert.NoError(t, err, "wiki page should be written")

	// log.md should contain an ingest record.
	logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md"))
	require.NoError(t, err)
	assert.Contains(t, string(logContent), "— ingest")
}

func TestStart_MovesToFailedOnError(t *testing.T) {
	brainDir := setupBrainDir(t)

	rawFile := filepath.Join(brainDir, "raw", "bad-file.md")
	require.NoError(t, os.WriteFile(rawFile, []byte("Some content."), 0o644))

	cfg := Config{
		BrainDir: brainDir,
		Interval: 50 * time.Millisecond,
		Pipeline: pipeline.Config{
			Complete:  errorComplete,
			ChunkSize: 0,
			Schema:    "# Schema\nThree page types.",
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	Start(ctx, cfg)

	// Poll until the file is moved to failed/.
	failedPath := filepath.Join(brainDir, "raw", "failed", "bad-file.md")
	var found bool
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if _, err := os.Stat(failedPath); err == nil {
			found = true
			break
		}
		time.Sleep(20 * time.Millisecond)
	}
	require.True(t, found, "file should be copied to failed/")

	// Original file should still exist (copy, not move — keeps Obsidian vault intact).
	_, err := os.Stat(rawFile)
	assert.NoError(t, err, "original file should remain in raw/")

	// A .failed marker should exist next to the original.
	_, err = os.Stat(rawFile + ".failed")
	assert.NoError(t, err, ".failed marker should be written")

	// log.md should contain a watcher error entry.
	logContent, err := os.ReadFile(filepath.Join(brainDir, "log.md"))
	require.NoError(t, err)
	assert.Contains(t, string(logContent), "— watcher error")
	assert.Contains(t, string(logContent), "bad-file.md")
}

func TestDeriveSource(t *testing.T) {
	tests := []struct {
		filename string
		want     string
	}{
		{"shape-up-book.md", "Shape Up Book"},
		{"raft-consensus.txt", "Raft Consensus"},
		{"my-note.md", "My Note"},
		{"single.md", "Single"},
		{"no-extension", "No Extension"},
	}

	for _, tc := range tests {
		t.Run(tc.filename, func(t *testing.T) {
			got := deriveSource(tc.filename)
			assert.Equal(t, tc.want, got)
		})
	}
}

func TestProcessDir_SkipsSubdirs(t *testing.T) {
	brainDir := setupBrainDir(t)

	// Create processed/, failed/, and rejected/ subdirs with files inside.
	for _, sub := range []string{"processed/2026-04-22", "failed", "rejected"} {
		require.NoError(t, os.MkdirAll(filepath.Join(brainDir, "raw", sub), 0o755))
	}

	processedFile := filepath.Join(brainDir, "raw", "processed", "2026-04-22", "old-file.md")
	failedFile := filepath.Join(brainDir, "raw", "failed", "broken-file.md")
	// hyperguild#81/#133: a human already declined this note via brain_reject.
	// If the watcher walks into rejected/, it re-ingests and publishes it to
	// the wiki — the exact opposite of the human's decision.
	rejectedFile := filepath.Join(brainDir, "raw", "rejected", "declined-file.md")
	require.NoError(t, os.WriteFile(processedFile, []byte("old"), 0o644))
	require.NoError(t, os.WriteFile(failedFile, []byte("broken"), 0o644))
	require.NoError(t, os.WriteFile(rejectedFile, []byte("---\nrejected_at: 2026-09-23T00:00:00Z\n---\ndeclined"), 0o644))

	// Also place a valid file in raw/ root that should be processed.
	validFile := filepath.Join(brainDir, "raw", "valid.md")
	require.NoError(t, os.WriteFile(validFile, []byte("valid content"), 0o644))

	date := time.Now().UTC().Format("2006-01-02")

	// Track which sources were passed to Complete.
	var processedSources []string
	completeFn := func(ctx context.Context, system, user string) (string, error) {
		// Record that this was called; return a minimal valid RawPage.
		raw := pipeline.RawPage{
			Title:   "Valid",
			Type:    "source",
			Subtype: "article",
			Content: "## Summary\n\nValid.\n",
		}
		b, _ := json.Marshal([]pipeline.RawPage{raw})
		processedSources = append(processedSources, "called")
		return string(b), nil
	}

	cfg := Config{
		BrainDir: brainDir,
		Interval: time.Hour, // not used; we call processDir directly
		Pipeline: pipeline.Config{
			Complete:  completeFn,
			ChunkSize: 0,
			Schema:    "# Schema\nThree page types.",
		},
	}

	errs := processDir(context.Background(), cfg, date)
	assert.Empty(t, errs, "no errors expected")

	// Complete should have been called exactly once (for valid.md, not for files in subdirs).
	assert.Len(t, processedSources, 1, "only the file in raw/ root should be processed")

	// Files in processed/, failed/, and rejected/ must remain untouched.
	_, err := os.Stat(processedFile)
	assert.NoError(t, err, "processed subdir file should be untouched")
	_, err = os.Stat(failedFile)
	assert.NoError(t, err, "failed subdir file should be untouched")
	_, err = os.Stat(rejectedFile)
	assert.NoError(t, err, "rejected subdir file should be untouched, never re-ingested")
}

// TestProcessDir_SkipsTunnelCandidateFiles guards against hyperguild#88:
// AutoTunnel's own human-review queue (brain/raw/tunnel-candidates-*.md)
// must never be re-ingested as if it were external source material — doing
// so fed the raw "(term: X)" log entries back through the LLM extraction
// pipeline, which then legitimately (from its own perspective) surfaced
// [[X]] as a wikilink for any term that happened to match a real page
// title, however generic (e.g. "mission").
func TestProcessDir_SkipsTunnelCandidateFiles(t *testing.T) {
	brainDir := setupBrainDir(t)

	tunnelFile := filepath.Join(brainDir, "raw", "tunnel-candidates-2026-07-19.md")
	require.NoError(t, os.WriteFile(tunnelFile, []byte(
		"# Tunnel candidates 2026-07-19\n\n- `wiki/homelab/decisions/foo.md` ↔ `wiki/telos/decisions/mission.md` (term: \"mission\")\n",
	), 0o644))

	var completeCalls int
	completeFn := func(ctx context.Context, system, user string) (string, error) {
		completeCalls++
		raw := pipeline.RawPage{Title: "Should not be written", Type: "source", Subtype: "article", Content: "## Summary\n\nx.\n"}
		b, _ := json.Marshal([]pipeline.RawPage{raw})
		return string(b), nil
	}

	cfg := Config{
		BrainDir: brainDir,
		Interval: time.Hour, // not used; we call processDir directly
		Pipeline: pipeline.Config{
			Complete:  completeFn,
			ChunkSize: 0,
			Schema:    "# Schema\nThree page types.",
		},
	}

	date := time.Now().UTC().Format("2006-01-02")
	errs := processDir(context.Background(), cfg, date)
	assert.Empty(t, errs)

	assert.Zero(t, completeCalls, "tunnel-candidates file must never reach the LLM extraction pipeline")

	// File must be left alone in raw/ — not moved to processed/, no marker written.
	_, err := os.Stat(tunnelFile + ".processed")
	assert.True(t, os.IsNotExist(err), "tunnel-candidates file should not get a .processed marker")
	_, err = os.Stat(filepath.Join(brainDir, "raw", "processed", date, "tunnel-candidates-2026-07-19.md"))
	assert.True(t, os.IsNotExist(err), "tunnel-candidates file should not be copied to processed/")
}

// hyperguild#153: raw/ is writable via Syncthing, and the watcher ingests
// whatever lands there with no API call at all. WalkDir yields symlinked
// files, and extract.Text followed them, so a symlink to any pod-readable
// file would be extracted into the brain. The watcher now ingests only
// regular files, read through an os.Root on raw/.
func TestProcessDir_DoesNotIngestSymlinks(t *testing.T) {
	brainDir := setupBrainDir(t)
	outside := filepath.Join(t.TempDir(), "secret.md")
	require.NoError(t, os.WriteFile(outside, []byte("SENTINEL-153-WATCHER"), 0o644))
	require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "leak.md")))

	var sawSentinel bool
	calls := 0
	cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
		Complete: func(_ context.Context, _, user string) (string, error) {
			calls++
			if strings.Contains(user, "SENTINEL-153-WATCHER") {
				sawSentinel = true
			}
			b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
			return string(b), nil
		},
		Schema: "# Schema",
	}}
	_ = processDir(context.Background(), cfg, time.Now().UTC().Format("2006-01-02"))
	assert.False(t, sawSentinel, "the symlink target must never reach the pipeline")
	assert.Zero(t, calls, "a symlink is not a note")
}