kennyragskill.py

"""KennyRAG: 3-step bulk search across blog-post dialogs - folder -> sections -> content, one kernel call per step.

KennyRAG finds content across many blog-post dialogs (Solveit .ipynb notebooks)
without loading whole dialog files. Three async functions narrow the search in three
stages: "which dialogs in this folder?" -> "which sections in these dialogs?" ->
"full text of the chosen sections." A search that would otherwise need 40,000-100,000
tokens of dialog loading costs roughly 5,000-7,000 tokens with this pipeline.

All three functions are async: every call needs await.

## The three functions and their granularity


| Function | Granularity | Purpose | Returns |
|---|---|---|---|
| get_first_notes(folder, depth=1, limit=1) | whole folder in ONE call | first note (intro summary) of every dialog | dict {dialog_name: summary} |
| get_section_summaries(dname, level=2) | ONE dialogue per call | every section heading + summary paragraph | str (joined by ---), or None |
| get_section_content(dname, title) | ONE section per call | full content of one section | str, or None |


get_first_notes is already batch by design: it takes a folder and processes every
dialog in it in a single call. The other two are deliberately single-target tools:
one call handles exactly one dialogue, or exactly one section inside one dialogue.
Calling them one by one for many targets is extremely inefficient - so stages 2 and 3
wrap them in bulk loops, as shown below.

## The core rule: one kernel call per stage

Every stage runs as ONE single top-level code block submitted to the Python kernel.
All repetitive work happens inside loops within that block. Never call the functions
one at a time from separate cells - that multiplies kernel round-trips and tokens.
This loop-based design is what makes KennyRAG fast and cheap.

## The three-stage optimized workflow

### Stage 1 - get_first_notes: filter the folder

One call scans the whole folder and returns {dialog_name: intro_summary} for every
dialog. Read the summaries and pick the interesting dialogues.

```py
notes = await get_first_notes('/visual_premise_cards/GeorgePolyaMath/public/posts', depth=1)
# agent reads notes, then picks, e.g.:
interesting = ['crw03_beggar_coding_root', 'crw05_pyskills_walkthru']
```

### Stage 2 - one loop over all interesting dialogues: find valuable sections

Write ONE loop block that calls get_section_summaries inside the loop for every
interesting dialogue, and submit the whole block as a single kernel call. The agent
reads the one concatenated output and decides which sections are valuable, then
builds the structured dict {dialog: [section titles]} that feeds stage 3.
```py
target_dlg = ['crw03_beggar_coding_root', 'crw05_pyskills_walkthru']  # from stage 1
base = '/visual_premise_cards/GeorgePolyaMath/public/posts'
for dlg in target_dlg:
    summ = await get_section_summaries(f'{base}/{dlg}', level=2)
    if summ:
        print(f"=== {dlg} ===\n{summ}\n")
# agent reads all summaries, then records the valuable sections:
# sec_by_dlg = {'crw03_beggar_coding_root': ['## The Original Claim',
#                                            '## The Mechanism of Collapse'],
#               'crw05_pyskills_walkthru': ['## The two safety dials']}
```

The kernel call happens once; the function runs many times inside the loop. All
results come back in one unified output.

### Stage 3 - double nested loop: extract full section contents

Take the dict from stage 2 and run ONE block with two nested loops: the outer loop
over dialogues, the inner loop over that dialogue's section titles. Call
get_section_content in the inner loop. All extracted contents aggregate into one
unified list for final synthesis.
```py
target = {'crw03_beggar_coding_root': ['## The Original Claim',
                                        '## The Mechanism of Collapse'],
            'crw05_pyskills_walkthru': ['## The two safety dials']}
base = '/visual_premise_cards/GeorgePolyaMath/public/posts'
all_content = []
for dlg, titles in target.items():
    for t in titles:
        content = await get_section_content(f'{base}/{dlg}', t)
        if content:
            all_content.append(f"### {dlg} - {t}\n{content}")
final = '\n\n'.join(all_content)   # one unified payload for the agent to synthesize
```

## Function details

### get_first_notes(folder, depth=1, limit=1)

Scans a folder and returns the first note from every dialog in it (not subfolders).
The first note of a blog-post dialog is its intro paragraph, summarizing the whole
post in 1-3 sentences - enough to judge relevance.

Parameters:
- folder (str): path to scan, relative to the data root (use leading /)
- depth (int): how many folder levels to scan (default 1 = the immediate folder)
- limit (int): how many notes to return per dialog (default 1)

Returns: dict of {dialog_name: first_note_summary}. Empty dict if no dialogs.

Example:

```py
notes = await get_first_notes('/visual_premise_cards/GeorgePolyaMath/public/posts', depth=1)
for name, summary in notes.items():
    print(f"{name}: {summary[:100]}")
```

### get_section_summaries(dname, level=2)

Returns one block of text: every section heading (at the given level) with the
summary paragraph directly below each heading, sections separated by ---. This is
the table of contents of ONE dialog: enough to decide which sections matter,
without loading section bodies.

Parameters:
- dname (str): dialog name/path (use leading / for absolute paths, no .ipynb)
- level (int): heading level to find (default 2 for '## '; pass 1 for '# ')

Returns: str with headings+summaries joined by ---, or None if no headings found.

Example:

```py
summaries = await get_section_summaries('/visual_premise_cards/GeorgePolyaMath/public/posts/crw03_beggar_coding_root', level=2)
if summaries is None:
    print("No level-2 headings - try level=1")
```

Output looks like:
```md
## The Original Claim
In that post, I defined beggar coding by what it looks like...
---
## The Mechanism of Collapse
It doesn't happen all at once...
```

### get_section_content(dname, title)

Extracts the full content of ONE section: all consecutive notes from the heading
note until the next heading at the same level. Use it after stage 2 has told you
which section you want. The title must match exactly, including the heading prefix
(e.g. '## '). The heading level is auto-detected from the number of # signs in the
title (minimum level 2).

Parameters:
- dname (str): dialog name/path
- title (str): exact section heading, e.g. '## What Escaping Actually Requires'

Returns: plain-text string with the full section content, or None if not found.

Example:

```py
content = await get_section_content('/visual_premise_cards/GeorgePolyaMath/public/posts/crw03_beggar_coding_root', '## What Escaping Actually Requires')
if content is not None:
    print(content)
```

## Complete usage demo
```py
# Stage 1 - which dialogs are relevant?
notes = await get_first_notes('/visual_premise_cards/GeorgePolyaMath/public/posts', depth=1)

# Stage 2 - which sections in the chosen dialogs? (one loop, one kernel call)
target_dlg = ['crw03_beggar_coding_root', 'crw05_pyskills_walkthru']
base = '/visual_premise_cards/GeorgePolyaMath/public/posts'
for dlg in target_dlg:
    summ = await get_section_summaries(f'{base}/{dlg}', level=2)
    if summ: print(f"=== {dlg} ===\n{summ}\n")

# Stage 3 - pull the exact sections (nested loop, one kernel call)
target = {'crw03_beggar_coding_root': ['## What Escaping Actually Requires']}
all_content = []
for dlg, titles in target.items():
    for t in titles:
        content = await get_section_content(f'{base}/{dlg}', t)
        if content: all_content.append(content)
```

## Main use cases

1. Answering questions about specific blog content (RAG): run stages 1-3, then
   answer from the retrieved sections.
2. Building a digest of a whole folder: use stage 1 for one-line post summaries,
   then stage 2 to expand the relevant posts into section outlines.
3. Cross-referencing: run stage 2 on several dialogs in one loop and compare
   section titles across dialogs.
4. Content screening: pull exact sections (stage 3) for human review or before
   republishing.
5. Feeding a long-term memory index: record which folders, dialogs, and sections
   were actually accessed (see the separate long-term memory skill), and use that
   slim JSON to prioritize the most-used content in future KennyRAG searches.

## Token efficiency


| Stage | What it reads | Estimated tokens |
|---|---|---|
| get_first_notes | first 100 tokens of each dialog | ~500 total |
| get_section_summaries | heading notes only (no code, no output) | ~3,000-5,000 |
| get_section_content | one section's full content | ~500-1,500 |


Without these tools, finding one section means loading whole dialogs (often
40,000-100,000 tokens each). The pipeline reduces this to roughly 5,000-7,000
tokens - and the one-kernel-call-per-stage rule keeps the tool-call overhead at
exactly three round-trips per search.

## Error handling


| Situation | Behavior |
|---|---|
| No dialogs in folder | get_first_notes returns empty dict |
| No headings at the level | get_section_summaries returns None |
| Title not found | get_section_content returns None |
| dname does not exist | underlying find_msgs warns; function returns empty/None |


Always check for None before using a return value:
```py
summaries = await get_section_summaries(dname)
if summaries is None:
    print("No sections found")
```

## Path conventions

- Leading / = absolute path from the Solveit data root: /visual_premise_cards/folder
- No leading / = relative to the current dialog's folder
- Never use full filesystem paths (e.g. /app/data/...)
- Dialog names never include the .ipynb extension

## Common pitfalls

1. Wrong heading level: if get_section_summaries returns None, check whether the
   dialog uses '## ' (level 2) or '# ' (level 1) headings and pass level accordingly.
2. Trailing whitespace in title: get_section_content matches titles exactly; copy
   titles without trailing spaces or newlines.
3. Forgetting await: all three functions are async - every call needs await.
4. Calling functions one at a time from separate cells: this breaks the one-kernel-
   call-per-stage rule and defeats the whole point of the loop-based workflow.
5. get_section_content title must include the heading prefix ('## ') exactly.
"""

