---
estimated_steps: 43
estimated_files: 6
skills_used: []
---

# T02: Prompt templates and pipeline stages 2-5 with orchestrator

Write the 4 prompt template files for stages 2-5 and implement all pipeline stage logic as Celery tasks in `pipeline/stages.py`. Each stage reads from PostgreSQL, loads its prompt template, calls the LLM client, parses the response, writes results back to PostgreSQL, and updates processing_status. The `run_pipeline` orchestrator chains stages and handles resumability.

## Failure Modes

| Dependency | On error | On timeout | On malformed response |
|------------|----------|-----------|----------------------|
| LLM API (primary) | Fall back to secondary endpoint | Fall back to secondary endpoint | Log raw response, retry once with 'output valid JSON' nudge, then fail stage |
| PostgreSQL | Task fails, Celery retries (max 3) | Task fails, Celery retries | N/A |
| Prompt template file | Task fails immediately with FileNotFoundError logged | N/A | N/A |

## Negative Tests

- **Malformed inputs**: LLM returns non-JSON text → caught by parse_response, logged with raw text
- **Error paths**: LLM returns valid JSON but wrong schema → Pydantic ValidationError caught, logged
- **Boundary conditions**: Empty transcript (0 segments) → stage 2 returns empty segmentation, stages 3-5 skip gracefully

## Steps

1. Create `prompts/stage2_segmentation.txt`: System prompt instructing the LLM to analyze transcript segments and identify topic boundaries. Input: full transcript text with segment indices. Output: JSON matching `SegmentationResult` schema — list of topic segments with start_index, end_index, topic_label, summary. Use XML-style `<transcript>` tags to fence user content from instructions.

2. Create `prompts/stage3_extraction.txt`: System prompt for extracting key moments from a topic segment. Input: segment text with timestamps. Output: JSON matching `ExtractionResult` — list of moments with title, summary, timestamps, content_type (technique/settings/reasoning/workflow), plugins mentioned, raw_transcript excerpt.

3. Create `prompts/stage4_classification.txt`: System prompt for classifying key moments against the canonical tag taxonomy. Input: moment title + summary + list of canonical categories/sub-topics from `canonical_tags.yaml`. Output: JSON matching `ClassificationResult` — topic_category, topic_tags for each moment.

4. Create `prompts/stage5_synthesis.txt`: System prompt for synthesizing a technique page from key moments. Input: all moments for a creator+topic group. Output: JSON matching `SynthesisResult` — title, slug, summary, body_sections (structured prose with sub-aspects), signal_chains, plugins, source_quality assessment.

5. Create `backend/pipeline/stages.py` with these components:
   a. Helper `_load_prompt(template_name: str) -> str` — reads from `settings.prompts_path / template_name`
   b. Helper `_get_sync_session()` — creates a sync SQLAlchemy session for Celery tasks using `create_engine()` (sync, not async) with the DATABASE_URL converted from `postgresql+asyncpg://` to `postgresql+psycopg2://` (or use `sqlalchemy.create_engine` with sync URL). **CRITICAL**: Celery is sync, so use `sqlalchemy.orm.Session`, not async.
   c. `@celery_app.task(bind=True, max_retries=3)` for `stage2_segmentation(self, video_id: str)`: Load transcript segments from DB ordered by segment_index, build prompt with full text, call LLM, parse SegmentationResult, update `topic_label` on each TranscriptSegment row.
   d. `@celery_app.task(bind=True, max_retries=3)` for `stage3_extraction(self, video_id: str)`: Group segments by topic_label, for each group call LLM to extract moments, create KeyMoment rows in DB, set `processing_status = extracted` on SourceVideo.
   e. `@celery_app.task(bind=True, max_retries=3)` for `stage4_classification(self, video_id: str)`: Load key moments, load canonical tags from `config/canonical_tags.yaml`, call LLM to classify, update topic_tags on moments. Stage 4 does NOT change processing_status.
   f. `@celery_app.task(bind=True, max_retries=3)` for `stage5_synthesis(self, video_id: str)`: Group key moments by (creator, topic_category), for each group call LLM to synthesize, create/update TechniquePage rows with body_sections, signal_chains, plugins, topic_tags, summary. Link KeyMoments to their TechniquePage via technique_page_id. Set `processing_status = reviewed` (or `published` if `settings.review_mode is False`).
   g. `@celery_app.task` for `run_pipeline(video_id: str)`: Check current processing_status, chain the appropriate stages (e.g., if `transcribed`, chain 2→3→4→5; if `extracted`, chain 4→5). Use `celery.chain()` for sequential execution.

