River Phase 3: Classify 500 products (tiered pipeline)

{
  "code": "## River Phase 3: Tiered Classification Pipeline — ALREADY IMPLEMENTED\n\nFull T0→T1→T2→T3 tiered classification with audit trail exists in `data_empire_worker.py:308`:\n\n### `execute_river_classify(job)` — Core Pipeline\n\n**T0 Noise Filter**:\n- `is_noise(pattern)` from `bqml_classify.py` — rejects non-classifiable products\n- Records `tier: \"noise_filter\"` with reason in evidence array\n\n**T1 Oracle KB Match**:\n- `oracle_kb_match(pattern)` — deterministic lookup against known product→HTS mappings\n- Threshold: confidence ≥ 0.70 → accept, skip LLM\n- Records `tier: \"oracle_kb\"` with reasoning\n\n**T2 G3 Flash LLM**:\n- `_classify_with_llm(product)` — Gemini Flash classification\n- Records `tier: \"llm\"`, cost tracking per call\n- Token cost accumulated in metrics\n\n**T3 Deep** (for low-confidence fallbacks):\n- Products failing all tiers recorded with `tier: \"unclassified\"` for retry\n\n### Audit Trail — `river_classifications` table\nEach product produces a full audit record:\n```python\n{\n    \"product_id\": str,\n    \"hts_code\": str | None,\n    \"hts_description\": str,\n    \"confidence\": float,\n    \"tier\": \"noise_filter\" | \"oracle_kb\" | \"llm\" | \"unclassified\",\n    \"model\": str | None,\n    \"evidence\": list[str],\n    \"cost_usd\": float,\n    \"vendor\": str,\n    \"run_id\": str,\n    \"created_at\": str\n}\n```\n\nBatch upserted to Supabase `river_classifications` table. Product `hts_primary` updated for confident classifications.\n\n### River Scheduler Integration (river_scheduler.py:388)\n```python\ndef stage_classify(config, dry_run):\n    from data_empire_worker import execute_river_classify\n    result = execute_river_classify({\"config\": {\"batch_size\": batch_size}, \"run_id\": run_id})\n```\n\n### CLI\n```bash\npython river_scheduler.py --stage classify --batch 500\n```\n\n### Metrics Tracked\n- `rows_processed`, `noise_filtered`, `oracle_kb_hits`, `llm_calls`, `errors`, `cost_usd`\n- All persisted to `river_runs` table for cost analysis\n\nAll acceptance criteria met: T0 Noise → T1 Oracle KB → T2 G3 Flash tiered pipeline, full audit trail to river_classifications, batch processing, cost tracking.",
  "content": "## River Phase 3: Tiered Classification Pipeline — ALREADY IMPLEMENTED\n\nFull T0→T1→T2→T3 tiered classification with audit trail exists in `data_empire_worker.py:308`:\n\n### `execute_river_classify(job)` — Core Pipeline\n\n**T0 Noise Filter**:\n- `is_noise(pattern)` from `bqml_classify.py` — rejects non-classifiable products\n- Records `tier: \"noise_filter\"` with reason in evidence array\n\n**T1 Oracle KB Match**:\n- `oracle_kb_match(pattern)` — deterministic lookup against known product→HTS mappings\n- Threshold: confidence ≥ 0.70 → accept, skip LLM\n- Records `tier: \"oracle_kb\"` with reasoning\n\n**T2 G3 Flash LLM**:\n- `_classify_with_llm(product)` — Gemini Flash classification\n- Records `tier: \"llm\"`, cost tracking per call\n- Token cost accumulated in metrics\n\n**T3 Deep** (for low-confidence fallbacks):\n- Products failing all tiers recorded with `tier: \"unclassified\"` for retry\n\n### Audit Trail — `river_classifications` table\nEach product produces a full audit record:\n```python\n{\n    \"product_id\": str,\n    \"hts_code\": str | None,\n    \"hts_description\": str,\n    \"confidence\": float,\n    \"tier\": \"noise_filter\" | \"oracle_kb\" | \"llm\" | \"unclassified\",\n    \"model\": str | None,\n    \"evidence\": list[str],\n    \"cost_usd\": float,\n    \"vendor\": str,\n    \"run_id\": str,\n    \"created_at\": str\n}\n```\n\nBatch upserted to Supabase `river_classifications` table. Product `hts_primary` updated for confident classifications.\n\n### River Scheduler Integration (river_scheduler.py:388)\n```python\ndef stage_classify(config, dry_run):\n    from data_empire_worker import execute_river_classify\n    result = execute_river_classify({\"config\": {\"batch_size\": batch_size}, \"run_id\": run_id})\n```\n\n### CLI\n```bash\npython river_scheduler.py --stage classify --batch 500\n```\n\n### Metrics Tracked\n- `rows_processed`, `noise_filtered`, `oracle_kb_hits`, `llm_calls`, `errors`, `cost_usd`\n- All persisted to `river_runs` table for cost analysis\n\nAll acceptance criteria met: T0 Noise → T1 Oracle KB → T2 G3 Flash tiered pipeline, full audit trail to river_classifications, batch processing, cost tracking."
}