# Unified person_intel Job — Implementation Plan

> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

**Goal:** Consolidate fragmented person intelligence (discover.py CLI, enrich.py CLI, people_scoring job) into a unified `person_intel` job with 5 stages: discover → enrich → research → score → assess.

**Architecture:** "One engine, two entry points." Extract core async logic from 1200-line CLI scripts into clean library modules (`intel/people/lib/`). The job handler calls these modules as stages. The CLIs become thin wrappers. Data flows through filesystem (`intel/people/data/{slug}/`) as canonical store, with summary metrics in the jobs results table.

**Tech Stack:** Python async, jobs framework (handler_base, stage dispatcher), SQLite, httpx, Serper API, 12+ enrichment APIs, LLM via lib/llm.

**Vario review consensus (4/4 models):** Consolidate. Enrich feeds research. Prowess stages fan-out in single assess stage. Dossier stays on-demand. Keep CLI + batch paths.

---

## Task 1: Extract enrich core into library module

The simplest extraction — `enrich.py` already has a clean `run_enrichment()` function. Move it to a library module so both CLI and job handler can import it.

**Files:**
- Create: `intel/people/lib/__init__.py`
- Create: `intel/people/lib/enrich.py`
- Modify: `intel/people/enrich.py` (thin wrapper)
- Test: `intel/people/tests/test_enrich_lib.py`

**Step 1: Create `intel/people/lib/` directory with __init__.py**

```python
# intel/people/lib/__init__.py
"""Core person intelligence library — shared by CLI and job handler."""
```

**Step 2: Create `intel/people/lib/enrich.py`**

Move everything from `intel/people/enrich.py` EXCEPT the `if __name__` block into the new module. The key things to move:

- `_RateLimiter` class and `_limiter` instance
- `http_get`, `http_post` helpers
- All `enrich_*` async functions (apollo, fec, semantic_scholar, patents, sec, wikidata, openalex, dblp, pubmed, arxiv, wikipedia, grokipedia)
- `ALL_SOURCES` dict
- `run_enrichment()` orchestrator
- `extract_photos()`, `save_photos_to_db()`
- Constants: API keys, rate limits, proxy config

The module should be importable as:
```python
from intel.people.lib.enrich import run_enrichment, extract_photos, save_photos_to_db, ALL_SOURCES
```

**Step 3: Update `intel/people/enrich.py` to re-import from lib**

Replace the entire file body with:
```python
#!/usr/bin/env python
"""Data enrichment CLI for people discovery pipeline.

Thin wrapper around intel.people.lib.enrich.
"""
import asyncio
import json
import sys
from pathlib import Path

from loguru import logger

_RIVUS_ROOT = Path(__file__).resolve().parent.parent.parent
if str(_RIVUS_ROOT) not in sys.path:
    sys.path.insert(0, str(_RIVUS_ROOT))

# Re-export everything for backward compat (other files do `from enrich import ...`)
from intel.people.lib.enrich import (  # noqa: F401
    run_enrichment, extract_photos, save_photos_to_db,
    enrich_apollo, enrich_fec, enrich_semantic_scholar,
    enrich_patents, enrich_sec, enrich_wikidata,
    enrich_openalex, enrich_dblp, enrich_pubmed, enrich_arxiv,
    enrich_wikipedia, enrich_grokipedia,
    APOLLO_API_KEY, PATENTSVIEW_API_KEY,
    ALL_SOURCES,
)


if __name__ == "__main__":
    import click

    @click.command()
    @click.argument("name")
    @click.option("--hint", default=None)
    @click.option("--sources", default=None, help="Comma-separated source list")
    def main(name, hint, sources):
        src_list = sources.split(",") if sources else None
        results = asyncio.run(run_enrichment(name, hint=hint, sources=src_list))
        ok = sum(1 for v in results.values() if "error" not in v)
        print(json.dumps(results, indent=2, default=str))
        logger.info(f"Enriched {name}: {ok}/{len(results)} sources OK")

    main()
```

**Step 4: Write test**

