Part 5 of 1729% complete

2026-08-21•7 min read

Build the runbook ingestion pipeline: Markdown to pgvector

Checkpoint tag: chapter-04-knowledge-ingestion — docs/runbooks is ingested idempotently, chunks carry tenant/service/version metadata, and you can inspect them with SQL.

What will be built

knowledge-ingestion becomes a batch job: it streams Markdown files from docs/runbooks, splits them on heading boundaries before applying token limits, computes deterministic chunk IDs, embeds each chunk through the configured EmbeddingModel, and upserts into opsdb.knowledge. Unchanged documents are skipped by checksum; superseded versions are retired; --ingest.dry-run=true reports the plan without writing.

Why it matters

RAG quality is decided here, not at query time. Chunks that split a procedure mid-step, IDs that change between runs, or metadata that can’t filter by tenant all show up downstream as wrong answers and cross-tenant leaks. And the incremental design matters operationally: re-embedding every document on every run turns a 30-second job into a 30-minute one and makes the embedding provider a billing dependency of your CI.

Production note: the prompt-sized files here could be read with Files.readString, but the parser streams line-by-line anyway (BufferedReader). The day someone drops a 200 MB export into the runbook directory, the job should degrade into “slow but bounded memory,” not OutOfMemoryError. Splitting an already-loaded giant string does not fix that — the load already happened.

Prerequisites and starting tag

Starting tag: chapter-03-basic-agent-api. Postgres (with pgvector) and Ollama running; nomic-embed-text pulled (the Compose Ollama entrypoint from Chapter 3 does this).

Concepts explained

Semantic before token. A Markdown document already declares its structure in headings. We split on ##-level sections first — a chunk never crosses a procedure boundary — then subdivide oversized sections at token boundaries. Reversing that order produces chunks that start mid-command.

Deterministic chunk IDs. chunk_id = sha256(documentId + version + headingPath + index). Re-ingesting an unchanged file produces identical IDs — the upsert is a no-op, not a delete/insert storm — and a chunk ID in a citation is stable enough to audit later.

Why our own tables, not PgVectorStore. Spring AI’s PgVectorStore is a good default, but it stores metadata as a JSON blob and manages its own schema. We need typed columns for tenant_id, doc_version, and service_id — the Chapter 5 retrieval filter is a plain WHERE clause with an index behind it, not a JSON predicate. The trade-off is real (we own the similarity SQL and the HNSW index), and it’s stated as an ADR in docs/adr/.

Files added or changed

