feat(sources): per-source chunk budget (SourceConnector.maxChunks)

CHANGELOG.md

@@ -1,4 +1,4 @@
-## [Unreleased] (detached send, server-owned queue, steering, push-on-append)
+## [Unreleased] (detached send, server-owned queue, steering, push-on-append; per-source chunk budget)
 
 ### Added
 - **Detached send (opt-in)** — `POST /api/agent/sessions/:id/messages` with `detach: true`
@@ -28,6 +28,12 @@
   failing after bounded retries (`ConversationDeps.finalizeRetryMs`, default 1s/5s/30s; the retry
   stops at shutdown and leaves the attempt reserved for recovery).
 
+- **`SourceConnector.maxChunks`** — per-source chunk budget for `ingest()` (default stays
+  `MAX_CHUNKS_PER_SOURCE` = 10,000). `RepoSourceConfig.maxChunks` passes it through
+  `repoConnector`. Symbol docs carry one chunk per symbol, so a mid-size repo passes the
+  default; the walk is alphabetical, so the overflow dropped the later top-level directories
+  (recorded in `skipped`, easy to miss).
+
 ### Fixed
 - Tool events without an invocation id get a per-turn counter id (`tc_<turn>_n<k>`, paired FIFO
   by name) instead of the bare tool name, in the conversation transport and workflow runs.

src/sources/connector.ts

@@ -26,6 +26,9 @@ export interface SourceConnector {
   readonly sourceId: string;
   /** fetch documents for the source's current scope; cheap when cached */
   fetch(signal?: AbortSignal | undefined): Promise<NormalizedDocument[]>;
+  /** per-source chunk budget for ingest() (default MAX_CHUNKS_PER_SOURCE).
+   *  Docs past it are recorded in `skipped`, not ingested. */
+  readonly maxChunks?: number;
 }
 
 export const MAX_TEXT_BYTES = 2 * 1024 * 1024; // D12: 2 MB text/doc
