hyperguild/ingestion/internal/pipeline/pipeline.go

// 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/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)
}

// 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
	}

	// 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, 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)

	return buildAndWrite(ctx, 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).
//
// 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, 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, conflicts, err := reviseOrCreate(dest, 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)
			}
			if err := os.WriteFile(dest, []byte(stamped), 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[:])
}

// readSourcePage returns the full on-disk content of the source page at
// wiki/sources/<sourceSlug>.md, or "" when it doesn't exist yet.
func readSourcePage(brainDir, sourceSlug string) string {
	raw, err := os.ReadFile(filepath.Join(brainDir, "wiki", "sources", sourceSlug+".md"))
	if err != nil {
		return ""
	}
	return string(raw)
}

// 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 {
	for _, line := range strings.Split(readSourcePage(brainDir, 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(dest string, page wiki.Page) (string, []string, error) {
	existing, err := os.ReadFile(dest)
	switch {
	case os.IsNotExist(err):
		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 := os.ReadFile(filepath.Join(brainDir, "schema.md"))
	if err != nil {
		return defaultSchema
	}
	return strings.TrimSpace(string(b))
}