# S03: LLM Extraction Pipeline + Qdrant Integration

**Goal:** A transcript JSON triggers stages 2-5: segmentation → extraction → classification → synthesis. Technique pages with key moments appear in DB. Qdrant has searchable embeddings. Pipeline is resumable per-video per-stage, uses configurable LLM endpoints with fallback, and loads prompt templates from disk.
**Demo:** After this: A transcript JSON triggers stages 2-5: segmentation → extraction → classification → synthesis. Technique pages with key moments appear in DB. Qdrant has searchable embeddings.

## Tasks
- [x] **T01: Extended Settings with 12 LLM/embedding/Qdrant config fields, created Celery app, built sync LLMClient with primary/fallback logic, and defined 8 Pydantic schemas for pipeline stages 2-5** — Extend Settings with LLM/embedding/Qdrant/prompt config vars, add openai and qdrant-client to requirements.txt, create the Celery app in worker.py, build the sync LLM client with primary/fallback logic, and define Pydantic schemas for all pipeline stage inputs/outputs. This task establishes all infrastructure the pipeline stages depend on.

## Steps

1. Add `openai>=1.0,<2.0`, `qdrant-client>=1.9,<2.0`, and `pyyaml>=6.0,<7.0` to `backend/requirements.txt`.

2. Extend `backend/config.py` `Settings` class with these fields (all with sensible defaults from .env.example):
   - `llm_api_url: str = 'http://localhost:11434/v1'`
   - `llm_api_key: str = 'sk-placeholder'`
   - `llm_model: str = 'qwen2.5:14b-q8_0'`
   - `llm_fallback_url: str = 'http://localhost:11434/v1'`
   - `llm_fallback_model: str = 'qwen2.5:14b-q8_0'`
   - `embedding_api_url: str = 'http://localhost:11434/v1'`
   - `embedding_model: str = 'nomic-embed-text'`
   - `embedding_dimensions: int = 768`
   - `qdrant_url: str = 'http://localhost:6333'`
   - `qdrant_collection: str = 'chrysopedia'`
   - `prompts_path: str = './prompts'`
   - `review_mode: bool = True`

3. Create `backend/worker.py`: Celery app instance using `config.get_settings().redis_url` as broker. Import and register tasks from `pipeline.stages` (import the module so decorators register). Worker must be importable as `celery -A worker worker`.

4. Create `backend/pipeline/__init__.py` (empty).

5. Create `backend/pipeline/schemas.py` with Pydantic models:
   - `TopicSegment(start_index, end_index, topic_label, summary)` — stage 2 output per segment group
   - `SegmentationResult(segments: list[TopicSegment])` — stage 2 full output
   - `ExtractedMoment(title, summary, start_time, end_time, content_type, plugins, raw_transcript)` — stage 3 output per moment
   - `ExtractionResult(moments: list[ExtractedMoment])` — stage 3 full output
   - `ClassifiedMoment(moment_index, topic_category, topic_tags, content_type_override)` — stage 4 output per moment
   - `ClassificationResult(classifications: list[ClassifiedMoment])` — stage 4 full output
   - `SynthesizedPage(title, slug, topic_category, topic_tags, summary, body_sections, signal_chains, plugins, source_quality)` — stage 5 output
   - `SynthesisResult(pages: list[SynthesizedPage])` — stage 5 full output
   All fields should use appropriate types: `content_type` as str (enum value), `body_sections` as dict, `signal_chains` as list[dict], etc.

6. Create `backend/pipeline/llm_client.py`:
   - Class `LLMClient` initialized with `settings: Settings`
   - Uses sync `openai.OpenAI(base_url=settings.llm_api_url, api_key=settings.llm_api_key)`
   - Method `complete(system_prompt: str, user_prompt: str, response_model: type[BaseModel] | None = None) -> str` that:
     a. Tries primary endpoint with `response_format={'type': 'json_object'}` if response_model is provided
     b. On connection/timeout error, tries fallback endpoint (`openai.OpenAI(base_url=settings.llm_fallback_url)` with `settings.llm_fallback_model`)
     c. Returns the raw completion text
     d. Logs WARNING on fallback, ERROR on total failure
   - Method `parse_response(text: str, model: type[T]) -> T` that validates JSON via Pydantic `model_validate_json()` with error handling
   - Catch `openai.APIConnectionError`, `openai.APITimeoutError` for fallback trigger
   - IMPORTANT: Use sync `openai.OpenAI`, not async — Celery tasks run in sync context

## Must-Haves

