feat: Add synthesis functionality with session summary and artifact ingestion

app.py

@@ -9,11 +9,14 @@ Provides a web UI for the FractalStat RAG system with GPU acceleration.
 import gradio as gr
 import os
 import time
-from typing import List, Dict, Any
 
 from warbler_cda.answer_generator import AnswerGenerator
 from warbler_cda.remote_pack_loader import RemotePackLoader
-from warbler_cda.summarization_ladder import SummarizationLadder
+from warbler_cda.synthesis_bridge import (
+    get_synthesis_session_summary,
+    ingest_synthesis_artifacts,
+    synthesize_text,
+)
 
 # Import the HuggingFace Spaces GPU decorator
 try:
@@ -328,6 +331,17 @@ def get_system_stats() -> str:
     output += f"**Total Documents:** {api.get_context_store_size():,}\n\n"
     output += f"**Document Types:** Scientific papers, novels, education, fiction, technical docs\n\n"
 
+    synthesis_summary = get_synthesis_session_summary(api)
+    output += "### 🧪 Synthesis Session Artifacts\n\n"
+    output += f"**Session Artifacts:** {synthesis_summary['total']}\n\n"
+    output += f"**Source Text Entries:** {synthesis_summary['source_text']}\n\n"
+    output += f"**Macro Distillations:** {synthesis_summary['macro_distillation']}\n\n"
+    if synthesis_summary["recent_ids"]:
+        output += "**Most Recent Artifact IDs:**\n"
+        for doc_id in synthesis_summary["recent_ids"]:
+            output += f"- {doc_id}\n"
+        output += "\n"
+
     # Query Performance
     output += "### ⚡ Query Performance\n\n"
     output += f"**Total Queries:** {metrics['retrieval_metrics']['total_queries']}\n\n"
@@ -407,72 +421,175 @@ def synthesize_document(
     document_text: str,
     micro_window_size: int = 5,
     macro_trigger_count: int = 3,
+    add_macro_to_search: bool = False,
+    add_source_to_search: bool = False,
 ) -> str:
