diff --git a/docs/raw-inbox.md b/docs/raw-inbox.md index 2814ff0..c2c8adf 100644 --- a/docs/raw-inbox.md +++ b/docs/raw-inbox.md @@ -12,6 +12,7 @@ with the machines you drop files from. (hyperguild#205, #208, #209) | a note whose text matches the client-content pattern, with no folder | parked in `raw/needs-review/`, not extracted | | a folder or `wing:` naming an unknown, reserved or non-ASCII wing, or two disagreeing signals | parked in `raw/needs-review/` | | a note over the size cap, or one that cannot be archived | stays in `raw/`, reported once, not retried until it changes | +| a note the model returned nothing usable for (no source page could be written) | fails: the note is kept in `raw/failed/` and the reason is in `log.md`; it is never archived as processed | Concepts and entities are shared across wings because their slug is the graph's identity. A source whose slug is taken gets a wing-qualified one (`notes-homelab`). @@ -39,6 +40,13 @@ server runs is usable without a restart. ## Known limits +- The client filter works on 1 MiB chunks that overlap by 4 KiB. A match that can + span more than 4 KiB (crafted padding between two words of a multi-word + pattern, exactly at a chunk edge) is not guaranteed to be found; this affects + notes over 1 MiB only. On a 64 MiB note the check takes about ten seconds. +- A pattern that matches the empty string (`.*`, `x*`) parks every unsignalled + note. That fails closed and only the operator's own configuration can do it. + - A deliberate folder into a sensitive wing places the note there, and the shared concept/entity pages then cite that source by title. Client material does not belong in the brain at all; the pattern only guards the unsignalled lane. diff --git a/ingestion/internal/api/promote.go b/ingestion/internal/api/promote.go index f056398..fe9115e 100644 --- a/ingestion/internal/api/promote.go +++ b/ingestion/internal/api/promote.go @@ -97,6 +97,16 @@ func listNotes(root *os.Root, queue string) ([]PendingNote, error) { } raw, info, readErr := readRawNote(root, e.Name()) if readErr != nil { + // Not skipped: a note that cannot be read (over the cap, say) must + // not silently vanish from the queue it is waiting in. + fi, ierr := root.Lstat(e.Name()) + if ierr != nil { + continue + } + out = append(out, PendingNote{ + Filename: e.Name(), CreatedAt: fi.ModTime().UTC().Format(time.RFC3339), SizeBytes: fi.Size(), + Excerpt: "(cannot be read: " + readErr.Error() + ")", Queue: queue, + }) continue } fm, body := parseFrontmatter(string(raw)) diff --git a/ingestion/internal/api/promote_queue_test.go b/ingestion/internal/api/promote_queue_test.go index 32a1b74..e7c1e6e 100644 --- a/ingestion/internal/api/promote_queue_test.go +++ b/ingestion/internal/api/promote_queue_test.go @@ -190,3 +190,21 @@ func TestReadCapped_StopsAtTheCap(t *testing.T) { assert.Contains(t, err.Error(), "exceeds") assert.LessOrEqual(t, e.n, int64(1000+1+32*1024), "pulled far more than the cap: %d", e.n) } + +// An .md over the read cap used to vanish from the listing (readRawNote failed +// and the entry was skipped). It must stay visible, with the reason. +func TestListPending_AnUnreadableNoteIsListedWithTheReason(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, "raw"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "raw", "huge.md"), []byte(strings.Repeat("x", 200)), 0o644)) + prev := maxRawNoteBytes + maxRawNoteBytes = 100 + t.Cleanup(func() { maxRawNoteBytes = prev }) + + got, err := ListPending(dir) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, "huge.md", got[0].Filename) + assert.Contains(t, got[0].Excerpt, "cannot be read") + assert.Contains(t, got[0].Excerpt, "exceeds") +} diff --git a/ingestion/internal/pipeline/pipeline.go b/ingestion/internal/pipeline/pipeline.go index f3ed9c4..f166a4a 100644 --- a/ingestion/internal/pipeline/pipeline.go +++ b/ingestion/internal/pipeline/pipeline.go @@ -113,7 +113,19 @@ func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryR meta.BodyHash = bodyHash applySourceMeta(allRaw, meta) - return buildAndWrite(ctx, cfg.Wing, allRaw, sourceSlug, date, brainDir, source, inventory, allWarnings, dryRun) + res, err := buildAndWrite(ctx, cfg.Wing, allRaw, sourceSlug, date, brainDir, source, inventory, allWarnings, dryRun) + if err != nil || cfg.Wing == "" { + return res, err + } + // In a wing every ingest must leave its source page: it is what ties the + // note to the wiki (and carries the hash the drift check reads). Without one + // nothing of the note exists, and the caller would archive it as processed. + for _, p := range res.Pages { + if isSourcePath(p) { + return res, nil + } + } + return res, fmt.Errorf("no source page was written for %q; nothing of the note was ingested (%s)", source, strings.Join(res.Warnings, "; ")) } // RunRaw runs the pipeline on pre-parsed RawPages, skipping the LLM extraction @@ -357,13 +369,19 @@ func resolveSourceSlug(brainDir, wing, base string) (string, error) { return "", fmt.Errorf("check source slug: %w", err) } own := "wiki/" + wing + "/sources/" + cand + ".md" - free := true + free, hasOwn := true, false for _, p := range existing { - if p != own { + if p == own { + hasOwn = true + } else { free = false - break } } + // The wing's own page is its identity: keep using it even if some other + // page shares the slug (that other page is the one dropSlugClashes skips). + if hasOwn { + return cand, nil + } if free { return cand, nil } @@ -399,12 +417,42 @@ func dropSlugClashes(brainDir string, pages []wiki.Page) ([]wiki.Page, []string, } slugOf := func(p wiki.Page) string { return strings.TrimSuffix(filepath.Base(p.Path), ".md") } - winner := map[string]wiki.Page{} + // Group by slug; where several pages with different paths want one slug, + // the source wins, then a page that already exists on disk (so its update + // is not lost), then entity over concept. + bySlug := map[string][]wiki.Page{} for _, p := range pages { - w, seen := winner[slugOf(p)] - if !seen || (w.Path != p.Path && rank(p) < rank(w)) { - winner[slugOf(p)] = p + bySlug[slugOf(p)] = append(bySlug[slugOf(p)], p) + } + winner := map[string]wiki.Page{} + for slug, group := range bySlug { + best := group[0] + if distinctPaths(group) > 1 { + existing, err := wiki.FindSlugs(brainDir, slug) + if err != nil { + return nil, nil, fmt.Errorf("check slug %q: %w", slug, err) + } + onDisk := func(p wiki.Page) int { + for _, e := range existing { + if e == p.Path { + return 0 + } + } + return 1 + } + for _, p := range group[1:] { + if rank(p) != rank(best) && (rank(p) == 0 || rank(best) == 0) { + if rank(p) < rank(best) { + best = p + } + continue + } + if onDisk(p) < onDisk(best) || (onDisk(p) == onDisk(best) && rank(p) < rank(best)) { + best = p + } + } } + winner[slug] = best } var kept []wiki.Page @@ -438,3 +486,11 @@ func dropSlugClashes(brainDir string, pages []wiki.Page) ([]wiki.Page, []string, } return kept, warnings, nil } + +func distinctPaths(pages []wiki.Page) int { + seen := map[string]bool{} + for _, p := range pages { + seen[p.Path] = true + } + return len(seen) +} diff --git a/ingestion/internal/pipeline/resolve.go b/ingestion/internal/pipeline/resolve.go index 21e04e3..528ad28 100644 --- a/ingestion/internal/pipeline/resolve.go +++ b/ingestion/internal/pipeline/resolve.go @@ -32,6 +32,13 @@ func Resolve(proposed []wiki.Page, inventory map[wiki.PageType][]wiki.Entry) []w out := make([]wiki.Page, 0, len(proposed)) for _, page := range proposed { + if isWingSourcePath(page.Path) { + // In a wing the source's path is its filename-derived slug. Matching + // by TITLE would merge two different notes the model happened to + // title alike ("Home", "Untitled") into one page. + out = append(out, page) + continue + } pt := pageTypeOf(page) title := extractTitle(page.Content) k := key{pt: pt, normalized: normalizeTitle(title)} @@ -97,3 +104,9 @@ func extractTitle(content string) string { } return "" } + +// isWingSourcePath reports whether path is wiki//sources/.md. +func isWingSourcePath(path string) bool { + parts := strings.Split(filepath.ToSlash(path), "/") + return len(parts) == 4 && parts[0] == "wiki" && parts[2] == "sources" +} diff --git a/ingestion/internal/pipeline/wing_layout_test.go b/ingestion/internal/pipeline/wing_layout_test.go index a881690..fe9915d 100644 --- a/ingestion/internal/pipeline/wing_layout_test.go +++ b/ingestion/internal/pipeline/wing_layout_test.go @@ -292,3 +292,83 @@ func TestRun_AnExistingSourceEntityPairDoesNotLockTheSourceOut(t *testing.T) { assert.Contains(t, read(t, dir, "wiki/homelab/sources/notes.md"), "fresh", "the source took the update") assert.Contains(t, read(t, dir, "wiki/entities/notes.md"), "stale entity", "the entity is untouched") } + +// Pass 3 of #211: a note must never leave the inbox as "processed" when nothing +// of it was written. +func TestRun_WithAWingAnIngestThatWritesNoSourcePageIsAnError(t *testing.T) { + tests := []struct { + name string + pages []RawPage + prep func(t *testing.T, dir string) + }{ + {"the model returned nothing", nil, nil}, + {"only pages of a type the layout has no place for", []RawPage{{Title: "Claim", Type: "claim", Content: "## x\n\ny\n"}}, nil}, + {"only an entity, and its slug is taken by a source", []RawPage{{Title: "S1", Type: "entity", Subtype: "tool", Content: "## Definition\n\nx\n"}}, + func(t *testing.T, dir string) { + require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki/web-clip/sources"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "wiki/web-clip/sources/s1.md"), []byte("---\ntitle: 'S1'\n---\nx"), 0o644)) + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dir := t.TempDir() + if tt.prep != nil { + tt.prep(t, dir) + } + res, err := Run(context.Background(), wingCfg("web-clip", tt.pages...), dir, "a note body", "other-note", false) + require.Error(t, err, "nothing was ingested; the caller must not archive it as done") + assert.Empty(t, res.Pages) + assert.Contains(t, err.Error(), "no source page") + }) + } +} + +func TestRun_WithAWingAnUnchangedSourceIsStillASkipNotAnError(t *testing.T) { + dir := t.TempDir() + _, err := Run(context.Background(), wingCfg("web-clip", sourceEntityConcept()...), dir, "same body", "talk", false) + require.NoError(t, err) + res, err := Run(context.Background(), wingCfg("web-clip", sourceEntityConcept()...), dir, "same body", "talk", false) + require.NoError(t, err) + assert.True(t, res.Skipped) +} + +// Title resolution must not merge two different notes into one source page just +// because the model titled both "Home": in a wing the source path is already +// deterministic (the filename), so the title must not redirect it. +func TestRun_WithAWingTwoNotesTheModelTitlesAlikeKeepTheirOwnSourcePages(t *testing.T) { + dir := t.TempDir() + home := func(body string) []RawPage { + return []RawPage{{Title: "Home", Type: "source", Subtype: "article", Content: "## Summary\n\n" + body + "\n"}} + } + _, err := Run(context.Background(), wingCfg("web-clip", home("note A text")...), dir, "body A", "note-a", false) + require.NoError(t, err) + _, err = Run(context.Background(), wingCfg("web-clip", home("note B text")...), dir, "body B", "note-b", false) + require.NoError(t, err) + + assert.True(t, exists(dir, "wiki/web-clip/sources/note-a.md")) + assert.True(t, exists(dir, "wiki/web-clip/sources/note-b.md"), "note B has its own page") + assert.Contains(t, read(t, dir, "wiki/web-clip/sources/note-a.md"), "note A text") + assert.Contains(t, read(t, dir, "wiki/web-clip/sources/note-b.md"), "note B text") + + res, err := Run(context.Background(), wingCfg("web-clip", home("note A text")...), dir, "body A", "note-a", false) + require.NoError(t, err) + assert.True(t, res.Skipped, "note A's hash was not overwritten by note B") +} + +// Both a concept and an entity named "Foo" arrive and the CONCEPT already exists +// on disk: the page that exists wins the tie, so its source reference is added. +func TestRun_ABatchTieBetweenAConceptAndAnEntityPrefersThePageAlreadyOnDisk(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, "wiki/concepts"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "wiki/concepts/foo.md"), + []byte("---\ntitle: 'Foo'\naliases:\n - 'Foo'\n---\n\n## Definition\n\nexisting concept\n\n## Sources\n"), 0o644)) + pages := []RawPage{ + {Title: "Talk", Type: "source", Subtype: "video", Content: "## Summary\n\nx\n\n## Concepts Introduced or Reinforced\n\n[[Foo]]\n"}, + {Title: "Foo", Type: "entity", Subtype: "tool", Content: "## Definition\n\nan entity\n"}, + {Title: "Foo", Type: "concept", Content: "## Definition\n\na concept\n\n## Sources\n"}, + } + _, err := Run(context.Background(), wingCfg("web-clip", pages...), dir, "body", "talk", false) + require.NoError(t, err) + assert.False(t, exists(dir, "wiki/entities/foo.md")) + assert.Contains(t, read(t, dir, "wiki/concepts/foo.md"), "talk", "the existing concept got the new source reference") +} diff --git a/ingestion/internal/watcher/watcher_wing_test.go b/ingestion/internal/watcher/watcher_wing_test.go index 16da56d..da865cd 100644 --- a/ingestion/internal/watcher/watcher_wing_test.go +++ b/ingestion/internal/watcher/watcher_wing_test.go @@ -339,3 +339,20 @@ func TestFrontmatterWing(t *testing.T) { func TestFrontmatterWing_ABOMDoesNotHideTheKey(t *testing.T) { assert.Equal(t, "homelab", frontmatterWing("\ufeff---\nwing: homelab\n---\nbody")) } + +// A note that produced nothing must not leave the inbox as "processed". +func TestProcessDir_AWingIngestThatWritesNothingFailsTheNoteVisibly(t *testing.T) { + var llmCalls atomic.Int32 + cfg, brainDir := wingTestCfg(t, router, &llmCalls) + cfg.Pipeline.Complete = func(context.Context, string, string) (string, error) { return "[]", nil } + put(t, brainDir, "empty-result.md", "A note the model returned nothing for.") + + require.Empty(t, processDir(context.Background(), cfg, "2026-10-20")) + + assert.True(t, has(brainDir, "raw/failed/empty-result.md"), "kept where a human looks") + processed, _ := filepath.Glob(filepath.Join(brainDir, "raw", "processed", "*", "*")) + assert.Empty(t, processed, "never archived as processed") + logb, err := os.ReadFile(filepath.Join(brainDir, "log.md")) + require.NoError(t, err) + assert.Contains(t, string(logb), "no source page") +} diff --git a/ingestion/internal/wing/router.go b/ingestion/internal/wing/router.go index 8d284ef..3df65ed 100644 --- a/ingestion/internal/wing/router.go +++ b/ingestion/internal/wing/router.go @@ -142,8 +142,10 @@ func matchesClient(re *regexp.Regexp, text string, chunk int) bool { if end == len(text) { return false } - // A cut inside a rune is harmless: the overlap carries every marker - // shorter than it whole into the next chunk. + // A cut inside a rune is harmless: the overlap carries every match that + // spans at most filterOverlap bytes whole into the next chunk. A match + // longer than that (a pattern like `acme.{5000}corp`, or padding + // between two words) can be missed on a chunk edge; documented limit. start = end - overlap } return false