hyperguild/ingestion/internal/pipeline/pipeline.go

// ingestion/internal/pipeline/pipeline.go
package pipeline

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"github.com/mathiasbq/hyperguild/ingestion/internal/brain"
	"github.com/mathiasbq/hyperguild/ingestion/internal/brainfs"
	"io/fs"
	"path/filepath"
	"strings"
	"time"

	"github.com/mathiasbq/hyperguild/ingestion/internal/provenance"
	"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)

	// Wing, when set, writes into wiki/<wing>/{sources,facts}/ and scopes the
	// inventory (title resolution, back-references) to that wing alone. Empty
	// keeps the retired wiki/{sources,concepts,entities}/ layout.
	Wing string
}

// 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) {
	if cfg.Wing != "" && brain.Sanitise(cfg.Wing) != cfg.Wing {
		return Result{}, fmt.Errorf("invalid wing %q: must be a plain lower-case wing name", cfg.Wing)
	}
	var inventory map[wiki.PageType][]wiki.Entry
	var err error
	if cfg.Wing != "" {
		inventory, err = wiki.LoadWingInventory(brainDir, cfg.Wing)
	} else {
		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)
	if cfg.Wing != "" {
		var rerr error
		if sourceSlug, rerr = resolveSourceSlug(brainDir, cfg.Wing, sourceSlug); rerr != nil {
			return Result{}, rerr
		}
	}
	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, cfg.Wing, sourceSlug) {
		return Result{Skipped: true, Warnings: []string{
			fmt.Sprintf("source %q unchanged since last ingest (hash %s) — extraction skipped", source, bodyHash[:12]),
		}}, nil
	}

	// The source's own page path is deterministic (sourceSlug, computed above)
	// even before extraction runs, unlike a concept/entity page's path, which
	// only exists once the model has produced a title. So page-aware
	// extraction (#106) is scoped to the source page: it's the one page every
	// chunk's prompt can be given in advance.
	existingPage := readSourcePage(brainDir, cfg.Wing, sourceSlug)

	chunks := Chunk(content, cfg.ChunkSize)

	var allRaw []RawPage
	var allWarnings []string

	for _, chunk := range chunks {
		userPrompt := BuildPrompt(schema, source, chunk, inventory, existingPage)
		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)

	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
// step. Use this when the caller has already produced the structured RawPage data
// (e.g. from a more capable model or manual curation).
//
// Every page written is stamped with provenance.FromContext(ctx)
// (hyperguild#145); a caller that knows the principal attaches it with
// provenance.WithStamp, everyone else stamps origin: unknown.
func RunRaw(ctx context.Context, 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(ctx, "", rawPages, sourceSlug, date, brainDir, source, inventory, nil, dryRun)
}

// buildAndWrite runs BuildPages through write for both Run and RunRaw.
func buildAndWrite(ctx context.Context, wing string, rawPages []RawPage, sourceSlug, date, brainDir, source string, inventory map[wiki.PageType][]wiki.Entry, warnings []string, dryRun bool) (Result, error) {
	pages, buildWarnings := BuildPagesIn(wing, 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)
	if wing != "" {
		var clashWarnings []string
		var cerr error
		merged, clashWarnings, cerr = dropSlugClashes(brainDir, merged)
		if cerr != nil {
			return Result{}, cerr
		}
		warnings = append(warnings, clashWarnings...)
	}

	var written []string
	for _, page := range merged {
		if !dryRun {
			content, conflicts, err := reviseOrCreate(brainDir, page)
			if err != nil {
				return Result{}, err
			}
			for _, heading := range conflicts {
				// Reported, not resolved. The on-disk version stands; whether
				// the disagreement is a genuine contradiction is the
				// Judge-gated enrich pass's call, not the write path's.
				warnings = append(warnings, fmt.Sprintf(
					"%s: source now disagrees on section %q — on-disk version kept",
					page.Path, heading))
			}
			// Stamped as the last step, after merge/revise, so a page that
			// was merged into is re-attributed to this ingest too.
			stamped, err := provenance.Apply(content, provenance.FromContext(ctx))
			if err != nil {
				return Result{}, fmt.Errorf("stamp %s: %w", page.Path, err)
			}
			// Written through the brain root, never through a symlinked
			// 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)
			}
		}
		written = append(written, page.Path)
	}

	if !dryRun {
		if wing != "" {
			if err := brain.BuildWingIndex(brainDir, wing); err != nil {
				warnings = append(warnings, fmt.Sprintf("rebuild wing index: %v", err))
			}
		} else 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[:])
}

