learning-platform/src/lib/extractionPipeline.js

import * as db from './db';
import { callLLM } from './llm';
import { EMIT_KNOWLEDGE_GRAPH_TOOL, EMIT_HANDBOOK_DELTA_TOOL } from './llmTools';
import { normalizeHandbookResult } from './llmSchemas';

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.

CRITICAL INSTRUCTIONS FOR COMPLETENESS:
- Extract EVERY SINGLE distinct role, process, and concept described or mentioned in the source text.
- DO NOT summarise, skip, truncate, or omit any items.
- If the document contains 29 roles, the topics array must contain exactly 29 role topics.
- Completeness is paramount. Failing to extract all topics loses critical company knowledge.
- Facts should be integrated into the descriptions of other topics — never extracted as standalone topics.
- Keep descriptions concise (max 3 sentences) so the response fits.

Topic IDs are lowercase kebab-case slugs specific to the topic (e.g. "software-engineer", "data-quality-review"). Do not use generic IDs like "role-1" or "concept-2".

Assign a learning_relevance to every topic:
- "core": fundamental company knowledge.
- "standard": normal learning topics.
- "peripheral": good to know, low priority.
- "exclude": pure operational reference (printer guides, wifi passwords) that should never be tested.

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 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.)
`;

const cachedSystem = (text) => [{ type: 'text', text, cache_control: { type: 'ephemeral' } }];

export async function analyzeHandbookDelta(fileContent, filePath) {
  const result = await callLLM({
    task: 'extract.handbook',
    tier: 'standard',
    system: cachedSystem(HANDBOOK_SYSTEM_PROMPT),
    user: `Analyze the following handbook file update (${filePath}):\n\n${fileContent}`,
    tools: [EMIT_HANDBOOK_DELTA_TOOL],
    toolChoice: { type: 'tool', name: EMIT_HANDBOOK_DELTA_TOOL.name },
    maxTokens: 8192,
  });

  const raw = result.toolUses[0]?.input;
  if (!raw) throw new Error('Handbook extraction did not emit a tool result.');
  const extractedData = normalizeHandbookResult(raw);

  await mergeKnowledgeGraph(extractedData);
  return { success: true, data: extractedData };
}

function chunkText(text, maxChunkSize = 4000) {
  const paragraphs = text.split(/\n+/);
  const chunks = [];
  let currentChunk = '';

  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;
    }
  }
  if (currentChunk) chunks.push(currentChunk.trim());
  return chunks;
}

export async function processSourceText(textContent, sourceName) {
  const existing = await db.getSources();
  const alreadyDone = existing.find(
    s => s.name === sourceName && s.status === 'completed'
  );
  if (alreadyDone) {
    throw new Error(`"${sourceName}" has already been processed. Delete the existing source first if you want to re-analyse it.`);
  }

  const rec = await db.addSource({ name: sourceName, status: 'processing' });
  const sourceId = rec.id;

  try {
    const chunks = chunkText(textContent, 4000);
    console.log(`[Pipeline] Split "${sourceName}" into ${chunks.length} chunks for processing.`);

    let allExtractedTopics = [];
    let 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));
      }

      console.log(`[Pipeline] Processing chunk ${i + 1}/${chunks.length} (${chunks[i].length} chars)...`);
      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]}`,
        tools: [EMIT_KNOWLEDGE_GRAPH_TOOL],
        toolChoice: { type: 'tool', name: EMIT_KNOWLEDGE_GRAPH_TOOL.name },
        maxTokens: 8192,
      });

      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)) {
        allExtractedTopics.push(...extractedData.topics);
      }
      if (extractedData.relations && Array.isArray(extractedData.relations)) {
        allExtractedRelations.push(...extractedData.relations);
      }
    }

    await mergeKnowledgeGraph({ topics: allExtractedTopics, relations: allExtractedRelations });
    await db.updateSourceStatus(sourceId, 'completed');

    return { success: true, data: { topics: allExtractedTopics, relations: allExtractedRelations } };

  } catch (error) {
    await db.updateSourceStatus(sourceId, 'failed', error.message);
    throw error;
  }
}

async function mergeKnowledgeGraph(newData) {
  const existingTopics = await db.getTopics();
  const existingRelations = await db.getRelations();

  const topicsMap = new Map(existingTopics.map(t => [t.id, t]));

  if (newData.topics && Array.isArray(newData.topics)) {
    for (const t of newData.topics) {
      if (topicsMap.has(t.id)) {
        const existing = topicsMap.get(t.id);
        topicsMap.set(t.id, {
          ...existing,
          ...t,
          description: t.description || existing.description,
        });
      } else {
        topicsMap.set(t.id, t);
      }
    }
  }

  const newRelations = [...existingRelations];
  if (newData.relations && Array.isArray(newData.relations)) {
    for (const r of newData.relations) {
      const isDup = newRelations.some(ex => ex.source === r.source && ex.target === r.target && ex.type === r.type);
      if (!isDup) {
        newRelations.push(r);
      }
    }
  }

  await db.saveTopics(Array.from(topicsMap.values()));
  await db.saveRelations(newRelations);
}