mathias/hyperguild
1
0
Fork 0
Code Issues 35 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

merge: land #195 (watcher/pipeline writes through the brain root, #158 part B) on top of inbox zero and the wing resolver

Browse Source 
Reconciled with v0.25.0 semantics:
- markers are only the fallback when the original cannot be removed, and are
  now written through the brain root; the refused set (size+mtime keyed) replaces
  the path-only 'stuck' set
- #195's tests updated for inbox zero (archive written, original retired, no
  marker), and the planted-marker-link test keeps its point: nothing is ever
  written through the link, and the note is processed once
- the 'cannot retire' test now counts archive copies; it counted LLM calls,
  which pipeline.Run's source-hash check hides (mutation: removing the
  remember step left it green)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
mathias
2026-10-20 20:02:15 +02:00
co-authored by Claude Sonnet 5.5
parent 858ade6504 22d0653b61
commit 2cc4f1b30d
11 changed files with 406 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ingestion/internal/watcher/watcher.go
+73 -57
View File
@@ -3,6 +3,7 @@ package watcher
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
@@ -14,6 +15,7 @@ import (
"unicode"
"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
"github.com/mathiasbq/hyperguild/ingestion/internal/extract"
"github.com/mathiasbq/hyperguild/ingestion/internal/pipeline"
"github.com/mathiasbq/hyperguild/ingestion/internal/wing"
@@ -25,11 +27,6 @@ type Config struct {
Interval time.Duration
Pipeline pipeline.Config
// stuck remembers notes that were ingested but could neither be removed
// 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.
@@ -39,9 +36,6 @@ type Config struct {
// Start launches the watcher in a background goroutine.
// It returns immediately. The watcher stops when ctx is cancelled.
func Start(ctx context.Context, cfg Config) {
if cfg.stuck == nil {
cfg.stuck = &sync.Map{}
}
go func() {
ticker := time.NewTicker(cfg.Interval)
defer ticker.Stop()
@@ -124,21 +118,29 @@ func processDir(ctx context.Context, cfg Config, date string) []error {
return nil
}
if cfg.stuck != nil {
if _, skip := cfg.stuck.Load(path); skip {
return nil
}
}
// Skip files that have already been processed or permanently failed.
if _, err := os.Stat(path + ".processed"); err == nil {
// Any entry at the marker path counts (Lstat, not Stat): a dangling
// 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
}
if _, err := os.Stat(path + ".failed"); err == nil {
if _, err := os.Lstat(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
}
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))
}
return nil
@@ -183,20 +185,16 @@ func processFile(ctx context.Context, cfg Config, root *os.Root, rawDir, path, d
}
_, 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")
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 {
// Copy to failed/ and leave a .failed marker so we don't retry. Both
// written through the brain root (hyperguild#158).
if cpErr := copyFromRoot(root, rel, cfg.BrainDir, "raw/failed/"+filename); cpErr != nil {
return fmt.Errorf("copy to failed: %w", cpErr)
}
if changedSince(root, rel, before) {
slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
return nil
}
retire(cfg, root, rel, path, ".failed", runErr.Error())
retire(cfg, root, rel, path, ".failed", runErr.Error(), before)
slog.Warn("watcher: file failed", "file", filename, "error", runErr)
@@ -208,25 +206,22 @@ 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.
processedDir := filepath.Join(cfg.BrainDir, "raw", "processed", date)
if err := os.MkdirAll(processedDir, 0o755); err != nil {
return fmt.Errorf("mkdir processed dir: %w", err)
}
dest := filepath.Join(processedDir, filename)
if _, err := os.Stat(dest); err == nil {
// Written through the brain root (hyperguild#158).
dest := "raw/processed/" + date + "/" + filename
if _, err := os.Lstat(filepath.Join(cfg.BrainDir, filepath.FromSlash(dest))); err == nil {
// Archive copy already exists; append timestamp to avoid overwriting.
ext := filepath.Ext(filename)
base := strings.TrimSuffix(filename, ext)
dest = filepath.Join(processedDir, base+"-"+time.Now().UTC().Format("150405")+ext)
dest = "raw/processed/" + date + "/" + base + "-" + time.Now().UTC().Format("150405") + ext
}
if err := copyFromRoot(root, rel, dest); err != nil {
if err := copyFromRoot(root, rel, cfg.BrainDir, dest); err != nil {
return fmt.Errorf("copy to processed: %w", err)
}
if changedSince(root, rel, before) {
slog.Warn("watcher: note changed during ingest; leaving it in raw/ for the next poll", "file", filename)
return nil
}
retire(cfg, root, rel, path, ".processed", date)
retire(cfg, root, rel, path, ".processed", date, before)
slog.Info("watcher: file processed", "file", filename, "source", source)
return nil
@@ -246,21 +241,47 @@ func changedSince(root *os.Root, rel string, before os.FileInfo) bool {
}
// retire removes an archived original from raw/ (inbox zero). If the removal
// fails it falls back to a marker file so the watcher still skips the file.
func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string) {
// fails it falls back to a marker written through the brain root (#158), and if
// even that fails the note is remembered, with the size and mtime it had, so it
// is not ingested again until it changes.
func retire(cfg Config, root *os.Root, rel, path, markerSuffix, markerBody string, before os.FileInfo) {
rmErr := root.Remove(rel)
if rmErr == nil {
return
}
slog.Error("watcher: failed to remove archived original; writing marker", "file", rel, "error", rmErr)
if err := os.WriteFile(path+markerSuffix, []byte(markerBody), 0o644); err != nil {
slog.Error("watcher: failed to write marker; skipping this note until restart", "file", rel, "error", err)
if cfg.stuck != nil {
cfg.stuck.Store(path, struct{}{})
}
if err := putBrainFile(cfg.BrainDir, "raw/"+filepath.ToSlash(rel)+markerSuffix, []byte(markerBody)); err != nil {
slog.Error("watcher: failed to write marker; skipping this note until it changes", "file", rel, "error", err)
rememberRefused(path, before)
}
}
// 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.
// "shape-up-book.md" → "Shape Up Book"
func deriveSource(filename string) string {
@@ -285,23 +306,22 @@ 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",
date, filename, runErr.Error())
logPath := filepath.Join(brainDir, "log.md")
f, err := os.OpenFile(logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return fmt.Errorf("open log: %w", err)
}
if _, err = f.WriteString(entry); err != nil {
f.Close() //nolint:errcheck
if err := brainfs.AppendFile(brainDir, "log.md", []byte(entry)); err != nil {
return fmt.Errorf("write log: %w", err)
}
return f.Close()
return nil
}
// putBrainFile creates or replaces a brain-relative file through the brain
// root, never through a symlinked directory or file (hyperguild#158).
func putBrainFile(brainDir, rel string, data []byte) error {
return brainfs.PutFile(brainDir, rel, data)
}
// 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
// file into raw/processed/ or raw/failed/ (hyperguild#153).
func copyFromRoot(root *os.Root, rel, dst string) error {
func copyFromRoot(root *os.Root, rel, brainDir, dstRel string) error {
fi, err := root.Lstat(rel)
if err != nil {
return err
@@ -314,15 +334,11 @@ func copyFromRoot(root *os.Root, rel, dst string) error {
return err
}
defer func() { _ = in.Close() }()
out, err := os.Create(dst)
data, err := io.ReadAll(in)
if err != nil {
return err
}
if _, err := io.Copy(out, in); err != nil {
_ = out.Close()
return err
}
return out.Close()
return putBrainFile(brainDir, dstRel, data)
}
// topDir is the first directory of a raw/-relative path, "" for a top-level file.
@@ -385,7 +401,7 @@ func parkForReview(cfg Config, root *os.Root, rel, path, filename, content strin
slog.Warn("watcher: note changed while it was being resolved; leaving the newer version in raw/", "file", filename)
return nil
}
retire(cfg, root, rel, path, ".processed", date)
retire(cfg, root, rel, path, ".processed", date, before)
return nil
}