6. Import `stages` module in `worker.py` to register tasks (ensure the import line `from pipeline import stages` is present after celery_app creation).

## Must-Haves

- [ ] 4 prompt template files in `prompts/` with clear instruction/data separation
- [ ] All 5 stage tasks + run_pipeline orchestrator defined in stages.py
- [ ] Celery tasks use SYNC SQLAlchemy sessions (not async)
- [ ] Stage 2 updates topic_label on TranscriptSegment rows
- [ ] Stage 3 creates KeyMoment rows and sets processing_status=extracted
- [ ] Stage 4 loads canonical_tags.yaml and classifies moments
- [ ] Stage 5 creates TechniquePage rows and sets processing_status=reviewed/published
- [ ] run_pipeline handles resumability based on current processing_status
- [ ] Prompts fence user content with XML-style tags

## Verification

- `test -f prompts/stage2_segmentation.txt && test -f prompts/stage3_extraction.txt && test -f prompts/stage4_classification.txt && test -f prompts/stage5_synthesis.txt` — all 4 prompt files exist
- `cd backend && python -c "from pipeline.stages import run_pipeline, stage2_segmentation, stage3_extraction, stage4_classification, stage5_synthesis; print('all stages imported')"` — imports succeed
- `cd backend && python -c "from worker import celery_app; print([t for t in celery_app.tasks if 'stage' in t or 'pipeline' in t])"` — shows registered tasks

## Observability Impact

- Signals added: INFO log at start/end of each stage with video_id and duration, WARNING on LLM fallback, ERROR on parse failure with raw response excerpt
- How a future agent inspects this: query `source_videos.processing_status` to see pipeline progress, check Celery worker logs for stage timing
- Failure state exposed: video stays at last successful processing_status, error logged with stage name and video_id

## Inputs

- ``backend/pipeline/schemas.py` — Pydantic models for stage I/O from T01`
- ``backend/pipeline/llm_client.py` — LLMClient with primary/fallback from T01`
- ``backend/config.py` — Settings with prompts_path, review_mode, llm_* fields from T01`
- ``backend/worker.py` — Celery app instance from T01`
- ``backend/models.py` — ORM models (TranscriptSegment, KeyMoment, TechniquePage, SourceVideo, ProcessingStatus)`
- ``config/canonical_tags.yaml` — canonical tag taxonomy for stage 4 classification`

## Expected Output

- ``prompts/stage2_segmentation.txt` — topic boundary detection prompt template`
- ``prompts/stage3_extraction.txt` — key moment extraction prompt template`
- ``prompts/stage4_classification.txt` — classification/tagging prompt template`
- ``prompts/stage5_synthesis.txt` — technique page synthesis prompt template`
- ``backend/pipeline/stages.py` — 5 Celery tasks (stages 2-5 + run_pipeline orchestrator)`
- ``backend/worker.py` — updated with pipeline.stages import for task registration`

## Verification

test -f prompts/stage2_segmentation.txt && test -f prompts/stage3_extraction.txt && test -f prompts/stage4_classification.txt && test -f prompts/stage5_synthesis.txt && cd backend && python -c "from pipeline.stages import run_pipeline, stage2_segmentation, stage3_extraction, stage4_classification, stage5_synthesis; print('all stages imported')"