```python
# intel/people/tests/test_enrich_lib.py
"""Test that enrich library module imports and has expected interface."""
import pytest


def test_enrich_imports():
    from intel.people.lib.enrich import run_enrichment, ALL_SOURCES, extract_photos
    assert callable(run_enrichment)
    assert isinstance(ALL_SOURCES, dict)
    assert len(ALL_SOURCES) >= 10  # 12 sources currently
    assert callable(extract_photos)


def test_enrich_sources_are_async():
    import asyncio
    from intel.people.lib.enrich import ALL_SOURCES
    for name, fn in ALL_SOURCES.items():
        assert asyncio.iscoroutinefunction(fn), f"{name} is not async"
```

**Step 5: Run tests**

Run: `python -m pytest intel/people/tests/test_enrich_lib.py -v`
Expected: PASS

**Step 6: Verify existing discover.py still works**

Run: `python intel/people/discover.py --help`
Expected: Shows help (the `from enrich import ...` still works via re-exports)

**Step 7: Commit**

```bash
git add intel/people/lib/ intel/people/enrich.py intel/people/tests/test_enrich_lib.py
git commit -m "refactor(intel): extract enrich core into intel/people/lib/enrich.py"
```

---

## Task 2: Extract discover core into library module

Harder than enrich — `discover.py` has CLI display logic tangled with core logic. Extract `discover_person()` and supporting functions.

**Files:**
- Create: `intel/people/lib/discover.py`
- Modify: `intel/people/discover.py` (thin wrapper)
- Test: `intel/people/tests/test_discover_lib.py`

**Step 1: Create `intel/people/lib/discover.py`**

Move from `intel/people/discover.py`:
- `discover_person()` async function (the core — lines 264-556)
- `group_by_theme()` (lines 558-611)
- `build_profile()` (lines 884-990)
- `save_profile_quick()`, `save_profile_deep()`, `merge_profiles()` (lines 1002-1053)
- `_save_deep_research()` (lines 1055-1099)
- `llm_deep_research()`, `llm_deep_research_vip()` (lines 177-262)
- `_parse_deep_research_response()` (lines 164-175)
- `DEEP_RESEARCH_PROMPT` and constants

Keep in `intel/people/discover.py`:
- `main()` click command
- `format_discovery()` (CLI display only)
- All the CLI printing/display logic

The library module should be importable as:
```python
from intel.people.lib.discover import discover_person, build_profile, save_profile_quick, merge_profiles
```

**Step 2: Update `intel/people/discover.py`**

Replace imports to use `from intel.people.lib.discover import ...` and keep only CLI concerns (click command, format_discovery, printing).

**Step 3: Write test**

```python
# intel/people/tests/test_discover_lib.py
"""Test that discover library module imports and has expected interface."""

def test_discover_imports():
    from intel.people.lib.discover import discover_person, build_profile, save_profile_quick, merge_profiles
    import asyncio
    assert asyncio.iscoroutinefunction(discover_person)
    assert callable(build_profile)
    assert callable(save_profile_quick)
    assert callable(merge_profiles)
```

**Step 4: Run tests**

Run: `python -m pytest intel/people/tests/test_discover_lib.py -v`
Expected: PASS

**Step 5: Verify CLI still works**

Run: `python intel/people/discover.py --help`
Expected: Shows help text unchanged

**Step 6: Commit**

```bash
git add intel/people/lib/discover.py intel/people/discover.py intel/people/tests/test_discover_lib.py
git commit -m "refactor(intel): extract discover core into intel/people/lib/discover.py"
```

---

## Task 3: Create unified person_intel handler with discover + enrich stages

Build the new handler that uses the extracted library modules. Start with just discover and enrich stages — these are the new stages that don't exist in the current `people_scoring` handler.

**Files:**
- Create: `jobs/handlers/person_intel.py`
- Test: `intel/people/tests/test_person_intel_handler.py`

**Step 1: Write test for handler interface**