- [ ] Settings has all 12 new fields with correct defaults
- [ ] `openai`, `qdrant-client`, `pyyaml` in requirements.txt
- [ ] `celery -A worker worker` is syntactically valid (worker.py creates Celery app and imports pipeline.stages)
- [ ] LLMClient uses sync openai.OpenAI, not AsyncOpenAI
- [ ] LLMClient has primary/fallback logic with proper exception handling
- [ ] All 8 Pydantic schema classes defined with correct field types

## Verification

- `cd backend && python -c "from config import Settings; s = Settings(); print(s.llm_api_url, s.qdrant_url, s.review_mode)"` prints defaults
- `cd backend && python -c "from pipeline.schemas import SegmentationResult, ExtractionResult, ClassificationResult, SynthesisResult; print('schemas ok')"` exits 0
- `cd backend && python -c "from pipeline.llm_client import LLMClient; print('client ok')"` exits 0
- `cd backend && python -c "from worker import celery_app; print(celery_app.main)"` exits 0
- `grep -q 'openai' backend/requirements.txt && grep -q 'qdrant-client' backend/requirements.txt`
  - Estimate: 1.5h
  - Files: backend/config.py, backend/requirements.txt, backend/worker.py, backend/pipeline/__init__.py, backend/pipeline/schemas.py, backend/pipeline/llm_client.py
  - Verify: cd backend && python -c "from config import Settings; s = Settings(); print(s.llm_api_url, s.qdrant_url, s.review_mode)" && python -c "from pipeline.schemas import SegmentationResult, ExtractionResult, ClassificationResult, SynthesisResult; print('schemas ok')" && python -c "from pipeline.llm_client import LLMClient; print('client ok')" && python -c "from worker import celery_app; print(celery_app.main)"
- [x] **T02: Created 4 prompt templates and implemented 5 Celery tasks (stages 2-5 + run_pipeline) with sync SQLAlchemy, retry logic, resumability, and Redis-based cross-stage data passing** — 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
  - Estimate: 2.5h
  - Files: prompts/stage2_segmentation.txt, prompts/stage3_extraction.txt, prompts/stage4_classification.txt, prompts/stage5_synthesis.txt, backend/pipeline/stages.py, backend/worker.py
  - Verify: 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')"
- [x] **T03: Created sync EmbeddingClient, QdrantManager with idempotent collection management and metadata-rich upserts, and wired stage6_embed_and_index into the Celery pipeline chain as a non-blocking side-effect stage** — Create the embedding client (sync OpenAI-compatible /v1/embeddings) and Qdrant client (collection management + upsert). Wire embedding generation into the pipeline after stage 5 synthesis, so technique page summaries and key moment summaries are embedded and upserted to Qdrant with metadata payloads.

## Failure Modes

| Dependency | On error | On timeout | On malformed response |
|------------|----------|-----------|----------------------|
| Embedding API | Log error, skip embedding for this batch, pipeline continues | Same as error | Validate vector dimensions match config, log mismatch |
| Qdrant server | Log error, skip upsert, pipeline continues (embeddings are not blocking) | Same as error | N/A |

## Steps

1. Create `backend/pipeline/embedding_client.py`:
   - Class `EmbeddingClient` initialized with `settings: Settings`
   - Uses sync `openai.OpenAI(base_url=settings.embedding_api_url, api_key=settings.llm_api_key)`
   - Method `embed(texts: list[str]) -> list[list[float]]` that calls `client.embeddings.create(model=settings.embedding_model, input=texts)` and returns the vector list
   - Handle `openai.APIConnectionError` gracefully — log and return empty list
   - Validate returned vector dimensions match `settings.embedding_dimensions`

2. Create `backend/pipeline/qdrant_client.py`:
   - Class `QdrantManager` initialized with `settings: Settings`
   - Uses sync `qdrant_client.QdrantClient(url=settings.qdrant_url)`
   - Method `ensure_collection()` — checks `collection_exists()`, creates with `VectorParams(size=settings.embedding_dimensions, distance=Distance.COSINE)` if not
   - Method `upsert_points(points: list[PointStruct])` — wraps `client.upsert()`
   - Method `upsert_technique_pages(pages: list[dict], vectors: list[list[float]])` — builds PointStruct list with payload (page_id, creator_id, title, topic_category, topic_tags, summary) and upserts
   - Method `upsert_key_moments(moments: list[dict], vectors: list[list[float]])` — builds PointStruct list with payload (moment_id, source_video_id, title, start_time, end_time, content_type) and upserts
   - Handle `qdrant_client.http.exceptions.UnexpectedResponse` for connection errors

