feat(pipeline): skip extraction when the source has not changed (#105) #126

Merged
mathias merged 4 commits from fix/105-source-drift-on-main into main 2026-09-23 05:09:30 +00:00
7 changed files with 243 additions and 8 deletions
Showing only changes of commit a5a328fea5 - Show all commits
+15 -2
View File
@@ -100,8 +100,21 @@ review gate.
Re-ingesting unchanged content is **byte-identical** — no diff, so brain-sync
does not commit noise that hides real edits.
Known limitation: because the on-disk frontmatter wins, `last_updated` keeps its
first-ingest date on a revised page.
Two fields are **generator-owned** and do get refreshed on a revise:
`last_updated` and `source_hash`. Everything else on disk wins. The boundary is
an explicit allowlist, so a field nobody has thought about stays human-owned.
## Re-ingesting an unchanged source is skipped
Ingest stores `source_hash` — sha256 of the ingested **body**, frontmatter
excluded — on the source page. A later ingest of the same source compares it
first and skips extraction entirely when it matches, reporting the skip rather
than doing it silently.
Body-only is deliberate: 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. A missing or unreadable hash always means "extract" — a missed skip
costs one extraction, a wrong skip loses an update.
## Phase 1 tools (available now)
+3
View File
@@ -76,6 +76,9 @@ func buildFrontmatter(rp RawPage, date string) string {
}
fmt.Fprintf(&sb, "date_ingested: %s\n", date)
fmt.Fprintf(&sb, "last_updated: %s\n", date)
if rp.SourceHash != "" {
fmt.Fprintf(&sb, "source_hash: %s\n", rp.SourceHash)
}
if rp.Source != "" {
fmt.Fprintf(&sb, "source: %s\n", yamlScalar(rp.Source))
}
+10
View File
@@ -20,6 +20,10 @@ type RawPage struct {
// ingested content's own frontmatter (see parseContentFrontmatter) — never
// set by the LLM. json:"-" keeps them immune to same-named keys the LLM
// might emit. Only meaningful for Type == "source".
// SourceHash is sha256 of the ingested BODY (frontmatter excluded), stamped
// so the next ingest of the same source can tell "unchanged" from "edited"
// without re-running extraction (#105).
SourceHash string `json:"-"`
Source string `json:"-"`
Author string `json:"-"`
Published string `json:"-"`
@@ -112,6 +116,11 @@ type sourceMeta struct {
Source string
Author string
Published string
// BodyHash is sha256 of the content with any leading frontmatter removed.
// 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
// whole corpus every time a crawler ran (#105).
BodyHash string
}
// parseContentFrontmatter extracts source/author/published from a leading
@@ -154,6 +163,7 @@ func applySourceMeta(pages []RawPage, meta sourceMeta) {
if pages[i].Type != "source" {
continue
}
pages[i].SourceHash = meta.BodyHash
pages[i].Source = meta.Source
pages[i].Author = meta.Author
pages[i].Published = meta.Published
+61 -1
View File
@@ -3,6 +3,8 @@ package pipeline
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"os"
"path/filepath"
@@ -26,6 +28,11 @@ type Config struct {
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/.
@@ -43,6 +50,18 @@ func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryR
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
@@ -59,7 +78,9 @@ func Run(ctx context.Context, cfg Config, brainDir, content, source string, dryR
allWarnings = append(allWarnings, warnings...)
}
applySourceMeta(allRaw, parseContentFrontmatter(content))
meta := parseContentFrontmatter(content)
meta.BodyHash = bodyHash
applySourceMeta(allRaw, meta)
return buildAndWrite(allRaw, sourceSlug, date, brainDir, source, inventory, allWarnings, dryRun)
}
@@ -119,6 +140,45 @@ func buildAndWrite(rawPages []RawPage, sourceSlug, date, brainDir, source string
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.
//
@@ -233,3 +233,63 @@ func TestRunRaw_ReIngestingUnchangedContentIsByteIdentical(t *testing.T) {
assert.Equal(t, string(first), string(second),
"re-ingesting unchanged content must leave the page byte-identical")
}
// --- #105: source drift — an unchanged re-ingest must not re-extract --------
func TestRun_SkipsExtractionWhenTheSourceIsUnchanged(t *testing.T) {
// #103 made a re-ingest harmless (byte-identical merge). It is still not
// free: the full LLM extraction runs over every chunk to produce a result
// that merges to nothing. This is the cost half.
dir := t.TempDir()
var calls int
cfg := Config{Complete: func(_ context.Context, _, _ string) (string, error) {
calls++
return `[{"title":"Shape Up","type":"source","subtype":"book","domain":"product","content":"## Summary\n\nA book.\n"}]`, nil
}}
content := "---\nsource: https://example.com/shape-up\n---\n\nThe body of the article.\n"
first, err := Run(context.Background(), cfg, dir, content, "shape-up", false)
require.NoError(t, err)
require.Equal(t, 1, calls, "first ingest must extract")
assert.False(t, first.Skipped)
second, err := Run(context.Background(), cfg, dir, content, "shape-up", false)
require.NoError(t, err)
assert.Equal(t, 1, calls, "unchanged content must not be extracted again")
assert.True(t, second.Skipped, "the skip must be reported, not silent")
}
func TestRun_ReExtractsWhenTheSourceBodyChanged(t *testing.T) {
dir := t.TempDir()
var calls int
cfg := Config{Complete: func(_ context.Context, _, _ string) (string, error) {
calls++
return `[{"title":"Shape Up","type":"source","subtype":"book","domain":"product","content":"## Summary\n\nA book.\n"}]`, nil
}}
_, err := Run(context.Background(), cfg, dir, "---\nsource: x\n---\n\nversion one\n", "shape-up", false)
require.NoError(t, err)
_, err = Run(context.Background(), cfg, dir, "---\nsource: x\n---\n\nversion two — the source was edited\n", "shape-up", false)
require.NoError(t, err)
assert.Equal(t, 2, calls, "changed content must be re-extracted")
}
func TestRun_FrontmatterOnlyChangeIsNotDrift(t *testing.T) {
// The hash covers the body, so re-fetching a page whose metadata moved
// (a re-crawl stamping a new date) is not treated as the source changing.
dir := t.TempDir()
var calls int
cfg := Config{Complete: func(_ context.Context, _, _ string) (string, error) {
calls++
return `[{"title":"Shape Up","type":"source","subtype":"book","domain":"product","content":"## Summary\n\nA book.\n"}]`, nil
}}
_, err := Run(context.Background(), cfg, dir, "---\nsource: x\nfetched: monday\n---\n\nsame body\n", "shape-up", false)
require.NoError(t, err)
_, err = Run(context.Background(), cfg, dir, "---\nsource: x\nfetched: tuesday\n---\n\nsame body\n", "shape-up", false)
require.NoError(t, err)
assert.Equal(t, 1, calls, "only the body counts as the source")
}
+51 -2
View File
@@ -38,7 +38,7 @@ type section struct {
// not present in a are appended.
func Merge(a, b Page) Page {
fmA, preambleA, secsA := parseSections(a.Content)
_, _, secsB := parseSections(b.Content)
fmB, _, secsB := parseSections(b.Content)
idx := make(map[string]int, len(secsA))
for i, s := range secsA {
@@ -61,7 +61,56 @@ func Merge(a, b Page) Page {
}
}
return Page{Path: a.Path, Content: rebuildContent(fmA, preambleA, secsA)}
return Page{Path: a.Path, Content: rebuildContent(refreshOwned(fmA, fmB), preambleA, secsA)}
}
// generatorOwned names the frontmatter fields the ingest pipeline may refresh
// on a page it is revising. Everything else on disk wins.
//
// #103 made the on-disk frontmatter win wholesale, which is what stops a
// regenerating writer from overwriting human-set metadata — a title someone
// fixed, or the claim-property envelope (brain#21). The cost was that fields
// the generator legitimately owns froze at their first-ingest values: when it
// last ran, and what it last saw.
//
// An explicit allowlist rather than a carve-out: the ownership boundary is
// nameable, and a field nobody has thought about stays human-owned by default,
// which is the safe direction.
var generatorOwned = map[string]bool{
"last_updated": true,
"source_hash": true,
}
// refreshOwned returns a's frontmatter with the generator-owned fields taken
// from b where b supplies them. Fields absent from b keep a's value, so a
// generator that stops emitting a field cannot blank it.
func refreshOwned(a, b string) string {
bVals := make(map[string]string)
for _, line := range strings.Split(b, "\n") {
key, value, found := strings.Cut(line, ":")
if !found || !generatorOwned[strings.TrimSpace(key)] {
continue
}
bVals[strings.TrimSpace(key)] = strings.TrimSpace(value)
}
if len(bVals) == 0 {
return a
}
out := make([]string, 0, len(strings.Split(a, "\n")))
for _, line := range strings.Split(a, "\n") {
key, _, found := strings.Cut(line, ":")
k := strings.TrimSpace(key)
if found && generatorOwned[k] {
if v, ok := bVals[k]; ok {
out = append(out, k+": "+v)
delete(bVals, k)
continue
}
}
out = append(out, line)
}
return strings.Join(out, "\n")
}
// parseSections splits markdown into its frontmatter, its preamble (content
+40
View File
@@ -174,3 +174,43 @@ title: X
assert.Contains(t, got, "Does this converge?", "existing question must survive")
assert.Contains(t, got, "What happens under load?", "new question must be added")
}
func TestMerge_GeneratorOwnedFrontmatterFieldsAreRefreshed(t *testing.T) {
// #103 made on-disk frontmatter win wholesale, which is what protects
// human-set fields from a regenerating writer. The cost was that fields the
// generator legitimately owns — when it last ran, what it last saw — froze
// at their first-ingest values. An explicit owned-set is the boundary:
// named fields the generator may refresh, everything else human-owned.
existing := Page{Path: "wiki/sources/x.md", Content: `---
title: 'X'
last_updated: 2026-01-01
source_hash: aaaa
half_life: durable
claim_provenance: asserted
---
## Summary
Original.
`}
fresh := Page{Path: "wiki/sources/x.md", Content: `---
title: 'Renamed by the model'
last_updated: 2026-08-22
source_hash: bbbb
---
## Summary
Rewritten by the model.
`}
got := Merge(existing, fresh).Content
assert.Contains(t, got, "last_updated: 2026-08-22", "generator owns last_updated")
assert.Contains(t, got, "source_hash: bbbb", "generator owns source_hash")
assert.Contains(t, got, "title: 'X'", "a human-set title must survive regeneration")
assert.Contains(t, got, "half_life: durable", "the claim envelope must survive regeneration")
assert.Contains(t, got, "claim_provenance: asserted")
assert.Contains(t, got, "Original.", "an existing section still wins over the model's rewrite")
assert.NotContains(t, got, "Renamed by the model")
}