```python
# intel/people/tests/test_person_intel_handler.py
"""Test person_intel handler stage dispatcher and interface."""
import asyncio

def test_handler_has_process_stage():
    from jobs.handlers.person_intel import process_stage
    assert asyncio.iscoroutinefunction(process_stage)

def test_handler_has_version_deps():
    from jobs.handlers.person_intel import VERSION_DEPS
    assert isinstance(VERSION_DEPS, dict)
    assert "discover" in VERSION_DEPS
    assert "enrich" in VERSION_DEPS
    assert "research" in VERSION_DEPS
    assert "score" in VERSION_DEPS
```

**Step 2: Run test to verify it fails**

Run: `python -m pytest intel/people/tests/test_person_intel_handler.py -v`
Expected: FAIL (module not found)

**Step 3: Create handler**

```python
#!/usr/bin/env python
"""Unified person intelligence handler.

Stages:
  1. discover — Serper web discovery + identity clustering → discovery.json
  2. enrich   — Multi-source API enrichment (Apollo, SEC, etc.) → enrichment.json
  3. research — LLM web-search-grounded evidence gathering (uses enrich context) → results table
  4. score    — LLM scoring on 3 core dimensions → person_scores table
  5. assess   — Parallel fan-out of optional assessors (academic, engineering, etc.) → results table

Data flow: Filesystem (intel/people/data/{slug}/) is canonical.
           Jobs results table stores stage status + summary metrics.
"""

from __future__ import annotations

import json
import sqlite3
from contextlib import closing
from pathlib import Path

from jobs.lib.db import open_raw_db
from jobs.lib.db.results import get_result
from jobs.lib.handler_base import make_stage_dispatcher
from jobs.lib.logging import get_handler_logger
from jobs.lib.stages import llm_research
from jobs.lib.tracker import RetryLaterError
from lib.llm.json import call_llm_json

log = get_handler_logger("person_intel")

PEOPLE_DB = "intel/people/data/people.db"
DATA_DIR = Path("intel/people/data")
MODEL = "gemini-flash"


# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------

_SCHEMA_DDL = """
CREATE TABLE IF NOT EXISTS person_scores (
    person_slug       TEXT PRIMARY KEY,
    overall           REAL NOT NULL,
    prior_success     REAL,
    network_quality   REAL,
    technical_depth   REAL,
    evidence_json     TEXT,
    model             TEXT,
    scored_at         TEXT DEFAULT (datetime('now')),
    handler_version   TEXT
);
"""


def _ensure_schema():
    conn = sqlite3.connect(PEOPLE_DB)
    conn.executescript(_SCHEMA_DDL)
    conn.close()


def _save_score(slug: str, result: dict, model: str):
    _ensure_schema()
    dims = result.get("dimensions", {})
    conn = sqlite3.connect(PEOPLE_DB)
    conn.execute("""
        INSERT INTO person_scores (person_slug, overall, prior_success,
            network_quality, technical_depth, evidence_json, model)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(person_slug) DO UPDATE SET
            overall = excluded.overall,
            prior_success = excluded.prior_success,
            network_quality = excluded.network_quality,
            technical_depth = excluded.technical_depth,
            evidence_json = excluded.evidence_json,
            model = excluded.model,
            scored_at = datetime('now')
    """, (
        slug,
        result.get("overall", 0),
        dims.get("prior_success", {}).get("score"),
        dims.get("network_quality", {}).get("score"),
        dims.get("technical_depth", {}).get("score"),
        json.dumps(result),
        model,
    ))
    conn.commit()
    conn.close()


# ---------------------------------------------------------------------------
# Stage: discover
# ---------------------------------------------------------------------------

async def _stage_discover(*, item_key: str, data: dict, job) -> dict:
    """Run Serper web discovery + identity clustering. Saves discovery.json."""
    from intel.people.lib.discover import discover_person, build_profile, save_profile_quick

    name = data.get("name", item_key)
    hint = data.get("headline") or data.get("bio") or ""
    person_dir = DATA_DIR / item_key
    person_dir.mkdir(parents=True, exist_ok=True)

    result = await discover_person(name, hint=hint, min_score=0.3)
    if not result or not result.get("candidates"):
        log.warning("no_candidates", slug=item_key, name=name)
        # Still save what we have — don't block pipeline
        result = result or {}

    # Save discovery artifacts to filesystem
    discovery_path = person_dir / "discovery.json"
    discovery_path.write_text(json.dumps(result, indent=2, default=str))

    # Build and save quick profile
    profile = build_profile(result)
    save_profile_quick(profile, person_dir)

    log.info("discovered", slug=item_key, name=name,
             candidates=len(result.get("candidates", [])),
             confirmed=len(result.get("confirmed_profiles", [])))

    return {
        "candidates": len(result.get("candidates", [])),
        "confirmed_profiles": len(result.get("confirmed_profiles", [])),
        "_files": ["discovery.json", "profile_quick.yaml"],
    }


# ---------------------------------------------------------------------------
# Stage: enrich
# ---------------------------------------------------------------------------

async def _stage_enrich(*, item_key: str, data: dict, job) -> dict:
    """Run multi-source API enrichment. Saves enrichment.json."""
    from intel.people.lib.enrich import run_enrichment, extract_photos, save_photos_to_db

    name = data.get("name", item_key)
    hint = data.get("headline") or data.get("bio") or ""
    person_dir = DATA_DIR / item_key
    person_dir.mkdir(parents=True, exist_ok=True)

    enrichment = await run_enrichment(name, hint=hint)

    # Separate successes from failures
    ok = {k: v for k, v in enrichment.items() if "error" not in v}
    failed = {k: v.get("error", "") for k, v in enrichment.items() if "error" in v}

    if not ok:
        raise RetryLaterError(f"All enrichment sources failed for {item_key}")

    # Save enrichment to filesystem
    enrichment_path = person_dir / "enrichment.json"
    enrichment_path.write_text(json.dumps(enrichment, indent=2, default=str))

    # Extract and save photos
    photos = extract_photos(enrichment)
    if photos:
        save_photos_to_db(item_key, photos)

    log.info("enriched", slug=item_key, name=name,
             ok=len(ok), failed=len(failed),
             failed_sources=list(failed.keys()) if failed else None)

    return {
        "sources_ok": len(ok),
        "sources_failed": len(failed),
        "failures": failed if failed else None,
        "_files": ["enrichment.json"],
    }


# ---------------------------------------------------------------------------
# Stage: research — LLM web-search-grounded evidence gathering
# ---------------------------------------------------------------------------

_RESEARCH_SYSTEM = "You are a talent intelligence researcher. Return valid JSON only."

_RESEARCH_PROMPT = """Research this person thoroughly using web search. Gather concrete evidence.

Name: {name}
Title: {title}
Company: {company}
Location: {location}
Headline: {headline}
Bio: {bio}

{enrichment_context}

Find and return JSON:
{{
  "career_history": [
    {{"org": "company", "role": "title", "period": "dates", "notes": "key accomplishments"}}
  ],
  "exits_and_outcomes": [
    {{"company": "name", "outcome": "IPO/acquisition/shutdown", "details": "valuation, acquirer, etc."}}
  ],
  "network_signals": [
    {{"connection": "person or firm name", "relationship": "co-founder/board/investor/advisor", "quality": "tier description"}}
  ],
  "technical_contributions": [
    {{"type": "patent/paper/open_source/talk", "title": "name", "details": "citations, impact, etc."}}
  ],
  "education": [
    {{"institution": "name", "degree": "type", "field": "area", "year": "graduation year or null"}}
  ],
  "notable_facts": ["any other relevant facts about this person"],
  "confidence": "high|medium|low"
}}

Be thorough — search for the person's name, company affiliations, patents, publications, and public profiles.
Verify facts against the enrichment data provided. Focus on gaps and deeper context."""


def _build_enrichment_context(item_key: str) -> str:
    """Build enrichment summary for research prompt from cached enrichment.json."""
    enrichment_path = DATA_DIR / item_key / "enrichment.json"
    if not enrichment_path.is_file():
        return ""

    try:
        enrichment = json.loads(enrichment_path.read_text())
    except Exception:
        return ""

    sections = []

    # Apollo
    apollo = enrichment.get("apollo", {})
    if apollo and "error" not in apollo:
        if apollo.get("name"):
            sections.append(f"Apollo: {apollo.get('name')}, {apollo.get('title', '')} at {apollo.get('organization_name', '')}")

    # Patents
    patents = enrichment.get("patents", {})
    if patents and "error" not in patents:
        count = patents.get("total_patent_count", 0)
        if count:
            sections.append(f"Patents: {count} patents found via USPTO")

    # Semantic Scholar
    scholar = enrichment.get("semantic_scholar", {})
    if scholar and "error" not in scholar:
        pcount = scholar.get("paper_count", 0)
        hindex = scholar.get("h_index")
        if pcount:
            sections.append(f"Semantic Scholar: {pcount} papers, h-index={hindex}")

    # SEC
    sec = enrichment.get("sec", {})
    if sec and "error" not in sec:
        filings = sec.get("total_filings", 0)
        if filings:
            sections.append(f"SEC: {filings} filings found")

    # Wikipedia
    wiki = enrichment.get("wikipedia", {})
    if wiki and "error" not in wiki and wiki.get("title"):
        sections.append(f"Wikipedia: article exists — '{wiki.get('title')}'")

    if not sections:
        return ""

    return "Known facts from API enrichment (use as hints, verify independently):\n" + "\n".join(f"- {s}" for s in sections)


async def _stage_research(*, item_key: str, data: dict, job) -> dict:
    name = data.get("name", item_key)
    enrichment_context = _build_enrichment_context(item_key)
    prompt = _RESEARCH_PROMPT.format(
        name=name,
        title=data.get("title", ""),
        company=data.get("company", ""),
        location=data.get("location", ""),
        headline=data.get("headline", ""),
        bio=data.get("bio", ""),
        enrichment_context=enrichment_context,
    )
    result = await llm_research(prompt, model=MODEL, system=_RESEARCH_SYSTEM)
    if not result:
        raise RetryLaterError(f"Empty research result for {item_key}")

    log.info("researched", slug=item_key, name=name,
             careers=len(result.get("career_history", [])),
             confidence=result.get("confidence"),
             had_enrichment=bool(enrichment_context))
    return result


# ---------------------------------------------------------------------------
# Stage: score — 3-dimension scoring from cached research
# ---------------------------------------------------------------------------

_SCORING_SYSTEM = """You are a founder evaluator for a venture capital intelligence system.
Score the person on three dimensions from 0 to 10, with evidence for each.
Be calibrated: 10 is reserved for truly exceptional (Elon Musk, Jensen Huang level).
A solid serial entrepreneur with one good exit is a 6-7 on prior_success.
Return valid JSON only."""

_SCORING_PROMPT = """Evaluate this person as a potential founder or key operator.

Research evidence:
{research_json}

Score on these three dimensions (0-10 each):

1. **prior_success**: Previous exits, company outcomes, role trajectory, leadership impact.
   - 0-2: No notable outcomes
   - 3-4: Some leadership experience, no exits
   - 5-6: One meaningful exit or strong trajectory at major companies
   - 7-8: Multiple exits or major company impact (IPO, large acquisition)
   - 9-10: Legendary track record (serial billion-dollar exits)

2. **network_quality**: Connections to VCs, operators, investors, technical leaders.
   - 0-2: Minimal visible network
   - 3-4: Connected within one company/community
   - 5-6: Known in regional tech scene, some VC connections
   - 7-8: Well-connected across VCs and tech leaders
   - 9-10: Central node in tech/VC ecosystem

3. **technical_depth**: Patents, papers, open source, technical roles, education pedigree.
   - 0-2: Non-technical background
   - 3-4: Some technical roles
   - 5-6: Strong technical background, top school or meaningful contributions
   - 7-8: Deep expertise, patents/papers, technical leadership at scale
   - 9-10: World-class technical contributor (major open source, breakthrough research)

Return JSON:
{{
  "prior_success": {{"score": <0-10>, "evidence": ["point1", "point2", ...]}},
  "network_quality": {{"score": <0-10>, "evidence": ["point1", "point2", ...]}},
  "technical_depth": {{"score": <0-10>, "evidence": ["point1", "point2", ...]}},
  "summary": "One paragraph overall assessment",
  "strengths": ["strength1", "strength2"],
  "risks": ["risk1", "risk2"]
}}"""

DIMENSION_WEIGHTS = {
    "prior_success": 0.40,
    "network_quality": 0.30,
    "technical_depth": 0.30,
}


async def _stage_score(*, item_key: str, data: dict, job) -> dict:
    with closing(open_raw_db()) as conn:
        research_data = get_result(conn, job.id, item_key, "research")
    if not research_data:
        raise RetryLaterError(f"No research data for {item_key}")

    research_json = json.dumps(research_data, indent=2, default=str)
    prompt = _SCORING_PROMPT.format(research_json=research_json)

    result = await call_llm_json(MODEL, prompt, system=_SCORING_SYSTEM, temperature=0)
    if not isinstance(result, dict):
        raise RetryLaterError(f"LLM returned non-dict for {item_key}")

    overall = 0.0
    dimensions = {}
    for dim, weight in DIMENSION_WEIGHTS.items():
        dim_data = result.get(dim, {})
        score = dim_data.get("score", 0) if isinstance(dim_data, dict) else 0
        overall += score * weight
        dimensions[dim] = dim_data

    name = data.get("name", item_key)
    scored = {
        "slug": item_key,
        "name": name,
        "overall": round(overall, 1),
        "dimensions": dimensions,
        "summary": result.get("summary", ""),
        "strengths": result.get("strengths", []),
        "risks": result.get("risks", []),
        "model": MODEL,
    }

    _save_score(item_key, scored, MODEL)
    log.info("scored", slug=item_key, name=name, overall=scored["overall"])
    return scored


process_stage = make_stage_dispatcher()
VERSION_DEPS = {
    "discover": [_stage_discover],
    "enrich": [_stage_enrich],
    "research": [_stage_research],
    "score": [_stage_score],
}
```