// readSourcePage returns the full on-disk content of the source page at
// wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet.
// It reads through the brain root, regular files only (hyperguild#157): the
// content goes into the LLM prompt.
func readSourcePage(brainDir, wing, sourceSlug string) string {
	rel := "wiki/sources/" + sourceSlug + ".md"
	if wing != "" {
		rel = "wiki/" + wing + "/sources/" + sourceSlug + ".md"
	}
	raw, err := readBrainFile(brainDir, rel)
	if err != nil {
		return ""
	}
	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,
// regular files only (hyperguild#157).
func readBrainFile(brainDir, rel string) ([]byte, error) {
	bfs, err := brainfs.Open(brainDir)
	if err != nil {
		return nil, err
	}
	defer func() { _ = bfs.Close() }()
	return bfs.ReadRegular(rel)
}

// 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, wing, sourceSlug string) string {
	for _, line := range strings.Split(readSourcePage(brainDir, wing, sourceSlug), "\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(brainDir string, page wiki.Page) (string, []string, error) {
	existing, err := readBrainFile(brainDir, page.Path)
	switch {
	case errors.Is(err, fs.ErrNotExist):
		return page.Content, nil, nil
	case err != nil:
		return "", nil, fmt.Errorf("read existing %s: %w", page.Path, err)
	}
	merged, conflicts := wiki.MergeReport(wiki.Page{Path: page.Path, Content: string(existing)}, page)
	return merged.Content, conflicts, 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 := readBrainFile(brainDir, "schema.md")
	if err != nil {
		return defaultSchema
	}
	return strings.TrimSpace(string(b))
}

// resolveSourceSlug returns the slug this wing's source page uses. A slug is a
// page's graph identity and must be unique brain-wide, but two wings dropping a
// file of the same name (notes.md, readme.md) is ordinary, so when the plain
// slug is taken by anything but this wing's own source page the wing name is
// appended (talk-homelab, then talk-homelab-2, ...). The choice is stable: a
// re-ingest finds the page it made, because that page is "its own".
func resolveSourceSlug(brainDir, wing, base string) (string, error) {
	for n := 1; n <= 100; n++ {
		cand := base
		switch n {
		case 1:
		case 2:
			cand = base + "-" + wing
		default:
			cand = fmt.Sprintf("%s-%s-%d", base, wing, n-1)
		}
		existing, err := wiki.FindSlugs(brainDir, cand)
		if err != nil {
			return "", fmt.Errorf("check source slug: %w", err)
		}
		own := "wiki/" + wing + "/sources/" + cand + ".md"
		free, hasOwn := true, false
		for _, p := range existing {
			if p == own {
				hasOwn = true
			} else {
				free = false
			}
		}
		// 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
		}
	}
	return "", fmt.Errorf("no free source slug for %q in wing %q", base, wing)
}

// dropSlugClashes removes, from the pages about to be written in a wing, any
// page whose slug belongs to a different page. A slug is the graph identity, so
// two pages must not share one; and a page that shares one cannot be updated
// again, because every later ingest sees the other as its clash.
//
// Two kinds of clash, both by slug, neither failing the ingest (one extracted
// page must not be able to lock a source out of re-ingest):
//   - inside the batch (a clip named notes.md whose model output also has an
//     entity "Notes"): the source wins over an entity, an entity over a concept;
//   - against the brain: an entity or concept whose slug is another page's. The
//     SOURCE page is exempt: it is the note's own output, and its slug was
//     already chosen free (or its own) by resolveSourceSlug.
//
// A page's own path is never a clash (a shared entity merging into itself).
func dropSlugClashes(brainDir string, pages []wiki.Page) ([]wiki.Page, []string, error) {
	rank := func(p wiki.Page) int {
		switch {
		case isSourcePath(p.Path):
			return 0
		case strings.HasPrefix(p.Path, "wiki/entities/"):
			return 1
		case strings.HasPrefix(p.Path, "wiki/concepts/"):
			return 2
		}
		return 3
	}
	slugOf := func(p wiki.Page) string { return strings.TrimSuffix(filepath.Base(p.Path), ".md") }

	// 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 {
		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
	var warnings []string
	for _, p := range pages {
		slug := slugOf(p)
		if w := winner[slug]; w.Path != p.Path {
			warnings = append(warnings, fmt.Sprintf("skipped %s: slug %q is also used by %s in this batch", p.Path, slug, w.Path))
			continue
		}
		if rank(p) == 0 {
			kept = append(kept, p)
			continue
		}
		existing, err := wiki.FindSlugs(brainDir, slug)
		if err != nil {
			return nil, nil, fmt.Errorf("check slug %q: %w", slug, err)
		}
		clash := ""
		for _, e := range existing {
			if e != p.Path {
				clash = e
				break
			}
		}
		if clash != "" {
			warnings = append(warnings, fmt.Sprintf("skipped %s: slug %q is already used by %s", p.Path, slug, clash))
			continue
		}
		kept = append(kept, p)
	}
	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)
}