3. Add a new Celery task `stage6_embed_and_index(video_id: str)` in `backend/pipeline/stages.py`:
   - Load all KeyMoments and TechniquePages created for this video
   - Build text strings for embedding (moment: title + summary; page: title + summary + topic_category)
   - Call EmbeddingClient.embed() to get vectors
   - Call QdrantManager.ensure_collection() then upsert_technique_pages() and upsert_key_moments()
   - This stage runs after stage 5 in the pipeline chain but does NOT update processing_status (it's a side-effect)
   - If embedding/Qdrant fails, log error but don't fail the pipeline — embeddings can be regenerated later

4. Update `run_pipeline` in `stages.py` to append `stage6_embed_and_index` to the chain.

## Must-Haves

- [ ] EmbeddingClient uses sync openai.OpenAI for /v1/embeddings calls
- [ ] QdrantManager creates collection only if not exists
- [ ] Qdrant points include metadata payloads (page_id/moment_id, creator_id, topic, timestamps)
- [ ] stage6 is non-blocking — embedding/Qdrant failures don't fail the pipeline
- [ ] Vector dimension from config, not hardcoded

## Verification

- `cd backend && python -c "from pipeline.embedding_client import EmbeddingClient; print('embed ok')"` exits 0
- `cd backend && python -c "from pipeline.qdrant_client import QdrantManager; print('qdrant ok')"` exits 0
- `cd backend && python -c "from pipeline.stages import stage6_embed_and_index; print('stage6 ok')"` exits 0

## Observability Impact

- Signals added: INFO log for embedding batch size and Qdrant upsert count, WARNING on embedding/Qdrant failures with error details
- How a future agent inspects this: check Qdrant dashboard at QDRANT_URL, query collection point count
- Failure state exposed: embedding failures logged but pipeline completes successfully
  - Estimate: 1.5h
  - Files: backend/pipeline/embedding_client.py, backend/pipeline/qdrant_client.py, backend/pipeline/stages.py
  - Verify: cd backend && python -c "from pipeline.embedding_client import EmbeddingClient; print('embed ok')" && python -c "from pipeline.qdrant_client import QdrantManager; print('qdrant ok')" && python -c "from pipeline.stages import stage6_embed_and_index; print('stage6 ok')"
- [x] **T04: Wired automatic run_pipeline.delay() dispatch after ingest commit and added POST /api/v1/pipeline/trigger/{video_id} for manual re-processing** — Connect the existing ingest endpoint to the pipeline by dispatching `run_pipeline.delay()` after successful commit, and add a manual re-trigger API endpoint for re-processing videos.

## Steps

1. Modify `backend/routers/ingest.py`:
   - After the `await db.commit()` and refresh calls (after the successful response is ready), dispatch the pipeline: `from pipeline.stages import run_pipeline; run_pipeline.delay(str(video.id))`
   - Wrap the dispatch in try/except to avoid failing the ingest response if Redis/Celery is down — log WARNING on dispatch failure
   - The dispatch must happen AFTER commit so the worker can find the data in PostgreSQL

2. Create `backend/routers/pipeline.py`:
   - Router with `prefix='/pipeline'`, tag `['pipeline']`
   - `POST /trigger/{video_id}`: Look up SourceVideo by id, verify it exists (404 if not), dispatch `run_pipeline.delay(str(video_id))`, return `{"status": "triggered", "video_id": str(video_id), "current_processing_status": video.processing_status.value}`
   - Uses `get_session` dependency for DB access

3. Mount the pipeline router in `backend/main.py` under `/api/v1`.

## Must-Haves

- [ ] Ingest endpoint dispatches run_pipeline.delay() after commit
- [ ] Pipeline dispatch failure does not fail the ingest response
- [ ] POST /api/v1/pipeline/trigger/{video_id} exists and returns status
- [ ] 404 returned for non-existent video_id
- [ ] Pipeline router mounted in main.py

## Verification

- `cd backend && python -c "from routers.pipeline import router; print([r.path for r in router.routes])"` shows `['/trigger/{video_id}']`
- `grep -q 'pipeline' backend/main.py` exits 0
- `grep -q 'run_pipeline' backend/routers/ingest.py` exits 0
  - Estimate: 45m
  - Files: backend/routers/ingest.py, backend/routers/pipeline.py, backend/main.py
  - Verify: cd backend && python -c "from routers.pipeline import router; print([r.path for r in router.routes])" && grep -q 'pipeline' backend/main.py && grep -q 'run_pipeline' backend/routers/ingest.py
- [x] **T05: Added 10 integration tests covering pipeline stages 2-6, trigger endpoint, ingest dispatch, resumability, and LLM fallback with mocked LLM/Qdrant and real PostgreSQL** — Write comprehensive integration tests with mocked LLM and Qdrant that verify the entire pipeline flow. Tests use the existing pytest-asyncio infrastructure from S02, extended with mocks for the OpenAI client and Qdrant client.

## Negative Tests

- **Malformed inputs**: LLM returns invalid JSON → pipeline handles gracefully, logs error
- **Error paths**: LLM primary endpoint connection refused → falls back to secondary
- **Boundary conditions**: Video with 0 segments → pipeline completes without errors, no moments created

## Steps

1. Create `backend/tests/fixtures/mock_llm_responses.py`:
   - Dictionary mapping stage names to valid JSON response strings matching each Pydantic schema
   - Stage 2: SegmentationResult with 2 topic segments covering the 5 sample transcript segments
   - Stage 3: ExtractionResult with 2 key moments (one technique, one settings)
   - Stage 4: ClassificationResult mapping moments to canonical tags
   - Stage 5: SynthesisResult with a technique page including body_sections, signal_chains, plugins
   - Embedding response: list of 768-dimensional vectors (can be random floats for testing)

2. Create `backend/tests/test_pipeline.py` with these tests using `unittest.mock.patch` on `openai.OpenAI` and `qdrant_client.QdrantClient`:
   a. `test_stage2_segmentation_updates_topic_labels`: Ingest sample transcript, run stage2 with mocked LLM, verify TranscriptSegment rows have topic_label set
   b. `test_stage3_extraction_creates_key_moments`: Run stages 2+3, verify KeyMoment rows created with correct fields, processing_status = extracted
   c. `test_stage4_classification_assigns_tags`: Run stages 2+3+4, verify moments have topic_category and topic_tags from canonical list
   d. `test_stage5_synthesis_creates_technique_pages`: Run full pipeline stages 2-5, verify TechniquePage rows created with body_sections, signal_chains, summary. Verify KeyMoments linked to TechniquePage via technique_page_id
   e. `test_stage6_embeds_and_upserts_to_qdrant`: Run full pipeline, verify EmbeddingClient.embed called with correct texts, QdrantManager.upsert called with correct payloads
   f. `test_run_pipeline_resumes_from_extracted`: Set video processing_status to extracted, run pipeline, verify only stages 4+5+6 execute (not 2+3)
   g. `test_pipeline_trigger_endpoint`: POST to /api/v1/pipeline/trigger/{video_id} with mocked Celery delay, verify 200 response
   h. `test_pipeline_trigger_404_for_missing_video`: POST to /api/v1/pipeline/trigger/{nonexistent_id}, verify 404
   i. `test_ingest_dispatches_pipeline`: POST transcript to ingest endpoint with mocked run_pipeline.delay, verify delay was called with video_id
   j. `test_llm_fallback_on_primary_failure`: Mock primary endpoint to raise APIConnectionError, verify fallback endpoint is called

   IMPORTANT: Pipeline stages use SYNC SQLAlchemy (not async). Tests that call stage functions directly need a real PostgreSQL test database. Use the existing conftest.py pattern but create sync engine/sessions for pipeline stage calls. Mock the LLM and Qdrant clients, not the DB.

3. Update `backend/tests/conftest.py` to add:
   - A fixture that creates a sync SQLAlchemy engine pointing to the test database (for pipeline stage tests)
   - A fixture that pre-ingests a sample transcript and returns the video_id (for pipeline tests to use)
   - Any necessary mock fixtures for LLMClient and QdrantManager

## Must-Haves

- [ ] 10 integration tests covering all pipeline stages, trigger endpoints, resumability, and fallback
- [ ] All LLM calls mocked with realistic response fixtures
- [ ] Qdrant calls mocked — no real Qdrant needed
- [ ] Pipeline stage tests use sync DB sessions (matching production Celery behavior)
- [ ] At least one negative test (LLM fallback on primary failure)
- [ ] All tests pass with `python -m pytest tests/test_pipeline.py -v`

## Verification

- `cd backend && python -m pytest tests/test_pipeline.py -v` — all tests pass
- `cd backend && python -m pytest tests/ -v` — all tests (including S02 ingest tests) still pass

## Observability Impact

- Signals added: test output shows per-test pass/fail with timing, mock call assertions verify logging would fire in production
- How a future agent inspects this: run `pytest tests/test_pipeline.py -v` to see all pipeline test results
  - Estimate: 2h
  - Files: backend/tests/fixtures/mock_llm_responses.py, backend/tests/test_pipeline.py, backend/tests/conftest.py
  - Verify: cd backend && python -m pytest tests/test_pipeline.py -v && python -m pytest tests/ -v