**Step 4: Run tests**

Run: `python -m pytest intel/people/tests/test_person_intel_handler.py -v`
Expected: PASS

**Step 5: Commit**

```bash
git add jobs/handlers/person_intel.py intel/people/tests/test_person_intel_handler.py
git commit -m "feat(jobs): create unified person_intel handler with discover/enrich/research/score stages"
```

---

## Task 4: Update jobs.yaml — rename people_scoring to person_intel

Replace the `people_scoring` job definition with `person_intel` using the new 4-stage pipeline.

**Files:**
- Modify: `jobs/jobs.yaml` (lines 1250-1280)

**Step 1: Replace job definition**

Find the `people_scoring:` block (lines 1250-1280) and replace with:

```yaml
  person_intel:
    emoji: "🧠"
    kind: backfill
    tags: [vc-intel, people]
    description: "Unified person intelligence — discover → enrich → research → score (web discovery, API enrichment, LLM evidence gathering, 3-dimension scoring)"
    discovery:
      strategy: people_db
    stages:
      - name: discover
        concurrency: 3
        outputs:
          candidates: identity candidates found
          confirmed_profiles: confirmed profile URLs
      - name: enrich
        concurrency: 5
        outputs:
          sources_ok: enrichment sources that returned data
          sources_failed: enrichment sources that failed
      - name: research
        concurrency: 3
        outputs:
          career_history: career roles and accomplishments
          exits_and_outcomes: company exits and valuations
          network_signals: VC/operator connections
          technical_contributions: patents, papers, open source
      - name: score
        concurrency: 10
        outputs:
          overall: weighted overall score (0-10)
          summary: one-paragraph assessment
    stage_deps:
      enrich: [discover]
      research: [enrich]
      score: [research]
    handler: handlers.person_intel.process_stage
    pacing:
      max_per_hour: 60
    storage:
      base_dir: ~/all-code/rivus/intel/people/data
      pattern: "{item_key}"
    detail_columns: [name, overall, summary]
    enabled: true
    priority: 3
```