-    """Generate a hierarchical synthesis artifact for a document-sized input."""
-    cleaned = document_text.strip()
-    if not cleaned:
-        return "Please provide document text to synthesize."
-
-    fragments: List[Dict[str, Any]] = []
-    units = [part.strip() for part in cleaned.splitlines() if part.strip()]
-    if len(units) < 2:
-        units = [sentence.strip() for sentence in cleaned.split(".") if sentence.strip()]
-
-    for index, unit in enumerate(units, start=1):
-        fragments.append(
-            {
-                "id": f"fragment_{index}",
-                "text": unit,
-                "heat": min(1.0, 0.4 + (len(unit) / 400.0)),
-            }
+    """Generate a synthesis artifact and optionally ingest it into retrieval."""
+    try:
+        artifacts = synthesize_text(
+            document_text=document_text,
+            micro_window_size=micro_window_size,
+            macro_trigger_count=macro_trigger_count,
         )
+    except ValueError as exc:
+        return str(exc)
+
+    output = artifacts.report_markdown
+
+    if add_macro_to_search or add_source_to_search:
+        ingested_ids = ingest_synthesis_artifacts(
+            api=api,
+            document_text=document_text,
+            artifacts=artifacts,
+            include_source=add_source_to_search,
+            include_macro=add_macro_to_search,
+        )
+        if ingested_ids:
+            output += "\n\n### Search Index Update\n\n"
+            output += "The following artifacts were added to the current searchable session index:\n\n"
+            for doc_id in ingested_ids:
+                output += f"- {doc_id}\n"
+            output += "\nYou can now query terms from these artifacts in the Query tab during this app session."
+        else:
+            output += "\n\n### Search Index Update\n\nNo new artifacts were added. This can happen if no macro distillation was produced."
+
+    return output
+
+
+@GPU
+def query_warbler_with_quarantine(
+    query_text: str,
+    max_results: int = 5,
+    use_hybrid: bool = True,
+    generate_answer: bool = True,
+) -> str:
+    """Query with Bob's quarantine reporting."""
+    if not query_text.strip():
+        return "Please enter a query."
+
+    start_time = time.time()
+    hybrid_available = embedding_provider is not None and EMBEDDING_BOOT_LIMIT != 0
+    effective_hybrid = use_hybrid and hybrid_available
+    query_mode = (
+        RetrievalMode.HYBRID_SEMANTIC_FRACTALSTAT
+        if effective_hybrid
+        else RetrievalMode.SEMANTIC_SIMILARITY
+    )
+
+    query = RetrievalQuery(
+        query_id=f"gradio_{int(time.time())}",
+        mode=query_mode,
+        semantic_query=query_text,
+        max_results=max_results * 2,
+        fractalstat_hybrid=effective_hybrid,
+        confidence_threshold=0.2 if not effective_hybrid else 0.3,
+    )
+
+    print(
+        f"DEBUG: Executing query '{query_text}' with mode={query_mode}, "
+        f"max_results={max_results}, hybrid={use_hybrid}"
+    )
+
+    assembly = api.retrieve_context(query)
+    elapsed_ms = (time.time() - start_time) * 1000
+
+    original_results = len(assembly.results)
+    quarantined_results = [result for result in assembly.results if result.conflict_flags]
+    retained_results = [result for result in assembly.results if not result.conflict_flags]
+    final_results = retained_results[:max_results]
+
+    print(
+        f"DEBUG: Query completed in {elapsed_ms:.0f}ms, found {original_results} raw results, "
+        f"Bob quarantined {len(quarantined_results)}, retained {len(final_results)} for display"
+    )
 
-    if not fragments:
-        return "Unable to derive synthesis fragments from the provided text."
+    fallback_triggered = False
+    if effective_hybrid and len(final_results) == 0 and len(quarantined_results) > 0:
+        print("DEBUG: Hybrid retained 0 results, checking quarantined pool for semantic fallback")
+        semantic_query = RetrievalQuery(
+            query_id=f"quarantine_fallback_{int(time.time())}",
+            mode=RetrievalMode.SEMANTIC_SIMILARITY,
+            semantic_query=query_text,
+            max_results=max_results,
+            confidence_threshold=0.3,
+            fractalstat_hybrid=False,
+        )
+        fallback_assembly = api.retrieve_context(semantic_query)
+        fallback_triggered = True
+        final_results = fallback_assembly.results[:max_results]
+        elapsed_ms += (time.time() - start_time - elapsed_ms / 1000) * 1000
+        print(
+            f"DEBUG: Quarantine fallback triggered - retrieved {len(final_results)} from semantic search"
+        )
 
-    ladder = SummarizationLadder(
-        config={
-            "micro_window_size": micro_window_size,
-            "macro_trigger_count": macro_trigger_count,
+    results_data = [
+        {
+            "id": result.content_id,
+            "content": result.content,
+            "relevance_score": result.relevance_score,
+            "semantic_similarity": result.semantic_similarity,
+            "fractalstat_resonance": result.fractalstat_resonance,
+            "metadata": result.metadata,
         }
+        for result in final_results
+    ]
+
+    generated_answer = None
+    if generate_answer:
+        generated_answer = answer_generator.generate_answer(query_text, results_data)
+
+    output = "## Query Results\n\n"
+    output += f"**Query:** {query_text}\n\n"
+    output += f"**Found:** {len(final_results)} results in {elapsed_ms:.0f}ms\n\n"
+    if len(quarantined_results) > 0:
+        output += (
+            f"**Risk Assessment:** {len(quarantined_results)} potentially conflicting results "
+            "quarantined by Bob the Skeptic\n\n"
+        )
+    elif len(final_results) < original_results:
+        output += f"**Risk Assessment:** {original_results - len(final_results)} duplicates removed\n\n"
+    else:
+        output += "**Risk Assessment:** No conflicts detected by Bob the Skeptic\n\n"
+
+    output += f"**Quality Score:** {assembly.assembly_quality:.3f}\n\n"
+
+    if generated_answer:
+        output += "### Generated Answer\n\n"
+        output += f"{generated_answer.answer}\n\n"
+        if generated_answer.citations:
+            citation_line = ", ".join(
+                f"[{citation['index']}] {citation.get('pack') or citation.get('id')}"
+                for citation in generated_answer.citations[:5]
+            )
+            output += f"**Sources:** {citation_line}\n\n"
+        output += f"**Generator:** {generated_answer.provider}\n\n"
+
+    if final_results:
+        output += "### Top Results\n\n"
+        for index, result in enumerate(final_results[:max_results], 1):
+            output += f"**{index}. Score: {result.relevance_score:.3f}**\n\n"
+            output += f"{result.content[:300]}...\n\n"
+            if effective_hybrid and result.fractalstat_resonance is not None:
+                output += f"- Semantic: {result.semantic_similarity:.3f}\n"
+                output += f"- FractalStat: {result.fractalstat_resonance:.3f}\n\n"
+            output += "---\n\n"
+    else:
+        output += "No results found.\n"
+
+    if fallback_triggered:
+        output += "*\\*Note: Used semantic search because hybrid results were quarantined\\*\n"
+
+    return output
+
+
+def query_from_synthesis_tab(query_text: str) -> str:
+    """Run a quick semantic follow-up query from the Synthesis tab."""
+    return query_warbler_with_quarantine(
+        query_text=query_text,
+        max_results=5,
+        use_hybrid=False,
+        generate_answer=True,
     )
-    report = ladder.process_fragments(fragments)
-
-    lines = ["## Synthesis Output", "", f"**Input Fragments:** {len(fragments)}", ""]
-
-    if ladder.macro_distillations:
-        latest_macro = ladder.macro_distillations[-1]
-        lines.extend([
-            "### Macro Distillation",
-            "",
-            latest_macro.distilled_essence,
-            "",
-        ])
-
-    if ladder.micro_summaries:
-        lines.append("### Micro Summaries")
-        lines.append("")
-        for micro in list(ladder.micro_summaries)[-3:]:
-            lines.append(f"- {micro.compressed_text}")
-        lines.append("")
-
-    metrics = ladder.get_compression_metrics()["current_state"]
-    lines.extend([
-        "### Metrics",
-        "",
-        f"- Active Micro Summaries: {metrics['micro_summaries_active']}",
-        f"- Macro Distillations: {metrics['macro_distillations_total']}",
-        f"- Compression Ratio: {metrics['compression_ratio']:.2f}",
-        "",
-    ])
-
-    if report["new_macro_distillations"]:
-        lines.append("### New Distillation IDs")
-        lines.append("")
-        for item in report["new_macro_distillations"]:
-            lines.append(f"- {item['distillation_id']}")
-
-    return "\n".join(lines)
 
 
 # Create Gradio interface
@@ -518,143 +635,6 @@ with gr.Blocks(title="Warbler CDA - FractalStat RAG") as demo:
             with gr.Column():
                 results_output = gr.Markdown(label="Results")
 
-        # Add Bob quarantine info to results
-        def format_results_with_quarantine(output: str, results_count: int, quarantined_count: int) -> str:
-            """Format results with Bob's quarantine information."""
-            if quarantined_count > 0:
-                status_line = f"**Bob the Skeptic**: {quarantined_count} conflicting results quarantined, {results_count} results retained\n\n"
-            elif quarantined_count == 0:
-                status_line = "**Bob the Skeptic**: No conflicting results detected\n\n"
-            else:  # quarantined_count is None when Bob not available
-                status_line = "**Bob the Skeptic**: Conflict detection unavailable\n\n"
-
-            return output.replace("## Query Results\n\n", "## Query Results\n\n" + status_line)
-
-        def query_warbler_with_quarantine(query_text: str, max_results: int = 5, use_hybrid: bool = True, generate_answer: bool = True) -> str:
-            """Query with Bob's quarantine reporting."""
-            if not query_text.strip():
-                return "Please enter a query."
-
-            start_time = time.time()
-            hybrid_available = embedding_provider is not None and EMBEDDING_BOOT_LIMIT != 0
-
-            # Create query - use hybrid mode when requested
-            effective_hybrid = use_hybrid and hybrid_available
-            query_mode = RetrievalMode.HYBRID_SEMANTIC_FRACTALSTAT if effective_hybrid else RetrievalMode.SEMANTIC_SIMILARITY
-
-            query = RetrievalQuery(
-                query_id=f"gradio_{int(time.time())}",
-                mode=query_mode,
-                semantic_query=query_text,
-                max_results=max_results * 2,  # Get extra results for quarantine prioritizing
-                fractalstat_hybrid=effective_hybrid,
-                confidence_threshold=0.2 if not effective_hybrid else 0.3,
-            )
-
-            # DEBUG: Log query details
-            print(f"DEBUG: Executing query '{query_text}' with mode={query_mode}, max_results={max_results}, hybrid={use_hybrid}")
-
-            # Execute query
-            assembly = api.retrieve_context(query)
-
-            elapsed_ms = (time.time() - start_time) * 1000
-
-            # Find quarantined results (those with conflict flags)
-            original_results = len(assembly.results)
-            quarantined_results = [r for r in assembly.results if r.conflict_flags]
-            retained_results = [r for r in assembly.results if not r.conflict_flags]
-
-            # Limit to requested max_results from retained results
-            final_results = retained_results[:max_results]
-
-            # DEBUG: Log results summary with Bob information
-            print(f"DEBUG: Query completed in {elapsed_ms:.0f}ms, "
-                  f"found {original_results} raw results, "
-                  f"Bob quarantined {len(quarantined_results)}, "
-                  f"retained {len(final_results)} for display")
-            if assembly.results:
-                print(f"DEBUG: Top 3 relevance scores: {[r.relevance_score for r in assembly.results[:3]]}")
-                print(f"DEBUG: Confidence threshold was: {query.confidence_threshold}")
-
-            # Hybrid Fallback: If hybrid mode and no final results from retained, try semantic on quarantine pool
-            fallback_triggered = False
-            if effective_hybrid and len(final_results) == 0 and len(quarantined_results) > 0:
-                print(f"DEBUG: Hybrid retained 0 results, checking quarantined pool for semantic fallback")
-                # Create semantic-only query for quarantined content
-                semantic_query = RetrievalQuery(
-                    query_id=f"quarantine_fallback_{int(time.time())}",
-                    mode=RetrievalMode.SEMANTIC_SIMILARITY,
-                    semantic_query=query_text,
-                    max_results=max_results,
-                    confidence_threshold=0.3,
-                    fractalstat_hybrid=False
-                )
-                # Note: In a full implementation, we'd re-query just the quarantined IDs
-                # For now, trigger full semantic search as fallback
-                fallback_assembly = api.retrieve_context(semantic_query)
-                fallback_triggered = True
-                final_results = fallback_assembly.results[:max_results]
-                elapsed_ms += (time.time() - start_time - elapsed_ms/1000) * 1000
-                print(f"DEBUG: Quarantine fallback triggered - retrieved {len(final_results)} from semantic search")
-
-            hybrid_fallback_used = fallback_triggered
-
-            results_data = [
-                {
-                    "id": result.content_id,
-                    "content": result.content,
-                    "relevance_score": result.relevance_score,
-                    "semantic_similarity": result.semantic_similarity,
-                    "fractalstat_resonance": result.fractalstat_resonance,
-                    "metadata": result.metadata,
-                }
-                for result in final_results
-            ]
-
-            generated_answer = None
-            if generate_answer:
-                generated_answer = answer_generator.generate_answer(query_text, results_data)
-
-            # Format results with Bob quarantine info
-            output = f"## Query Results\n\n"
-            output += f"**Query:** {query_text}\n\n"
-            output += f"**Found:** {len(final_results)} results in {elapsed_ms:.0f}ms\n\n"
-            if len(quarantined_results) > 0:
-                output += f"**Risk Assessment:** {len(quarantined_results)} potentially conflicting results quarantined by Bob the Skeptic\n\n"
-            elif len(final_results) < original_results:
-                output += f"**Risk Assessment:** {original_results - len(final_results)} duplicates removed\n\n"
-            else:
-                output += "**Risk Assessment:** No conflicts detected by Bob the Skeptic\n\n"
-
-            output += f"**Quality Score:** {assembly.assembly_quality:.3f}\n\n"
-
-            if generated_answer:
-                output += "### Generated Answer\n\n"
-                output += f"{generated_answer.answer}\n\n"
-                if generated_answer.citations:
-                    citation_line = ", ".join(
-                        f"[{c['index']}] {c.get('pack') or c.get('id')}" for c in generated_answer.citations[:5]
-                    )
-                    output += f"**Sources:** {citation_line}\n\n"
-                output += f"**Generator:** {generated_answer.provider}\n\n"
-
-            if final_results:
-                output += "### Top Results\n\n"
-                for i, result in enumerate(final_results[:max_results], 1):
-                    output += f"**{i}. Score: {result.relevance_score:.3f}**\n\n"
-                    output += f"{result.content[:300]}...\n\n"
-                    if effective_hybrid and result.fractalstat_resonance is not None:
-                        output += f"- Semantic: {result.semantic_similarity:.3f}\n"
-                        output += f"- FractalStat: {result.fractalstat_resonance:.3f}\n\n"
-                    output += "---\n\n"
-            else:
-                output += "No results found.\n"
-
-            if hybrid_fallback_used:
-                output += "*\\*Note: Used semantic search because hybrid results were quarantined\\*\n"
-
-            return output
-
         query_btn.click(  # pylint: disable=E1101
             fn=query_warbler_with_quarantine,
             inputs=[query_input, max_results, use_hybrid, generate_answer_toggle],
@@ -699,15 +679,45 @@ with gr.Blocks(title="Warbler CDA - FractalStat RAG") as demo:
                 step=1,
                 label="Macro Trigger Count",
             )
+        with gr.Row():
+            synthesis_add_macro = gr.Checkbox(
+                label="Add macro distillation to current search session",
+                value=False,
+            )
+            synthesis_add_source = gr.Checkbox(
+                label="Also add original source text to current search session",
+                value=False,
+            )
         synthesis_btn = gr.Button("Synthesize", variant="primary")
         synthesis_output = gr.Markdown(
-            "Paste a document and click **Synthesize** to generate a TLDA-style macro distillation."
+            "Paste a document and click **Synthesize** to generate a TLDA-style macro distillation. Optionally add it to the current search session."
         )
         synthesis_btn.click(
             fn=synthesize_document,
-            inputs=[synthesis_input, synthesis_micro_window, synthesis_macro_trigger],
+            inputs=[
+                synthesis_input,
+                synthesis_micro_window,
+                synthesis_macro_trigger,
+                synthesis_add_macro,
+                synthesis_add_source,
+            ],
             outputs=synthesis_output,
         )
+        gr.Markdown("### Query This Synthesis")
+        synthesis_query_input = gr.Textbox(
+            label="Follow-up Query",
+            placeholder="Ask about the newly added synthesis artifacts without leaving this tab...",
+            lines=2,
+        )
+        synthesis_query_btn = gr.Button("Query Current Session", variant="secondary")
+        synthesis_query_output = gr.Markdown(
+            "If you added synthesis artifacts to search, run a follow-up query here to inspect them immediately."
+        )
+        synthesis_query_btn.click(
+            fn=query_from_synthesis_tab,
+            inputs=[synthesis_query_input],
+            outputs=synthesis_query_output,
+        )
     
     with gr.Tab("About"):
         gr.HTML("""

tests/test_synthesis_bridge.py

@@ -0,0 +1,99 @@
+from warbler_cda.retrieval_api import RetrievalAPI, RetrievalMode, RetrievalQuery
+from warbler_cda.synthesis_bridge import (
+    build_synthesis_fragments,
+    get_synthesis_session_summary,
+    ingest_synthesis_artifacts,
+    synthesize_text,
+)
+
+
+def test_build_synthesis_fragments_prefers_line_structure():
+    fragments = build_synthesis_fragments("line one\n\nline two\nline three")
+
+    assert len(fragments) == 3
+    assert fragments[0]["id"] == "fragment_1"
+    assert fragments[1]["text"] == "line two"
+
+
+
+def test_synthesize_text_generates_macro_for_longer_input():
+    text = "\n".join(
+        [
+            "Ten of Hearts is the weekly prompt.",
+            "Use the card and its symbols as inspiration.",
+            "The jam allows joining late and taking breaks.",
+            "Projects should be submitted within the relevant week.",
+            "AI use is discouraged and should be labeled if used.",
+            "Bigotry and harassment are prohibited.",
+            "New submissions and reworks are allowed.",
+            "Participants should be kind to one another.",
+            "Submissions should connect to the card theme.",
+            "The focus is on making games you want to make.",
+        ]
+    )
+
+    artifacts = synthesize_text(text, micro_window_size=3, macro_trigger_count=2)
+
+    assert artifacts.fragments
+    assert artifacts.macro_distillation is not None
+    assert artifacts.metrics["macro_distillations_total"] >= 1
+    assert "## Synthesis Output" in artifacts.report_markdown
+
+
+
+def test_ingested_synthesis_source_becomes_queryable():
+    api = RetrievalAPI(config={"enable_fractalstat_hybrid": False, "enable_lexical_index": True})
+    text = "\n".join(
+        [
+            "Card game jam entries should fit on a playing card.",
+            "Participants can use Ten of Hearts as inspiration.",
+            "The jam values quantity of output and kindness.",
+            "Submitting within the week is encouraged.",
+            "Decks of cards can be used as a randomizer.",
+        ]
+    )
+    artifacts = synthesize_text(text, micro_window_size=3, macro_trigger_count=2)
+
+    ingested_ids = ingest_synthesis_artifacts(
+        api=api,
+        document_text=text,
+        artifacts=artifacts,
+        include_source=True,
+        include_macro=True,
+    )
+
+    assert len(ingested_ids) >= 1
+    query = RetrievalQuery(
+        query_id="bridge-test",
+        mode=RetrievalMode.SEMANTIC_SIMILARITY,
+        semantic_query="playing card kindness",
+        fractalstat_hybrid=False,
+        confidence_threshold=0.0,
+        max_results=5,
+    )
+
+    assembly = api.retrieve_context(query)
+
+    assert assembly.results
+    assert any(result.content_id in ingested_ids for result in assembly.results)
+
+
+def test_synthesis_session_summary_counts_ingested_artifacts():
+    api = RetrievalAPI(config={"enable_fractalstat_hybrid": False, "enable_lexical_index": True})
+    text = "alpha\nbeta\ngamma\ndelta\nepsilon\nzeta"
+    artifacts = synthesize_text(text, micro_window_size=3, macro_trigger_count=2)
+
+    ingest_synthesis_artifacts(
+        api=api,
+        document_text=text,
+        artifacts=artifacts,
+        include_source=True,
+        include_macro=True,
+    )
+
+    summary = get_synthesis_session_summary(api)
+
+    assert summary["total"] >= 1
+    assert summary["source_text"] == 1
+    assert summary["macro_distillation"] >= 1
+    assert summary["recent_ids"]

warbler_cda/synthesis_bridge.py

@@ -0,0 +1,188 @@
+"""Utilities for TLDA-style synthesis and optional retrieval ingestion."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import hashlib
+import time
+from typing import Any, Dict, List, Optional
+
+from warbler_cda.summarization_ladder import SummarizationLadder
+
+
+@dataclass
+class SynthesisArtifacts:
+    """Structured artifacts produced by a synthesis run."""
+
+    report_markdown: str
+    fragments: List[Dict[str, Any]]
+    micro_summaries: List[str]
+    macro_distillation: Optional[str]
+    metrics: Dict[str, Any]
+
+
+def build_synthesis_fragments(document_text: str) -> List[Dict[str, Any]]:
+    """Split text into fragments suitable for the summarization ladder."""
+    cleaned = document_text.strip()
+    if not cleaned:
+        return []
+
+    units = [part.strip() for part in cleaned.splitlines() if part.strip()]
+    if len(units) < 2:
+        units = [sentence.strip() for sentence in cleaned.split(".") if sentence.strip()]
+
+    fragments: List[Dict[str, Any]] = []
+    for index, unit in enumerate(units, start=1):
+        fragments.append(
+            {
+                "id": f"fragment_{index}",
+                "text": unit,
+                "heat": min(1.0, 0.4 + (len(unit) / 400.0)),
+            }
+        )
+
+    return fragments
+
+
+def synthesize_text(
+    document_text: str,
+    micro_window_size: int = 5,
+    macro_trigger_count: int = 3,
+) -> SynthesisArtifacts:
+    """Generate hierarchical synthesis artifacts and a markdown report."""
+    fragments = build_synthesis_fragments(document_text)
+    if not fragments:
+        raise ValueError("Please provide document text to synthesize.")
+
+    ladder = SummarizationLadder(
+        config={
+            "micro_window_size": micro_window_size,
+            "macro_trigger_count": macro_trigger_count,
+        }
+    )
+    report = ladder.process_fragments(fragments)
+
+    micro_summaries = [summary.compressed_text for summary in list(ladder.micro_summaries)]
+    macro_distillation = (
+        ladder.macro_distillations[-1].distilled_essence if ladder.macro_distillations else None
+    )
+    metrics = ladder.get_compression_metrics()["current_state"]
+
+    lines = ["## Synthesis Output", "", f"**Input Fragments:** {len(fragments)}", ""]
+
+    if macro_distillation:
+        lines.extend(["### Macro Distillation", "", macro_distillation, ""])
+
+    if micro_summaries:
+        lines.append("### Micro Summaries")
+        lines.append("")
+        for micro in micro_summaries[-3:]:
+            lines.append(f"- {micro}")
+        lines.append("")
+
+    lines.extend(
+        [
+            "### Metrics",
+            "",
+            f"- Active Micro Summaries: {metrics['micro_summaries_active']}",
+            f"- Macro Distillations: {metrics['macro_distillations_total']}",
+            f"- Compression Ratio: {metrics['compression_ratio']:.2f}",
+            "",
+        ]
+    )
+
+    if report["new_macro_distillations"]:
+        lines.append("### New Distillation IDs")
+        lines.append("")
+        for item in report["new_macro_distillations"]:
+            lines.append(f"- {item['distillation_id']}")
+
+    return SynthesisArtifacts(
+        report_markdown="\n".join(lines),
+        fragments=fragments,
+        micro_summaries=micro_summaries,
+        macro_distillation=macro_distillation,
+        metrics=metrics,
+    )
+
+
+def ingest_synthesis_artifacts(
+    api: Any,
+    document_text: str,
+    artifacts: SynthesisArtifacts,
+    include_source: bool = False,
+    include_macro: bool = True,
+) -> List[str]:
+    """Persist source and/or synthesis artifacts into the retrieval context store."""
+    ingested_ids: List[str] = []
+    timestamp = int(time.time() * 1000)
+    content_hash = hashlib.sha256(document_text.encode()).hexdigest()[:12]
+    base_metadata = {
+        "source": "synthesis_tab",
+        "content_hash": content_hash,
+        "artifact_type": "synthesis_session",
+    }
+
+    if include_source:
+        source_id = f"synthesis-source-{timestamp}-{content_hash}"
+        added = api.add_document(
+            source_id,
+            document_text,
+            {
+                **base_metadata,
+                "artifact_type": "source_text",
+                "fragment_count": len(artifacts.fragments),
+            },
+            auto_embed=False,
+        )
+        if added:
+            ingested_ids.append(source_id)
+
+    if include_macro and artifacts.macro_distillation:
+        macro_id = f"synthesis-macro-{timestamp}-{content_hash}"
+        added = api.add_document(
+            macro_id,
+            artifacts.macro_distillation,
+            {
+                **base_metadata,
+                "artifact_type": "macro_distillation",
+                "fragment_count": len(artifacts.fragments),
+                "micro_summary_count": len(artifacts.micro_summaries),
+            },
+            auto_embed=False,
+        )
+        if added:
+            ingested_ids.append(macro_id)
+
+    return ingested_ids
+
+
+def get_synthesis_session_summary(api: Any) -> Dict[str, Any]:
+    """Summarize synthesis artifacts currently stored in the active retrieval session."""
+    context_store = getattr(api, "_context_store", {}) or {}
+
+    summary = {
+        "total": 0,
+        "source_text": 0,
+        "macro_distillation": 0,
+        "recent_ids": [],
+    }
+
+    synthesis_entries = []
+    for doc_id, doc_data in context_store.items():
+        metadata = doc_data.get("metadata", {})
+        if metadata.get("source") != "synthesis_tab":
+            continue
+
+        summary["total"] += 1
+        artifact_type = metadata.get("artifact_type")
+        if artifact_type == "source_text":
+            summary["source_text"] += 1
+        elif artifact_type == "macro_distillation":
+            summary["macro_distillation"] += 1
+
+        synthesis_entries.append((doc_data.get("added_at", 0), doc_id))
+
+    synthesis_entries.sort(reverse=True)
+    summary["recent_ids"] = [doc_id for _, doc_id in synthesis_entries[:5]]
+    return summary