From aeb197d5f4b2343db0b8e2d968361c0eb5740ead Mon Sep 17 00:00:00 2001 From: RaymondVerhoef Date: Wed, 20 May 2026 17:56:45 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20phase=203=20of=20AI=20pipeline=20harden?= =?UTF-8?q?ing=20=E2=80=94=20extraction=20quality?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace stateless one-shot extraction with a stateful, paced, cancellable pipeline. Six subtasks: - 3.1 Sentence-aware chunking with 800-char overlap (was paragraph-only at 4000 chars). Hard-split fallback for runaway sentences. - 3.2 Stateful extraction: chunks 2+ receive an "already-extracted topic IDs" hint capped at 200 IDs, so the model reuses IDs instead of inventing variants like software-developer vs software-engineer. - 3.3 Token-bucket limiter in llmRetry.js (extractionLimiter, 5 req/min). callLLM awaits the limiter before fetch; 429+Retry-After calls pauseUntil. Replaces hard setTimeout(12000) and setTimeout(15000). - 3.4 relevance_locked column on topics — admin edits to relevance are sticky across re-extraction. Migration + merge respects the flag + unlock checkbox in KnowledgeGraph edit form. - 3.5 Unify relation vocabulary — handbook prompt no longer mentions legacy "executes"; one-shot migration rewrites existing executes rows to executed_by with source/target swapped. - 3.6 Cancellation — Cancel button on UploadZone wired to an AbortController threaded into callLLM; aborted runs persist status = "cancelled" rather than "failed". Tests: 16 new unit tests for chunkText, buildKnownIdsHint, and createLimiter. All 61 tests pass, 0 lint errors, build clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- ...0500000_updated_topics_relevance_locked.js | 23 +++ .../1780500001_normalize_relation_types.js | 43 ++++++ src/components/admin/KnowledgeGraph.jsx | 44 ++++-- src/components/admin/UploadZone.jsx | 31 +++- src/lib/__tests__/extractionPipeline.test.js | 89 +++++++++++ src/lib/__tests__/llmRetry.test.js | 72 +++++++++ src/lib/db.js | 1 + src/lib/extractionPipeline.js | 145 ++++++++++++++---- src/lib/llm.js | 6 + src/lib/llmRetry.js | 103 +++++++++++++ 10 files changed, 509 insertions(+), 48 deletions(-) create mode 100644 pb_migrations/1780500000_updated_topics_relevance_locked.js create mode 100644 pb_migrations/1780500001_normalize_relation_types.js create mode 100644 src/lib/__tests__/extractionPipeline.test.js create mode 100644 src/lib/__tests__/llmRetry.test.js diff --git a/pb_migrations/1780500000_updated_topics_relevance_locked.js b/pb_migrations/1780500000_updated_topics_relevance_locked.js new file mode 100644 index 0000000..5f604a9 --- /dev/null +++ b/pb_migrations/1780500000_updated_topics_relevance_locked.js @@ -0,0 +1,23 @@ +/// +migrate((app) => { + const collection = app.findCollectionByNameOrId("pbc_2800040823") + + // add field — relevance_locked is set to true whenever an admin edits + // learning_relevance via the UI; mergeKnowledgeGraph must never overwrite + // learning_relevance on a locked topic during re-extraction. + collection.fields.addAt(5, new Field({ + "hidden": false, + "id": "bool_relevance_locked", + "name": "relevance_locked", + "presentable": false, + "required": false, + "system": false, + "type": "bool" + })) + + return app.save(collection) +}, (app) => { + const collection = app.findCollectionByNameOrId("pbc_2800040823") + collection.fields.removeById("bool_relevance_locked") + return app.save(collection) +}) diff --git a/pb_migrations/1780500001_normalize_relation_types.js b/pb_migrations/1780500001_normalize_relation_types.js new file mode 100644 index 0000000..fd18acb --- /dev/null +++ b/pb_migrations/1780500001_normalize_relation_types.js @@ -0,0 +1,43 @@ +/// +// One-shot data migration: rewrite legacy "executes" relations to the +// canonical "executed_by" vocabulary by swapping source and target. +// Previously `role --executes--> process`; canonical is +// `process --executed_by--> role`. +migrate((app) => { + const records = app.findRecordsByFilter( + "pbc_1883724256", // relations collection + 'type = "executes"', + '', + 0, + 0, + ) + + for (const rec of records) { + const source = rec.get("source") + const target = rec.get("target") + rec.set("type", "executed_by") + rec.set("source", target) + rec.set("target", source) + app.save(rec) + } +}, (app) => { + // Reverse: turn executed_by back into executes (best-effort — only those + // created before this migration would have been "executes"; rolling back + // will affect any newer executed_by rows too). + const records = app.findRecordsByFilter( + "pbc_1883724256", + 'type = "executed_by"', + '', + 0, + 0, + ) + + for (const rec of records) { + const source = rec.get("source") + const target = rec.get("target") + rec.set("type", "executes") + rec.set("source", target) + rec.set("target", source) + app.save(rec) + } +}) diff --git a/src/components/admin/KnowledgeGraph.jsx b/src/components/admin/KnowledgeGraph.jsx index 40898ab..d363226 100644 --- a/src/components/admin/KnowledgeGraph.jsx +++ b/src/components/admin/KnowledgeGraph.jsx @@ -180,10 +180,17 @@ const KnowledgeGraph = () => { }; const saveEdit = async () => { - await db.upsertTopic({ ...selectedNode, ...editData }); - const updated = topics.map(t => t.id === selectedNode.id ? { ...t, ...editData } : t); + // If the admin touched learning_relevance, lock it so re-extraction won't overwrite the choice. + // But an explicit relevance_locked in editData (the unlock checkbox) always wins. + const relevanceChanged = + editData.learning_relevance !== undefined && + editData.learning_relevance !== selectedNode.learning_relevance; + const next = { ...editData }; + if (relevanceChanged && next.relevance_locked === undefined) next.relevance_locked = true; + await db.upsertTopic({ ...selectedNode, ...next }); + const updated = topics.map(t => t.id === selectedNode.id ? { ...t, ...next } : t); setTopics(updated); - setSelectedNode({ ...selectedNode, ...editData }); + setSelectedNode({ ...selectedNode, ...next }); setIsEditing(false); }; @@ -265,22 +272,11 @@ const KnowledgeGraph = () => { setSyncProgress(`Processing ${count} of ${filesToProcess.length}: ${file.name}...`); try { const rawContent = await getFileContent('respellion', 'employee-handbook', file.path); + // Pacing is handled centrally by extractionLimiter inside analyzeHandbookDelta. await analyzeHandbookDelta(rawContent, file.path); await db.updateHandbookSyncState(file.path, file.sha); - - // To respect Anthropic's 5 requests per minute rate limit on this tier, - // we pause for 15 seconds before processing the next file. - if (count < filesToProcess.length) { - setSyncProgress(`Waiting 15s to avoid rate limits... (${count}/${filesToProcess.length})`); - await new Promise(resolve => setTimeout(resolve, 15000)); - } } catch (err) { console.error('Failed to process file:', file.path, err); - // We continue processing other files even if one fails, but still wait to avoid further rate limits - if (count < filesToProcess.length) { - setSyncProgress(`Error on ${file.name}. Waiting 15s... (${count}/${filesToProcess.length})`); - await new Promise(resolve => setTimeout(resolve, 15000)); - } } } setSyncProgress('Sync Complete! Click "Analyze & Optimize Graph" above to clean up and merge.'); @@ -369,7 +365,7 @@ ${JSON.stringify({ topics: compactTopics, relations: compactRelations })}`; if (actions.relevanceUpdates && Array.isArray(actions.relevanceUpdates)) { for (const update of actions.relevanceUpdates) { const topicIndex = updatedTopics.findIndex(t => t.id === update.id); - if (topicIndex !== -1) { + if (topicIndex !== -1 && !updatedTopics[topicIndex].relevance_locked) { updatedTopics[topicIndex] = { ...updatedTopics[topicIndex], learning_relevance: update.learning_relevance }; } } @@ -532,6 +528,17 @@ ${JSON.stringify({ topics: compactTopics, relations: compactRelations })}`; + {selectedNode.relevance_locked && ( + + )}
@@ -554,9 +561,12 @@ ${JSON.stringify({ topics: compactTopics, relations: compactRelations })}`;

Type & Relevance

-
+
{selectedNode.type} {selectedNode.learning_relevance || 'standard'} + {selectedNode.relevance_locked && ( + locked + )}
diff --git a/src/components/admin/UploadZone.jsx b/src/components/admin/UploadZone.jsx index d64a8a8..63d4029 100644 --- a/src/components/admin/UploadZone.jsx +++ b/src/components/admin/UploadZone.jsx @@ -1,5 +1,5 @@ import { useState, useRef } from 'react'; -import { UploadCloud, AlertCircle, CheckCircle } from 'lucide-react'; +import { UploadCloud, AlertCircle, CheckCircle, X } from 'lucide-react'; import { processSourceText } from '../../lib/extractionPipeline'; import Card from '../ui/Card'; import Button from '../ui/Button'; @@ -12,24 +12,36 @@ const UploadZone = ({ onUploadComplete }) => { const [status, setStatus] = useState(null); const fileInputRef = useRef(null); + const abortRef = useRef(null); // ── File upload (drag & drop / browse) ──────────────────────────────────── const processFile = async (file) => { setIsProcessing(true); setStatus(null); + const controller = new AbortController(); + abortRef.current = controller; try { const text = await file.text(); - await processSourceText(text, file.name); + await processSourceText(text, file.name, { signal: controller.signal }); setStatus({ type: 'success', msg: `Successfully processed: ${file.name}` }); if (onUploadComplete) onUploadComplete(); } catch (error) { - setStatus({ type: 'error', msg: `Error processing file: ${error.message}` }); + if (error?.name === 'AbortError') { + setStatus({ type: 'error', msg: `Cancelled: ${file.name}` }); + } else { + setStatus({ type: 'error', msg: `Error processing file: ${error.message}` }); + } } finally { + abortRef.current = null; setIsProcessing(false); } }; + const cancelProcessing = () => { + abortRef.current?.abort(new DOMException('Cancelled by user', 'AbortError')); + }; + const handleDragOver = (e) => { e.preventDefault(); setIsDragging(true); }; const handleDragLeave = () => setIsDragging(false); @@ -80,6 +92,19 @@ const UploadZone = ({ onUploadComplete }) => { /> + {/* Cancel sits outside the upload card so pointer-events-none doesn't disable it. */} + {isProcessing && ( +
+ +
+ )} + {/* ─── Status messages ─── */} {status && (
({ + pb: { collection: () => ({ create: () => ({ catch: () => {} }) }) }, +})); + +import { chunkText, buildKnownIdsHint, MAX_CHUNK_CHARS, OVERLAP_CHARS } from '../extractionPipeline'; + +describe('chunkText', () => { + let warnSpy; + beforeEach(() => { warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); }); + afterEach(() => { warnSpy.mockRestore(); }); + + it('returns the original text as a single chunk when below maxChars', () => { + const result = chunkText('A short paragraph.'); + expect(result).toEqual(['A short paragraph.']); + }); + + it('returns empty array for empty/whitespace input', () => { + expect(chunkText('')).toEqual([]); + expect(chunkText(' \n ')).toEqual([]); + }); + + it('splits along sentence boundaries with overlap between adjacent chunks', () => { + const sentence = 'This is a sentence with exactly a known length. '; + const text = sentence.repeat(100); // ~5000 chars + const chunks = chunkText(text, { maxChars: 600, overlapChars: 150 }); + + expect(chunks.length).toBeGreaterThan(1); + for (const c of chunks) { + expect(c.length).toBeLessThanOrEqual(600); + } + // Adjacent chunks share trailing text — the overlap should be non-empty. + for (let i = 1; i < chunks.length; i++) { + const tail = chunks[i - 1].slice(-150); + // The new chunk must begin with content that appears at the tail of the prior chunk. + const firstHundred = chunks[i].slice(0, 100); + // At least one word from the tail should appear in the head of the next chunk. + const words = tail.split(/\s+/).filter((w) => w.length > 3); + const shared = words.some((w) => firstHundred.includes(w)); + expect(shared).toBe(true); + } + }); + + it('hard-splits a single sentence that exceeds maxChars and logs a warning', () => { + const huge = 'word '.repeat(400).trim() + '.'; // ~2000 chars, no sentence break + const chunks = chunkText(huge, { maxChars: 500, overlapChars: 50 }); + expect(chunks.length).toBeGreaterThan(1); + expect(warnSpy).toHaveBeenCalled(); + for (const c of chunks) expect(c.length).toBeLessThanOrEqual(500); + }); + + it('handles paragraph-only splits when no sentence punctuation is present', () => { + const paragraphs = Array.from({ length: 10 }, (_, i) => `paragraph ${i} content here`).join('\n\n'); + const chunks = chunkText(paragraphs, { maxChars: 80, overlapChars: 20 }); + expect(chunks.length).toBeGreaterThan(1); + }); + + it('uses the documented defaults', () => { + expect(MAX_CHUNK_CHARS).toBe(8000); + expect(OVERLAP_CHARS).toBe(800); + }); +}); + +describe('buildKnownIdsHint', () => { + it('returns empty string when no IDs are known', () => { + expect(buildKnownIdsHint([])).toBe(''); + expect(buildKnownIdsHint(undefined)).toBe(''); + expect(buildKnownIdsHint(null)).toBe(''); + }); + + it('formats the known IDs as a bulleted list with a leading instruction', () => { + const hint = buildKnownIdsHint(['software-engineer', 'onboarding-buddy']); + expect(hint).toContain('Already-extracted topic IDs'); + expect(hint).toContain('- software-engineer'); + expect(hint).toContain('- onboarding-buddy'); + expect(hint.endsWith('\n')).toBe(true); + }); + + it('caps the hint at the 200 most recent IDs', () => { + const ids = Array.from({ length: 250 }, (_, i) => `topic-${i}`); + const hint = buildKnownIdsHint(ids); + // The newest IDs must appear; the oldest must not. + expect(hint).toContain('topic-249'); + expect(hint).toContain('topic-50'); + expect(hint).not.toContain('topic-49'); + expect(hint).not.toContain('topic-0\n'); + }); +}); diff --git a/src/lib/__tests__/llmRetry.test.js b/src/lib/__tests__/llmRetry.test.js new file mode 100644 index 0000000..2529459 --- /dev/null +++ b/src/lib/__tests__/llmRetry.test.js @@ -0,0 +1,72 @@ +import { describe, expect, it, vi, afterEach } from 'vitest'; +import { createLimiter, extractionLimiter } from '../llmRetry'; + +afterEach(() => { vi.useRealTimers(); }); + +describe('createLimiter', () => { + it('rejects an invalid rps', () => { + expect(() => createLimiter({ rps: 0 })).toThrow(); + expect(() => createLimiter({ rps: -1 })).toThrow(); + }); + + it('rejects an invalid burst', () => { + expect(() => createLimiter({ rps: 1, burst: 0 })).toThrow(); + }); + + it('lets the first call through immediately (initial burst token)', async () => { + const limiter = createLimiter({ rps: 1, burst: 1 }); + const start = Date.now(); + await limiter.acquire(); + expect(Date.now() - start).toBeLessThan(50); + }); + + it('queues subsequent calls to respect the spacing', async () => { + vi.useFakeTimers(); + const limiter = createLimiter({ rps: 10, burst: 1 }); // 100ms spacing + await limiter.acquire(); // consume initial token + + let resolved = false; + const p = limiter.acquire().then(() => { resolved = true; }); + + await vi.advanceTimersByTimeAsync(50); + expect(resolved).toBe(false); + + await vi.advanceTimersByTimeAsync(100); + await p; + expect(resolved).toBe(true); + }); + + it('honours pauseUntil — no acquire returns before the pause expires', async () => { + vi.useFakeTimers(); + const limiter = createLimiter({ rps: 100, burst: 5 }); + limiter.pauseUntil(Date.now() + 1000); + + let resolved = false; + const p = limiter.acquire().then(() => { resolved = true; }); + + await vi.advanceTimersByTimeAsync(500); + expect(resolved).toBe(false); + + await vi.advanceTimersByTimeAsync(600); + await p; + expect(resolved).toBe(true); + }); + + it('aborts a queued acquire when the signal fires', async () => { + const limiter = createLimiter({ rps: 1, burst: 1 }); + await limiter.acquire(); // consume + + const ctl = new AbortController(); + const p = limiter.acquire({ signal: ctl.signal }); + ctl.abort(); + + await expect(p).rejects.toBeInstanceOf(DOMException); + }); +}); + +describe('extractionLimiter', () => { + it('is exported and exposes the limiter shape', () => { + expect(typeof extractionLimiter.acquire).toBe('function'); + expect(typeof extractionLimiter.pauseUntil).toBe('function'); + }); +}); diff --git a/src/lib/db.js b/src/lib/db.js index 5d0536a..331566b 100644 --- a/src/lib/db.js +++ b/src/lib/db.js @@ -18,6 +18,7 @@ export async function saveTopics(topics) { type: t.type, description: t.description, learning_relevance: t.learning_relevance || 'standard', + relevance_locked: t.relevance_locked === true, }, { requestKey: null }); } } diff --git a/src/lib/extractionPipeline.js b/src/lib/extractionPipeline.js index e553af9..84bc259 100644 --- a/src/lib/extractionPipeline.js +++ b/src/lib/extractionPipeline.js @@ -1,8 +1,28 @@ import * as db from './db'; import { callLLM } from './llm'; +import { extractionLimiter } from './llmRetry'; import { EMIT_KNOWLEDGE_GRAPH_TOOL, EMIT_HANDBOOK_DELTA_TOOL } from './llmTools'; import { normalizeHandbookResult } from './llmSchemas'; +const MAX_KNOWN_IDS_HINT = 200; + +/** + * Build the "already-extracted topic IDs" hint that prepends every chunk + * after the first. Capped at the most-recent `MAX_KNOWN_IDS_HINT` IDs so + * the prompt stays a bounded size; the model uses this list to reuse IDs + * rather than invent variants like `software-developer` for + * `software-engineer`. + */ +export function buildKnownIdsHint(ids) { + if (!ids || !ids.length) return ''; + const recent = ids.slice(-MAX_KNOWN_IDS_HINT); + return [ + 'Already-extracted topic IDs (do NOT create new IDs for these — reuse them if the same concept appears here):', + ...recent.map((id) => `- ${id}`), + '', + ].join('\n'); +} + const EXTRACTION_SYSTEM_PROMPT = `You are an AI knowledge extractor for Respellion, an IT company built on radical transparency. You receive a source text. Extract every distinct concept, role, and process from it and emit them through the emit_knowledge_graph tool. @@ -28,17 +48,17 @@ Relation types: related_to | depends_on | part_of | executed_by. const HANDBOOK_SYSTEM_PROMPT = `You are analysing an update to the Respellion Employee Handbook. Emit the extracted topics and relations through the emit_handbook_delta tool. CRITICAL INSTRUCTIONS: -- Every process must have a role attached (the role that executes it). +- Every process must have a role attached. Express this as: process --executed_by--> role. - Every concept must connect to a process or role. - Mark handbook topics with metadata.source = "github_handbook". - Assign learning_relevance using the same scale as extraction: core | standard | peripheral | exclude. -Relation types: related_to | depends_on | part_of | executed_by. (Legacy "executes" relations are normalised by the client into executed_by with source/target swapped.) +Relation types: related_to | depends_on | part_of | executed_by. `; const cachedSystem = (text) => [{ type: 'text', text, cache_control: { type: 'ephemeral' } }]; -export async function analyzeHandbookDelta(fileContent, filePath) { +export async function analyzeHandbookDelta(fileContent, filePath, { signal } = {}) { const result = await callLLM({ task: 'extract.handbook', tier: 'standard', @@ -47,6 +67,8 @@ export async function analyzeHandbookDelta(fileContent, filePath) { tools: [EMIT_HANDBOOK_DELTA_TOOL], toolChoice: { type: 'tool', name: EMIT_HANDBOOK_DELTA_TOOL.name }, maxTokens: 8192, + limiter: extractionLimiter, + signal, }); const raw = result.toolUses[0]?.input; @@ -57,24 +79,79 @@ export async function analyzeHandbookDelta(fileContent, filePath) { return { success: true, data: extractedData }; } -function chunkText(text, maxChunkSize = 4000) { - const paragraphs = text.split(/\n+/); - const chunks = []; - let currentChunk = ''; +/** + * Sentence-aware chunker with overlap. + * + * Targets ~2000 input tokens per chunk (`MAX_CHUNK_CHARS / 4`). Splits on + * sentence boundaries first, then falls back to paragraph boundaries, and + * hard-splits inside an oversized sentence as a last resort. Adjacent chunks + * share `overlapChars` of trailing text to preserve cross-boundary context + * for the model. + * + * Exported for unit tests; callers in this module use it directly. + * + * @param {string} text + * @param {{ maxChars?: number, overlapChars?: number }} [opts] + * @returns {string[]} + */ +export const MAX_CHUNK_CHARS = 8000; +export const OVERLAP_CHARS = 800; - for (const para of paragraphs) { - if ((currentChunk + '\n' + para).length > maxChunkSize) { - if (currentChunk) chunks.push(currentChunk.trim()); - currentChunk = para; - } else { - currentChunk = currentChunk ? currentChunk + '\n' + para : para; +export function chunkText(text, { maxChars = MAX_CHUNK_CHARS, overlapChars = OVERLAP_CHARS } = {}) { + if (typeof text !== 'string' || !text.trim()) return []; + const trimmed = text.trim(); + if (trimmed.length <= maxChars) return [trimmed]; + + const units = splitIntoChunkableUnits(trimmed, maxChars); + if (units.length === 0) return []; + + const chunks = []; + let buf = ''; + let bufLen = 0; // length of new (non-overlap) content added since last flush + + for (const unit of units) { + const wouldOverflow = (buf ? buf.length + 1 + unit.length : unit.length) > maxChars; + if (wouldOverflow && bufLen > 0) { + chunks.push(buf.trim()); + const overlap = buf.length > overlapChars ? buf.slice(-overlapChars) : ''; + buf = overlap; + bufLen = 0; } + // If the overlap + unit still won't fit, drop the overlap so the unit fits cleanly. + if (buf && (buf.length + 1 + unit.length) > maxChars) { + buf = ''; + } + buf = buf ? buf + ' ' + unit : unit; + bufLen += unit.length + (bufLen > 0 ? 1 : 0); } - if (currentChunk) chunks.push(currentChunk.trim()); + if (bufLen > 0 && buf.trim()) chunks.push(buf.trim()); return chunks; } -export async function processSourceText(textContent, sourceName) { +function splitIntoChunkableUnits(text, maxChars) { + const paragraphs = text.split(/\n\s*\n+/); + const units = []; + for (const para of paragraphs) { + const trimmedPara = para.trim(); + if (!trimmedPara) continue; + const sentences = trimmedPara.split(/(?<=[.!?])\s+/); + for (const s of sentences) { + const sentence = s.trim(); + if (!sentence) continue; + if (sentence.length <= maxChars) { + units.push(sentence); + } else { + for (let i = 0; i < sentence.length; i += maxChars) { + units.push(sentence.slice(i, i + maxChars)); + } + console.warn(`[chunkText] Hard-split a sentence of ${sentence.length} chars (exceeds maxChars=${maxChars}).`); + } + } + } + return units; +} + +export async function processSourceText(textContent, sourceName, { signal } = {}) { const existing = await db.getSources(); const alreadyDone = existing.find( s => s.name === sourceName && s.status === 'completed' @@ -87,36 +164,42 @@ export async function processSourceText(textContent, sourceName) { const sourceId = rec.id; try { - const chunks = chunkText(textContent, 4000); + const chunks = chunkText(textContent); console.log(`[Pipeline] Split "${sourceName}" into ${chunks.length} chunks for processing.`); - let allExtractedTopics = []; - let allExtractedRelations = []; + const existingTopics = await db.getTopics(); + const knownIds = existingTopics.map((t) => t.id); + + const allExtractedTopics = []; + const allExtractedRelations = []; for (let i = 0; i < chunks.length; i++) { - if (i > 0) { - console.log(`[Pipeline] Pacing delay (12s) to prevent rate limits before chunk ${i + 1}/${chunks.length}...`); - await new Promise(r => setTimeout(r, 12000)); - } + if (signal?.aborted) throw signal.reason ?? new DOMException('Aborted', 'AbortError'); console.log(`[Pipeline] Processing chunk ${i + 1}/${chunks.length} (${chunks[i].length} chars)...`); + const hint = i > 0 ? buildKnownIdsHint(knownIds) : ''; const result = await callLLM({ task: 'extract.source', tier: 'standard', system: cachedSystem(EXTRACTION_SYSTEM_PROMPT), - user: `Analyze this part of the document (${i + 1}/${chunks.length}):\n\n${chunks[i]}`, + user: `${hint}Analyze this part of the document (${i + 1}/${chunks.length}):\n\n${chunks[i]}`, tools: [EMIT_KNOWLEDGE_GRAPH_TOOL], toolChoice: { type: 'tool', name: EMIT_KNOWLEDGE_GRAPH_TOOL.name }, maxTokens: 8192, + limiter: extractionLimiter, + signal, }); const extractedData = result.toolUses[0]?.input; if (!extractedData) throw new Error(`Extraction did not emit a tool result for chunk ${i + 1}.`); - if (extractedData.topics && Array.isArray(extractedData.topics)) { + if (Array.isArray(extractedData.topics)) { allExtractedTopics.push(...extractedData.topics); + for (const t of extractedData.topics) { + if (t?.id && !knownIds.includes(t.id)) knownIds.push(t.id); + } } - if (extractedData.relations && Array.isArray(extractedData.relations)) { + if (Array.isArray(extractedData.relations)) { allExtractedRelations.push(...extractedData.relations); } } @@ -127,7 +210,8 @@ export async function processSourceText(textContent, sourceName) { return { success: true, data: { topics: allExtractedTopics, relations: allExtractedRelations } }; } catch (error) { - await db.updateSourceStatus(sourceId, 'failed', error.message); + const isAbort = error?.name === 'AbortError'; + await db.updateSourceStatus(sourceId, isAbort ? 'cancelled' : 'failed', isAbort ? 'cancelled by user' : error.message); throw error; } } @@ -142,11 +226,16 @@ async function mergeKnowledgeGraph(newData) { for (const t of newData.topics) { if (topicsMap.has(t.id)) { const existing = topicsMap.get(t.id); - topicsMap.set(t.id, { + const merged = { ...existing, ...t, description: t.description || existing.description, - }); + }; + if (existing.relevance_locked) { + merged.learning_relevance = existing.learning_relevance; + merged.relevance_locked = true; + } + topicsMap.set(t.id, merged); } else { topicsMap.set(t.id, t); } diff --git a/src/lib/llm.js b/src/lib/llm.js index 0edea82..c6acd5c 100644 --- a/src/lib/llm.js +++ b/src/lib/llm.js @@ -272,6 +272,7 @@ function validateToolInputs(toolUses, task, toolSchemas) { * @property {number} [maxTokens=4096] * @property {number} [temperature=0] * @property {AbortSignal} [signal] + * @property {{ acquire: (opts?:{signal?:AbortSignal}) => Promise, pauseUntil: (untilMs:number) => void }} [limiter] */ /** @@ -291,6 +292,7 @@ export async function callLLM(options) { maxTokens = 4096, temperature = 0, signal, + limiter, } = options; if (!task) throw new Error('callLLM requires a `task` label.'); @@ -318,6 +320,7 @@ export async function callLLM(options) { try { result = await withRetry( async () => { + if (limiter) await limiter.acquire({ signal }); const timeoutCtl = signal ? null : new AbortController(); const timer = timeoutCtl ? setTimeout(() => timeoutCtl.abort(new DOMException('Timeout', 'AbortError')), DEFAULT_TIMEOUT_MS) : null; const fetchSignal = linkSignals(signal, timeoutCtl?.signal); @@ -337,6 +340,9 @@ export async function callLLM(options) { const errBody = await response.json().catch(() => ({})); if (isRetryableStatus(response.status)) { const retryAfterMs = parseRetryAfter(response.headers.get('Retry-After')); + if (response.status === 429 && retryAfterMs != null && limiter) { + limiter.pauseUntil(Date.now() + retryAfterMs); + } throw new RetryableError(response.status, retryAfterMs, `HTTP ${response.status}`); } throw new LLMHttpError(response.status, response.statusText, errBody); diff --git a/src/lib/llmRetry.js b/src/lib/llmRetry.js index 33281f5..b2d2710 100644 --- a/src/lib/llmRetry.js +++ b/src/lib/llmRetry.js @@ -62,6 +62,109 @@ function sleep(ms, signal) { }); } +/** + * Token-bucket rate limiter shared by callers that hit the same upstream + * quota (e.g. handbook extraction loops). Replaces the static + * `setTimeout(..., 12000)` / 15s sleeps that Phase 1 relied on. The bucket + * refills continuously; `acquire` resolves either immediately (token + * available) or after the next refill tick. + * + * `rps` is "requests per second" (use fractional values for per-minute + * limits: `5/60` for 5 req/min). `burst` is the maximum number of tokens + * the bucket can hold; default 1 means strict spacing. + * + * Call `pauseUntil(timestampMs)` after a 429 with a `Retry-After` hint — + * no acquire returns before that timestamp. + * + * @param {{ rps?: number, burst?: number }} [opts] + */ +export function createLimiter({ rps = 1, burst = 1 } = {}) { + if (rps <= 0) throw new Error('createLimiter: rps must be > 0'); + if (burst < 1) throw new Error('createLimiter: burst must be >= 1'); + const intervalMs = 1000 / rps; + let tokens = burst; + let lastRefill = Date.now(); + let pausedUntil = 0; + const waiters = []; + + function refill(now) { + const elapsed = now - lastRefill; + if (elapsed <= 0) return; + const earned = elapsed / intervalMs; + if (earned >= 1) { + tokens = Math.min(burst, tokens + Math.floor(earned)); + lastRefill = now; + } + } + + function drain() { + while (waiters.length) { + const now = Date.now(); + if (now < pausedUntil) { + scheduleWake(pausedUntil - now); + return; + } + refill(now); + if (tokens >= 1) { + tokens -= 1; + const w = waiters.shift(); + w.resolve(); + } else { + const wait = Math.max(intervalMs - (now - lastRefill), 0); + scheduleWake(wait); + return; + } + } + } + + let wakeTimer = null; + function scheduleWake(ms) { + if (wakeTimer) return; + wakeTimer = setTimeout(() => { + wakeTimer = null; + drain(); + }, ms); + } + + return { + /** @param {{signal?:AbortSignal}} [opts] */ + async acquire({ signal } = {}) { + if (signal?.aborted) throw signal.reason ?? new DOMException('Aborted', 'AbortError'); + return new Promise((resolve, reject) => { + const entry = { resolve, reject }; + const onAbort = () => { + const i = waiters.indexOf(entry); + if (i !== -1) waiters.splice(i, 1); + reject(signal.reason ?? new DOMException('Aborted', 'AbortError')); + }; + if (signal) signal.addEventListener('abort', onAbort, { once: true }); + const wrapped = { + resolve: () => { if (signal) signal.removeEventListener('abort', onAbort); resolve(); }, + reject, + }; + waiters.push(wrapped); + drain(); + }); + }, + /** Block all `acquire`s until `untilMs` (epoch milliseconds). */ + pauseUntil(untilMs) { + if (untilMs > pausedUntil) pausedUntil = untilMs; + drain(); + }, + /** Inspect state — primarily for tests. */ + _state() { + return { tokens, pausedUntil, waiters: waiters.length }; + }, + }; +} + +/** + * Shared limiter for the multi-call extraction loops (source chunks, + * handbook file sync). 5 requests/minute matches the lowest published + * Anthropic tier so we stay well clear of 429. + */ +export const extractionLimiter = createLimiter({ rps: 5 / 60, burst: 1 }); + /** * Run `fn(attempt)` with retry. `fn` may throw a `RetryableError` to request * a retry, or any other error to fail immediately. -- 2.49.1