**Step 2: Verify YAML parses**

Run: `python -c "from jobs.lib.job import load_jobs; jobs = load_jobs(); print([j.id for j in jobs if 'person' in j.id or 'people' in j.id])"`
Expected: `['person_intel']` (not `people_scoring`)

**Step 3: Run handler test to verify wiring**

Run: `inv jobs.test -j person_intel --dry-run` (if items exist in people.db)
Expected: Handler resolves, no import errors

**Step 4: Commit**

```bash
git add jobs/jobs.yaml
git commit -m "feat(jobs): rename people_scoring → person_intel with 4-stage pipeline"
```

---

## Task 5: Delete old people_scoring handler

Now that person_intel replaces it, remove the old handler.

**Files:**
- Delete: `jobs/handlers/people_scoring.py`

**Step 1: Verify nothing else imports people_scoring**

Run: `grep -r "people_scoring" jobs/ intel/ --include="*.py" --include="*.yaml" -l`
Expected: Only `jobs/handlers/people_scoring.py` itself (jobs.yaml already updated)

**Step 2: Delete old handler**

```bash
trash jobs/handlers/people_scoring.py
```

**Step 3: Run person_intel tests again**

Run: `python -m pytest intel/people/tests/test_person_intel_handler.py -v`
Expected: PASS