@@ -57,6 +60,7 @@ export async function ingest(
   // the source row must exist before documents reference it (FK) — upsert it
   await storage.upsertSource({ id: connector.sourceId, kind: connector.kind, config_json: '{}', status: 'active', last_synced_at: Date.now() });
   let totalChunks = 0;
+  const maxChunks = connector.maxChunks ?? MAX_CHUNKS_PER_SOURCE;
 
   for (const doc of docs) {
     if (opts.signal?.aborted) break;
@@ -92,8 +96,8 @@ export async function ingest(
     let chunks: Array<Omit<ChunkRow, 'id' | 'doc_sha256'>> = [];
     if (isText) {
       chunks = (doc.chunks ?? structuralChunks(doc.content!, doc.kind)).map((c) => ({ ordinal: c.ordinal, heading_path: c.headingPath, content: c.content, tokens: c.tokens }));
-      if (totalChunks + chunks.length > MAX_CHUNKS_PER_SOURCE) {
-        chunks = chunks.slice(0, Math.max(0, MAX_CHUNKS_PER_SOURCE - totalChunks));
+      if (totalChunks + chunks.length > maxChunks) {
+        chunks = chunks.slice(0, Math.max(0, maxChunks - totalChunks));
         if (chunks.length === 0) { out.skipped.push({ uri: doc.uri, reason: 'chunk cap reached for source' }); continue; }
       }
       totalChunks += chunks.length;

src/sources/repo.ts

@@ -20,6 +20,11 @@ export interface RepoSourceConfig {
   /** max file bytes read per file (skip monsters) */
   maxFileBytes?: number;
   maxFiles?: number;
+  /** chunk budget for ingest() (default MAX_CHUNKS_PER_SOURCE = 10,000).
+   *  Symbol docs carry one chunk per symbol, so a mid-size repo can pass the
+   *  default — the walk is alphabetical, so the overflow silently drops the
+   *  LATER top-level directories (they land in `skipped`). */
+  maxChunks?: number;
   /** include the tree doc (default true) */
   treeDoc?: boolean;
   /** emit a per-symbol-chunks doc per file (plan Stage G). With
@@ -106,6 +111,7 @@ export function repoConnector(sourceId: string, cfg: RepoSourceConfig): SourceCo
   return {
     kind: 'repo',
     sourceId,
+    ...(cfg.maxChunks !== undefined ? { maxChunks: cfg.maxChunks } : {}),
     async fetch() {
       let root = cfg.origin;
       let tmpClone: string | null = null;

test/ingest-chunk-budget.test.ts

@@ -0,0 +1,65 @@
+import { afterAll, describe, expect, it } from 'vitest';
+import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { ingest, MAX_CHUNKS_PER_SOURCE, type SourceConnector } from '../src/sources/connector.ts';
+import { repoConnector } from '../src/sources/repo.ts';
+import { SqliteStorage } from '../src/memory/sqlite.ts';
+
+/** Per-source chunk budget: ingest() honours `connector.maxChunks` (default
+ *  MAX_CHUNKS_PER_SOURCE), and repoConnector passes `cfg.maxChunks` through. */
+
+const dir = mkdtempSync(join(tmpdir(), 'butler-budget-'));
+afterAll(() => { rmSync(dir, { recursive: true, force: true }); });
+
+/** n docs with one prechunked chunk each — chunk count is exact */
+function fixed(sourceId: string, n: number, maxChunks?: number): SourceConnector {
+  return {
+    kind: 'file',
+    sourceId,
+    ...(maxChunks !== undefined ? { maxChunks } : {}),
+    async fetch() {
+      return Array.from({ length: n }, (_, i) => ({
+        uri: `file://${sourceId}/${i}`, title: `${i}`, content: `${sourceId} doc ${i}`, kind: 'plain' as const,
+        chunks: [{ ordinal: 0, headingPath: `${i}`, content: `${sourceId} doc ${i}`, tokens: 2 }],
+      }));
+    },
+  };
+}
+
+describe('ingest chunk budget', () => {
+  it('default stays MAX_CHUNKS_PER_SOURCE (10,000): doc 10,001 is skipped', async () => {
+    expect(MAX_CHUNKS_PER_SOURCE).toBe(10_000);
+    const storage = SqliteStorage.memory();
+    const r = await ingest(fixed('dflt', 10_001), storage, { artifactsDir: dir });
+    expect(r.documents).toHaveLength(10_000);
+    expect(r.skipped).toEqual([{ uri: 'file://dflt/10000', reason: 'chunk cap reached for source' }]);
+    storage.close();
+  });
+
+  it('a connector maxChunks lowers or raises the budget', async () => {
+    const storage = SqliteStorage.memory();
+    const low = await ingest(fixed('low', 5, 3), storage, { artifactsDir: dir });
+    expect(low.documents.map((d) => d.uri)).toEqual(['file://low/0', 'file://low/1', 'file://low/2']);
+    expect(low.skipped.map((s) => s.uri)).toEqual(['file://low/3', 'file://low/4']);
+    const high = await ingest(fixed('high', 10_005, 20_000), storage, { artifactsDir: dir });
+    expect(high.documents).toHaveLength(10_005);
+    expect(high.skipped).toEqual([]);
+    storage.close();
+  });
+
+  it('repoConnector passes cfg.maxChunks through; the alphabetical walk drops LATER files at the cap', async () => {
+    const root = join(dir, 'repo');
+    for (const d of ['a', 'z']) {
+      mkdirSync(join(root, d), { recursive: true });
+      writeFileSync(join(root, d, 'x.md'), `# One\n${d}1\n# Two\n${d}2\n# Three\n${d}3\n`); // distinct content: identical text would hit the content-address cache
+    }
+    const cfg = { origin: root, treeDoc: false, symbols: false };
+    expect(repoConnector('r0', cfg).maxChunks).toBeUndefined();
+    const storage = SqliteStorage.memory();
+    const capped = await ingest(repoConnector('r1', { ...cfg, maxChunks: 3 }), storage, { artifactsDir: dir });
+    expect(capped.documents.map((d) => d.uri)).toEqual(['repo://a/x.md']);
+    expect(capped.skipped.map((s) => s.uri)).toEqual(['repo://z/x.md']);
+    storage.close();
+  });
+});