__all__ = ['get_first_notes', 'get_section_summaries', 'get_section_content']

from dialoghelper.core import find_msgs, list_dialogs

async def get_first_notes(folder, depth=1, limit=1):
    """Get the first note from every dialog in a folder (not subfolders).

    Returns {dialog_name: first_note_content}. Use the summaries to pick
    which dialogs deserve a deeper look in stage 2.
    """
    result = await list_dialogs(subpath=folder, depth=depth)
    items = result.get('items', [])
    results = {}
    for name in items:
        if name.endswith('/'):  # skip subfolders
            continue
        msgs = await find_msgs(msg_type='note', dname=f"/{folder}/{name}", limit=limit)
        if msgs:
            results[name] = msgs[0]['content']
    return results


async def get_section_summaries(dname, level=2):
    """One block of text: every section heading + its summary paragraph, joined by ---.

    Returns None if no headings found at the given level.
    """
    pattern = f'^{"#" * level} [^#]'
    headings = await find_msgs(dname=dname, re_pattern=pattern, msg_type='note')
    if not headings:
        return None
    return '\n\n---\n\n'.join(h['content'] for h in headings)


async def get_section_content(dname, title):
    """Full content of one section (all notes from the heading to the next heading).

    Returns plain text, or None if the title is not found. Title must match
    exactly, including the '## ' prefix; heading level is auto-detected.
    """
    level = len(title) - len(title.lstrip('#'))
    level = max(level, 2)
    result = await get_sections_from_dialog(
        dname=dname, sections=[title], exact_match=True,
        level=level, as_xml=True,
    )
    if not result or result.strip() == '<sections></sections>':
        return None
    return result.replace('<sections>', '').replace('</sections>', '').strip()