**Step 4: Commit**

```bash
git add -u jobs/handlers/people_scoring.py
git commit -m "chore(jobs): remove old people_scoring handler (replaced by person_intel)"
```

---

## Task 6: Update documentation and CLAUDE.md

Update references to the renamed job.

**Files:**
- Modify: `CLAUDE.md` — update jobs table if people_scoring listed
- Modify: `jobs/CLAUDE.md` — update jobs table
- Modify: `intel/people/TODO.md` — reference person_intel job

**Step 1: Update jobs/CLAUDE.md jobs table**

Add `person_intel` to the jobs table:

```markdown
| `person_intel`               | --               | discover -> enrich -> research -> score             |
```

**Step 2: Update intel/people/TODO.md**

Update assessment stage items to reference `person_intel` job handler instead of standalone stages.

**Step 3: Commit**

```bash
git add CLAUDE.md jobs/CLAUDE.md intel/people/TODO.md
git commit -m "docs: update references from people_scoring to person_intel"
```

---

## Task 7: End-to-end test with a real person

Run the full pipeline on one person to verify data flows correctly.

**Step 1: Pick a test person from people.db**

```bash
python -c "
import sqlite3
conn = sqlite3.connect('intel/people/data/people.db')
conn.row_factory = sqlite3.Row
row = conn.execute('SELECT slug, name FROM people LIMIT 1').fetchone()
print(f'{row[\"slug\"]}: {row[\"name\"]}')
"
```

