diff --git a/CHANGELOG.md b/CHANGELOG.md index f3df032..8443bc7 100644 --- a/CHANGELOG.md +++ b/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__n`, paired FIFO by name) instead of the bare tool name, in the conversation transport and workflow runs. diff --git a/src/sources/connector.ts b/src/sources/connector.ts index 5d3b519..ef36c2d 100644 --- a/src/sources/connector.ts +++ b/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; + /** 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> = []; 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; diff --git a/src/sources/repo.ts b/src/sources/repo.ts index e1c2f1d..d876228 100644 --- a/src/sources/repo.ts +++ b/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; diff --git a/test/ingest-chunk-budget.test.ts b/test/ingest-chunk-budget.test.ts new file mode 100644 index 0000000..ca557f5 --- /dev/null +++ b/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(); + }); +});