learning_ai_common_plat/services/mcp-server/src/modules/a2a/reflection-synthesis-pipeline.ts

/**
 * ReflectionSynthesisAgent — A2A pipeline for MindLyst weekly reflection synthesis.
 *
 * Agent roster (3 steps):
 *   1. ReflectionCollectorAgent  — fetch recent reflections
 *   2. ThemeExtractionAgent      — run reflection-enrichment extraction on concatenated reflection text
 *   3. SynthesisReportAgent      — surface recurring themes, propose weekly summary text
 *
 * MCP tools:
 *   mindlyst.reflections.synthesize(limit?) — run pipeline
 */

import { randomUUID } from 'node:crypto';
import { z } from 'zod';
import { registerTool } from '../tools/registry.js';
import type { McpToolRequest } from '../tools/types.js';
import { mindlystReflectionsList, type ReflectionDoc } from '../../lib/mindlyst-client.js';
import { extractionRun, type ExtractionItem } from '../../lib/extraction-client.js';

// ── Types ──────────────────────────────────────────────────────────────────────

interface ThemeFrequency {
  theme: string;
  occurrences: number;
  sources: string[];
}

export interface ReflectionSynthesisReport {
  runId: string;
  productId: 'mindlyst';
  reflectionsAnalyzed: number;
  topThemes: ThemeFrequency[];
  highlights: string[];
  challenges: string[];
  extractedEntities: ExtractionItem[];
  proposedSummary: string;
  generatedAt: string;
}

// ── Step 1: ReflectionCollectorAgent ──────────────────────────────────────────

async function collectReflections(
  limit: number,
  opts: { token?: string; requestId?: string }
): Promise<ReflectionDoc[]> {
  try {
    const result = await mindlystReflectionsList({ limit }, opts);
    return result.items;
  } catch {
    return [];
  }
}

// ── Step 2: ThemeExtractionAgent ──────────────────────────────────────────────

async function extractThemes(
  reflections: ReflectionDoc[],
  opts: { token?: string; requestId?: string }
): Promise<{ entities: ExtractionItem[]; rawText: string }> {
  if (reflections.length === 0) return { entities: [], rawText: '' };

  const allThemes = reflections.flatMap(r => r.themes);
  const allHighlights = reflections.flatMap(r => r.highlights);
  const allChallenges = reflections.flatMap(r => r.challenges);
  const summaries = reflections.map(r => r.summary).filter(Boolean) as string[];

  const rawText = [
    'Themes: ' + allThemes.join(', '),
    'Highlights: ' + allHighlights.join('. '),
    'Challenges: ' + allChallenges.join('. '),
    summaries.length > 0 ? 'Summaries: ' + summaries.join(' ') : '',
  ]
    .filter(Boolean)
    .join('\n');

  try {
    const result = await extractionRun({ text: rawText, taskId: 'reflection-enrichment' }, opts);
    return { entities: result.extractions, rawText };
  } catch {
    return { entities: [], rawText };
  }
}

// ── Step 3: SynthesisReportAgent ──────────────────────────────────────────────

function buildThemeFrequencies(reflections: ReflectionDoc[]): ThemeFrequency[] {
  const frequencyMap = new Map<string, { count: number; sources: string[] }>();

  for (const reflection of reflections) {
    for (const theme of reflection.themes) {
      const entry = frequencyMap.get(theme) ?? { count: 0, sources: [] };
      entry.count++;
      entry.sources.push(reflection.weekStartDate);
      frequencyMap.set(theme, entry);
    }
  }

  return Array.from(frequencyMap.entries())
    .map(([theme, { count, sources }]) => ({ theme, occurrences: count, sources }))
    .sort((a, b) => b.occurrences - a.occurrences)
    .slice(0, 10);
}

function buildProposedSummary(
  topThemes: ThemeFrequency[],
  highlights: string[],
  challenges: string[],
  reflectionCount: number
): string {
  if (reflectionCount === 0) return 'No reflections found to synthesize.';

  const topThemeNames = topThemes
    .slice(0, 3)
    .map(t => t.theme)
    .join(', ');
  const summaryParts: string[] = [
    `Across ${reflectionCount} reflection(s), recurring themes include: ${topThemeNames || 'none identified'}.`,
  ];
  if (highlights.length > 0) {
    summaryParts.push(`Key highlights: ${highlights.slice(0, 3).join('; ')}.`);
  }
  if (challenges.length > 0) {
    summaryParts.push(`Challenges to address: ${challenges.slice(0, 2).join('; ')}.`);
  }
  return summaryParts.join(' ');
}

// ── Pipeline runner ────────────────────────────────────────────────────────────

async function runReflectionSynthesisPipeline(
  limit: number,
  req: McpToolRequest
): Promise<ReflectionSynthesisReport> {
  const runId = randomUUID();
  const opts = { token: req.headers.authorization?.slice(7), requestId: req.id };

  req.log.info({ runId, stepId: 'collect', limit }, 'ReflectionCollectorAgent start');
  const reflections = await collectReflections(limit, opts);
  req.log.info(
    { runId, stepId: 'collect', count: reflections.length },
    'ReflectionCollectorAgent done'
  );

  req.log.info(
    { runId, stepId: 'extract', count: reflections.length },
    'ThemeExtractionAgent start'
  );
  const { entities } = await extractThemes(reflections, opts);
  req.log.info(
    { runId, stepId: 'extract', entityCount: entities.length },
    'ThemeExtractionAgent done'
  );

  req.log.info({ runId, stepId: 'synthesize' }, 'SynthesisReportAgent start');
  const topThemes = buildThemeFrequencies(reflections);
  const allHighlights = reflections.flatMap(r => r.highlights);
  const allChallenges = reflections.flatMap(r => r.challenges);
  const proposedSummary = buildProposedSummary(
    topThemes,
    allHighlights,
    allChallenges,
    reflections.length
  );

  const report: ReflectionSynthesisReport = {
    runId,
    productId: 'mindlyst',
    reflectionsAnalyzed: reflections.length,
    topThemes,
    highlights: allHighlights,
    challenges: allChallenges,
    extractedEntities: entities,
    proposedSummary,
    generatedAt: new Date().toISOString(),
  };
  req.log.info(
    { runId, stepId: 'synthesize', themeCount: topThemes.length },
    'SynthesisReportAgent done'
  );

  return report;
}

// ── MCP tool registration ─────────────────────────────────────────────────────

registerTool({
  name: 'mindlyst.reflections.synthesize',
  description:
    'A2A pipeline: collects recent MindLyst reflections, runs reflection-enrichment extraction to surface entities and patterns, identifies recurring themes, and proposes a weekly summary text. Returns top themes by frequency, extracted entities, and a draft weekly summary. Requires admin role.',
  requiredRole: 'admin',
  inputSchema: z.object({
    limit: z.coerce
      .number()
      .int()
      .min(1)
      .max(52)
      .default(8)
      .describe('Number of recent reflections to include (default 8, max 52 = 1 year)'),
  }),
  async execute(args, req) {
    return runReflectionSynthesisPipeline(args.limit, req);
  },
});