---
estimated_steps: 67
estimated_files: 1
skills_used: []
---

# T01: Add compose helpers and wire compose-or-create logic into stage5_synthesis

Add two helper functions to stages.py and modify the stage5_synthesis per-category loop to detect existing pages and branch to the compose path.

## Steps

1. Add `TechniquePageVideo` to the imports from `models` at line ~27.

2. Add `_build_compose_user_prompt(existing_page, existing_moments, new_moments, creator_name)` helper function:
   - Takes an existing `TechniquePage` ORM object, a list of `KeyMoment` ORM objects (existing), a list of `(KeyMoment, dict)` tuples (new moments with classification), and creator name string
   - Serialize existing page to dict matching SynthesizedPage shape: title, slug, topic_category, summary, body_sections, signal_chains, plugins, source_quality
   - Format existing moments as `[0]-[N-1]` using `_build_moments_text()` pattern but from plain KeyMoment objects (not tuples with cls_info — existing moments don't have classification data, use empty dict)
   - Format new moments as `[N]-[N+M-1]` using `_build_moments_text()` with offset indices applied
   - Build XML-tagged user prompt: `<existing_page>`, `<existing_moments>`, `<new_moments>`, `<creator>` tags (same structure as test_harness.py's `build_compose_prompt()`)
   - Return the user prompt string

3. Add `_compose_into_existing(existing_page, existing_moments, new_moment_group, category, creator_name, system_prompt, llm, model_override, modality, hard_limit, video_id, run_id)` helper function:
   - Load compose system prompt via `_load_prompt('stage5_compose.txt', video_id=video_id)`
   - Call `_build_compose_user_prompt()` to build user prompt
   - Estimate tokens via `estimate_max_tokens()`
   - Call `llm.complete()` with compose system prompt, response_model=SynthesisResult, same callback/param pattern as `_synthesize_chunk()`
   - Parse via `_safe_parse_llm_response()` and return SynthesisResult

4. Modify the per-category loop in `stage5_synthesis()` (around line 1200):
   - **Before** the existing chunked synthesis block, add compose detection:
     ```
     existing_page = session.execute(
         select(TechniquePage).where(
             TechniquePage.creator_id == video.creator_id,
             func.lower(TechniquePage.topic_category) == func.lower(category),
         )
     ).scalars().first()
     ```
   - If `existing_page` is found, load its linked moments:
     ```
     existing_moments = session.execute(
         select(KeyMoment)
         .where(KeyMoment.technique_page_id == existing_page.id)
         .order_by(KeyMoment.start_time)
     ).scalars().all()
     ```
   - If existing_page AND existing_moments → compose path: call `_compose_into_existing()`, use result.pages as synthesized_pages
   - Log INFO: 'Stage 5: Composing into existing page \'%s\' (%d existing moments + %d new moments)'
   - If >1 page matches, log WARNING about multiple matches and proceed with first
   - If no existing_page → fall through to existing synthesis block (unchanged)
   - Wrap in `else` so existing chunked synthesis only runs when not composing

5. In the persist block (around line 1380), after the `if existing:` / `else:` branch that creates/updates the page:
   - Set `page.body_sections_format = 'v2'` on every page (both new and updated)
   - Add TechniquePageVideo INSERT:
     ```python
     from sqlalchemy.dialects.postgresql import insert as pg_insert
     stmt = pg_insert(TechniquePageVideo.__table__).values(
         technique_page_id=page.id,
         source_video_id=video.id,
     ).on_conflict_do_nothing()
     session.execute(stmt)
     ```

## Must-Haves

- [ ] `_build_compose_user_prompt()` produces XML-tagged prompt with correct offset indices
- [ ] `_compose_into_existing()` calls LLM with compose system prompt and returns SynthesisResult
- [ ] Compose-or-create decision queries DB by creator_id + LOWER(topic_category)
- [ ] Existing synthesis path unchanged when no existing page found
- [ ] body_sections_format = 'v2' set on all pages
- [ ] TechniquePageVideo row inserted for every page+video combination
- [ ] Case-insensitive category matching (func.lower)

## Failure Modes

| Dependency | On error | On timeout | On malformed response |
|------------|----------|-----------|----------------------|
| LLM (compose) | _safe_parse_llm_response retries once, then raises (existing retry mechanism) | Celery task retry (max_retries=3) | SynthesisResult validation rejects, retry with feedback |
| DB (existing page query) | Exception propagates to stage-level handler, triggers retry | Same | N/A |

## Negative Tests

- No existing page for creator+category → falls through to standard synthesis (no compose)
- Existing page found but zero linked moments → should still compose (empty existing_moments list)
- Multiple pages match creator+category → uses first, logs warning

## Inputs

- ``backend/pipeline/stages.py` — existing stage5_synthesis function to modify`
- ``backend/pipeline/test_harness.py` — reference implementation of build_compose_prompt() for prompt structure`
- ``backend/models.py` — TechniquePageVideo model, body_sections_format column`
- ``prompts/stage5_compose.txt` — compose system prompt loaded by _compose_into_existing()`
- ``backend/pipeline/schemas.py` — SynthesisResult schema for LLM response parsing`

## Expected Output

- ``backend/pipeline/stages.py` — modified with _build_compose_user_prompt(), _compose_into_existing(), compose-or-create branch, TechniquePageVideo INSERT, body_sections_format='v2'`

## Verification

cd /home/aux/projects/content-to-kb-automator && python -c "from pipeline.stages import _build_compose_user_prompt, _compose_into_existing; print('imports OK')" && grep -q 'body_sections_format' backend/pipeline/stages.py && grep -q 'TechniquePageVideo' backend/pipeline/stages.py && grep -q 'stage5_compose' backend/pipeline/stages.py
