Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac84ddf5a9 | ||
|
|
db18785bfc | ||
|
|
138d6e2545 | ||
|
|
41b7105b31 |
@@ -17,7 +17,6 @@ import (
|
|||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/audit"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/audit"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/capture"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/capture"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/capturehttp"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/classification"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/classification"
|
||||||
@@ -90,15 +89,11 @@ func (s *claudeSink) Ingest(ctx context.Context, b claudewatcher.Batch) error {
|
|||||||
sb.WriteString("\n\n")
|
sb.WriteString("\n\n")
|
||||||
}
|
}
|
||||||
slug := "session-" + b.Host + "-" + b.SessionID
|
slug := "session-" + b.Host + "-" + b.SessionID
|
||||||
dest := "archive/claude-sessions/" + b.Host + "/" + slug + ".md"
|
dest := filepath.Join(s.brainDir, "archive", "claude-sessions", b.Host, slug+".md")
|
||||||
// Written through the brain root, never through a symlinked directory
|
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
||||||
// or file (hyperguild#158).
|
return fmt.Errorf("create claude-sessions archive dir: %w", err)
|
||||||
bfs, err := brainfs.Open(s.brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("open brain: %w", err)
|
|
||||||
}
|
}
|
||||||
defer func() { _ = bfs.Close() }()
|
if err := os.WriteFile(dest, []byte(sb.String()), 0o644); err != nil {
|
||||||
if err := bfs.PutRegular(dest, []byte(sb.String())); err != nil {
|
|
||||||
return fmt.Errorf("write claude session archive: %w", err)
|
return fmt.Errorf("write claude session archive: %w", err)
|
||||||
}
|
}
|
||||||
s.logger.Debug("claude session archived (non-indexed)", "path", dest)
|
s.logger.Debug("claude session archived (non-indexed)", "path", dest)
|
||||||
|
|||||||
@@ -76,27 +76,3 @@ func TestAlertingWarnings(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// hyperguild#158: the claude-session archive was written with os.MkdirAll +
|
|
||||||
// os.WriteFile by path, so a symlinked archive directory redirected it.
|
|
||||||
func TestClaudeSinkDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) {
|
|
||||||
for name, mk := range map[string]func(t *testing.T, dir string) (link, target string){
|
|
||||||
"outside": func(t *testing.T, _ string) (string, string) { o := t.TempDir(); return o, o },
|
|
||||||
"in-brain": func(t *testing.T, dir string) (string, string) {
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755))
|
|
||||||
return filepath.Join("..", "raw"), filepath.Join(dir, "raw")
|
|
||||||
},
|
|
||||||
} {
|
|
||||||
t.Run(name, func(t *testing.T) {
|
|
||||||
dir := t.TempDir()
|
|
||||||
link, target := mk(t, dir)
|
|
||||||
require.NoError(t, os.Symlink(link, filepath.Join(dir, "archive")))
|
|
||||||
sink := &claudeSink{brainDir: dir, logger: slog.New(slog.NewTextHandler(os.Stderr, nil))}
|
|
||||||
err := sink.Ingest(context.Background(), claudewatcher.Batch{Host: "koala", SessionID: "abc",
|
|
||||||
Turns: []claudewatcher.Turn{{Type: "assistant", Content: "x"}}})
|
|
||||||
assert.Error(t, err)
|
|
||||||
ents, _ := os.ReadDir(target)
|
|
||||||
assert.Empty(t, ents, "the archive was written at the link's target")
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"syscall"
|
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -64,13 +63,3 @@ func TestStartupSymlinkCheckRefusesAnUnlistableBrainRoot(t *testing.T) {
|
|||||||
t.Cleanup(func() { _ = os.Chmod(brain, 0o755) })
|
t.Cleanup(func() { _ = os.Chmod(brain, 0o755) })
|
||||||
assert.Error(t, startupSymlinkCheck(brain))
|
assert.Error(t, startupSymlinkCheck(brain))
|
||||||
}
|
}
|
||||||
|
|
||||||
// hyperguild#179: a FIFO in the brain is refused at startup with a clear
|
|
||||||
// message, instead of hanging or exiting later in a reader.
|
|
||||||
func TestStartupSymlinkCheckRefusesABrainWithAFIFO(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(brain, "classification.yaml"), 0o644))
|
|
||||||
err := startupSymlinkCheck(brain)
|
|
||||||
require.Error(t, err)
|
|
||||||
assert.Contains(t, err.Error(), "classification.yaml")
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,34 +0,0 @@
|
|||||||
package api_test
|
|
||||||
|
|
||||||
import (
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/api"
|
|
||||||
)
|
|
||||||
|
|
||||||
// hyperguild#178: an unreadable directory under wiki/ made the link check
|
|
||||||
// fail and brain_write refuse the note, losing the learning. The note is now
|
|
||||||
// written with its links as given.
|
|
||||||
func TestWriteNoteKeepsTheLearningWhenTheLinkCheckCannotIndex(t *testing.T) {
|
|
||||||
if os.Geteuid() == 0 {
|
|
||||||
t.Skip("root ignores directory permissions")
|
|
||||||
}
|
|
||||||
brain := t.TempDir()
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w", "facts"), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "w", "facts", "real.md"), []byte("x"), 0o644))
|
|
||||||
locked := filepath.Join(brain, "wiki", "other", "locked")
|
|
||||||
require.NoError(t, os.MkdirAll(locked, 0o755))
|
|
||||||
require.NoError(t, os.Chmod(locked, 0o000))
|
|
||||||
t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
|
|
||||||
|
|
||||||
rel, err := api.WriteNote(brain, api.WriteNoteOptions{Wing: "w", Hall: "facts", Filename: "n", Content: "learning [[real]] and [[gone]]"})
|
|
||||||
require.NoError(t, err)
|
|
||||||
got, err := os.ReadFile(filepath.Join(brain, filepath.FromSlash(rel)))
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Contains(t, string(got), "[[gone]]", "links kept as given when the index is partial")
|
|
||||||
}
|
|
||||||
@@ -47,12 +47,7 @@ var (
|
|||||||
func StripUnresolvedLinks(brainDir, content string) (string, []string, error) {
|
func StripUnresolvedLinks(brainDir, content string) (string, []string, error) {
|
||||||
index, err := entryIndex(brainDir)
|
index, err := entryIndex(brainDir)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// A partial index cannot tell an invented link from one into the
|
return "", nil, err
|
||||||
// part it could not read, so unwrap nothing, and do not refuse the
|
|
||||||
// write either: this runs before every brain_write, and a refused
|
|
||||||
// write is a lost learning (hyperguild#178).
|
|
||||||
slog.Warn("link check skipped: cannot index the brain", "err", err)
|
|
||||||
return content, nil, nil
|
|
||||||
}
|
}
|
||||||
if len(index) == 0 {
|
if len(index) == 0 {
|
||||||
return content, nil, nil
|
return content, nil, nil
|
||||||
|
|||||||
@@ -1,41 +0,0 @@
|
|||||||
package brain_test
|
|
||||||
|
|
||||||
import (
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
|
|
||||||
)
|
|
||||||
|
|
||||||
// hyperguild#178: StripUnresolvedLinks runs before every brain_write and
|
|
||||||
// brain_update. An unreadable directory under knowledge/ or wiki/ made it
|
|
||||||
// error, the write was refused and the learning lost. When the link index
|
|
||||||
// cannot be built, the content is now returned unchanged (no link is
|
|
||||||
// unwrapped, none is invented) and no error stops the write.
|
|
||||||
func TestStripUnresolvedLinksKeepsContentWhenTheIndexCannotBeBuilt(t *testing.T) {
|
|
||||||
if os.Geteuid() == 0 {
|
|
||||||
t.Skip("root ignores directory permissions")
|
|
||||||
}
|
|
||||||
b := t.TempDir()
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(b, "wiki", "w", "facts"), 0o755))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(b, "wiki", "w", "facts", "real.md"), []byte("x"), 0o644))
|
|
||||||
locked := filepath.Join(b, "wiki", "other", "locked")
|
|
||||||
require.NoError(t, os.MkdirAll(locked, 0o755))
|
|
||||||
require.NoError(t, os.Chmod(locked, 0o000))
|
|
||||||
t.Cleanup(func() { _ = os.Chmod(locked, 0o755) })
|
|
||||||
|
|
||||||
in := "learning [[real]] and [[gone]]"
|
|
||||||
out, unlinked, err := brain.StripUnresolvedLinks(b, in)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, in, out, "content unchanged: no link unwrapped on a partial index")
|
|
||||||
assert.Empty(t, unlinked)
|
|
||||||
|
|
||||||
require.NoError(t, os.Chmod(locked, 0o755))
|
|
||||||
out, _, err = brain.StripUnresolvedLinks(b, in)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.NotContains(t, out, "[[gone]]", "positive control: with a readable tree the strip runs")
|
|
||||||
}
|
|
||||||
@@ -109,7 +109,7 @@ func (b *Brain) ReadDirRegular(relDir string) ([]string, error) {
|
|||||||
// after following a symlink, and a ".." after a symlinked directory would
|
// after following a symlink, and a ".." after a symlinked directory would
|
||||||
// otherwise pass the check and redirect the read or write.
|
// otherwise pass the check and redirect the read or write.
|
||||||
func cleanRel(rel string) error {
|
func cleanRel(rel string) error {
|
||||||
if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) {
|
if rel == "" || path.Clean(rel) != rel || !filepath.IsLocal(rel) || strings.Contains(rel, "\\") {
|
||||||
return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular)
|
return fmt.Errorf("%q: not a clean brain-relative path: %w", rel, ErrNotRegular)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -243,27 +243,6 @@ func (b *Brain) RealParent(rel string) error {
|
|||||||
return b.realDir(path.Dir(rel))
|
return b.realDir(path.Dir(rel))
|
||||||
}
|
}
|
||||||
|
|
||||||
// PutFile opens the brain dir as a root and calls PutRegular, for one-off
|
|
||||||
// writers that hold no Brain.
|
|
||||||
func PutFile(brainDir, rel string, data []byte) error {
|
|
||||||
b, err := Open(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
defer func() { _ = b.Close() }()
|
|
||||||
return b.PutRegular(rel, data)
|
|
||||||
}
|
|
||||||
|
|
||||||
// AppendFile opens the brain dir as a root and calls AppendRegular.
|
|
||||||
func AppendFile(brainDir, rel string, data []byte) error {
|
|
||||||
b, err := Open(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
defer func() { _ = b.Close() }()
|
|
||||||
return b.AppendRegular(rel, data)
|
|
||||||
}
|
|
||||||
|
|
||||||
// mkdirReal makes sure every component of rel exists as a real directory,
|
// mkdirReal makes sure every component of rel exists as a real directory,
|
||||||
// creating missing ones through the root; an existing component that is
|
// creating missing ones through the root; an existing component that is
|
||||||
// not a real directory is ErrNotRegular.
|
// not a real directory is ErrNotRegular.
|
||||||
|
|||||||
@@ -529,20 +529,3 @@ func TestCreateRegularCreatesOnlyNewFilesThroughRealDirectories(t *testing.T) {
|
|||||||
got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md"))
|
got, _ = os.ReadFile(filepath.Join(dir, "wiki", "new", "facts", "n.md"))
|
||||||
assert.Equal(t, "one", string(got))
|
assert.Equal(t, "one", string(got))
|
||||||
}
|
}
|
||||||
|
|
||||||
// On Linux a backslash is an ordinary filename byte, not a separator, so a
|
|
||||||
// name containing one stays a single in-root component (hyperguild#158: the
|
|
||||||
// watcher retried such a note forever while cleanRel refused it).
|
|
||||||
func TestReadAndPutRegularTreatABackslashAsAFilenameByte(t *testing.T) {
|
|
||||||
dir := t.TempDir()
|
|
||||||
b, err := Open(dir)
|
|
||||||
require.NoError(t, err)
|
|
||||||
defer func() { _ = b.Close() }()
|
|
||||||
require.NoError(t, b.PutRegular(`a\..\b.md`, []byte("x")))
|
|
||||||
got, err := os.ReadFile(filepath.Join(dir, `a\..\b.md`))
|
|
||||||
require.NoError(t, err, "the name was not kept as one component")
|
|
||||||
assert.Equal(t, "x", string(got))
|
|
||||||
got, err = b.ReadRegular(`a\..\b.md`)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "x", string(got))
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -88,6 +88,7 @@ func TestNoPackageReferencesExtractTextOutsideExtract(t *testing.T) {
|
|||||||
// osOpenAllowed names the functions in guarded files that may call os.Open*
|
// osOpenAllowed names the functions in guarded files that may call os.Open*
|
||||||
// on a path that is not a raw/ read.
|
// on a path that is not a raw/ read.
|
||||||
var osOpenAllowed = map[string]bool{
|
var osOpenAllowed = map[string]bool{
|
||||||
|
"internal/watcher/watcher.go:appendWatcherLog": true, // ingest log append, not a raw/ read
|
||||||
// Text reads by path; outside this package only TextIn's temp copy reaches
|
// Text reads by path; outside this package only TextIn's temp copy reaches
|
||||||
// it (TestNoPackageReferencesExtractTextOutsideExtract enforces that).
|
// it (TestNoPackageReferencesExtractTextOutsideExtract enforces that).
|
||||||
"internal/extract/extract.go:Text": true,
|
"internal/extract/extract.go:Text": true,
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
|
||||||
@@ -88,7 +90,8 @@ func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
for relPath, page := range pending {
|
for relPath, page := range pending {
|
||||||
if err := bfs.PutRegular(relPath, []byte(page.Content)); err != nil {
|
dest := filepath.Join(brainDir, filepath.FromSlash(relPath))
|
||||||
|
if err := os.WriteFile(dest, []byte(page.Content), 0o644); err != nil {
|
||||||
return 0, fmt.Errorf("write %s: %w", relPath, err)
|
return 0, fmt.Errorf("write %s: %w", relPath, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,6 +9,8 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -126,6 +128,10 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
|
|||||||
var written []string
|
var written []string
|
||||||
for _, page := range merged {
|
for _, page := range merged {
|
||||||
if !dryRun {
|
if !dryRun {
|
||||||
|
dest := filepath.Join(brainDir, filepath.FromSlash(page.Path))
|
||||||
|
if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
|
||||||
|
return Result{}, fmt.Errorf("mkdir for %s: %w", page.Path, err)
|
||||||
|
}
|
||||||
content, conflicts, err := reviseOrCreate(brainDir, page)
|
content, conflicts, err := reviseOrCreate(brainDir, page)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Result{}, err
|
return Result{}, err
|
||||||
@@ -144,9 +150,7 @@ func buildAndWrite(ctx context.Context, rawPages []RawPage, sourceSlug, date, br
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
|
return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
|
||||||
}
|
}
|
||||||
// Written through the brain root, never through a symlinked
|
if err := os.WriteFile(dest, []byte(stamped), 0o644); err != nil {
|
||||||
// directory or page (hyperguild#158).
|
|
||||||
if err := writeBrainFile(brainDir, page.Path, []byte(stamped)); err != nil {
|
|
||||||
return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
|
return Result{}, fmt.Errorf("write %s: %w", page.Path, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -197,17 +201,6 @@ func readSourcePage(brainDir, sourceSlug string) string {
|
|||||||
return string(raw)
|
return string(raw)
|
||||||
}
|
}
|
||||||
|
|
||||||
// writeBrainFile creates or replaces one brain-relative file through the
|
|
||||||
// brain root (hyperguild#158).
|
|
||||||
func writeBrainFile(brainDir, rel string, data []byte) error {
|
|
||||||
bfs, err := brainfs.Open(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
defer func() { _ = bfs.Close() }()
|
|
||||||
return bfs.PutRegular(rel, data)
|
|
||||||
}
|
|
||||||
|
|
||||||
// readBrainFile reads one brain-relative file through the brain root,
|
// readBrainFile reads one brain-relative file through the brain root,
|
||||||
// regular files only (hyperguild#157).
|
// regular files only (hyperguild#157).
|
||||||
func readBrainFile(brainDir, rel string) ([]byte, error) {
|
func readBrainFile(brainDir, rel string) ([]byte, error) {
|
||||||
|
|||||||
@@ -165,53 +165,3 @@ func TestBackfillRefsDoesNotCopyAPageSwappedInBehindASymlink(t *testing.T) {
|
|||||||
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
|
require.Contains(t, read(), "[[s|S]]", "positive control: the regular page gets the source ref")
|
||||||
brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
|
brainfstest.AssertNoLeakFrom(t, page, inside, target, 3*time.Second, read)
|
||||||
}
|
}
|
||||||
|
|
||||||
// hyperguild#158: BackfillRefs read a regular concept page, a symlink could
|
|
||||||
// be swapped in, and os.WriteFile then followed it and overwrote the link's
|
|
||||||
// target. The target is not restored between rounds here, so any
|
|
||||||
// write-through shows.
|
|
||||||
func TestBackfillRefsNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "raw")
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "sources", "s.md"), []byte("---\ntitle: S\n---\n[[c|C]]\n"), 0o644))
|
|
||||||
page := filepath.Join(brain, "wiki", "concepts", "c.md")
|
|
||||||
target := filepath.Join(brain, "raw", "secret.md")
|
|
||||||
require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
|
|
||||||
inside := "---\ntitle: C\n---\nbody\n"
|
|
||||||
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
|
|
||||||
brainfstest.StartSwap(t, page, inside, target, false)
|
|
||||||
ok := 0
|
|
||||||
for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
|
|
||||||
if _, err := BackfillRefs(context.Background(), brain); err == nil {
|
|
||||||
ok++
|
|
||||||
}
|
|
||||||
}
|
|
||||||
got, _ := os.ReadFile(target)
|
|
||||||
assert.Equal(t, "untouched", string(got), "BackfillRefs wrote through a swapped-in symlink")
|
|
||||||
assert.Positive(t, ok, "positive control: some backfills ran")
|
|
||||||
}
|
|
||||||
|
|
||||||
// hyperguild#158: buildAndWrite wrote each page with os.WriteFile by path, so
|
|
||||||
// a page read as regular and swapped for a symlink before the write was
|
|
||||||
// written through. The target is not restored between rounds, so any
|
|
||||||
// write-through shows.
|
|
||||||
func TestBuildAndWriteNeverWritesThroughAPageSwappedForASymlink(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
mkdirs(t, brain, "wiki/sources", "wiki/concepts", "wiki/entities", "raw")
|
|
||||||
page := filepath.Join(brain, "wiki", "concepts", "zq-concept.md")
|
|
||||||
target := filepath.Join(brain, "raw", "secret.md")
|
|
||||||
require.NoError(t, os.WriteFile(target, []byte("untouched"), 0o644))
|
|
||||||
inside := "---\ntitle: Zq Concept\n---\nbody\n"
|
|
||||||
require.NoError(t, os.WriteFile(page, []byte(inside), 0o644))
|
|
||||||
pages := []RawPage{{Title: "Zq Concept", Type: "concept", Content: "## About\n\nmore\n"}}
|
|
||||||
brainfstest.StartSwap(t, page, inside, target, false)
|
|
||||||
ok := 0
|
|
||||||
for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); {
|
|
||||||
if _, err := RunRaw(context.Background(), brain, "zq source", pages, false); err == nil {
|
|
||||||
ok++
|
|
||||||
}
|
|
||||||
}
|
|
||||||
got, _ := os.ReadFile(target)
|
|
||||||
assert.Equal(t, "untouched", string(got), "buildAndWrite wrote through a swapped-in symlink")
|
|
||||||
assert.Positive(t, ok, "positive control: some runs returned without error")
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,104 +0,0 @@
|
|||||||
package symlinkguard
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"net"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"syscall"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
)
|
|
||||||
|
|
||||||
// hyperguild#179: the tripwire looked only for symlinks, so a FIFO (or
|
|
||||||
// socket) in the brain passed it. A FIFO at classification.yaml then made
|
|
||||||
// startup exit; at a note path it could hang readers that skip the handle
|
|
||||||
// check.
|
|
||||||
func TestScanReportsFIFOsAndSocketsAsSpecialFiles(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(brain, "wiki", "w"), 0o755))
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(brain, "classification.yaml"), 0o644))
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(brain, "wiki", "w", "f.md"), 0o644))
|
|
||||||
l, err := net.Listen("unix", filepath.Join(brain, "s.sock"))
|
|
||||||
require.NoError(t, err)
|
|
||||||
defer func() { _ = l.Close() }()
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "wiki", "w", "n.md"), []byte("n"), 0o644))
|
|
||||||
|
|
||||||
r, err := Scan(brain)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, []string{"classification.yaml", "s.sock", "wiki/w/f.md"}, r.Special)
|
|
||||||
assert.False(t, r.Clean())
|
|
||||||
assert.Contains(t, r.String(), "special file")
|
|
||||||
}
|
|
||||||
|
|
||||||
// Watch removes a FIFO as it removes symlinks, and leaves a regular file
|
|
||||||
// (which Scan never reports) alone.
|
|
||||||
func TestWatchRemovesAFIFOAndLeavesAnUnreportedRegularFile(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
fifo := filepath.Join(brain, "f.md")
|
|
||||||
require.NoError(t, syscall.Mkfifo(fifo, 0o644))
|
|
||||||
regular := filepath.Join(brain, "n.md")
|
|
||||||
require.NoError(t, os.WriteFile(regular, []byte("n"), 0o644))
|
|
||||||
n := &recNotifier{}
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
defer cancel()
|
|
||||||
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
||||||
require.Eventually(t, func() bool {
|
|
||||||
_, err := os.Lstat(fifo)
|
|
||||||
return os.IsNotExist(err)
|
|
||||||
}, 2*time.Second, 10*time.Millisecond, "the FIFO is removed")
|
|
||||||
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
|
|
||||||
b, err := os.ReadFile(regular)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "n", string(b))
|
|
||||||
n.mu.Lock()
|
|
||||||
defer n.mu.Unlock()
|
|
||||||
assert.Contains(t, n.msgs[0], "f.md")
|
|
||||||
}
|
|
||||||
|
|
||||||
// removeSpecial re-checks the type: a path that is a regular file or a
|
|
||||||
// directory at the re-check is not removed. (A swap between the re-check and
|
|
||||||
// the unlink can still remove one; Linux has no conditional unlink. The same
|
|
||||||
// window exists in removeSymlinks.)
|
|
||||||
func TestRemoveSpecialRefusesRegularFilesAndDirectories(t *testing.T) {
|
|
||||||
brain := t.TempDir()
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brain, "n.md"), []byte("n"), 0o644))
|
|
||||||
require.NoError(t, os.MkdirAll(filepath.Join(brain, "d"), 0o755))
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(brain, "f.md"), 0o644))
|
|
||||||
removed, failed := removeSpecial(brain, []string{"n.md", "d", "f.md"})
|
|
||||||
assert.Equal(t, []string{"f.md"}, removed)
|
|
||||||
assert.Equal(t, []string{"n.md", "d"}, failed)
|
|
||||||
_, err := os.Stat(filepath.Join(brain, "n.md"))
|
|
||||||
assert.NoError(t, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// A special file Watch cannot remove (its directory is read-only) stays
|
|
||||||
// reported: the persistent-finding alert names it.
|
|
||||||
func TestWatchReportsASpecialFileItCannotRemove(t *testing.T) {
|
|
||||||
if os.Geteuid() == 0 {
|
|
||||||
t.Skip("root ignores directory permissions")
|
|
||||||
}
|
|
||||||
brain := t.TempDir()
|
|
||||||
ro := filepath.Join(brain, "ro")
|
|
||||||
require.NoError(t, os.MkdirAll(ro, 0o755))
|
|
||||||
require.NoError(t, syscall.Mkfifo(filepath.Join(ro, "f.md"), 0o644))
|
|
||||||
require.NoError(t, os.Chmod(ro, 0o555))
|
|
||||||
t.Cleanup(func() { _ = os.Chmod(ro, 0o755) })
|
|
||||||
n := &recNotifier{}
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
defer cancel()
|
|
||||||
go Watch(ctx, brain, 10*time.Millisecond, n)
|
|
||||||
require.Eventually(t, func() bool { return n.count() >= 1 }, 2*time.Second, 10*time.Millisecond)
|
|
||||||
_, err := os.Lstat(filepath.Join(ro, "f.md"))
|
|
||||||
require.NoError(t, err, "the FIFO could not be removed")
|
|
||||||
n.mu.Lock()
|
|
||||||
defer n.mu.Unlock()
|
|
||||||
assert.Contains(t, n.msgs[0], "ro/f.md")
|
|
||||||
assert.Contains(t, n.msgs[0], "special file")
|
|
||||||
for _, m := range n.msgs {
|
|
||||||
assert.NotContains(t, m, "removed", "a file it could not remove was reported as removed")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -46,16 +46,10 @@ type Notifier interface {
|
|||||||
type Result struct {
|
type Result struct {
|
||||||
Symlinks []string
|
Symlinks []string
|
||||||
Unreadable []string
|
Unreadable []string
|
||||||
// Special holds FIFOs, sockets and devices (hyperguild#179): a reader
|
|
||||||
// that opens one by path can hang, and a FIFO at classification.yaml
|
|
||||||
// stops startup.
|
|
||||||
Special []string
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Clean reports whether the scan found nothing to act on.
|
// Clean reports whether the scan found nothing to act on.
|
||||||
func (r Result) Clean() bool {
|
func (r Result) Clean() bool { return len(r.Symlinks) == 0 && len(r.Unreadable) == 0 }
|
||||||
return len(r.Symlinks) == 0 && len(r.Unreadable) == 0 && len(r.Special) == 0
|
|
||||||
}
|
|
||||||
|
|
||||||
// String describes the findings for a log line or alert. Paths are quoted, so
|
// String describes the findings for a log line or alert. Paths are quoted, so
|
||||||
// a name containing a newline cannot blur two findings together.
|
// a name containing a newline cannot blur two findings together.
|
||||||
@@ -65,9 +59,6 @@ func (r Result) String() string {
|
|||||||
if len(r.Symlinks) > 0 {
|
if len(r.Symlinks) > 0 {
|
||||||
parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks)))
|
parts = append(parts, fmt.Sprintf("%d symlink(s): %s", len(r.Symlinks), q(r.Symlinks)))
|
||||||
}
|
}
|
||||||
if len(r.Special) > 0 {
|
|
||||||
parts = append(parts, fmt.Sprintf("%d special file(s) (FIFO, socket or device; delete them): %s", len(r.Special), q(r.Special)))
|
|
||||||
}
|
|
||||||
if len(r.Unreadable) > 0 {
|
if len(r.Unreadable) > 0 {
|
||||||
parts = append(parts, fmt.Sprintf("%d unreadable dir(s): %s", len(r.Unreadable), q(r.Unreadable)))
|
parts = append(parts, fmt.Sprintf("%d unreadable dir(s): %s", len(r.Unreadable), q(r.Unreadable)))
|
||||||
}
|
}
|
||||||
@@ -122,16 +113,12 @@ func Scan(brainDir string) (Result, error) {
|
|||||||
if d.IsDir() && p == rootGit {
|
if d.IsDir() && p == rootGit {
|
||||||
return filepath.SkipDir
|
return filepath.SkipDir
|
||||||
}
|
}
|
||||||
switch {
|
if d.Type()&os.ModeSymlink != 0 {
|
||||||
case d.Type()&os.ModeSymlink != 0:
|
|
||||||
r.Symlinks = append(r.Symlinks, rel)
|
r.Symlinks = append(r.Symlinks, rel)
|
||||||
case !d.IsDir() && !d.Type().IsRegular():
|
|
||||||
r.Special = append(r.Special, rel)
|
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
sort.Strings(r.Symlinks)
|
sort.Strings(r.Symlinks)
|
||||||
sort.Strings(r.Special)
|
|
||||||
sort.Strings(r.Unreadable)
|
sort.Strings(r.Unreadable)
|
||||||
if walkErr != nil {
|
if walkErr != nil {
|
||||||
// Only the root's own error reaches here (deeper errors are recorded
|
// Only the root's own error reaches here (deeper errors are recorded
|
||||||
@@ -195,20 +182,14 @@ func Watch(ctx context.Context, brainDir string, interval time.Duration, n Notif
|
|||||||
q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s",
|
q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#157): removed %d symlink(s) from the brain dir: %s",
|
||||||
len(removed), Result{}.quoted(removed)))
|
len(removed), Result{}.quoted(removed)))
|
||||||
}
|
}
|
||||||
sRemoved, sFailed := removeSpecial(brainDir, r.Special)
|
key = Result{Symlinks: failed, Unreadable: r.Unreadable}.String()
|
||||||
if len(sRemoved) > 0 {
|
|
||||||
slog.Error("brain symlink tripwire removed special files", "count", len(sRemoved))
|
|
||||||
q.push(fmt.Sprintf("brain symlink tripwire (hyperguild#179): removed %d special file(s) (FIFO, socket or device) from the brain dir: %s",
|
|
||||||
len(sRemoved), Result{}.quoted(sRemoved)))
|
|
||||||
}
|
|
||||||
key = Result{Symlinks: failed, Unreadable: r.Unreadable, Special: sFailed}.String()
|
|
||||||
}
|
}
|
||||||
if key != reported {
|
if key != reported {
|
||||||
reported = key
|
reported = key
|
||||||
if key != "" {
|
if key != "" {
|
||||||
slog.Error("brain symlink tripwire", "finding", key)
|
slog.Error("brain symlink tripwire", "finding", key)
|
||||||
q.push("brain symlink tripwire (hyperguild#157, #179): " + key +
|
q.push("brain symlink tripwire (hyperguild#157): " + key +
|
||||||
". Symlinks can lead readers out of the brain and special files can hang them; remove them.")
|
". Readers may follow symlinks out of the brain; fix them.")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -341,40 +322,6 @@ func permanent(err error) bool {
|
|||||||
return errors.As(err, &p) && p.Permanent()
|
return errors.As(err, &p) && p.Permanent()
|
||||||
}
|
}
|
||||||
|
|
||||||
// removeSpecial unlinks each brain-relative FIFO, socket or device through
|
|
||||||
// an os.Root on the brain dir, after re-checking that it is still one: a
|
|
||||||
// path that is a regular file, a directory or a symlink at the re-check is
|
|
||||||
// left alone and reported as failed. The re-check narrows, but cannot close,
|
|
||||||
// the window before the unlink (Linux has no conditional unlink); the same
|
|
||||||
// window exists in removeSymlinks, and only a local writer can use it.
|
|
||||||
func removeSpecial(brainDir string, rels []string) (removed, failed []string) {
|
|
||||||
if len(rels) == 0 {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
resolved, err := filepath.EvalSymlinks(brainDir)
|
|
||||||
if err != nil {
|
|
||||||
return nil, rels
|
|
||||||
}
|
|
||||||
root, err := os.OpenRoot(resolved)
|
|
||||||
if err != nil {
|
|
||||||
return nil, rels
|
|
||||||
}
|
|
||||||
defer func() { _ = root.Close() }()
|
|
||||||
for _, rel := range rels {
|
|
||||||
fi, err := root.Lstat(rel)
|
|
||||||
if err != nil || fi.IsDir() || fi.Mode().IsRegular() || fi.Mode()&os.ModeSymlink != 0 {
|
|
||||||
failed = append(failed, rel)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if err := root.Remove(rel); err != nil {
|
|
||||||
failed = append(failed, rel)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
removed = append(removed, rel)
|
|
||||||
}
|
|
||||||
return removed, failed
|
|
||||||
}
|
|
||||||
|
|
||||||
// removeSymlinks unlinks each brain-relative symlink through an os.Root on
|
// removeSymlinks unlinks each brain-relative symlink through an os.Root on
|
||||||
// the brain dir, after re-checking it is still a symlink. Removing a symlink
|
// the brain dir, after re-checking it is still a symlink. Removing a symlink
|
||||||
// never touches its target. It returns what was removed and what was not.
|
// never touches its target. It returns what was removed and what was not.
|
||||||
|
|||||||
@@ -3,18 +3,15 @@ package watcher
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
|
||||||
"time"
|
"time"
|
||||||
"unicode"
|
"unicode"
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
||||||
)
|
)
|
||||||
@@ -112,28 +109,14 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Skip files that have already been processed or permanently failed.
|
// Skip files that have already been processed or permanently failed.
|
||||||
// Any entry at the marker path counts (Lstat, not Stat): a dangling
|
if _, err := os.Stat(path + ".processed"); err == nil {
|
||||||
// marker symlink made os.Stat report nothing and the note was
|
|
||||||
// reprocessed every interval (hyperguild#158).
|
|
||||||
if _, err := os.Lstat(path + ".processed"); err == nil {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
if _, err := os.Lstat(path + ".failed"); err == nil {
|
if _, err := os.Stat(path + ".failed"); err == nil {
|
||||||
return nil
|
|
||||||
}
|
|
||||||
// A note whose archive copy or marker was refused is not retried
|
|
||||||
// until it changes: a refusal has no marker to stop the next round,
|
|
||||||
// and each round can cost an LLM call (hyperguild#158).
|
|
||||||
info, ierr := d.Info()
|
|
||||||
if ierr == nil && refusedUnchanged(path, info) {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil {
|
if err := processFile(ctx, cfg, root, rawDir, path, date); err != nil {
|
||||||
if ierr == nil && errors.Is(err, brainfs.ErrNotRegular) {
|
|
||||||
rememberRefused(path, info)
|
|
||||||
slog.Error("watcher: note refused; not retried until it changes", "file", filepath.Base(path), "error", err)
|
|
||||||
}
|
|
||||||
errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
|
errs = append(errs, fmt.Errorf("process %s: %w", filepath.Base(path), err))
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -162,15 +145,16 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
|
|
||||||
_, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
|
_, runErr := pipeline.Run(ctx, cfg.Pipeline, cfg.BrainDir, content, source, false)
|
||||||
if runErr != nil {
|
if runErr != nil {
|
||||||
// Copy to failed/ and leave a .failed marker so we don't retry. Both
|
// Copy to failed/ and leave a .failed marker so we don't retry.
|
||||||
// written through the brain root (hyperguild#158).
|
failedDir := filepath.Join(cfg.BrainDir, "raw", "failed")
|
||||||
if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil {
|
if mkErr := os.MkdirAll(failedDir, 0o755); mkErr != nil {
|
||||||
|
return fmt.Errorf("mkdir failed dir: %w", mkErr)
|
||||||
|
}
|
||||||
|
dest := filepath.Join(failedDir, filename)
|
||||||
|
if cpErr := copyFromRoot(root, rel, dest); cpErr != nil {
|
||||||
return fmt.Errorf("copy to failed: %w", cpErr)
|
return fmt.Errorf("copy to failed: %w", cpErr)
|
||||||
}
|
}
|
||||||
if mkErr := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".failed", []byte(runErr.Error())); mkErr != nil {
|
if mkErr := os.WriteFile(path+".failed", []byte(runErr.Error()), 0o644); mkErr != nil {
|
||||||
if errors.Is(mkErr, brainfs.ErrNotRegular) {
|
|
||||||
return fmt.Errorf("write .failed marker: %w", mkErr)
|
|
||||||
}
|
|
||||||
slog.Error("watcher: failed to write .failed marker", "error", mkErr)
|
slog.Error("watcher: failed to write .failed marker", "error", mkErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -184,21 +168,21 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest.
|
// Copy to processed/YYYY-MM-DD/ and leave a .processed marker so we don't re-ingest.
|
||||||
// Written through the brain root (hyperguild#158).
|
processedDir := filepath.Join(cfg.BrainDir, "raw", "processed", date)
|
||||||
dest := "raw/processed/" + date + "/" + filename
|
if err := os.MkdirAll(processedDir, 0o755); err != nil {
|
||||||
if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
|
return fmt.Errorf("mkdir processed dir: %w", err)
|
||||||
|
}
|
||||||
|
dest := filepath.Join(processedDir, filename)
|
||||||
|
if _, err := os.Stat(dest); err == nil {
|
||||||
// Archive copy already exists; append timestamp to avoid overwriting.
|
// Archive copy already exists; append timestamp to avoid overwriting.
|
||||||
ext := filepath.Ext(filename)
|
ext := filepath.Ext(filename)
|
||||||
base := strings.TrimSuffix(filename, ext)
|
base := strings.TrimSuffix(filename, ext)
|
||||||
dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext
|
dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext)
|
||||||
}
|
}
|
||||||
if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil {
|
if err := copyFromRoot(root, rel, dest); err != nil {
|
||||||
return fmt.Errorf("copy to processed: %w", err)
|
return fmt.Errorf("copy to processed: %w", err)
|
||||||
}
|
}
|
||||||
if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+".processed", []byte(date)); err != nil {
|
if err := os.WriteFile(path+".processed", []byte(date), 0o644); err != nil {
|
||||||
if errors.Is(err, brainfs.ErrNotRegular) {
|
|
||||||
return fmt.Errorf("write .processed marker: %w", err)
|
|
||||||
}
|
|
||||||
slog.Error("watcher: failed to write .processed marker", "error", err)
|
slog.Error("watcher: failed to write .processed marker", "error", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -206,32 +190,6 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// refused remembers notes whose archive copy or marker write was refused,
|
|
||||||
// by path, with the size and mtime they had, for the life of the process.
|
|
||||||
var refused sync.Map
|
|
||||||
|
|
||||||
type refusedStamp struct {
|
|
||||||
size int64
|
|
||||||
mtime time.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
func rememberRefused(path string, info os.FileInfo) {
|
|
||||||
refused.Store(path, refusedStamp{size: info.Size(), mtime: info.ModTime()})
|
|
||||||
}
|
|
||||||
|
|
||||||
func refusedUnchanged(path string, info os.FileInfo) bool {
|
|
||||||
v, ok := refused.Load(path)
|
|
||||||
if !ok {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
st := v.(refusedStamp)
|
|
||||||
if st.size == info.Size() && st.mtime.Equal(info.ModTime()) {
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
refused.Delete(path)
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
// deriveSource turns a filename into a human-readable source name.
|
// deriveSource turns a filename into a human-readable source name.
|
||||||
// "shape-up-book.md" → "Shape Up Book"
|
// "shape-up-book.md" → "Shape Up Book"
|
||||||
func deriveSource(filename string) string {
|
func deriveSource(filename string) string {
|
||||||
@@ -256,22 +214,23 @@ func appendWatcherLog(brainDir, filename string, runErr error, date string) erro
|
|||||||
entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
|
entry := fmt.Sprintf("## %s — watcher error\n\n- **File:** %s\n- **Error:** %s\n\n",
|
||||||
date, filename, runErr.Error())
|
date, filename, runErr.Error())
|
||||||
|
|
||||||
if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil {
|
logPath := filepath.Join(brainDir, "log.md")
|
||||||
return fmt.Errorf("write log: %w", err)
|
f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
||||||
}
|
if err != nil {
|
||||||
return nil
|
return fmt.Errorf("open log: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// putBrainFile creates or replaces a brain-relative file through the brain
|
if _, err = f.WriteString(entry); err != nil {
|
||||||
// root, never through a symlinked directory or file (hyperguild#158).
|
f.Close() //nolint:errcheck
|
||||||
func putBrainFile(brainDir, rel string, data []byte) error {
|
return fmt.Errorf("write log: %w", err)
|
||||||
return brainfs.PutFile(brainDir, rel, data)
|
}
|
||||||
|
return f.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
// copyFromRoot archives a raw/ file read through the root, refusing anything
|
// copyFromRoot archives a raw/ file read through the root, refusing anything
|
||||||
// that is not a regular file, so a swapped-in symlink cannot copy an outside
|
// that is not a regular file, so a swapped-in symlink cannot copy an outside
|
||||||
// file into raw/processed/ or raw/failed/ (hyperguild#153).
|
// file into raw/processed/ or raw/failed/ (hyperguild#153).
|
||||||
func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
|
func copyFromRoot(root *os.Root, rel, dst string) error {
|
||||||
fi, err := root.Lstat(rel)
|
fi, err := root.Lstat(rel)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -284,9 +243,13 @@ func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer func() { _ = in.Close() }()
|
defer func() { _ = in.Close() }()
|
||||||
data, err := io.ReadAll(in)
|
out, err := os.Create(dst)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
return putBrainFile(brainDir, dstRel, data)
|
if _, err := io.Copy(out, in); err != nil {
|
||||||
|
_ = out.Close()
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return out.Close()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,186 +0,0 @@
|
|||||||
package watcher
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
|
||||||
"github.com/stretchr/testify/require"
|
|
||||||
|
|
||||||
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
|
|
||||||
)
|
|
||||||
|
|
||||||
func okPipeline() pipeline.Config {
|
|
||||||
return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
|
|
||||||
b, _ := json.Marshal([]pipeline.RawPage{{Title: "X", Type: "source", Subtype: "article", Content: "## S\n\nx\n"}})
|
|
||||||
return string(b), nil
|
|
||||||
}, Schema: "# Schema"}
|
|
||||||
}
|
|
||||||
|
|
||||||
func failingPipeline() pipeline.Config {
|
|
||||||
return pipeline.Config{Complete: func(context.Context, string, string) (string, error) {
|
|
||||||
return "", errors.New("model down")
|
|
||||||
}, Schema: "# Schema"}
|
|
||||||
}
|
|
||||||
|
|
||||||
// hyperguild#158: the watcher wrote the archive copy under raw/processed or
|
|
||||||
// raw/failed, and its log.md entry, by path, so a symlinked archive
|
|
||||||
// directory or log redirected the write elsewhere in the brain or outside
|
|
||||||
// it. (A pre-planted marker symlink makes the watcher skip the note, so the
|
|
||||||
// marker write itself is reachable only by a race.)
|
|
||||||
func TestProcessDirDoesNotWriteThroughASymlinkedArchiveDir(t *testing.T) {
|
|
||||||
for name, pcfg := range map[string]pipeline.Config{"processed": okPipeline(), "failed": failingPipeline()} {
|
|
||||||
t.Run(name, func(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
outside := t.TempDir()
|
|
||||||
victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", name)))
|
|
||||||
_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pcfg}, "2026-10-05")
|
|
||||||
got, _ := os.ReadFile(victim)
|
|
||||||
assert.Equal(t, "untouched", string(got))
|
|
||||||
ents, _ := os.ReadDir(outside)
|
|
||||||
assert.Empty(t, ents, "the archive copy was written outside the brain")
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestProcessDirWritesMarkersAndArchiveInRealDirectories(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
|
||||||
errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
|
|
||||||
assert.Empty(t, errs)
|
|
||||||
_, err := os.Stat(filepath.Join(brainDir, "raw", "n.md.processed"))
|
|
||||||
assert.NoError(t, err, "positive control: marker written")
|
|
||||||
_, err = os.Stat(filepath.Join(brainDir, "raw", "processed", "2026-10-05", "n.md"))
|
|
||||||
assert.NoError(t, err, "positive control: archive copy written")
|
|
||||||
}
|
|
||||||
|
|
||||||
// The log.md entry for a failed note is appended through the brain root: with
|
|
||||||
// a real raw/failed (so the archive copy succeeds and the log is reached), a
|
|
||||||
// symlinked log.md is not written through.
|
|
||||||
func TestProcessDirDoesNotAppendTheLogThroughASymlink(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
victim := filepath.Join(brainDir, "wiki", "concepts", "v.md")
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(filepath.Join("wiki", "concepts", "v.md"), filepath.Join(brainDir, "log.md")))
|
|
||||||
_ = processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: failingPipeline()}, "2026-10-05")
|
|
||||||
_, err := os.Stat(filepath.Join(brainDir, "raw", "failed", "n.md"))
|
|
||||||
require.NoError(t, err, "positive control: the archive copy was written, so the log append was reached")
|
|
||||||
got, _ := os.ReadFile(victim)
|
|
||||||
assert.Equal(t, "untouched", string(got), "the log was appended through a symlink")
|
|
||||||
}
|
|
||||||
|
|
||||||
// A marker path holding anything (here a dangling symlink) counts as a
|
|
||||||
// marker: the note is skipped and nothing is created at the link's target.
|
|
||||||
// (os.Stat followed the link, saw nothing, and processed the note again.)
|
|
||||||
func TestProcessDirTreatsAnyEntryAtTheMarkerPathAsAMarker(t *testing.T) {
|
|
||||||
for _, kind := range []string{"processed", "failed"} {
|
|
||||||
t.Run(kind, func(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
outside := filepath.Join(t.TempDir(), "created")
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(outside, filepath.Join(brainDir, "raw", "n.md."+kind)))
|
|
||||||
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"}}
|
|
||||||
_ = processDir(context.Background(), cfg, "2026-10-05")
|
|
||||||
assert.Zero(t, calls, "the note was processed despite an entry at its marker path")
|
|
||||||
_, err := os.Lstat(outside)
|
|
||||||
assert.True(t, os.IsNotExist(err), "a file was created at the marker link's target")
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// The marker write itself is reachable while pipeline.Run is in flight
|
|
||||||
// (in production, an LLM call): a link planted at the marker path then is
|
|
||||||
// refused, and the note is remembered, so removing the link (as
|
|
||||||
// symlinkguard does) does not make the watcher call the model again.
|
|
||||||
func TestProcessDirRefusesAMarkerLinkPlantedDuringTheRunAndDoesNotRetry(t *testing.T) {
|
|
||||||
for _, kind := range []string{"processed", "failed"} {
|
|
||||||
for _, target := range []string{"outside", "in-brain"} {
|
|
||||||
t.Run(kind+"/"+target, func(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
raw := filepath.Join(brainDir, "raw")
|
|
||||||
marker := filepath.Join(raw, "n.md."+kind)
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(raw, "n.md"), []byte("note"), 0o644))
|
|
||||||
// The in-brain target is relative, so os.Root itself would
|
|
||||||
// follow it; only the regular-file check refuses it.
|
|
||||||
victim := filepath.Join(raw, "victim")
|
|
||||||
link := filepath.Join(t.TempDir(), "created")
|
|
||||||
if target == "in-brain" {
|
|
||||||
require.NoError(t, os.WriteFile(victim, []byte("untouched"), 0o644))
|
|
||||||
link = "victim"
|
|
||||||
}
|
|
||||||
ok := okPipeline().Complete
|
|
||||||
calls := 0
|
|
||||||
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{Schema: "# Schema",
|
|
||||||
Complete: func(ctx context.Context, sys, user string) (string, error) {
|
|
||||||
calls++
|
|
||||||
require.NoError(t, os.Symlink(link, marker))
|
|
||||||
if kind == "failed" {
|
|
||||||
return "", errors.New("model down")
|
|
||||||
}
|
|
||||||
return ok(ctx, sys, user)
|
|
||||||
}}}
|
|
||||||
for i := 0; i < 3; i++ {
|
|
||||||
errs := processDir(context.Background(), cfg, "2026-10-05")
|
|
||||||
if i == 0 {
|
|
||||||
require.NotEmpty(t, errs, "the planted link was not refused")
|
|
||||||
}
|
|
||||||
_ = os.Remove(marker)
|
|
||||||
}
|
|
||||||
assert.Equal(t, 1, calls, "the note was retried after its marker write was refused")
|
|
||||||
if target == "in-brain" {
|
|
||||||
got, err := os.ReadFile(victim)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.Equal(t, "untouched", string(got), "the marker was written through the link")
|
|
||||||
} else {
|
|
||||||
_, err := os.Lstat(link)
|
|
||||||
assert.True(t, os.IsNotExist(err), "the marker was written through the link")
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// A refused archive copy must not retry every interval: the note is
|
|
||||||
// remembered and skipped until it changes (a symlinked raw/failed cost one
|
|
||||||
// LLM call per round before).
|
|
||||||
func TestProcessDirDoesNotRetryANoteWhoseArchiveCopyIsRefused(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note"), 0o644))
|
|
||||||
require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(brainDir, "raw", "failed")))
|
|
||||||
calls := 0
|
|
||||||
cfg := Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: pipeline.Config{
|
|
||||||
Complete: func(context.Context, string, string) (string, error) { calls++; return "", errors.New("model down") }, Schema: "# Schema"}}
|
|
||||||
for i := 0; i < 3; i++ {
|
|
||||||
_ = processDir(context.Background(), cfg, "2026-10-05")
|
|
||||||
}
|
|
||||||
assert.Equal(t, 1, calls, "the refused note was retried every round")
|
|
||||||
|
|
||||||
// Once the note changes it is tried again.
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "n.md"), []byte("note, edited"), 0o644))
|
|
||||||
future := time.Now().Add(2 * time.Second)
|
|
||||||
require.NoError(t, os.Chtimes(filepath.Join(brainDir, "raw", "n.md"), future, future))
|
|
||||||
_ = processDir(context.Background(), cfg, "2026-10-05")
|
|
||||||
assert.Equal(t, 2, calls, "positive control: a changed note is tried again")
|
|
||||||
}
|
|
||||||
|
|
||||||
// A note whose name has a backslash is legal on Linux and must be processed
|
|
||||||
// (cleanRel refused it, so it was never archived or marked).
|
|
||||||
func TestProcessDirProcessesANoteWithABackslashInItsName(t *testing.T) {
|
|
||||||
brainDir := setupBrainDir(t)
|
|
||||||
require.NoError(t, os.WriteFile(filepath.Join(brainDir, "raw", "a\\b.md"), []byte("note"), 0o644))
|
|
||||||
errs := processDir(context.Background(), Config{BrainDir: brainDir, Interval: time.Hour, Pipeline: okPipeline()}, "2026-10-05")
|
|
||||||
assert.Empty(t, errs)
|
|
||||||
_, err := os.Stat(filepath.Join(brainDir, "raw", "a\\b.md.processed"))
|
|
||||||
assert.NoError(t, err)
|
|
||||||
}
|
|
||||||
Reference in New Issue
Block a user