# Pipeline Stages Context

## Transcript format conventions

- **Participant codes**: Segments use `[p1]`, `[p2]` etc. — never generic role labels like `[PARTICIPANT]`
- **Speaker labels**: Original Whisper labels (`Speaker A`, `Speaker B`) kept in parentheses in raw transcripts only
- **Timecodes**: `MM:SS` for segments under 1 hour, `HH:MM:SS` at or above 1 hour. Mixed formats in the same file is correct
- **Timecodes are floats internally**: All data structures store seconds as `float`. String formatting happens only at output. Never parse formatted timecodes back within the same session
- **`.txt` is canonical**: The parser (`load_transcripts_from_dir`) reads only `.txt` files. `.md` files are human-readable companions, not parsed back
- **Legacy format support**: Parser also accepts old-format files with `[PARTICIPANT]`/`[RESEARCHER]` role labels

## Output directory structure

```
output/
├── raw_transcripts/          # Stage 6 output
│   ├── p1_raw.txt            # Plain text (canonical, machine-readable)
│   ├── p1_raw.md             # Markdown companion (human-readable)
│   ├── p2_raw.txt
│   └── p2_raw.md
├── cooked_transcripts/       # Stage 7 output (PII-redacted, only when --redact-pii)
│   ├── p1_cooked.txt
│   ├── p1_cooked.md
│   ├── p2_cooked.txt
│   └── p2_cooked.md
├── intermediate/             # JSON snapshots (if write_intermediate=True)
├── temp/                     # Extracted audio, working files
├── people.yaml               # Participant registry (computed stats + human-editable fields)
├── research_report.md        # Final markdown report
├── research_report.html      # Final HTML report with interactive features
├── transcript_p1.html        # Per-participant transcript page (one per participant)
├── transcript_p2.html
└── ...
```

## Stage 5b: Speaker identification

`identify_speakers.py` runs a two-pass speaker role assignment: heuristic first, then LLM refinement.

- **Heuristic pass** (`identify_speaker_roles_heuristic`): scores speakers by question ratio and researcher-phrase hits. Assigns `RESEARCHER`, `PARTICIPANT`, or `OBSERVER`. Fast, no LLM needed
- **LLM pass** (`identify_speaker_roles_llm`): sends first ~5 minutes to the LLM for refined role assignment. Also extracts `person_name` and `job_title` for each speaker when mentioned in the transcript
- **Return type**: `identify_speaker_roles_llm()` returns `list[SpeakerInfo]` — a dataclass with `speaker_label`, `role`, `person_name`, `job_title`. Still mutates segments in place for role assignment (existing behaviour). Returns empty list on exception
- **`SpeakerInfo` import**: defined in `identify_speakers.py`. Other modules import it under `TYPE_CHECKING` to avoid circular imports (e.g. `people.py` uses `if TYPE_CHECKING: from bristlenose.stages.identify_speakers import SpeakerInfo`)
- **Structured output**: `SpeakerRoleItem` in `llm/structured.py` has `person_name` and `job_title` fields (both default `""` for backward compatibility with existing LLM responses)

## LLM concurrency in stages

Stages 8 and 9 accept a `concurrency: int = 1` kwarg, passed by the pipeline from `settings.llm_concurrency`. The pattern:

```python
semaphore = asyncio.Semaphore(concurrency)
async def _process(transcript):
    async with semaphore:
        return await _segment_single(transcript, llm_client)
return list(await asyncio.gather(*(_process(t) for t in transcripts)))
```

- **Default 1**: behaves identically to the old sequential loop (backward compatible, useful for tests)
- **Pipeline passes 3**: `concurrency=self.settings.llm_concurrency` in both `run()` and `run_analysis_only()`
- **Error isolation**: each `_process()` closure has its own try/except. A failed participant doesn't cancel siblings (asyncio.gather default behaviour — if one raises, others still complete since the exception is caught inside `_process`)
- **No cross-stage semaphore**: each stage creates its own semaphore. Stages still execute sequentially in the pipeline

## Platform-aware session grouping (Stage 1)

`ingest.py` groups input files into sessions using a two-pass strategy that handles Teams, Zoom, and Google Meet naming conventions.

- **`_normalise_stem(stem)`**: strips platform suffixes before stem matching. Expects lowercased input:
  - Teams: `{YYYYMMDD}_{HHMMSS}-Meeting Recording`, `-meeting transcript`
  - Zoom cloud: `Audio Transcript_` prefix, `_{MeetingID}_{Month_DD_YYYY}` tail (9–11 digit IDs)
  - Google Meet: `({YYYY-MM-DD at ...})` parenthetical, `- Transcript` suffix
  - Legacy: `_transcript`, `_subtitles`, `_captions`, `_sub`, `_srt`
- **`_is_zoom_local_dir(dir_name)`**: matches `YYYY-MM-DD HH.MM.SS Topic MeetingID` pattern
- **`group_into_sessions()`**: Pass 1 groups Zoom local folder files by directory; Pass 2 groups remaining files by normalised stem
- **Regex patterns**: 6 compiled module-level patterns (`_TEAMS_SUFFIX_RE`, `_ZOOM_CLOUD_TAIL_RE`, `_ZOOM_CLOUD_PREFIX_RE`, `_ZOOM_LOCAL_DIR_RE`, `_GMEET_PAREN_RE`, `_GMEET_TRANSCRIPT_SUFFIX_RE`)
- **Tests**: `tests/test_ingest.py` — 35 tests covering normalisation, Zoom dir detection, session grouping for all platforms, false positive prevention

## Concurrent audio extraction (Stage 2)

`extract_audio_for_sessions()` is async — video files are extracted in parallel via `asyncio.to_thread()` (wrapping blocking `subprocess.run` FFmpeg calls) bounded by `asyncio.Semaphore(4)`.

- **Default 4**: fixed constant, not hardware-adaptive. The bottleneck is the macOS media engine (shared hardware decode), not CPU cores. 4 works well from M1 to M4 Ultra
- **`_extract_one()` helper**: runs `has_audio_stream()` + `extract_audio_from_video()` inside the semaphore. Both are blocking subprocess calls wrapped in `asyncio.to_thread()`
- **Platform transcript skip**: when `session.has_existing_transcript=True`, FFmpeg extraction is skipped entirely — the pipeline will use the parsed transcript and never call Whisper, so audio decode is unnecessary
- **Error isolation**: same pattern as LLM stages — a failed extraction doesn't cancel siblings
- **VideoToolbox**: `utils/audio.py` passes `-hwaccel videotoolbox` on macOS, so concurrent extractions share the hardware media engine for H.264/HEVC decode
- **`concurrency` kwarg**: exposed but not yet wired to config (unlike `llm_concurrency`). Default of 4 is sufficient; config wiring deferred until there's a real need
- **Tests**: `tests/test_extract_audio.py` — 2 tests for extraction skip behaviour

## Duplicate timecode helpers

Both `models.py` and `utils/timecodes.py` define `format_timecode()` and `parse_timecode()`. They behave identically. Stage files import from one or the other — both are fine. The `utils/timecodes.py` version has a more sophisticated parser (SRT/VTT milliseconds support).