async def get_sections_from_dialog(
    dname:str=None,              # Dialog name (None = current). Path relative to solveit root if starts with '/', else relative to current dialog
    sections:list=None,          # List of indices (int) or titles (str) to select. None = all sections. Supports negative indices (-1 = last). Examples: [0, 2], ['# My Section'], [0, -1, '# Title']
    level:int=1,                 # Heading level to find (1 = H1 '# ', 2 = H2 '## ', 3 = H3 '### ')
    as_xml:bool=False,           # If True, return XML string for AI context (no prints). If False, print XML for human (return None)
    exact_match:bool=False,      # If True, title must match exactly. If False (default), substring matching finds ALL matches (e.g., 'Moved' finds all sections titles containing 'Moved')
    skip_sections:list=None,     # List of section indices to skip. None = use auto_skip_template. Use [] to skip nothing
    auto_skip_template:bool=True # Auto-detect and skip template heading (matches '# Template (synced' or '# My Root TEMPLATE')
)->str|None:                     # Returns XML string wrapped in <sections> if as_xml=True, else None (prints for human)
    """Extract content for specific sections from any dialog.

    Finds all headings at specified level, applies skip filter, selects by index or title,
    then uses find_msgs(header_section=...) to extract full section content.
    Useful for loading specific sections into AI context.

    Examples:
        get_sections_from_dialog(sections=[0, -1], as_xml=True)  # First and last (after skip) as XML
        get_sections_from_dialog(sections=['# My Section'], as_xml=True)  # By title substring
        get_sections_from_dialog(sections=['Moved'])  # Substring match — finds ALL section titles containing 'Moved'
        get_sections_from_dialog(sections=['# Exact Title'], exact_match=True)  # Exact match — finds only exact title
        get_sections_from_dialog(skip_sections=[])  # Include template section
        get_sections_from_dialog(skip_sections=[0, -1])  # Skip first AND last sections
        get_sections_from_dialog(dname='/myGems/other_dialog', sections=[1, 2])  # From another dialog
        get_sections_from_dialog(level=2)  # Get H2 headings instead of H1
    """
    
    # Helper: detect template heading
    def _is_template(title):
        return title.startswith('# Template (synced') or title.startswith('# My Root TEMPLATE')
    
    # STEP 1: Find all headings at specified level
    pattern = f'^{"#"*level} [^#]'
    headings = [m for m in await find_msgs(dname=dname, re_pattern=pattern, msg_type='note')]
    
    if not as_xml: print(f"📚 Found {len(headings)} level-{level} headings in dialog")
    
    if not headings: 
        if as_xml: return '<sections></sections>'
        print("⚠️ No headings found"); return None
    
    # STEP 2: Apply skip filter
    if skip_sections is not None:
        if skip_sections:
            skip_indices = set()
            for s in skip_sections:
                if s < 0: skip_indices.add(len(headings) + s)
                else: skip_indices.add(s)
            headings = [h for i, h in enumerate(headings) if i not in skip_indices]
            if not as_xml: print(f"⏭️ Skipped {len(skip_indices)} section(s), {len(headings)} remaining")
    elif auto_skip_template:
        original_count = len(headings)
        headings = [h for h in headings if not _is_template(h['content'].strip().split('\n')[0])]
        skipped = original_count - len(headings)
        if skipped and not as_xml: print(f"⏭️ Auto-skipped template section, {len(headings)} remaining")
    
    if not headings:
        if as_xml: return '<sections></sections>'
        print("⚠️ No headings remaining after skip"); return None
    
    # STEP 3: Select sections by index or title
    if sections is None: selected = headings
    else:
        selected = []
        for s in sections:
            if isinstance(s, int): selected.append(headings[s])
            else:
                for h in headings:
                    title = h['content'].strip().split('\n')[0]
                    if (exact_match and s == title) or (not exact_match and s in title):
                        selected.append(h)
                        if exact_match: break  # Only break for exact match
    
    if not selected:
        if as_xml: return '<sections></sections>'
        print(f"⚠️ No matching sections found for: {sections}"); return None
    
    if not as_xml: print(f"📋 Selected {len(selected)} section(s)")
    
    # STEP 4: Extract content and build XML
    xml_parts = []
    
    for h in selected:
        title = h['content'].strip().split('\n')[0]
        
        section_xml = await find_msgs(dname=dname, header_section=title, as_xml=True)
        section_xml_clean = section_xml.replace('<msgs>', '').replace('</msgs>', '').strip()
        xml_parts.append(section_xml_clean)
        
        if not as_xml:
            section_msgs = await find_msgs(dname=dname, header_section=title)
            notes = sum(1 for m in section_msgs if m.get('msg_type')=='note')
            codes = sum(1 for m in section_msgs if m.get('msg_type')=='code')
            prompts = sum(1 for m in section_msgs if m.get('msg_type')=='prompt')
            print(f"  → {title[:60]}... | 📝{notes} notes, 💻{codes} code, 💬{prompts} prompts")
    
    # STEP 5: Return or print XML
    xml_output = f"<sections>\n{chr(10).join(xml_parts)}\n</sections>"
    
    if as_xml:
        return xml_output
    
    print("-" * 60)
    print(xml_output)
    return None