1
docs/runbooks/*.md (seeded runbooks)
2
knowledge-ingestion/build.gradle.kts
3
knowledge-ingestion/src/main/resources/application.yml
4
knowledge-ingestion/src/main/resources/db/migration/V1__knowledge_init.sql
5
knowledge-ingestion/src/main/java/in/o612/eng/opsagent/ingest/
6
    IngestionApplication.java (exists), IngestionRunner.java
7
    parse/MarkdownSectionParser.java, parse/RunbookFrontMatter.java
8
    chunk/Chunker.java
9
    embed/EmbeddingGateway.java
10
    store/KnowledgeStore.java
11
    IngestionProperties.java
12
knowledge-ingestion/src/test/java/...

Complete code

application.yml

1
spring:
2
  application:
3
    name: knowledge-ingestion
4
  datasource:
5
    url: ${OPS_DB_URL:jdbc:postgresql://localhost:5432/opsdb}
6
    username: ${OPS_DB_USER:agent}
7
    password: ${OPS_DB_PASSWORD:agent-dev-password}
8
  flyway:
9
    schemas: knowledge
10
    default-schema: knowledge
11
    locations: classpath:db/migration
12
  ai:
13
    ollama:
14
      base-url: ${OLLAMA_BASE_URL:http://localhost:11434}
15
      embedding:
16
        options:
17
          model: nomic-embed-text
18
  main:
19
    web-application-type: none
20

21
ingest:
22
  source-dir: ${RUNBOOK_DIR:docs/runbooks}
23
  dry-run: false
24
  max-chunk-tokens: 512
25
  overlap-tokens: 64

V1__knowledge_init.sql

1
CREATE SCHEMA IF NOT EXISTS knowledge;
2
CREATE EXTENSION IF NOT EXISTS vector;
3

4
CREATE TABLE knowledge.documents (
5
    document_id    TEXT NOT NULL,
6
    version        TEXT NOT NULL,
7
    tenant_id      TEXT NOT NULL,
8
    service_id     TEXT,
9
    title          TEXT NOT NULL,
10
    effective_date DATE,
11
    checksum       TEXT NOT NULL,
12
    source_uri     TEXT NOT NULL,
13
    superseded     BOOLEAN NOT NULL DEFAULT FALSE,
14
    PRIMARY KEY (document_id, version)
15
);
16

17
CREATE TABLE knowledge.document_chunks (
18
    chunk_id      TEXT PRIMARY KEY,
19
    document_id   TEXT NOT NULL,
20
    version       TEXT NOT NULL,
21
    chunk_index   INT  NOT NULL,
22
    heading_path  TEXT NOT NULL,
23
    content       TEXT NOT NULL,
24
    tenant_id     TEXT NOT NULL,
25
    service_id    TEXT,
26
    embedding     vector(768) NOT NULL
27
);
28

29
CREATE INDEX chunks_tenant_service_idx
30
    ON knowledge.document_chunks (tenant_id, service_id);
31

32
CREATE INDEX chunks_embedding_hnsw
33
    ON knowledge.document_chunks
34
    USING hnsw (embedding vector_cosine_ops);
35

36
CREATE TABLE knowledge.ingestion_runs (
37
    id            BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
38
    started_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
39
    dry_run       BOOLEAN NOT NULL,
40
    scanned       INT NOT NULL,
41
    embedded      INT NOT NULL,
42
    skipped       INT NOT NULL,
43
    superseded    INT NOT NULL
44
);

RunbookFrontMatter + MarkdownSectionParser — front matter is key: value between --- fences; sections split on ATX headings, tracking the heading path (Deployment > Rollback):

1
package in.o612.eng.opsagent.ingest.parse;
2

3
import java.io.BufferedReader;
4
import java.io.IOException;
5
import java.io.Reader;
6
import java.util.ArrayDeque;
7
import java.util.ArrayList;
8
import java.util.Deque;
9
import java.util.List;
10

11
public class MarkdownSectionParser {
12

13
    public record Section(List<String> headingPath, String body) {}
14

15
    public List<Section> parse(Reader source) throws IOException {
16
        var sections = new ArrayList<Section>();
17
        var path = new ArrayDeque<String>();
18
        var body = new StringBuilder();
19
        boolean inFence = false;
20
        String currentHeading = null;
21

22
        try (var reader = new BufferedReader(source)) {
23
            String line;
24
            while ((line = reader.readLine()) != null) {
25
                if (line.strip().startsWith("```")) {
26
                    inFence = !inFence;                 // headings inside code are not headings
27
                    body.append(line).append('\n');
28
                    continue;
29
                }
30
                var m = HEADING.matcher(line);
31
                if (!inFence && m.matches()) {
32
                    flush(sections, path, body, currentHeading);
33
                    int level = m.group(1).length();
34
                    String text = m.group(2).strip();
35
                    while (path.size() >= level) path.pollLast();
36
                    path.addLast(text);
37
                    currentHeading = text;
38
                    continue;
39
                }
40
                body.append(line).append('\n');
41
            }
42
        }
43
        flush(sections, path, body, currentHeading);
44
        return sections;
45
    }
46

47
    private static final java.util.regex.Pattern HEADING =
48
            java.util.regex.Pattern.compile("^(#{1,6})\\s+(.+)$");
49

50
    private void flush(List<Section> out, Deque<String> path, StringBuilder body, String heading) {
51
        var text = body.toString().strip();
52
        if (!text.isEmpty()) {
53
            out.add(new Section(List.copyOf(path), text));
54
        }
55
        body.setLength(0);
56
    }
57
}

Add the tokenizer dependency — jtokkit = "1.1.0" in the catalog’s [versions], jtokkit = { module = "com.knuddels:jtokkit", version.ref = "jtokkit" } under [libraries], then implementation(libs.jtokkit) in knowledge-ingestion/build.gradle.kts.

Chunker — subdivide oversized sections on paragraph boundaries using a token count (jtokkit’s cl100k_base, close enough to every embedding model’s splitter for budgeting purposes):

1
package in.o612.eng.opsagent.ingest.chunk;
2

3
import com.knuddels.jtokkit.Encodings;
4
import com.knuddels.jtokkit.api.Encoding;
5
import in.o612.eng.opsagent.ingest.parse.MarkdownSectionParser.Section;
6

7
import java.util.ArrayList;
8
import java.util.List;
9

10
public class Chunker {
11

12
    public record Chunk(List<String> headingPath, int index, String content) {}
13

14
    private final Encoding encoding = Encodings.newDefaultEncodingRegistry()
15
            .getEncoding(com.knuddels.jtokkit.api.EncodingType.CL100K_BASE);
16
    private final int maxTokens;
17
    private final int overlapTokens;
18

19
    public Chunker(int maxTokens, int overlapTokens) {
20
        this.maxTokens = maxTokens;
21
        this.overlapTokens = overlapTokens;
22
    }
23

24
    public List<Chunk> chunk(Section section) {
25
        if (count(section.body()) <= maxTokens) {
26
            return List.of(new Chunk(section.headingPath(), 0, section.body()));
27
        }
28
        var chunks = new ArrayList<Chunk>();
29
        var current = new StringBuilder();
30
        int index = 0;
31
        for (String para : section.body().split("\n\n")) {
32
            if (count(current + para) > maxTokens && current.length() > 0) {
33
                chunks.add(new Chunk(section.headingPath(), index++, current.toString().strip()));
34
                current = new StringBuilder(tailTokens(current.toString())).append(para).append("\n\n");
35
            } else {
36
                current.append(para).append("\n\n");
37
            }
38
        }
39
        if (current.length() > 0) {
40
            chunks.add(new Chunk(section.headingPath(), index, current.toString().strip()));
41
        }
42
        return chunks;
43
    }
44

45
    private int count(String s) { return encoding.countTokens(s); }
46

47
    private String tailTokens(String s) {
48
        // carry the last ~overlapTokens worth of text into the next chunk
49
        var ids = encoding.encode(s);
50
        if (ids.size() <= overlapTokens) return s;
51
        return encoding.decode(ids.subList(ids.size() - overlapTokens, ids.size()));
52
    }
53
}

IngestionRunner — the job itself. Plan first, write second; dry-run stops after the plan:

1
package in.o612.eng.opsagent.ingest;
2

3
import in.o612.eng.opsagent.ingest.chunk.Chunker;
4
import in.o612.eng.opsagent.ingest.embed.EmbeddingGateway;
5
import in.o612.eng.opsagent.ingest.parse.MarkdownSectionParser;
6
import in.o612.eng.opsagent.ingest.parse.RunbookFrontMatter;
7
import in.o612.eng.opsagent.ingest.store.KnowledgeStore;
8
import org.springframework.boot.ApplicationArguments;
9
import org.springframework.boot.ApplicationRunner;
10
import org.springframework.stereotype.Component;
11

12
import java.io.FileReader;
13
import java.nio.charset.StandardCharsets;
14
import java.nio.file.Files;
15
import java.nio.file.Path;
16
import java.security.MessageDigest;
17
import java.util.HexFormat;
18
import java.util.List;
19
import java.util.stream.Stream;
20

21
@Component
22
public class IngestionRunner implements ApplicationRunner {
23

24
    private final MarkdownSectionParser parser = new MarkdownSectionParser();
25
    private final Chunker chunker;
26
    private final EmbeddingGateway embeddings;
27
    private final KnowledgeStore store;
28
    private final IngestionProperties props;
29

30
    public IngestionRunner(Chunker chunker, EmbeddingGateway embeddings,
31
                           KnowledgeStore store, IngestionProperties props) {
32
        this.chunker = chunker;
33
        this.embeddings = embeddings;
34
        this.store = store;
35
        this.props = props;
36
    }
37

38
    @Override
39
    public void run(ApplicationArguments args) throws Exception {
40
        int scanned = 0, embedded = 0, skipped = 0, superseded = 0;
41
        try (Stream<Path> files = Files.list(Path.of(props.sourceDir()))) {
42
            List<Path> runbooks = files.filter(p -> p.toString().endsWith(".md")).sorted().toList();
43
            for (Path file : runbooks) {
44
                scanned++;
45
                String checksum = sha256(file);
46
                var fm = RunbookFrontMatter.read(file);
47
                if (store.isCurrent(fm.documentId(), fm.version(), checksum)) {
48
                    skipped++;
49
                    continue;
50
                }
51
                if (props.dryRun()) {
52
                    System.out.printf("[dry-run] would ingest %s v%s (%s)%n",
53
                            fm.documentId(), fm.version(), file);
54
                    continue;
55
                }
56
                try (var reader = new FileReader(file.toFile(), StandardCharsets.UTF_8)) {
57
                    int idx = 0;
58
                    for (var section : parser.parse(reader)) {
59
                        for (var chunk : chunker.chunk(section)) {
60
                            String chunkId = chunkId(fm.documentId(), fm.version(),
61
                                    String.join(">", chunk.headingPath()), idx);
62
                            float[] vector = embeddings.embed(chunk.content());
63
                            store.upsertChunk(chunkId, fm, chunk, vector, idx++);
64
                        }
65
                    }
66
                }
67
                superseded += store.supersedeOlderVersions(fm.documentId(), fm.version(), checksum);
68
                store.recordDocument(fm, checksum, file.toUri().toString());
69
                embedded++;
70
            }
71
        }
72
        store.recordRun(props.dryRun(), scanned, embedded, skipped, superseded);
73
        System.out.printf("ingest complete: scanned=%d embedded=%d skipped=%d superseded=%d%n",
74
                scanned, embedded, skipped, superseded);
75
    }
76

77
    private static String chunkId(String doc, String version, String headingPath, int index) {
78
        return "chk-" + sha256(doc + "|" + version + "|" + headingPath + "|" + index).substring(0, 24);
79
    }
80

81
    private static String sha256(Path file) throws Exception {
82
        try (var in = Files.newInputStream(file)) {
83
            return HexFormat.of().formatHex(
84
                    MessageDigest.getInstance("SHA-256").digest(in.readAllBytes()));
85
        }
86
    }
87

88
    private static String sha256(String s) {
89
        try {
90
            return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256")
91
                    .digest(s.getBytes(StandardCharsets.UTF_8)));
92
        } catch (java.security.NoSuchAlgorithmException e) { throw new IllegalStateException(e); }
93
    }
94
}

EmbeddingGateway wraps EmbeddingModel.embedForResponse(List.of(text)) behind a port so tests can substitute a fixed-vector stub — one more place where tests never touch a model.

KnowledgeStore upserts with embedding = cast(:v as vector), and supersedeOlderVersions marks previous documents rows and deletes their chunks — a runbook’s old procedure must stop being retrievable the moment the new one lands.

A seeded runbook (docs/runbooks/payment-gateway-degraded.md) with front matter document_id: rb-payment-gateway-degraded, version: "3", tenant_id: acme, service_id: payment-gateway, effective_date: 2026-08-01, then ## sections for Symptoms, Diagnosis, Mitigation, Rollback. The repo carries three runbooks covering both tenants and one deliberately stale version: "2" twin of the same document to exercise supersede.

Commands to build and run

1
./gradlew :knowledge-ingestion:bootRun --args='--ingest.dry-run=true'
2
./gradlew :knowledge-ingestion:bootRun
3
psql "postgresql://agent:agent-dev-password@localhost:5432/opsdb" \
4
  -c "select chunk_id, heading_path, left(content,60) from knowledge.document_chunks"
5
./gradlew :knowledge-ingestion:bootRun   # again -> skipped=N, embedded=0

Automated tests

MarkdownSectionParserTest: headings inside ``` fences are not split; heading paths nest correctly; a document with no headings yields one section.
ChunkerTest: sections under the token limit pass through untouched; oversized sections split on paragraph boundaries with overlap; chunk count is deterministic.
IngestionIT (Testcontainers pgvector/pgvector:0.8.6-pg17 + stub EmbeddingGateway returning fixed vectors): first run inserts N chunks; second run inserts zero; editing a runbook re-embeds only that document; superseded versions return no chunks.
DeterminismTest: same file → same chunk IDs on repeated runs.

Failure-injection lab

Point OLLAMA_BASE_URL at a dead port mid-run: the job fails with a partial batch committed. Re-run — checksums make the retry safe; only unwritten chunks embed. This is why recordDocument happens after all its chunks.
Flip a runbook’s tenant_id front matter and re-ingest: the new chunks carry the new tenant while the old version’s chunks are gone — verify no acme chunk survives for the globex-owned doc.

Security considerations

Runbook content is untrusted input even though we wrote it — Chapter 5 wraps retrieved text in delimiters and Chapter 13 includes a poisoned-document eval case. document_chunks never stores secrets: ingestion logs counts and IDs, never content. The HNSW index is built on the column itself; a WHERE tenant_id = ? filter in the query keeps cross-tenant recall impossible rather than merely unlikely.

Troubleshooting

type "vector" does not exist: pgvector extension missing — CREATE EXTENSION vector is in V1; confirm the migration ran against opsdb, not simdb.
Dimension mismatch (expected 768): you swapped embedding models without updating the column; embedding dimensions are part of the schema contract — changing the model is a migration, not a config tweak.
Dry-run prints nothing: --ingest.dry-run=true with an unchanged checksum set still exits quietly; delete a chunk row to force work.

Checkpoint verification checklist

dry-run reports the plan and writes nothing (verify ingestion_runs row).
Second full run: skipped=3, embedded=0.
Chunk IDs identical across runs; document_chunks queryable by tenant_id.
Superseded version contributes zero chunks.

Commit message and Git tag

1
feat(ingestion): streaming markdown-to-pgvector pipeline with incremental re-ingest

git tag chapter-04-knowledge-ingestion

What comes next

Chapter 5 turns these chunks into answers: the retrieval port, tenant-filtered similarity search, citation mapping, and — just as important — the abstention path when evidence isn’t there.

Project State Ledger — `chapter-04-knowledge-ingestion`

Tables: knowledge.documents, knowledge.document_chunks (vector(768), HNSW), knowledge.ingestion_runs (V1__knowledge_init.sql)
Pipeline: stream → front matter → heading sections → token-boundary chunker → embed → upsert; checksum skip; supersede on new version; ingestion_runs audit row
Ports: EmbeddingGateway (stubbable)
Chunk ID: chk- + sha256(docId|version|headingPath|index)[:24]
Config: ingest.source-dir, ingest.dry-run, ingest.max-chunk-tokens=512, ingest.overlap-tokens=64
Test doubles: fixed-vector embedding stub; pgvector Testcontainer
Next: chapter-05-rag-citations

JavaPostgresAISpring Boot