// ingestion/internal/pipeline/pipeline.go package pipeline import ( "context" "crypto/sha256" "encoding/hex" "fmt" "os" "path/filepath" "strings" "time" "github.com/mathiasbq/hyperguild/ingestion/internal/wiki" ) // CompleteFunc is the function signature for LLM calls. type CompleteFunc func(ctx context.Context, system, user string) (string, error) // Config holds pipeline configuration. type Config struct { Complete CompleteFunc ChunkSize int // 0 = no chunking Schema string // overrides brain/schema.md when set (useful in tests) } // Result is the outcome of a pipeline run. type Result struct { Pages []string // relative paths written (or would-be written in dry-run) Warnings []string // Skipped reports that the source was byte-identical to the last ingest, // so no extraction ran and nothing was written. Reported rather than // silent: a run that does nothing must say so, or "the pipeline is fine" // and "the pipeline stopped working" look the same (#105, #107). Skipped bool } // Run ingests content and writes structured wiki pages to brainDir/wiki/. // In dry-run mode, pages are returned but not written to disk. func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryRun bool) (Result, error) { inventory, err := wiki.LoadInventory(brainDir) if err != nil { return Result{}, fmt.Errorf("load inventory: %w", err) } schema := cfg.Schema if schema == "" { schema = loadSchema(brainDir) } sourceSlug := wiki.Slug(source) date := time.Now().UTC().Format("2006-01-02") // Drift check BEFORE any extraction: an unchanged source costs a full LLM // pass over every chunk to produce a result that merges to nothing (#105). // The merge is already idempotent (#103), so this is cost and signal, not // correctness — which is why it is safe for it to be an early return. bodyHash := hashBody(content) if !dryRun && bodyHash != "" && bodyHash == storedSourceHash(brainDir, sourceSlug) { return Result{Skipped: true, Warnings: []string{ fmt.Sprintf("source %q unchanged since last ingest (hash %s) — extraction skipped", source, bodyHash[:12]), }}, nil } chunks := Chunk(content, cfg.ChunkSize) var allRaw []RawPage var allWarnings []string for _, chunk := range chunks { userPrompt := BuildPrompt(schema, source, chunk, inventory) output, err := cfg.Complete(ctx, systemPrompt, userPrompt) if err != nil { return Result{}, fmt.Errorf("LLM call: %w", err) } raw, warnings := ParseRawPages(output) allRaw = append(allRaw, raw...) allWarnings = append(allWarnings, warnings...) } meta := parseContentFrontmatter(content) meta.BodyHash = bodyHash applySourceMeta(allRaw, meta) return buildAndWrite(allRaw, sourceSlug, date, brainDir, source, inventory, allWarnings, dryRun) } // RunRaw runs the pipeline on pre-parsed RawPages, skipping the LLM extraction // step. Use this when the caller has already produced the structured RawPage data // (e.g. from a more capable model or manual curation). func RunRaw(brainDir, source string, rawPages []RawPage, dryRun bool) (Result, error) { inventory, err := wiki.LoadInventory(brainDir) if err != nil { return Result{}, fmt.Errorf("load inventory: %w", err) } sourceSlug := wiki.Slug(source) date := time.Now().UTC().Format("2006-01-02") return buildAndWrite(rawPages, sourceSlug, date, brainDir, source, inventory, nil, dryRun) } // buildAndWrite runs BuildPages through write for both Run and RunRaw. func buildAndWrite(rawPages []RawPage, sourceSlug, date, brainDir, source string, inventory map[wiki.PageType][]wiki.Entry, warnings []string, dryRun bool) (Result, error) { pages, buildWarnings := BuildPages(rawPages, sourceSlug, date) warnings = append(warnings, buildWarnings...) resolved := Resolve(pages, inventory) canonicalized, linkWarnings := CanonicalizeLinks(resolved, inventory) warnings = append(warnings, linkWarnings...) withRefs := injectSourceRefs(canonicalized, inventory, brainDir) merged := mergeAll(withRefs) var written []string for _, page := range merged { 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, err := reviseOrCreate(dest, page) if err != nil { return Result{}, err } if err := os.WriteFile(dest, []byte(content), 0o644); err != nil { return Result{}, fmt.Errorf("write %s: %w", page.Path, err) } } written = append(written, page.Path) } if !dryRun { if err := wiki.RebuildIndex(brainDir, date); err != nil { warnings = append(warnings, fmt.Sprintf("rebuild index: %v", err)) } if err := wiki.AppendLog(brainDir, source, written, warnings, date); err != nil { warnings = append(warnings, fmt.Sprintf("append log: %v", err)) } } return Result{Pages: written, Warnings: warnings}, nil } // hashBody returns sha256 of content with any leading frontmatter block // removed, or "" for empty content. // // Body-only on purpose: a re-crawl that stamps a new fetch date has not changed // the source, and treating that as drift would re-extract the corpus every time // a crawler ran (#105). func hashBody(content string) string { body := content if strings.HasPrefix(body, "---\n") { if _, rest, found := strings.Cut(body[4:], "\n---"); found { body = strings.TrimPrefix(rest, "\n") } } if strings.TrimSpace(body) == "" { return "" } sum := sha256.Sum256([]byte(body)) return hex.EncodeToString(sum[:]) } // storedSourceHash reads source_hash from the source page a previous ingest // wrote, or "" when there is no such page or no such field. // // Absence always means "extract". That is the fail-safe direction: a missed // skip costs one extraction, while a wrong skip loses an update silently — and // silence is the failure mode this repo keeps paying for. func storedSourceHash(brainDir, sourceSlug string) string { raw, err := os.ReadFile(filepath.Join(brainDir, "wiki", "sources", sourceSlug+".md")) if err != nil { return "" } for _, line := range strings.Split(string(raw), "\n") { if key, value, found := strings.Cut(line, ":"); found && strings.TrimSpace(key) == "source_hash" { return strings.TrimSpace(value) } } return "" } // reviseOrCreate returns the bytes to write for one extracted page: merged into // whatever is already on disk, or the page as-is when the path is new. // // This is the compile half of compile-on-ingest. Before it, the write was an // unconditional os.WriteFile, so re-ingesting a source replaced its page // wholesale and discarded everything accumulated since — including human edits // (hyperguild#43: "re-ingest overwrites wholesale"). // // The merge is deliberately asymmetric in the existing page's favour: // wiki.Merge keeps the on-disk frontmatter and the on-disk version of any // section both sides carry, adds sections only the extraction produced, and // unions bullet lists. So a freshly extracted page can add to a note but never // silently rewrite what a human put there — which is what makes it safe to run // this on every ingest rather than behind a review gate. // // Known limitation: because the on-disk frontmatter wins, `last_updated` keeps // its first-ingest date on a revised page. Left deliberately — the alternative // is this function editing frontmatter it does not own, and no lint reads that // field today. func reviseOrCreate(dest string, page wiki.Page) (string, error) { existing, err := os.ReadFile(dest) switch { case os.IsNotExist(err): return page.Content, nil case err != nil: return "", fmt.Errorf("read existing %s: %w", page.Path, err) } return wiki.Merge(wiki.Page{Path: page.Path, Content: string(existing)}, page).Content, nil } // mergeAll deduplicates pages by path, merging content from later occurrences. func mergeAll(pages []wiki.Page) []wiki.Page { order := make([]string, 0, len(pages)) byPath := make(map[string]wiki.Page, len(pages)) for _, p := range pages { if _, seen := byPath[p.Path]; !seen { order = append(order, p.Path) byPath[p.Path] = p } else { byPath[p.Path] = wiki.Merge(byPath[p.Path], p) } } result := make([]wiki.Page, 0, len(order)) for _, path := range order { result = append(result, byPath[path]) } return result } const defaultSchema = `# Brain Wiki Schema Three page types: wiki/sources/, wiki/concepts/, wiki/entities/. See brain/schema.md for the full schema. ` func loadSchema(brainDir string) string { b, err := os.ReadFile(filepath.Join(brainDir, "schema.md")) if err != nil { return defaultSchema } return strings.TrimSpace(string(b)) }