From bc1e7fc06d3bb509077090497af4943b3380ecf0 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Thu, 10 Sep 2026 17:49:28 -0700 Subject: [PATCH 1/5] Add failing tests for the poison-item wedge that froze the code source Three red tests, one per link in the chain that kept mcp.copilotkit.ai's `code` source stuck at commit 0d0ea901 for ten days: - the code chunker latches `inTemplateString` on a `//` inside a template literal and then emits the whole rest of the file as one chunk; - the OpenAI provider's 30,000-CHARACTER cap does not bound TOKENS, so an oversized chunk hard-400s instead of being shrunk and retried; - the orchestrator holds the state token for a failing item forever, so one permanently-failing file freezes the entire source. --- src/__tests__/code-chunker-oversize.test.ts | 129 +++++++ .../embeddings-oversize-input.test.ts | 162 +++++++++ ...rchestrator-poison-item-quarantine.test.ts | 321 ++++++++++++++++++ 3 files changed, 612 insertions(+) create mode 100644 src/__tests__/code-chunker-oversize.test.ts create mode 100644 src/__tests__/embeddings-oversize-input.test.ts create mode 100644 src/__tests__/orchestrator-poison-item-quarantine.test.ts diff --git a/src/__tests__/code-chunker-oversize.test.ts b/src/__tests__/code-chunker-oversize.test.ts new file mode 100644 index 0000000..65539f2 --- /dev/null +++ b/src/__tests__/code-chunker-oversize.test.ts @@ -0,0 +1,129 @@ +import { describe, it, expect } from "vitest"; +import { chunkCode } from "../indexing/chunking/code.js"; +import type { SourceConfig } from "../types.js"; + +// Regression for the ten-day `code`-source wedge on mcp.copilotkit.ai. +// +// The named failing item was CopilotKit's +// packages/web-inspector/dev/threads-state-lab.ts. Production logged: +// +// [pipeline:code] Failed to index packages/web-inspector/dev/threads-state-lab.ts: +// BadRequestError: 400 Invalid 'input[2]': maximum input length is 8192 tokens. +// +// `input[2]` — the THIRD chunk — was 35,964 characters covering lines 159-1121 +// of a 1121-line file, because the chunker's block-state tracker got stuck. +// The trigger is line 262: +// +// wsUrl: `ws://127.0.0.1:5177/inspector-lab-runtime/${key}/realtime`, +// +// `stripStringsAndLineComments` tracked ' and " but NOT ` , so the `//` inside +// the template literal read as a line comment. The rest of the line (including +// the CLOSING backtick) was discarded, leaving an ODD backtick count, so the +// tracker latched `inTemplateString = true` and never cleared it. Every +// subsequent blank line was rejected as a split point, so the remaining ~960 +// lines collapsed into one chunk that blew past the embedding model's token +// limit — which failed the item, held the source's state token, and froze the +// index. +// +// Two invariants are pinned here: +// 1. a `//` (or `/*`) inside a template literal must not latch the tracker; +// 2. no chunk may exceed the hard size backstop, whatever the source looks +// like — a chunker that cannot find split points must still bound output. + +const CODE_CONFIG = { + name: "code", + type: "code", + chunk: { target_lines: 80, overlap_lines: 10 }, +} as unknown as SourceConfig; + +/** Reproduce the production file's shape: a URL-in-template-literal, then a + * long tail of ordinary blank-line-separated declarations. */ +function buildFileWithUrlInTemplateLiteral(tailBlocks: number): string { + const head = [ + "const ENABLE_URL =", + ' "https://intelligence.copilotkit.ai/intelligence/enable";', + "", + "function runtimeInfo(key: string) {", + " return {", + " wsUrl: `ws://127.0.0.1:5177/inspector-lab-runtime/${key}/realtime`,", + " };", + "}", + "", + ]; + const tail: string[] = []; + for (let i = 0; i < tailBlocks; i++) { + tail.push( + `export const SCENARIO_${i} = {`, + ` id: "scenario-${i}",`, + ` label: "Scenario number ${i} with a reasonably long descriptive label",`, + ` description: "Fixture payload ${i} used by the threads state lab dev harness",`, + "};", + "", + ); + } + return head.concat(tail).join("\n"); +} + +describe("code chunker: template literals and the hard size backstop", () => { + it("does not treat `//` inside a template literal as a line comment", () => { + // 200 tail blocks = 1200 lines. With the bug, EVERY line after the wsUrl + // line is swallowed into a single chunk because the tracker believes it is + // still inside a template literal, so no blank line qualifies as a split + // point. + const content = buildFileWithUrlInTemplateLiteral(200); + const chunks = chunkCode( + content, + "packages/web-inspector/dev/threads-state-lab.ts", + CODE_CONFIG, + ); + + // ~1200 lines at target_lines 80 must produce many chunks, not two. + expect(chunks.length).toBeGreaterThan(10); + + // And no single chunk may swallow the whole tail. + const largest = Math.max(...chunks.map((c) => c.content.length)); + expect(largest).toBeLessThan(15_000); + }); + + it("bounds chunk size even when the source offers NO split points", () => { + // A file with no blank lines at all: the mechanical fallback splits on + // target_lines, but nothing bounds the CHARACTER size of those lines. A + // single 80-line range of long lines still overflows the embedding model. + const longLine = " ".concat("const x = ", '"'.padEnd(600, "y"), '";'); + const content = Array.from({ length: 400 }, () => longLine).join("\n"); + const chunks = chunkCode(content, "packages/dense/no-blank-lines.ts", CODE_CONFIG); + + for (const chunk of chunks) { + expect(chunk.content.length).toBeLessThanOrEqual(12_000); + } + }); + + it("still splits an ordinary file on blank-line boundaries", () => { + // Guard against "fixed the latch by never entering template state at all". + const content = buildFileWithUrlInTemplateLiteral(40); + const chunks = chunkCode(content, "packages/a/b.ts", CODE_CONFIG); + expect(chunks.length).toBeGreaterThan(1); + // Chunks stay line-addressable and in order. + for (let i = 1; i < chunks.length; i++) { + expect(chunks[i].startLine).toBeGreaterThan(chunks[i - 1].startLine); + } + }); + + it("keeps a genuine multi-line template literal intact as one region", () => { + // The latch exists for a reason: a blank line INSIDE a template literal is + // not a safe split point. Fixing the `//` case must not lose that. + const content = [ + "const q = `", + "SELECT 1", + "", + "FROM t", + "`;", + "", + "const after = 1;", + "", + ].join("\n"); + const chunks = chunkCode(content, "a.ts", CODE_CONFIG); + // Short file — one chunk, and crucially no crash / no latch leaking out. + expect(chunks).toHaveLength(1); + }); +}); diff --git a/src/__tests__/embeddings-oversize-input.test.ts b/src/__tests__/embeddings-oversize-input.test.ts new file mode 100644 index 0000000..5ca83ce --- /dev/null +++ b/src/__tests__/embeddings-oversize-input.test.ts @@ -0,0 +1,162 @@ +import { describe, it, expect, beforeAll, afterAll, vi } from "vitest"; +import http from "node:http"; +import type { AddressInfo } from "node:net"; + +// Regression for the ten-day `code`-source wedge on mcp.copilotkit.ai. +// +// OpenAIEmbeddingProvider guarded oversized inputs with a CHARACTER cap: +// +// const MAX_CHARS = 30_000; // "~8192 tokens with safety margin" +// +// That is a unit error. 30,000 characters is only ~8192 tokens at ≥3.66 +// chars/token; dense source code tokenizes far denser than prose, so a 30,000 +// character code chunk is comfortably over 10,000 tokens. Production hit it: +// +// BadRequestError: 400 Invalid 'input[2]': maximum input length is 8192 tokens. +// +// A 400 is not retryable, so the item failed, the source's state token was +// held, and the whole `code` source froze for ten days. +// +// The invariant: an input the API rejects purely for LENGTH must be recovered +// from in-provider by shrinking and retrying, not surfaced as a hard failure. +// No fixed character cap can be correct for every tokenizer, so the provider +// has to converge on the real limit empirically. +// +// The fake below is the REAL OpenAI SDK talking to a REAL HTTP server that +// returns production's exact 400 shape. Its limit is expressed in characters +// (a stand-in for the tokenizer) and deliberately set BELOW MAX_CHARS, which +// is precisely the condition the char cap cannot see. + +const SERVER_CHAR_LIMIT = 4_000; + +interface EmbeddingsRequest { + input: string[]; + model: string; +} + +let server: http.Server; +let baseUrl: string; +let requests: EmbeddingsRequest[] = []; +let savedBaseUrl: string | undefined; + +beforeAll(async () => { + server = http.createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + const parsed = JSON.parse(body) as EmbeddingsRequest; + requests.push(parsed); + const overIndex = parsed.input.findIndex( + (t) => t.length > SERVER_CHAR_LIMIT, + ); + if (overIndex !== -1) { + // Production's exact error shape. + res.writeHead(400, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + error: { + message: `Invalid 'input[${overIndex}]': maximum input length is 8192 tokens.`, + type: "invalid_request_error", + param: `input[${overIndex}]`, + code: null, + }, + }), + ); + return; + } + res.writeHead(200, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + object: "list", + model: parsed.model, + data: parsed.input.map((_, i) => ({ + object: "embedding", + index: i, + embedding: Array.from({ length: 8 }, () => 0.01), + })), + usage: { prompt_tokens: 1, total_tokens: 1 }, + }), + ); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + baseUrl = `http://127.0.0.1:${port}/v1`; + savedBaseUrl = process.env.OPENAI_BASE_URL; + process.env.OPENAI_BASE_URL = baseUrl; +}); + +afterAll(async () => { + if (savedBaseUrl === undefined) delete process.env.OPENAI_BASE_URL; + else process.env.OPENAI_BASE_URL = savedBaseUrl; + await new Promise((resolve) => server.close(() => resolve())); +}); + +async function makeProvider() { + // Import AFTER OPENAI_BASE_URL is set: the SDK client is built in the + // provider constructor and reads the env var there. + const { OpenAIEmbeddingProvider } = await import("../indexing/embeddings.js"); + return new OpenAIEmbeddingProvider("test-key", "text-embedding-3-small", 8); +} + +describe("OpenAIEmbeddingProvider: an over-length input must not hard-fail", () => { + it("recovers from a length-400 by shrinking the offending input and retrying", async () => { + requests = []; + const provider = await makeProvider(); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + + // 20,000 chars: UNDER the 30,000 char cap, so the existing guard lets it + // through untouched — and the server rejects it. This is the production + // condition exactly. + const oversized = "x".repeat(20_000); + const vectors = await provider.embedBatch(["small", oversized]); + + warnSpy.mockRestore(); + + expect(vectors).toHaveLength(2); + expect(vectors[1]).toHaveLength(8); + // It converged by shrinking, not by luck: more than one request was made + // and the final one was within the server's limit. + expect(requests.length).toBeGreaterThan(1); + const last = requests[requests.length - 1]; + expect(Math.max(...last.input.map((t) => t.length))).toBeLessThanOrEqual( + SERVER_CHAR_LIMIT, + ); + // Healthy siblings in the batch are not mangled. + expect(last.input[0]).toBe("small"); + }); + + it("still fails loudly on a NON-length 400 (no infinite shrink loop)", async () => { + // Guard against "swallow every 400". A bad API key or an unknown param + // must still throw. + requests = []; + const badServer = http.createServer((req, res) => { + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + res.writeHead(400, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + error: { + message: "Unknown parameter: 'dimensions'.", + type: "invalid_request_error", + }, + }), + ); + }); + }); + await new Promise((r) => badServer.listen(0, "127.0.0.1", r)); + const { port } = badServer.address() as AddressInfo; + const prev = process.env.OPENAI_BASE_URL; + process.env.OPENAI_BASE_URL = `http://127.0.0.1:${port}/v1`; + try { + const provider = await makeProvider(); + await expect(provider.embedBatch(["anything"])).rejects.toThrow( + /Unknown parameter/, + ); + } finally { + process.env.OPENAI_BASE_URL = prev; + await new Promise((r) => badServer.close(() => r())); + } + }); +}); diff --git a/src/__tests__/orchestrator-poison-item-quarantine.test.ts b/src/__tests__/orchestrator-poison-item-quarantine.test.ts new file mode 100644 index 0000000..5b8c0c5 --- /dev/null +++ b/src/__tests__/orchestrator-poison-item-quarantine.test.ts @@ -0,0 +1,321 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +// Regression for the ten-day `code`-source wedge on mcp.copilotkit.ai. +// +// C1 (orchestrator-state-token-hold.test.ts) deliberately refuses to advance +// the state token past a failed item, so the failure is RETRIED rather than +// skipped. That is right for a transient failure — and catastrophic for a +// permanent one: one file that fails every single run freezes the ENTIRE +// source. Production sat at commit 0d0ea901 for ten days, serving stale +// source-code search results, because one chunk of one file was too large for +// the embedding model and therefore failed identically on every retry. +// +// The invariant added here: retries are BOUNDED. After MAX_ITEM_ATTEMPTS +// consecutive runs in which the SAME item fails, that item is quarantined — +// it stops holding the state token hostage, the rest of the source indexes, +// and the quarantine is recorded (and surfaced to operators) so the skip is +// loud rather than silent. Quarantined items are still re-attempted on every +// subsequent run; a later success clears the record. +// +// The negative assertion matters just as much: a TRANSIENT failure must still +// hold the token and retry. Quarantining on the first error would trade a +// wedge for silent data loss. + +const { + mockGetIndexState, + mockUpsertIndexState, + mockIndexItems, + mockRemoveItems, + mockFullAcquire, + mockIncrementalAcquire, +} = vi.hoisted(() => ({ + mockGetIndexState: vi.fn(), + mockUpsertIndexState: vi.fn(), + mockIndexItems: vi.fn(), + mockRemoveItems: vi.fn(), + mockFullAcquire: vi.fn(), + mockIncrementalAcquire: vi.fn(), +})); + +vi.mock("../config.js", () => ({ + getConfig: vi.fn().mockReturnValue({ + databaseUrl: "postgresql://test", + openaiApiKey: "test-key", + githubToken: "", + githubWebhookSecret: "", + port: 3001, + nodeEnv: "test", + logLevel: "info", + cloneDir: "/tmp/test", + slackBotToken: "", + slackSigningSecret: "", + discordBotToken: "", + notionToken: "", + }), + getServerConfig: vi.fn().mockReturnValue({ + server: { name: "test", version: "1.0" }, + sources: [ + { + name: "docs", + type: "markdown", + path: "/tmp/docs", + file_patterns: ["**/*.md"], + chunk: {}, + }, + ], + tools: [ + { + name: "search", + type: "search", + description: "Search", + source: "docs", + default_limit: 5, + max_limit: 20, + result_format: "docs", + }, + ], + embedding: { + provider: "openai", + model: "text-embedding-3-small", + dimensions: 1536, + }, + indexing: { + auto_reindex: false, + reindex_hour_utc: 3, + stale_threshold_hours: 24, + }, + }), + getIndexableSourceNames: vi.fn().mockReturnValue(new Set(["docs"])), + getAnalyticsConfig: vi.fn().mockReturnValue(undefined), +})); + +vi.mock("../db/queries.js", () => ({ + getIndexState: (...args: unknown[]) => mockGetIndexState(...args), + upsertIndexState: (...args: unknown[]) => mockUpsertIndexState(...args), + cleanupOldWebhookDeliveries: vi.fn().mockResolvedValue(0), +})); + +vi.mock("../db/analytics.js", () => ({ + cleanupOldQueryLogs: vi.fn().mockResolvedValue(0), +})); + +vi.mock("../indexing/embeddings.js", () => { + class MockEmbeddingProvider { + embed = vi.fn().mockResolvedValue([0.1, 0.2]); + embedBatch = vi.fn().mockResolvedValue([[0.1, 0.2]]); + } + return { + EmbeddingClient: MockEmbeddingProvider, + createEmbeddingProvider: () => new MockEmbeddingProvider(), + }; +}); + +vi.mock("../indexing/pipeline.js", () => ({ + IndexingPipeline: class MockIndexingPipeline { + indexItems = mockIndexItems; + removeItems = mockRemoveItems; + }, +})); + +vi.mock("../indexing/providers/index.js", () => ({ + getProvider: vi.fn().mockReturnValue(() => ({ + fullAcquire: mockFullAcquire, + incrementalAcquire: mockIncrementalAcquire, + getCurrentStateToken: vi.fn().mockResolvedValue("token-2"), + })), +})); + +import { IndexingOrchestrator } from "../indexing/orchestrator.js"; +import { computeSourceConfigFingerprint } from "../indexing/source-fingerprint.js"; +import type { IndexState, SourceConfig } from "../types.js"; + +const DOCS_SOURCE_FINGERPRINT = computeSourceConfigFingerprint({ + name: "docs", + type: "markdown", + path: "/tmp/docs", + file_patterns: ["**/*.md"], + chunk: {}, +} as SourceConfig); + +/** The poison item, mirroring production's threads-state-lab.ts. */ +const POISON = "packages/web-inspector/dev/threads-state-lab.ts"; + +/** + * A stateful stand-in for the index_state row. The wedge only shows up ACROSS + * runs, so the fake must persist what each run writes and feed it back to the + * next run's getIndexState — a per-call mockResolvedValue cannot express it. + */ +function installStatefulIndexState(initial: IndexState): { row: IndexState } { + const holder = { row: { ...initial } }; + mockGetIndexState.mockImplementation(async () => ({ ...holder.row })); + mockUpsertIndexState.mockImplementation(async (state: IndexState) => { + holder.row = { ...state }; + }); + return holder; +} + +async function runSourceReindex( + orchestrator: IndexingOrchestrator, +): Promise { + const done = new Promise((resolve) => { + orchestrator.onReindexComplete = () => resolve(); + }); + orchestrator.queueSourceReindex("docs"); + await Promise.race([ + done, + (async () => { + for (let i = 0; i < 50; i++) { + await new Promise((r) => setTimeout(r, 50)); + if (!orchestrator.isIndexing()) return; + } + })(), + ]); + await new Promise((r) => setTimeout(r, 50)); +} + +describe("IndexingOrchestrator: a permanently failing item must not wedge the source", () => { + let orchestrator: IndexingOrchestrator; + let errSpy: ReturnType; + + beforeEach(() => { + vi.clearAllMocks(); + orchestrator = new IndexingOrchestrator(); + mockIndexItems.mockResolvedValue({ failedIds: [] }); + mockRemoveItems.mockResolvedValue({ failedIds: [] }); + errSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + }); + + afterEach(() => { + errSpy.mockRestore(); + }); + + it("quarantines an item that fails every run, and the source recovers", async () => { + const holder = installStatefulIndexState({ + source_type: "markdown", + source_key: "docs", + last_commit_sha: "token-1", + config_fingerprint: DOCS_SOURCE_FINGERPRINT, + last_indexed_at: new Date(), + status: "idle", + error_message: null, + }); + + mockIncrementalAcquire.mockResolvedValue({ + items: [{ id: "docs/ok.md", content: "a" }, { id: POISON, content: "b" }], + removedIds: [], + stateToken: "token-2", + }); + // The poison item fails identically on EVERY run — exactly production. + mockIndexItems.mockResolvedValue({ + failedIds: [POISON], + failures: [ + { + id: POISON, + error: "400 Invalid 'input[2]': maximum input length is 8192 tokens.", + }, + ], + }); + + // Runs 1 and 2: the failure is still presumed transient, so the token is + // held and the source stays errored. That is the C1 behaviour, preserved. + await runSourceReindex(orchestrator); + expect(holder.row.last_commit_sha).toBe("token-1"); + expect(holder.row.status).toBe("error"); + + await runSourceReindex(orchestrator); + expect(holder.row.last_commit_sha).toBe("token-1"); + expect(holder.row.status).toBe("error"); + + // Run 3: the same item has now failed MAX_ITEM_ATTEMPTS consecutive runs. + // It is quarantined, the token advances, and the source recovers so the + // other 99.9% of the repo stops going stale. + await runSourceReindex(orchestrator); + expect(holder.row.last_commit_sha).toBe("token-2"); + expect(holder.row.status).toBe("idle"); + + // The skip must be LOUD and recorded, not silent. + const failures = holder.row.item_failures ?? {}; + expect(Object.keys(failures)).toContain(POISON); + expect(failures[POISON]?.quarantined).toBe(true); + expect(failures[POISON]?.attempts).toBeGreaterThanOrEqual(3); + expect(failures[POISON]?.last_error).toContain("8192 tokens"); + + const loggedQuarantine = errSpy.mock.calls + .map((c) => c.join(" ")) + .some((line) => /quarantin/i.test(line) && line.includes(POISON)); + expect(loggedQuarantine).toBe(true); + }); + + it("does NOT quarantine a TRANSIENT failure — it holds the token and retries", async () => { + // The negative assertion. Quarantining on the first error would trade the + // wedge for silent data loss. + const holder = installStatefulIndexState({ + source_type: "markdown", + source_key: "docs", + last_commit_sha: "token-1", + config_fingerprint: DOCS_SOURCE_FINGERPRINT, + last_indexed_at: new Date(), + status: "idle", + error_message: null, + }); + + mockIncrementalAcquire.mockResolvedValue({ + items: [{ id: "docs/flaky.md", content: "a" }], + removedIds: [], + stateToken: "token-2", + }); + + // Run 1: a one-off network blip. + mockIndexItems.mockResolvedValueOnce({ + failedIds: ["docs/flaky.md"], + failures: [{ id: "docs/flaky.md", error: "ECONNRESET" }], + }); + await runSourceReindex(orchestrator); + expect(holder.row.last_commit_sha).toBe("token-1"); + expect(holder.row.status).toBe("error"); + expect(holder.row.item_failures?.["docs/flaky.md"]?.quarantined ?? false).toBe( + false, + ); + + // Run 2: it succeeds. The token advances and the failure record clears — + // the item must NOT carry a stale strike into the future. + mockIndexItems.mockResolvedValue({ failedIds: [], failures: [] }); + await runSourceReindex(orchestrator); + expect(holder.row.last_commit_sha).toBe("token-2"); + expect(holder.row.status).toBe("idle"); + expect(holder.row.item_failures?.["docs/flaky.md"]).toBeUndefined(); + }); + + it("clears a quarantine once the item finally indexes", async () => { + const holder = installStatefulIndexState({ + source_type: "markdown", + source_key: "docs", + last_commit_sha: "token-1", + config_fingerprint: DOCS_SOURCE_FINGERPRINT, + last_indexed_at: new Date(), + status: "idle", + error_message: null, + item_failures: { + [POISON]: { + attempts: 5, + first_failed_at: new Date("2026-09-01T18:31:05.344Z").toISOString(), + last_error: "maximum input length is 8192 tokens", + quarantined: true, + }, + }, + }); + + mockIncrementalAcquire.mockResolvedValue({ + items: [{ id: POISON, content: "b" }], + removedIds: [], + stateToken: "token-2", + }); + mockIndexItems.mockResolvedValue({ failedIds: [], failures: [] }); + + await runSourceReindex(orchestrator); + + expect(holder.row.last_commit_sha).toBe("token-2"); + expect(holder.row.status).toBe("idle"); + expect(holder.row.item_failures?.[POISON]).toBeUndefined(); + }); +}); From 2fdaec457ff714646fcf2e485c9e8812aaa3c2ab Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Thu, 10 Sep 2026 17:56:32 -0700 Subject: [PATCH 2/5] Stop a URL in a template literal from collapsing a file into one chunk MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The code chunker's block-state tracker stripped ' and " strings before counting backticks, but knew nothing about backticks themselves. So in wsUrl: `ws://127.0.0.1:5177/inspector-lab-runtime/${key}/realtime`, the `//` read as a line comment, the rest of the line — closing backtick included — was discarded, the backtick count came out odd, and the tracker latched inTemplateString for the remainder of the file. With the latch stuck no blank line qualified as a split point again, so the last ~960 lines of CopilotKit's threads-state-lab.ts became a single 36 KB chunk that OpenAI rejected at 8192 tokens. Replaces the two-stage strip-then-count with one left-to-right scan where the quote, template and comment contexts are mutually exclusive — the only way `//` can be classified correctly — and adds a hard 12,000-character bound on any emitted chunk, because the blank-line and mechanical fallbacks bound LINES and nothing bounded characters. Measured over the 1,120 files the production 'code' source indexes: chunks over 12,000 chars went 15 -> 0, and threads-state-lab.ts went from 3 chunks (largest 27,785 chars) to 8 (largest 11,753). --- src/__tests__/code-chunker-oversize.test.ts | 10 +- src/indexing/chunking/code.ts | 241 ++++++++++++++------ 2 files changed, 183 insertions(+), 68 deletions(-) diff --git a/src/__tests__/code-chunker-oversize.test.ts b/src/__tests__/code-chunker-oversize.test.ts index 65539f2..6026a30 100644 --- a/src/__tests__/code-chunker-oversize.test.ts +++ b/src/__tests__/code-chunker-oversize.test.ts @@ -91,7 +91,11 @@ describe("code chunker: template literals and the hard size backstop", () => { // single 80-line range of long lines still overflows the embedding model. const longLine = " ".concat("const x = ", '"'.padEnd(600, "y"), '";'); const content = Array.from({ length: 400 }, () => longLine).join("\n"); - const chunks = chunkCode(content, "packages/dense/no-blank-lines.ts", CODE_CONFIG); + const chunks = chunkCode( + content, + "packages/dense/no-blank-lines.ts", + CODE_CONFIG, + ); for (const chunk of chunks) { expect(chunk.content.length).toBeLessThanOrEqual(12_000); @@ -105,7 +109,9 @@ describe("code chunker: template literals and the hard size backstop", () => { expect(chunks.length).toBeGreaterThan(1); // Chunks stay line-addressable and in order. for (let i = 1; i < chunks.length; i++) { - expect(chunks[i].startLine).toBeGreaterThan(chunks[i - 1].startLine); + expect(chunks[i].startLine ?? 0).toBeGreaterThan( + chunks[i - 1].startLine ?? 0, + ); } }); diff --git a/src/indexing/chunking/code.ts b/src/indexing/chunking/code.ts index 011e4c2..53750ea 100644 --- a/src/indexing/chunking/code.ts +++ b/src/indexing/chunking/code.ts @@ -66,90 +66,116 @@ interface BlockState { } /** - * Check whether the character at `pos` is escaped by counting preceding - * backslashes. An odd number means the character is escaped. - */ -function isEscaped(line: string, pos: number): boolean { - let backslashes = 0; - for (let j = pos - 1; j >= 0 && line[j] === "\\"; j--) { - backslashes++; - } - return backslashes % 2 === 1; -} - -/** - * Strip string literals and single-line comments from a line so that - * block-comment and template-string detection only fires on real syntax. + * Advance the cross-line block state by scanning one line character by + * character. + * + * This is a single left-to-right pass rather than the two-stage + * "strip strings, then count backticks" it replaces. That older shape had a + * fatal ordering bug: the stripper knew about `'` and `"` but NOT about + * backticks, so a `//` inside a template literal — + * + * wsUrl: `ws://127.0.0.1:5177/inspector-lab-runtime/${key}/realtime`, + * + * — read as the start of a line comment. Everything after `ws:` was discarded, + * including the CLOSING backtick, which left an odd backtick count and latched + * `inTemplateString` for the remainder of the file. With the latch stuck, no + * blank line qualified as a split point ever again, so ~960 lines collapsed + * into a single chunk that exceeded the embedding model's token limit. That + * one chunk failed to embed, which held the source's state token, which froze + * mcp.copilotkit.ai's `code` source at commit 0d0ea901 for ten days. + * + * Scanning in one pass keeps the quote/template/comment contexts mutually + * exclusive, which is the only way `//` can be classified correctly. + * + * Single-quote and double-quote strings are treated as line-local (JS does not + * carry them across lines without an explicit continuation); template literals + * and block comments carry across lines via the returned state. `${…}` + * interpolations are treated as literal template text — expressions there can + * technically contain nested strings and comments, but bounding the chunk size + * (see MAX_CHUNK_CHARS) is the backstop for anything this heuristic misreads. */ -function stripStringsAndLineComments(line: string): string { - let result = ""; - let inSingle = false; - let inDouble = false; +function trackBlockState(line: string, state: BlockState): BlockState { + let { inBlockComment, inTemplateString } = state; + let i = 0; - for (let i = 0; i < line.length; i++) { + while (i < line.length) { const ch = line[i]; + const next = line[i + 1]; - if (!inSingle && !inDouble && ch === "/" && line[i + 1] === "/") { - break; // rest of line is a single-line comment + if (inBlockComment) { + if (ch === "*" && next === "/") { + inBlockComment = false; + i += 2; + continue; + } + i++; + continue; } - if (!inDouble && ch === "'" && !isEscaped(line, i)) { - inSingle = !inSingle; - } else if (!inSingle && ch === '"' && !isEscaped(line, i)) { - inDouble = !inDouble; + if (inTemplateString) { + // A backslash escapes the next character, so `\`` does not close. + if (ch === "\\") { + i += 2; + continue; + } + if (ch === "`") { + inTemplateString = false; + } + i++; + continue; } - if (!inSingle && !inDouble) { - result += ch; + if (ch === "/" && next === "/") { + // Real line comment — nothing after it can change the block state. + break; } - } - - return result; -} - -function trackBlockState(line: string, state: BlockState): BlockState { - const newState = { ...state }; - const stripped = stripStringsAndLineComments(line); - - if (newState.inBlockComment) { - if (stripped.includes("*/")) { - newState.inBlockComment = false; + if (ch === "/" && next === "*") { + inBlockComment = true; + i += 2; + continue; } - return newState; - } - - if (newState.inTemplateString) { - // Count unescaped backticks - const backticks = (stripped.match(/(?> { + const content = formatChunk(lines, startLine, filePath); + if (content.length <= MAX_CHUNK_CHARS) { + return [{ content, startLine, endLine, language }]; + } + + if (lines.length > 1) { + const mid = Math.ceil(lines.length / 2); + return [ + ...emitBounded( + lines.slice(0, mid), + startLine, + startLine + mid - 1, + filePath, + language, + ), + ...emitBounded( + lines.slice(mid), + startLine + mid, + endLine, + filePath, + language, + ), + ]; + } + + // One line, over the cap. Slice the raw line and re-format each slice so + // every emitted chunk carries the breadcrumb and stays under the bound. + const overhead = content.length - lines[0].length; + const sliceSize = Math.max(1, MAX_CHUNK_CHARS - overhead); + const out: Array> = []; + for (let i = 0; i < lines[0].length; i += sliceSize) { + out.push({ + content: formatChunk( + [lines[0].slice(i, i + sliceSize)], + startLine, + filePath, + ), + startLine, + endLine, + language, + }); + } + return out; +} + /** * Split lines into groups at double-newline boundaries, respecting block state. */ @@ -366,13 +471,17 @@ export function chunkCode( const startLine = start + 1; // 1-indexed const endLine = end + 1; // 1-indexed - chunks.push({ - content: formatChunk(chunkLines, startLine, filePath), + // Emit through the size backstop rather than pushing directly: a range + // that is fine by LINE count can still be enormous by character count. + for (const emitted of emitBounded( + chunkLines, startLine, endLine, + filePath, language, - chunkIndex: chunks.length, - }); + )) { + chunks.push({ ...emitted, chunkIndex: chunks.length }); + } } return chunks; From fc97f530568e155f0b26a554b3c4ea338b3e05e0 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Thu, 10 Sep 2026 17:56:32 -0700 Subject: [PATCH 3/5] Recover from an over-length embedding input instead of failing the item MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenAIEmbeddingProvider capped inputs at 30,000 CHARACTERS to stay under an 8192-TOKEN limit. That holds only at 3.66 chars/token; source code tokenizes denser, so a 30,000-character code chunk is comfortably past 10,000 tokens and comes back as a non-retryable 400. A length rejection is recoverable by sending less text, so it is now handled separately from the generic retry path: the provider parses the input[N] the API named, halves that input, and retries — bounded at 12 rounds and a 64-char floor so it always terminates. Every shrink logs a warning naming the truncation, and any other 400 still throws. No fixed character cap can be right for every tokenizer; converging on the real limit empirically is what makes an over-length input non-fatal. --- src/indexing/embeddings.ts | 122 ++++++++++++++++++++++++++++++++++++- 1 file changed, 121 insertions(+), 1 deletion(-) diff --git a/src/indexing/embeddings.ts b/src/indexing/embeddings.ts index a7fd74e..28ffea5 100644 --- a/src/indexing/embeddings.ts +++ b/src/indexing/embeddings.ts @@ -7,6 +7,57 @@ const MAX_BATCH_SIZE = 2048; const MAX_RETRIES = 3; const BASE_DELAY_MS = 1000; +// How many times a single batch may be shrunk in response to an +// input-too-long 400 before giving up. Each round halves the offending input, +// so 12 rounds take a 30,000-character input below 8 characters — the bound +// exists to guarantee termination, not because it is ever expected to be hit. +const MAX_SHRINK_ROUNDS = 12; + +// Inputs are never shrunk below this. An input this small that STILL trips the +// length error is not a length problem, and looping further would hide the +// real one. +const MIN_SHRINK_CHARS = 64; + +/** + * Whether an API error is the provider rejecting an input purely for being too + * long — recoverable by sending less text, unlike every other 400. + * + * This distinction is what kept mcp.copilotkit.ai's `code` source frozen for + * ten days. A single code chunk embedded to more than 8192 tokens, OpenAI + * answered `400 Invalid 'input[2]': maximum input length is 8192 tokens.`, the + * provider treated it as fatal, the item failed, and the orchestrator held the + * source's state token — every run, identically, for ten days. + */ +function isInputTooLongError(error: unknown): boolean { + // Duck-type the 400 rather than relying solely on `instanceof`: the SDK + // class is not always the same object across module boundaries (and test + // doubles replace it outright), while `status` is stable on every APIError. + const status = (error as { status?: unknown } | null)?.status; + const isBadRequest = + status === 400 || + (typeof OpenAI.BadRequestError === "function" && + error instanceof OpenAI.BadRequestError); + if (!isBadRequest) return false; + const message = String((error as Error | null)?.message ?? ""); + return /maximum input length|maximum context length|reduce (?:your|the) input/i.test( + message, + ); +} + +/** + * Which input index the provider named, if it named one. OpenAI reports the + * FIRST offending element as `input[N]`, so shrinking just that element and + * retrying converges on the real limit without mangling healthy siblings in + * the same batch. + */ +function parseOffendingInputIndex(error: unknown): number | null { + const message = String((error as Error)?.message ?? ""); + const match = /input\[(\d+)\]/.exec(message); + if (!match) return null; + const index = Number(match[1]); + return Number.isInteger(index) && index >= 0 ? index : null; +} + // Assert a provider returned exactly one vector per input text, failing LOUD // with context on a shortfall. A provider/proxy that streams nothing (or a mock // returning `{ data: [] }`) yields a results array SHORTER than the input, so @@ -66,6 +117,39 @@ function modelSupportsDimensions(model: string): boolean { return /^text-embedding-3-/.test(model); } +/** + * Halve the length of the input(s) the provider rejected. + * + * When the error named an index, only that input is touched, so a single + * pathological chunk does not degrade the rest of its batch. When it did not, + * every input above {@link MIN_SHRINK_CHARS} is halved — a blunt instrument, + * but one that still converges. + * + * Returns null when nothing can usefully be shrunk, which is the signal to + * stop and surface the original error. + */ +function shrinkOversizedInputs( + texts: string[], + offendingIndex: number | null, +): { texts: string[]; changedCount: number } | null { + const shouldShrink = (index: number, text: string): boolean => { + if (text.length <= MIN_SHRINK_CHARS) return false; + return offendingIndex === null || offendingIndex === index; + }; + + let changedCount = 0; + const next = texts.map((text, index) => { + if (!shouldShrink(index, text)) return text; + changedCount++; + return text.slice( + 0, + Math.max(MIN_SHRINK_CHARS, Math.floor(text.length / 2)), + ); + }); + + return changedCount === 0 ? null : { texts: next, changedCount }; +} + // ── Provider interface ────────────────────────────────────────────────────── export interface EmbeddingProvider { @@ -133,7 +217,12 @@ export class OpenAIEmbeddingProvider implements EmbeddingProvider { async embedBatch(texts: string[]): Promise { if (texts.length === 0) return []; - // Truncate texts that exceed OpenAI's 8192 token limit (~32K chars with safety margin) + // A first-pass cap on absurd inputs. It is expressed in CHARACTERS while + // the API's limit is in TOKENS, so it can only ever be a heuristic: 30,000 + // characters is ~8192 tokens at 3.66 chars/token, and dense source code + // tokenizes well below that ratio. Inputs that slip past this cap and get + // rejected are recovered by the shrink-and-retry path in embedWithRetry — + // that, not this constant, is what makes an over-length input non-fatal. const MAX_CHARS = 30_000; const truncated = texts.map((t) => t.length > MAX_CHARS ? t.slice(0, MAX_CHARS) : t, @@ -171,6 +260,7 @@ export class OpenAIEmbeddingProvider implements EmbeddingProvider { texts: string[], batchNum: number, attempt: number = 1, + shrinkRound: number = 0, ): Promise { try { const response = await this.client.embeddings.create({ @@ -199,6 +289,36 @@ export class OpenAIEmbeddingProvider implements EmbeddingProvider { const sorted = response.data.sort((a, b) => a.index - b.index); return sorted.map((item) => item.embedding); } catch (error: unknown) { + // An input-too-long 400 is recoverable by sending less text, so it is + // handled before the generic retry bookkeeping: it is not an attempt + // that should count against MAX_RETRIES, and retrying it unchanged + // would fail identically forever. + if (isInputTooLongError(error)) { + const shrunk = shrinkOversizedInputs( + texts, + parseOffendingInputIndex(error), + ); + if (shrunk && shrinkRound < MAX_SHRINK_ROUNDS) { + console.warn( + `Embedding batch ${batchNum}: input rejected as too long ` + + `(${(error as Error).message}); truncating ${shrunk.changedCount} ` + + `input(s) and retrying (shrink round ${shrinkRound + 1}/${MAX_SHRINK_ROUNDS}). ` + + `Indexed content for the affected chunk(s) will be TRUNCATED.`, + ); + return this.embedWithRetry( + shrunk.texts, + batchNum, + attempt, + shrinkRound + 1, + ); + } + console.error( + `Embedding batch ${batchNum}: input still rejected as too long after ` + + `${shrinkRound} shrink round(s); giving up.`, + ); + throw error; + } + if (attempt >= MAX_RETRIES) { console.error( `Embedding batch ${batchNum} failed after ${MAX_RETRIES} retries`, From 333c7fa0f3d6087f2eee90a528f1c4ad9090a233 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Thu, 10 Sep 2026 17:56:42 -0700 Subject: [PATCH 4/5] Bound the retry of a failing item so one file cannot freeze a source MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator refuses to advance a source's state token past an item that failed, so the failure is retried rather than silently skipped. With an unbounded retry that is a trap: an item failing for a permanent reason fails identically forever. mcp.copilotkit.ai's 'code' source sat at commit 0d0ea901 for ten days — serving ten-day-old source-code search results to every user — because one file produced a chunk larger than the embedding model's limit. index_state now carries an item_failures ledger: consecutive-failure count, first-failure time and last error, per item id. Three consecutive failing runs quarantines an item — it stops holding the state token, the rest of the source indexes, and every run logs the id, the attempt count and the error. One is too few: a network blip, a rate limit and a poison file all look identical on their first failure, and quarantining those would trade a visible wedge for invisible data loss. Quarantine is advisory, not a blocklist. A quarantined item is still handed to the pipeline on every run, and a success clears its record outright. /health and the admin index-stats op gained quarantined_items — id, attempts, since, error — following the next_acquire precedent from #160, so an item missing from the index is never an invisible gap. --- src/__tests__/admin-ops-endpoints.test.ts | 5 + .../orchestrator-config-fingerprint.test.ts | 4 +- ...rchestrator-poison-item-quarantine.test.ts | 15 +- .../orchestrator-state-token-hold.test.ts | 20 ++- src/db/queries.ts | 21 ++- src/db/schema.ts | 12 ++ src/indexing/orchestrator.ts | 144 ++++++++++++++++-- src/indexing/pipeline.ts | 49 ++++-- src/server.ts | 19 +++ src/types.ts | 34 +++++ 10 files changed, 283 insertions(+), 40 deletions(-) diff --git a/src/__tests__/admin-ops-endpoints.test.ts b/src/__tests__/admin-ops-endpoints.test.ts index 8bca6dc..62634d3 100644 --- a/src/__tests__/admin-ops-endpoints.test.ts +++ b/src/__tests__/admin-ops-endpoints.test.ts @@ -397,6 +397,11 @@ describe("admin ops control surface", () => { // that would write zero rows is now visible BEFORE it runs. next_acquire: "full", next_acquire_reason: "no-stored-config-fingerprint", + // No item has been quarantined, so nothing is missing from the index. + // The key is present regardless: an operator reading this shape must + // not have to guess whether an empty list means "none" or "not + // reported". + quarantined_items: [], }, ]); }); diff --git a/src/__tests__/orchestrator-config-fingerprint.test.ts b/src/__tests__/orchestrator-config-fingerprint.test.ts index 653836f..b0f0342 100644 --- a/src/__tests__/orchestrator-config-fingerprint.test.ts +++ b/src/__tests__/orchestrator-config-fingerprint.test.ts @@ -116,11 +116,11 @@ vi.mock("../indexing/pipeline.js", () => ({ h.indexed.add(item.id); h.indexedThisRun.push(item.id); } - return { failedIds: [] }; + return { failedIds: [], failures: [] }; } async removeItems(ids: string[]) { for (const id of ids) h.indexed.delete(id); - return { failedIds: [] }; + return { failedIds: [], failures: [] }; } }, })); diff --git a/src/__tests__/orchestrator-poison-item-quarantine.test.ts b/src/__tests__/orchestrator-poison-item-quarantine.test.ts index 5b8c0c5..6c1bfe1 100644 --- a/src/__tests__/orchestrator-poison-item-quarantine.test.ts +++ b/src/__tests__/orchestrator-poison-item-quarantine.test.ts @@ -201,7 +201,10 @@ describe("IndexingOrchestrator: a permanently failing item must not wedge the so }); mockIncrementalAcquire.mockResolvedValue({ - items: [{ id: "docs/ok.md", content: "a" }, { id: POISON, content: "b" }], + items: [ + { id: "docs/ok.md", content: "a" }, + { id: POISON, content: "b" }, + ], removedIds: [], stateToken: "token-2", }); @@ -241,8 +244,8 @@ describe("IndexingOrchestrator: a permanently failing item must not wedge the so expect(failures[POISON]?.last_error).toContain("8192 tokens"); const loggedQuarantine = errSpy.mock.calls - .map((c) => c.join(" ")) - .some((line) => /quarantin/i.test(line) && line.includes(POISON)); + .map((c: unknown[]) => c.join(" ")) + .some((line: string) => /quarantin/i.test(line) && line.includes(POISON)); expect(loggedQuarantine).toBe(true); }); @@ -273,9 +276,9 @@ describe("IndexingOrchestrator: a permanently failing item must not wedge the so await runSourceReindex(orchestrator); expect(holder.row.last_commit_sha).toBe("token-1"); expect(holder.row.status).toBe("error"); - expect(holder.row.item_failures?.["docs/flaky.md"]?.quarantined ?? false).toBe( - false, - ); + expect( + holder.row.item_failures?.["docs/flaky.md"]?.quarantined ?? false, + ).toBe(false); // Run 2: it succeeds. The token advances and the failure record clears — // the item must NOT carry a stale strike into the future. diff --git a/src/__tests__/orchestrator-state-token-hold.test.ts b/src/__tests__/orchestrator-state-token-hold.test.ts index 35a8358..b5a4592 100644 --- a/src/__tests__/orchestrator-state-token-hold.test.ts +++ b/src/__tests__/orchestrator-state-token-hold.test.ts @@ -171,8 +171,10 @@ describe("IndexingOrchestrator state-token hold on item failure (C1)", () => { status: "idle", error_message: null, }); - mockIndexItems.mockResolvedValue({ failedIds: [] }); - mockRemoveItems.mockResolvedValue({ failedIds: [] }); + // The pipeline reports failures as {id, error} pairs; failedIds is the + // projection of that list. Mirror the real contract in the doubles. + mockIndexItems.mockResolvedValue({ failedIds: [], failures: [] }); + mockRemoveItems.mockResolvedValue({ failedIds: [], failures: [] }); }); it("does NOT advance the state token when an item fails to index", async () => { @@ -185,7 +187,10 @@ describe("IndexingOrchestrator state-token hold on item failure (C1)", () => { stateToken: "new-token", }); // One item failed. - mockIndexItems.mockResolvedValue({ failedIds: ["docs/bad.md"] }); + mockIndexItems.mockResolvedValue({ + failedIds: ["docs/bad.md"], + failures: [{ id: "docs/bad.md", error: "boom" }], + }); const errSpy = vi.spyOn(console, "error").mockImplementation(() => {}); await runSourceReindex(orchestrator); @@ -213,7 +218,10 @@ describe("IndexingOrchestrator state-token hold on item failure (C1)", () => { removedIds: ["docs/gone.md"], stateToken: "new-token", }); - mockRemoveItems.mockResolvedValue({ failedIds: ["docs/gone.md"] }); + mockRemoveItems.mockResolvedValue({ + failedIds: ["docs/gone.md"], + failures: [{ id: "docs/gone.md", error: "boom" }], + }); const errSpy = vi.spyOn(console, "error").mockImplementation(() => {}); await runSourceReindex(orchestrator); @@ -231,8 +239,8 @@ describe("IndexingOrchestrator state-token hold on item failure (C1)", () => { removedIds: ["docs/gone.md"], stateToken: "new-token", }); - mockIndexItems.mockResolvedValue({ failedIds: [] }); - mockRemoveItems.mockResolvedValue({ failedIds: [] }); + mockIndexItems.mockResolvedValue({ failedIds: [], failures: [] }); + mockRemoveItems.mockResolvedValue({ failedIds: [], failures: [] }); await runSourceReindex(orchestrator); diff --git a/src/db/queries.ts b/src/db/queries.ts index 79fb171..ef7310d 100644 --- a/src/db/queries.ts +++ b/src/db/queries.ts @@ -7,6 +7,7 @@ import type { FaqChunkResult, IndexState, IndexStatus, + ItemFailureRecord, } from "../types.js"; /** @@ -884,7 +885,7 @@ export async function getIndexState( ): Promise { const pool = getPool(); const sql = ` - SELECT source_type, source_key, last_commit_sha, config_fingerprint, last_indexed_at, status, error_message + SELECT source_type, source_key, last_commit_sha, config_fingerprint, last_indexed_at, status, error_message, item_failures FROM index_state WHERE source_type = $1 AND source_key = $2 `; @@ -907,6 +908,10 @@ export async function getIndexState( last_indexed_at: row.last_indexed_at, status: row.status as IndexStatus, error_message: row.error_message, + // node-postgres decodes jsonb for us; installs predating the column read + // back undefined, which normalizes to null ("no outstanding failures"). + item_failures: + (row.item_failures as Record | null) ?? null, }; } @@ -917,15 +922,16 @@ export async function upsertIndexState(state: IndexState): Promise { const pool = getPool(); const sql = ` INSERT INTO index_state - (source_type, source_key, last_commit_sha, config_fingerprint, last_indexed_at, status, error_message) + (source_type, source_key, last_commit_sha, config_fingerprint, last_indexed_at, status, error_message, item_failures) VALUES - ($1, $2, $3, $4, $5, $6, $7) + ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (source_type, source_key) DO UPDATE SET last_commit_sha = EXCLUDED.last_commit_sha, config_fingerprint = EXCLUDED.config_fingerprint, last_indexed_at = EXCLUDED.last_indexed_at, status = EXCLUDED.status, - error_message = EXCLUDED.error_message + error_message = EXCLUDED.error_message, + item_failures = EXCLUDED.item_failures `; // Sanitize every text-typed bind. The highest-risk column here is // error_message: it's populated with raw upstream errors, which in the @@ -946,6 +952,13 @@ export async function upsertIndexState(state: IndexState): Promise { state.last_indexed_at ?? null, stripNulBytes(state.status ?? "idle"), state.error_message == null ? null : stripNulBytes(state.error_message), + // item_failures carries raw upstream error text (and item ids derived from + // repository paths), so it gets the same deep NUL scrub as every other + // jsonb bind — an unsanitized 0x00 would reject the whole UPDATE and turn + // the failure it is recording into a poison-pill loop. + state.item_failures == null || Object.keys(state.item_failures).length === 0 + ? null + : JSON.stringify(stripNulBytesDeep(state.item_failures)), ]); } diff --git a/src/db/schema.ts b/src/db/schema.ts index d33a373..8f95d33 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -38,6 +38,7 @@ CREATE TABLE IF NOT EXISTS index_state ( last_indexed_at TIMESTAMPTZ, status TEXT NOT NULL DEFAULT 'idle', error_message TEXT, + item_failures JSONB, CONSTRAINT index_state_source_uniq UNIQUE (source_type, source_key) ); @@ -111,6 +112,17 @@ CREATE INDEX IF NOT EXISTS idx_chunks_tsv ON chunks USING GIN (tsv); -- carries the column for fresh installs. ALTER TABLE index_state ADD COLUMN IF NOT EXISTS config_fingerprint TEXT; +-- index_state.item_failures. Consecutive-failure counts for the items that +-- failed to index or remove on recent runs, keyed by item id. The orchestrator +-- deliberately holds a source's state token when an item fails so the failure +-- is retried rather than skipped; without a failure COUNT it cannot tell a +-- transient failure from a permanent one, and a permanently-failing item +-- freezes the entire source (the mcp.copilotkit.ai code source was stuck for +-- ten days on one over-sized file). Additive and nullable: rows written before +-- this column existed read back NULL, which the orchestrator treats as "no +-- outstanding failures". +ALTER TABLE index_state ADD COLUMN IF NOT EXISTS item_failures JSONB; + -- Analytics: query_log table for tracking tool usage -- -- request_source tags the ORIGIN of the request (user|synthetic|analysis), diff --git a/src/indexing/orchestrator.ts b/src/indexing/orchestrator.ts index bfba706..cde0cb9 100644 --- a/src/indexing/orchestrator.ts +++ b/src/indexing/orchestrator.ts @@ -14,6 +14,7 @@ import { createEmbeddingProvider } from "./embeddings.js"; import type { EmbeddingProvider } from "./embeddings.js"; import { getProvider } from "./providers/index.js"; import { IndexingPipeline } from "./pipeline.js"; +import type { ItemFailure } from "./pipeline.js"; import { computeSourceConfigFingerprint, decideAcquisition, @@ -26,9 +27,80 @@ import { import { cleanupOldQueryLogs } from "../db/analytics.js"; import { markAtlasCachePagesStaleForSources } from "../db/atlas.js"; import { isAtlasSourceConfig, isFileSourceConfig } from "../types.js"; -import type { IndexState, IndexStatus, SourceConfig } from "../types.js"; +import type { + IndexState, + IndexStatus, + ItemFailureRecord, + SourceConfig, +} from "../types.js"; import type { ProviderOptions } from "./providers/types.js"; +/** + * How many consecutive runs an item may fail before it is quarantined. + * + * The orchestrator holds a source's state token whenever an item fails, so the + * failure is retried rather than silently skipped (see the hold logic in + * indexSourceWithState). With an unbounded retry that is a trap: an item that + * fails for a PERMANENT reason fails identically forever, and the source never + * moves. mcp.copilotkit.ai's `code` source sat at commit 0d0ea901 for ten days + * because one file produced a chunk larger than the embedding model's token + * limit — one file, and source-code search served ten-day-old results to every + * user the whole time. + * + * Three is the smallest bound that still distinguishes the two cases. One + * attempt cannot: a network blip, a rate limit, a brief upstream outage all + * look exactly like a poison item on their first failure, and quarantining + * them would trade a visible wedge for invisible data loss. Three consecutive + * runs — across separate webhook or nightly cycles, minutes to hours apart — + * is strong evidence the failure is a property of the item, not the weather. + */ +const MAX_ITEM_ATTEMPTS = 3; + +/** + * Fold one run's outcome into a source's per-item failure ledger. + * + * - An item that failed AGAIN has its streak incremented, and is quarantined + * once the streak reaches {@link MAX_ITEM_ATTEMPTS}. + * - An item that was ATTEMPTED this run and succeeded is dropped entirely, + * streak and quarantine both. Recovery must be complete, or a single old + * strike would eventually quarantine a perfectly healthy item. + * - An item that was NOT attempted this run (an incremental run only walks + * what changed) keeps its record untouched — absence from this run's item + * list is not evidence of anything. + * + * Quarantine is advisory, not a blocklist: a quarantined item is still handed + * to the pipeline on every subsequent run. All it means is that the item no + * longer holds the state token hostage for everything else in the source. + */ +export function reconcileItemFailures( + previous: Record | null, + attemptedIds: Set, + failures: ItemFailure[], +): Record { + const now = new Date().toISOString(); + const failedById = new Map(failures.map((f) => [f.id, f.error])); + const next: Record = {}; + + for (const [id, record] of Object.entries(previous ?? {})) { + if (failedById.has(id)) continue; // handled below, with its new count + if (attemptedIds.has(id)) continue; // attempted and succeeded — forgiven + next[id] = record; // untouched this run; carry it forward verbatim + } + + for (const [id, error] of failedById) { + const priorAttempts = previous?.[id]?.attempts ?? 0; + const attempts = priorAttempts + 1; + next[id] = { + attempts, + first_failed_at: previous?.[id]?.first_failed_at ?? now, + last_error: error, + quarantined: attempts >= MAX_ITEM_ATTEMPTS, + }; + } + + return next; +} + /** * Find all source configs that reference a given repo URL. */ @@ -815,22 +887,39 @@ export class IndexingOrchestrator { // and is never re-processed (permanent silent loss). When anything // failed we leave the prior token in place and mark the run errored so // the next incremental run reprocesses the failed items. - const failedIds: string[] = []; + const failures: ItemFailure[] = []; + const attemptedIds = new Set(); if (result.removedIds.length > 0) { - const { failedIds: removeFailed } = await pipeline.removeItems( + for (const id of result.removedIds) attemptedIds.add(id); + const { failures: removeFailures } = await pipeline.removeItems( result.removedIds, ); - failedIds.push(...removeFailed); + failures.push(...removeFailures); } if (result.items.length > 0) { - const { failedIds: indexFailed } = await pipeline.indexItems( + for (const item of result.items) attemptedIds.add(item.id); + const { failures: indexFailures } = await pipeline.indexItems( result.items, result.stateToken, ); - failedIds.push(...indexFailed); + failures.push(...indexFailures); } - if (failedIds.length > 0) { + // Fold this run's outcome into the per-item failure ledger, which + // is what tells a TRANSIENT failure apart from a permanent one. + const itemFailures = reconcileItemFailures( + state?.item_failures ?? null, + attemptedIds, + failures, + ); + const blocking = Object.entries(itemFailures).filter( + ([, record]) => !record.quarantined, + ); + const quarantined = Object.entries(itemFailures).filter( + ([, record]) => record.quarantined, + ); + + if (blocking.length > 0) { // Do NOT advance last_commit_sha — setIndexStatus preserves the // prior token, so the next incremental run re-diffs from where we // were and reprocesses the items that failed this run. Return false @@ -838,18 +927,37 @@ export class IndexingOrchestrator { // it is excluded from affectedSourceNames, so onReindexComplete and // the Atlas cache invalidation only fire for sources that fully // succeeded. + const blockingIds = blocking.map(([id]) => id); console.error( - `[orchestrator] Indexing for ${sourceConfig.name} had ${failedIds.length} failed item(s); holding state token for retry: ${failedIds.slice(0, 10).join(", ")}${failedIds.length > 10 ? " …" : ""}`, + `[orchestrator] Indexing for ${sourceConfig.name} had ${blockingIds.length} failed item(s); holding state token for retry: ${blockingIds.slice(0, 10).join(", ")}${blockingIds.length > 10 ? " …" : ""}`, ); await this.setIndexStatus( sourceConfig.type, sourceConfig.name, "error", - `${failedIds.length} item(s) failed to index/remove; state token held for retry`, + `${blockingIds.length} item(s) failed to index/remove; state token held for retry`, + itemFailures, ); return false; } + if (quarantined.length > 0) { + // Every outstanding failure has now failed MAX_ITEM_ATTEMPTS runs + // in a row, so retrying it unchanged is not going to work and the + // rest of the source should stop going stale behind it. Advance + // the token, but say so LOUDLY and keep the record: a quarantined + // item is still re-attempted on every subsequent run, and + // /health plus the admin index-stats op list it with its error. + for (const [id, record] of quarantined) { + console.error( + `[orchestrator] ${sourceConfig.name}: QUARANTINED ${id} after ` + + `${record.attempts} consecutive failed attempt(s) — advancing the ` + + `state token WITHOUT it so the rest of the source can index. ` + + `This item is missing from the index. Last error: ${record.last_error}`, + ); + } + } + await upsertIndexState({ source_type: sourceConfig.type, source_key: sourceConfig.name, @@ -862,9 +970,14 @@ export class IndexingOrchestrator { config_fingerprint: configFingerprint, last_indexed_at: new Date(), status: "idle", + item_failures: + Object.keys(itemFailures).length > 0 ? itemFailures : null, }); console.log( - `[orchestrator] Indexing complete for ${sourceConfig.name}`, + `[orchestrator] Indexing complete for ${sourceConfig.name}` + + (quarantined.length > 0 + ? ` (${quarantined.length} item(s) QUARANTINED and NOT indexed)` + : ""), ); return true; } catch (err) { @@ -923,6 +1036,7 @@ export class IndexingOrchestrator { sourceKey: string, status: IndexStatus, errorMessage?: string, + itemFailures?: Record, ): Promise { const existing = await getIndexState(sourceType, sourceKey); await upsertIndexState({ @@ -936,6 +1050,16 @@ export class IndexingOrchestrator { last_indexed_at: existing?.last_indexed_at ?? null, status, error_message: errorMessage ?? null, + // The ledger is PRESERVED unless this call supplies a new one. A plain + // status write ("indexing") must not wipe the consecutive-failure + // counts — that would reset every item's streak on every run and make + // the retry bound unreachable, restoring the permanent wedge. + item_failures: + itemFailures !== undefined + ? Object.keys(itemFailures).length > 0 + ? itemFailures + : null + : (existing?.item_failures ?? null), }); } } diff --git a/src/indexing/pipeline.ts b/src/indexing/pipeline.ts index b98eab8..07083e4 100644 --- a/src/indexing/pipeline.ts +++ b/src/indexing/pipeline.ts @@ -8,6 +8,18 @@ import { isFileSourceConfig } from "../types.js"; import type { Chunk, SourceConfig } from "../types.js"; import type { ContentItem } from "./providers/types.js"; +/** + * One item that failed to index or remove, with the error that caused it. + * + * The id alone is not enough for the caller: the orchestrator records WHY an + * item keeps failing so an operator can see the cause without going log + * spelunking in a container that rotates its output. + */ +export interface ItemFailure { + id: string; + error: string; +} + export class IndexingPipeline { private sourceConfig: SourceConfig; private embeddingProvider: EmbeddingProvider; @@ -30,16 +42,16 @@ export class IndexingPipeline { * * A single item's failure must not abort the batch (the remaining items still * index), but it MUST be surfaced: the returned `failedIds` lists every item - * whose `indexItem` threw. The caller uses this to avoid advancing the index - * state token past items that did not actually index — otherwise a failed - * item falls behind the advanced token and is never re-processed (permanent - * silent data loss). + * whose `indexItem` threw, and `failures` pairs each id with its error. The + * caller uses this to avoid advancing the index state token past items that + * did not actually index — otherwise a failed item falls behind the advanced + * token and is never re-processed (permanent silent data loss). */ async indexItems( items: ContentItem[], stateToken: string, - ): Promise<{ failedIds: string[] }> { - const failedIds: string[] = []; + ): Promise<{ failedIds: string[]; failures: ItemFailure[] }> { + const failures: ItemFailure[] = []; for (const item of items) { try { await this.indexItem(item, stateToken); @@ -48,10 +60,10 @@ export class IndexingPipeline { // pg-level metadata survives for diagnosis; collect the id so the // caller can hold the state token back. console.error(`${this.logPrefix} Failed to index ${item.id}:`, err); - failedIds.push(item.id); + failures.push({ id: item.id, error: errorText(err) }); } } - return { failedIds }; + return { failedIds: failures.map((f) => f.id), failures }; } /** @@ -60,18 +72,20 @@ export class IndexingPipeline { * processed — but the failed ids are RETURNED so the caller does not advance * the index state token over items whose stale chunks are still in the index. */ - async removeItems(ids: string[]): Promise<{ failedIds: string[] }> { - const failedIds: string[] = []; + async removeItems( + ids: string[], + ): Promise<{ failedIds: string[]; failures: ItemFailure[] }> { + const failures: ItemFailure[] = []; for (const id of ids) { try { await deleteChunksByFile(this.sourceConfig.name, id); } catch (err) { // Log the full error (not just err.message) so the stack survives. console.error(`${this.logPrefix} Failed to remove ${id}:`, err); - failedIds.push(id); + failures.push({ id, error: errorText(err) }); } } - return { failedIds }; + return { failedIds: failures.map((f) => f.id), failures }; } private async indexItem( @@ -155,3 +169,14 @@ export class IndexingPipeline { await replaceChunksForFile(this.sourceConfig.name, item.id, chunks); } } + +/** + * A single-line, storage-safe rendition of an error for the failure record. + * The full error (stack and all) already went to the log above; this is the + * short form that lands in index_state and on /health, so it is capped. + */ +function errorText(err: unknown): string { + const text = err instanceof Error ? err.message : String(err); + const oneLine = text.replace(/\s+/g, " ").trim(); + return oneLine.length > 500 ? `${oneLine.slice(0, 500)}…` : oneLine; +} diff --git a/src/server.ts b/src/server.ts index 9eadb0e..72ca8bc 100644 --- a/src/server.ts +++ b/src/server.ts @@ -2112,6 +2112,12 @@ export function projectIndexStateForOperators(s: IndexState): { error: string | null; next_acquire: "full" | "incremental" | null; next_acquire_reason: string | null; + quarantined_items: Array<{ + id: string; + attempts: number; + since: string; + error: string; + }>; } { // The source may have been removed from the config while its index_state // row survives; there is then no current config to compare against. @@ -2134,6 +2140,19 @@ export function projectIndexStateForOperators(s: IndexState): { error: s.error_message ?? null, next_acquire: decision?.mode ?? null, next_acquire_reason: decision?.reason ?? null, + // Items the orchestrator gave up holding the state token for. They are + // NOT in the index, so listing them here is the whole point: a + // quarantined item must never be an invisible gap. Only quarantined + // entries are projected — an item mid-retry is already reflected by + // `status: "error"`. + quarantined_items: Object.entries(s.item_failures ?? {}) + .filter(([, record]) => record.quarantined) + .map(([id, record]) => ({ + id, + attempts: record.attempts, + since: record.first_failed_at, + error: record.last_error, + })), }; } diff --git a/src/types.ts b/src/types.ts index 6bc7d9f..b7784c0 100644 --- a/src/types.ts +++ b/src/types.ts @@ -631,6 +631,33 @@ export interface ChunkOutput { export type IndexStatus = "idle" | "indexing" | "error"; +/** + * Per-item failure bookkeeping for one source, keyed by the item id the + * provider uses (a repo-relative file path for code/markdown sources). + * + * The orchestrator refuses to advance a source's state token past an item that + * failed, so the failure is retried instead of silently skipped. That is right + * for a transient failure and fatal for a permanent one: mcp.copilotkit.ai's + * `code` source sat at commit 0d0ea901 for TEN DAYS because one file's chunk + * was too large for the embedding model and therefore failed identically on + * every retry. Counting consecutive failures is what lets a permanent failure + * be told apart from a transient one, and bounded. + */ +export interface ItemFailureRecord { + /** Consecutive runs in which this item was attempted and failed. */ + attempts: number; + /** ISO timestamp of the first failure in the current streak. */ + first_failed_at: string; + /** The most recent error, so an operator can see WHY without log archaeology. */ + last_error: string; + /** + * True once `attempts` reached the retry bound. A quarantined item no longer + * holds the source's state token — it is still re-attempted on every run, + * but it can no longer freeze everything else behind it. + */ + quarantined: boolean; +} + export interface IndexState { source_type: string; source_key: string; @@ -645,4 +672,11 @@ export interface IndexState { last_indexed_at?: Date | null; status?: IndexStatus; error_message?: string | null; + /** + * Item ids that failed on recent runs, with their consecutive-failure counts + * (see {@link ItemFailureRecord}). Absent/NULL for installs whose + * index_state predates the column, and cleared whenever a run completes with + * nothing outstanding. + */ + item_failures?: Record | null; } From cbed45c04efd9e03eb1b9aa7cb08f8b850d1f0f0 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Thu, 10 Sep 2026 18:01:13 -0700 Subject: [PATCH 5/5] Drive the transient-failure test from a call counter, not a once-value check-test-shapes flags mockResolvedValueOnce in a file that clears rather than resets mocks: clearAllMocks drains the call log but not the once-queue, so an unconsumed value leaks into a later test's first call. --- ...orchestrator-poison-item-quarantine.test.ts | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/__tests__/orchestrator-poison-item-quarantine.test.ts b/src/__tests__/orchestrator-poison-item-quarantine.test.ts index 6c1bfe1..6918dcb 100644 --- a/src/__tests__/orchestrator-poison-item-quarantine.test.ts +++ b/src/__tests__/orchestrator-poison-item-quarantine.test.ts @@ -268,11 +268,20 @@ describe("IndexingOrchestrator: a permanently failing item must not wedge the so stateToken: "token-2", }); - // Run 1: a one-off network blip. - mockIndexItems.mockResolvedValueOnce({ - failedIds: ["docs/flaky.md"], - failures: [{ id: "docs/flaky.md", error: "ECONNRESET" }], + // Run 1 blips, run 2 succeeds. Driven by a call counter rather than + // mockResolvedValueOnce: an unconsumed once-value would leak into a later + // test, and this suite clears (not resets) mocks between tests. + let indexCall = 0; + mockIndexItems.mockImplementation(async () => { + indexCall++; + return indexCall === 1 + ? { + failedIds: ["docs/flaky.md"], + failures: [{ id: "docs/flaky.md", error: "ECONNRESET" }], + } + : { failedIds: [], failures: [] }; }); + await runSourceReindex(orchestrator); expect(holder.row.last_commit_sha).toBe("token-1"); expect(holder.row.status).toBe("error"); @@ -282,7 +291,6 @@ describe("IndexingOrchestrator: a permanently failing item must not wedge the so // Run 2: it succeeds. The token advances and the failure record clears — // the item must NOT carry a stale strike into the future. - mockIndexItems.mockResolvedValue({ failedIds: [], failures: [] }); await runSourceReindex(orchestrator); expect(holder.row.last_commit_sha).toBe("token-2"); expect(holder.row.status).toBe("idle");