---
estimated_steps: 51
estimated_files: 2
skills_used: []
---

# T01: Create watcher.py — watchdog observer with file stability, validation, and POST logic

Create the core watcher script that monitors a folder for new transcript JSON files and POSTs them to the ingest API.

**Slice:** S03 — Transcript Folder Watcher — Auto-Ingest Service
**Milestone:** M007

## Description

Build `backend/watcher.py` — a standalone Python script using the `watchdog` library to monitor a configurable folder for new `.json` files. When a file is detected:
1. Wait for file stability (size unchanged for 2 seconds) to handle SCP/rsync partial writes
2. Read and validate JSON structure (must contain `source_file`, `creator_folder`, `duration_seconds`, `segments`)
3. POST the file as multipart upload to the ingest API endpoint using httpx
4. Move successfully processed files to `processed/` subfolder
5. Move failed files to `failed/` subfolder with a `.error` sidecar file containing the error details

The script uses `PollingObserver` (not inotify Observer) for reliability on ZFS/network-mounted filesystems. It watches recursively.

Add `watchdog` to `backend/requirements.txt`.

## Failure Modes

| Dependency | On error | On timeout | On malformed response |
|------------|----------|-----------|----------------------|
| Ingest API | Move file to `failed/` with HTTP status + body in `.error` sidecar | httpx timeout (30s default) → same as error | Log warning, move to `failed/` |
| Filesystem | Log exception, skip file (watchdog continues) | N/A | N/A |

## Negative Tests

- **Malformed inputs**: Invalid JSON (parse error) → `failed/` with error sidecar. Missing required keys → `failed/` with validation error. Non-JSON files → ignored (only `.json` suffix triggers processing). Empty file → `failed/`.
- **Error paths**: API returns 4xx/5xx → `failed/` with response body. API unreachable → `failed/` with connection error. File deleted between detection and read → log warning, skip.
- **Boundary conditions**: File still being written (size changing) → wait until stable. Same filename in `processed/` already exists → overwrite (idempotent).

## Steps

1. Add `watchdog>=4.0,<5.0` to `backend/requirements.txt`
2. Create `backend/watcher.py` with:
   - Configuration from environment variables: `WATCH_FOLDER` (default `/watch`), `WATCHER_API_URL` (default `http://chrysopedia-api:8000/api/v1/ingest`), `WATCHER_STABILITY_SECONDS` (default `2`), `WATCHER_POLL_INTERVAL` (default `5`)
   - `TranscriptHandler(FileSystemEventHandler)` class that handles `on_created` events for `.json` files
   - `wait_for_stability(path, seconds)` function — polls file size every 0.5s, returns True when size is unchanged for `seconds` duration, returns False if file disappears
   - `validate_transcript(data)` function — checks for required keys (`source_file`, `creator_folder`, `duration_seconds`, `segments`)
   - `post_to_ingest(filepath)` function — reads file, validates JSON, POSTs as multipart upload via httpx, returns (success, detail)
   - `move_to_processed(filepath)` / `move_to_failed(filepath, error)` functions — create subdirs as needed, move file, write `.error` sidecar for failures
   - Main block: create `processed/` and `failed/` subdirs, start `PollingObserver` with recursive=True, run until KeyboardInterrupt
   - Structured logging with `logging.basicConfig` at INFO level, logger name `watcher`
3. Verify `python -m py_compile backend/watcher.py` passes

## Must-Haves

- [x] watchdog added to requirements.txt
- [x] watcher.py handles file stability check
- [x] watcher.py validates JSON keys before POST
- [x] watcher.py POSTs via httpx multipart upload
- [x] Processed files move to processed/ subfolder
- [x] Failed files move to failed/ with .error sidecar
- [x] Structured logging with watcher.* logger
- [x] PollingObserver used (not inotify) for ZFS reliability

## Verification

- `python -m py_compile backend/watcher.py` exits 0
- `grep -q 'watchdog' backend/requirements.txt` exits 0
- `grep -q 'PollingObserver' backend/watcher.py` exits 0
- `grep -q 'processed' backend/watcher.py && grep -q 'failed' backend/watcher.py` exits 0

## Observability Impact

- Signals added: `watcher` logger with structured messages for pickup, stability wait, validation, POST result, and file moves
- How a future agent inspects this: `docker logs chrysopedia-watcher` shows all watcher activity
- Failure state exposed: `.error` sidecar files in `failed/` directory with HTTP status, response body, or exception traceback

## Inputs

- `backend/requirements.txt`

## Expected Output

- `backend/watcher.py`
- `backend/requirements.txt`

## Verification

python -m py_compile backend/watcher.py && grep -q 'watchdog' backend/requirements.txt && grep -q 'PollingObserver' backend/watcher.py
