387 lines
14 KiB
Go
387 lines
14 KiB
Go
// ingestion/internal/watcher/watcher_test.go
|
|
package watcher
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"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_IngestsTheRealNoteAndSkipsTheSymlinkBesideIt(t *testing.T) {
|
|
brainDir := setupBrainDir(t)
|
|
outside := filepath.Join(t.TempDir(), "secret.md")
|
|
require.NoError(t, os.WriteFile(outside, []byte("SENTINEL-153-WATCHER"), 0o644))
|
|
// "a-" sorts before "real": a walk that stopped at the first symlink
|
|
// (filepath.SkipAll — one synced link halting all ingestion) would fail
|
|
// the call count below (PR #156 review).
|
|
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "a-leak.md")))
|
|
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "real.md"), []byte("real note"), 0o644))
|
|
|
|
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",
|
|
}}
|
|
errs := processDir(context.Background(), cfg, time.Now().UTC().Format("2006-01-02"))
|
|
assert.Empty(t, errs)
|
|
assert.False(t, sawSentinel, "the symlink target must never reach the pipeline")
|
|
assert.Equal(t, 1, calls, "exactly the real note is ingested; the symlink is not a note")
|
|
}
|
|
|
|
// A symlinked raw/ is refused outright: os.OpenRoot would otherwise follow
|
|
// it and confine reads to wherever it points.
|
|
func TestProcessDir_RefusesASymlinkedRawDir(t *testing.T) {
|
|
brainDir := t.TempDir()
|
|
elsewhere := t.TempDir()
|
|
require.NoError(t, os.WriteFile(filepath.Join(elsewhere, "x.md"), []byte("SENTINEL-153-RAWLINK"), 0o644))
|
|
require.NoError(t, os.Symlink(elsewhere, filepath.Join(brainDir, "raw")))
|
|
calls := 0
|
|
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
|
|
Complete: func(context.Context, string, string) (string, error) { calls++; return "[]", nil },
|
|
Schema: "# Schema",
|
|
}}
|
|
errs := processDir(context.Background(), cfg, time.Now().UTC().Format("2006-01-02"))
|
|
assert.NotEmpty(t, errs, "a symlinked raw/ must be reported, not silently ingested")
|
|
assert.Zero(t, calls)
|
|
}
|
|
|
|
// PR #156: a FIFO swapped into raw/ between the watcher's checks and its
|
|
// read or archive copy must not hang the watcher, which runs unattended.
|
|
func TestProcessDir_DoesNotHangUnderAFIFOSwapRace(t *testing.T) {
|
|
brainDir := setupBrainDir(t)
|
|
raw := filepath.Join(brainDir, "raw")
|
|
var stop atomic.Bool
|
|
go func() {
|
|
for i := 0; !stop.Load(); i++ {
|
|
tmp := filepath.Join(raw, ".swap")
|
|
_ = os.Remove(tmp)
|
|
if i%2 == 0 {
|
|
_ = os.WriteFile(tmp, []byte("x"), 0o644)
|
|
} else {
|
|
_ = syscall.Mkfifo(tmp, 0o644)
|
|
}
|
|
_ = os.Rename(tmp, filepath.Join(raw, "r.md"))
|
|
}
|
|
}()
|
|
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
|
|
Complete: func(context.Context, string, string) (string, error) {
|
|
b, _ := json.Marshal([]pipeline.RawPage{{Title: "R", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
|
|
return string(b), nil
|
|
},
|
|
Schema: "# Schema",
|
|
}}
|
|
var iters atomic.Int64
|
|
done := make(chan struct{})
|
|
go func() {
|
|
for deadline := time.Now().Add(1500 * time.Millisecond); time.Now().Before(deadline); {
|
|
_ = os.Remove(filepath.Join(raw, "r.md.processed"))
|
|
_ = os.Remove(filepath.Join(raw, "r.md.failed"))
|
|
_ = processDir(context.Background(), cfg, "2026-10-03")
|
|
iters.Add(1)
|
|
}
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
stop.Store(true)
|
|
case <-time.After(8 * time.Second):
|
|
stop.Store(true)
|
|
if f, err := os.OpenFile(filepath.Join(raw, "r.md"), os.O_WRONLY|syscall.O_NONBLOCK, 0); err == nil {
|
|
_ = f.Close()
|
|
}
|
|
t.Fatalf("the watcher hung on a swapped-in FIFO after %d iterations", iters.Load())
|
|
}
|
|
}
|