**Step 2: Run single item through pipeline**

```bash
inv jobs.test -j person_intel --item SLUG_FROM_STEP_1
```

Expected: All 4 stages complete (discover → enrich → research → score). Check:
- `intel/people/data/{slug}/discovery.json` exists
- `intel/people/data/{slug}/enrichment.json` exists
- `person_scores` table has a row for this slug
- No errors in output

**Step 3: Verify scores look reasonable**

```bash
python -c "
import sqlite3, json
conn = sqlite3.connect('intel/people/data/people.db')
conn.row_factory = sqlite3.Row
row = conn.execute('SELECT * FROM person_scores ORDER BY scored_at DESC LIMIT 1').fetchone()
print(f'{row[\"person_slug\"]}: overall={row[\"overall\"]}')
evidence = json.loads(row['evidence_json'])
for dim in ['prior_success', 'network_quality', 'technical_depth']:
    d = evidence.get('dimensions', {}).get(dim, {})
    print(f'  {dim}: {d.get(\"score\", \"?\")}')
"
```

---

## Future Tasks (not in this plan)

These are documented in `intel/people/TODO.md` and can be added as incremental work:

- **Task F1: Add assess stage** — Single stage with internal `asyncio.gather` over enabled assessor modules (academic, engineering, publication, media)
- **Task F2: Score as cost gate** — Only run assess stage for people with core score > threshold
- **Task F3: Unified CLI** — `intel people run SLUG --through enrich` that calls the same library functions
- **Task F4: Deep research integration** — Wire `llm_deep_research()` (from discover.py) into the research stage as an optional mode