mathiasandClaude Opus 5.5
eedb110be5
Ingest pipeline readers (readSourcePage and loadSchema feed the LLM prompt; reviseOrCreate; injectSourceRefs; BackfillRefs list and reads) and classification.Load read through brainfs: symlinked or special files no longer reach the prompt, written pages or the I1 sovereignty config, and a FIFO no longer hangs ingest or startup. A symlinked/special classification.yaml is a hard error; an absent one still yields defaults. Refs #157. Adversarial review: BLOCKED, BLOCKED, PASS. Follow-ups: #158 (write paths), #179 (FIFO tripwire). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Mathias <mthbqv@gmail.com>
101 lines
2.7 KiB
Go
101 lines
2.7 KiB
Go
// ingestion/internal/pipeline/backfill.go
|
|
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
|
|
"io/fs"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"github.com/mathiasbq/hyperguild/ingestion/internal/wiki"
|
|
)
|
|
|
|
// BackfillRefs walks wiki/sources/ and injects source back-references into every
|
|
// concept and entity page that each source links to.
|
|
// Changes for all sources are accumulated in memory before writing, so multiple
|
|
// sources referencing the same concept are merged in one pass.
|
|
// Deduplication is handled by wiki.Merge — running this multiple times is safe.
|
|
// Returns the number of concept/entity pages written.
|
|
func BackfillRefs(ctx context.Context, brainDir string) (int, error) {
|
|
inventory, err := wiki.LoadInventory(brainDir)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("load inventory: %w", err)
|
|
}
|
|
|
|
// Listed and read through the brain root, regular files only
|
|
// (hyperguild#157).
|
|
bfs, err := brainfs.Open(brainDir)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("open brain: %w", err)
|
|
}
|
|
defer func() { _ = bfs.Close() }()
|
|
names, err := bfs.ReadDirRegular("wiki/sources")
|
|
if err != nil {
|
|
if errors.Is(err, fs.ErrNotExist) {
|
|
return 0, nil
|
|
}
|
|
return 0, fmt.Errorf("read sources dir: %w", err)
|
|
}
|
|
|
|
// Accumulate all changes before writing: relPath → updated Page.
|
|
// Collecting first means two sources that both link the same concept
|
|
// get both refs merged before a single write.
|
|
pending := make(map[string]wiki.Page)
|
|
|
|
for _, name := range names {
|
|
if ctx.Err() != nil {
|
|
return 0, ctx.Err()
|
|
}
|
|
if !strings.HasSuffix(name, ".md") {
|
|
continue
|
|
}
|
|
|
|
b, err := bfs.ReadRegular("wiki/sources/" + name)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
sourceContent := string(b)
|
|
sourceSlug := strings.TrimSuffix(name, ".md")
|
|
sourceTitle := extractTitle(sourceContent)
|
|
if sourceTitle == "" {
|
|
sourceTitle = sourceSlug
|
|
}
|
|
sourceRef := "- [[" + sourceSlug + "|" + sourceTitle + "]]"
|
|
|
|
for slug := range extractWikilinks(sourceContent) {
|
|
if slug == sourceSlug {
|
|
continue
|
|
}
|
|
pt, ok := findInInventory(slug, inventory)
|
|
if !ok {
|
|
continue
|
|
}
|
|
relPath := "wiki/" + string(pt) + "/" + slug + ".md"
|
|
|
|
// Start from already-accumulated version if we've seen this page.
|
|
page, seen := pending[relPath]
|
|
if !seen {
|
|
raw, err := bfs.ReadRegular(relPath)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
page = wiki.Page{Path: relPath, Content: string(raw)}
|
|
}
|
|
pending[relPath] = addSourceRef(page, sourceRef)
|
|
}
|
|
}
|
|
|
|
for relPath, page := range pending {
|
|
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 len(pending), nil
|
|
}
|