src/redstack/features/career.py

from __future__ import annotations

import math
import unicodedata
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import date
from typing import Final

from redstack.domain.enums import EvidenceKind
from redstack.domain.provenance import EvidenceRef
from redstack.domain.source import RawCandidate, RawPosition
from redstack.features.view import (
    DURATION_SATURATION_MONTHS,
    STALE_HALF_LIFE_DAYS,
    CellEmission,
    FeatureCell,
    FeatureId,
    bounded_log_scale,
    cell,
    clamp_unit,
    days_between,
    make_evidence,
    recency_unit,
)

__feature_version__ = "1.1.0"

_CAREER = EvidenceKind.CAREER_FIELD
_PROFILE = EvidenceKind.PROFILE_FIELD

# --------------------------------------------------------------------------- #
# Lexicons (lowercased substring matches over NFC-normalized text).           #
# --------------------------------------------------------------------------- #

# Title seniority ladder: (rank, tokens). Highest matching rank wins; an IC
# title with no ladder token defaults to rank 2 ("mid").
_TITLE_LADDER: Final[tuple[tuple[int, tuple[str, ...]], ...]] = (
    (0, ("intern", "trainee", "apprentice")),
    (1, ("junior", "jr ", "jr.", "associate", "entry", "graduate")),
    (2, ("engineer", "developer", "analyst", "scientist", "programmer", "sde")),
    (3, ("senior", "sr ", "sr.", "lead", "specialist", "ii", "iii")),
    (4, ("staff", "principal", "architect", "manager", "head", "director")),
    (
        5,
        (
            "vp",
            "vice president",
            "chief",
            "cto",
            "ceo",
            "founder",
            "co-founder",
            "partner",
        ),
    ),
)
_DEFAULT_TITLE_RANK: Final[int] = 2
_MAX_RANK: Final[float] = 5.0

# Scope cues found in *descriptions* (the corroborating evidence).
_SCOPE_CUES: Final[tuple[tuple[int, tuple[str, ...]], ...]] = (
    (5, ("founded", "co-founded", "started the company", "p&l", "org of")),
    (
        4,
        (
            "managed a team",
            "led a team",
            "led the team",
            "people management",
            "direct reports",
            "headcount",
            "hired",
            "owned the",
            "set the strategy",
            "roadmap ownership",
            "architected",
            "org-wide",
        ),
    ),
    (
        3,
        (
            "led",
            "owned",
            "drove",
            "mentored",
            "designed",
            "spearheaded",
            "end-to-end",
            "cross-functional",
        ),
    ),
    (
        2,
        (
            "built",
            "implemented",
            "developed",
            "shipped",
            "wrote",
            "coded",
            "delivered",
            "contributed",
        ),
    ),
)
_DEFAULT_SCOPE_RANK: Final[int] = 1

_CONSULTING_FIRMS: Final[tuple[str, ...]] = (
    "tcs",
    "tata consultancy",
    "infosys",
    "wipro",
    "accenture",
    "cognizant",
    "capgemini",
    "deloitte",
    "hcl",
    "tech mahindra",
    "mindtree",
    "mphasis",
    "ibm global services",
    "ltimindtree",
    "persistent systems",
)
_CONSULTING_CUES: Final[tuple[str, ...]] = (
    "consulting",
    "consultancy",
    "client",
    "clients",
    "staff augmentation",
    "outsourc",
    "managed services",
    "system integrat",
    "body shop",
    "billable",
    "engagement",
    "sow",
    "client-facing",
)
_PRODUCT_CUES: Final[tuple[str, ...]] = (
    "product",
    "saas",
    "platform",
    "our app",
    "our users",
    "feature flag",
    "a/b test",
    "growth",
    "consumer",
    "in-house",
    "proprietary product",
    "product-led",
    "user base",
    "mau",
    "dau",
)

_CODING_CUES: Final[tuple[str, ...]] = (
    "python",
    "java",
    "c++",
    "golang",
    "rust",
    "scala",
    "typescript",
    "pytorch",
    "tensorflow",
    "spark",
    "kubernetes",
    "sql",
    "wrote",
    "coded",
    "implemented",
    "built",
    "developed",
    "refactored",
    "debugged",
    "unit test",
    "pull request",
    "code review",
    "commit",
)
_ARCH_ONLY_CUES: Final[tuple[str, ...]] = (
    "oversaw",
    "strategy",
    "roadmap",
    "stakeholder",
    "governance",
    "steering",
    "high-level design",
    "review boards",
    "vendor management",
    "budget",
    "presentation",
)
_RESEARCH_CUES: Final[tuple[str, ...]] = (
    "research",
    "published",
    "publication",
    "paper",
    "novel approach",
    "phd",
    "thesis",
    "state-of-the-art",
    "neurips",
    "icml",
    "acl",
    "cvpr",
    "prototype only",
    "proof of concept",
    "experimental study",
)
_MGMT_CUES: Final[tuple[str, ...]] = (
    "managed a team",
    "people management",
    "direct reports",
    "headcount",
    "hiring",
    "performance reviews",
    "1:1s",
    "team of",
    "line manager",
    "managed engineers",
    "delegated",
)
_PRODUCTION_CUES: Final[tuple[str, ...]] = (
    "production",
    "in prod",
    "deployed",
    "serving",
    "live traffic",
    "at scale",
    "latency",
    "throughput",
    "uptime",
    "sla",
    "ci/cd",
    "monitoring",
    "on-call",
    "rollout",
    "millions of",
    "qps",
    "p99",
)
_TECH_BREADTH_TOKENS: Final[tuple[str, ...]] = (
    "python",
    "java",
    "c++",
    "golang",
    "rust",
    "scala",
    "typescript",
    "pytorch",
    "tensorflow",
    "spark",
    "kubernetes",
    "sql",
    "kafka",
    "airflow",
    "ray",
    "onnx",
)


def _text(value: str) -> str:
    return " ".join(unicodedata.normalize("NFC", value).lower().split())


def _contains_any(text: str, tokens: Sequence[str]) -> bool:
    return any(token in text for token in tokens)


def _count_any(text: str, tokens: Sequence[str]) -> int:
    return sum(1 for token in tokens if token in text)


def _ladder_rank(
    text: str, ladder: tuple[tuple[int, tuple[str, ...]], ...], default: int
) -> int:
    best = default
    for rank, tokens in ladder:
        if _contains_any(text, tokens):
            best = max(best, rank)
    return best


def _saturating(count: int, *, saturation: float) -> float:
    """Diminishing-returns presence score for a token hit-count."""
    return bounded_log_scale(float(count), saturation=saturation)


# --------------------------------------------------------------------------- #
# Per-position derived view (immutable).                                      #
# --------------------------------------------------------------------------- #


@dataclass(frozen=True, slots=True)
class _PosView:
    index: int
    duration_months: float
    tenure_weight: float
    recency_weight: float
    combined_weight: float
    title_rank: int
    scope_rank: int
    level: float
    size_ordinal: int
    description_present: bool
    product_score: float
    consulting_score: float
    coding_score: float
    arch_score: float
    research_score: float
    mgmt_score: float
    production_score: float
    hands_on_score: float
    tech_tokens: frozenset[str]
    start_date: date
    end_ref: date


def _build_pos_view(index: int, pos: RawPosition, as_of: date) -> _PosView:
    title_text = _text(pos.title)
    desc_text = _text(pos.description)
    company_text = _text(pos.company)
    industry_text = _text(pos.industry)
    blob = f"{desc_text} {industry_text} {company_text}"
    description_present = bool(desc_text)

    title_rank = _ladder_rank(title_text, _TITLE_LADDER, _DEFAULT_TITLE_RANK)
    scope_rank = (
        _ladder_rank(desc_text, _SCOPE_CUES, _DEFAULT_SCOPE_RANK)
        if description_present
        else _DEFAULT_SCOPE_RANK
    )
    # Corroborated level: description-dominant. Uncorroborated titles are
    # discounted (only 50% of the claimed rank counts) so a fabricated
    # "Principal" with an empty description cannot lift trajectory metrics.
    if description_present:
        level = 0.6 * float(scope_rank) + 0.4 * float(title_rank)
    else:
        level = 0.5 * float(title_rank)

    # Classification scores (description/industry dominant; company name is a
    # strong consulting tell via the known-firm list).
    consulting_hits = _count_any(blob, _CONSULTING_CUES)
    is_known_consulting = _contains_any(company_text, _CONSULTING_FIRMS)
    product_hits = _count_any(blob, _PRODUCT_CUES)
    consulting_score = clamp_unit(
        max(
            _saturating(consulting_hits, saturation=2.0),
            1.0 if is_known_consulting else 0.0,
        )
    )
    product_score = clamp_unit(
        _saturating(product_hits, saturation=2.0) * (1.0 - 0.5 * consulting_score)
    )

    coding_score = _saturating(_count_any(blob, _CODING_CUES), saturation=3.0)
    arch_score = _saturating(_count_any(blob, _ARCH_ONLY_CUES), saturation=2.0)
    research_score = _saturating(_count_any(blob, _RESEARCH_CUES), saturation=2.0)
    mgmt_score = _saturating(_count_any(blob, _MGMT_CUES), saturation=2.0)
    production_score = _saturating(_count_any(blob, _PRODUCTION_CUES), saturation=2.0)
    # Hands-on engineering: coding presence net of architecture-only signalling.
    hands_on_score = clamp_unit(coding_score - 0.5 * arch_score)
    tech_tokens = frozenset(t for t in _TECH_BREADTH_TOKENS if t in blob)

    duration_months = float(pos.duration_months)
    tenure_weight = duration_months if duration_months > 0.0 else 0.0
    # Recency: days since the role ended (current → 0 → fully recent).
    end_ref = as_of if (pos.is_current or pos.end_date is None) else pos.end_date
    days_since = float(days_between(as_of, end_ref))
    recency_weight = recency_unit(days_since, half_life_days=STALE_HALF_LIFE_DAYS)
    combined_weight = tenure_weight * recency_weight

    return _PosView(
        index=index,
        duration_months=duration_months,
        tenure_weight=tenure_weight,
        recency_weight=recency_weight,
        combined_weight=combined_weight,
        title_rank=title_rank,
        scope_rank=scope_rank,
        level=level,
        size_ordinal=pos.company_size.ordinal,
        description_present=description_present,
        product_score=product_score,
        consulting_score=consulting_score,
        coding_score=coding_score,
        arch_score=arch_score,
        research_score=research_score,
        mgmt_score=mgmt_score,
        production_score=production_score,
        hands_on_score=hands_on_score,
        tech_tokens=tech_tokens,
        start_date=pos.start_date,
        end_ref=end_ref,
    )


# --------------------------------------------------------------------------- #
# Weighting + small numeric utilities.                                        #
# --------------------------------------------------------------------------- #


def _weighted(values: tuple[float, ...], weights: tuple[float, ...]) -> float:
    """Weighted mean; falls back to the unweighted mean if all weights are 0."""
    total = math.fsum(weights)
    if total <= 0.0:
        if not values:
            return 0.0
        return math.fsum(values) / len(values)
    return math.fsum(v * w for v, w in zip(values, weights)) / total


def _argmax_contribution(
    views: tuple[_PosView, ...], scores: tuple[float, ...], weights: tuple[float, ...]
) -> int:
    """Index of the position contributing most (score×weight); ties → lowest index."""
    best_index = views[0].index
    best_value = -1.0
    for view, score, weight in zip(views, scores, weights):
        contribution = score * (weight if weight > 0.0 else 1.0)
        if contribution > best_value:
            best_value = contribution
            best_index = view.index
    return best_index


def _union_months(views: tuple[_PosView, ...]) -> float:
    """Total non-overlapping employed months across positions (years×12)."""
    intervals: list[tuple[date, date]] = []
    for view in views:
        if view.end_ref > view.start_date:
            intervals.append((view.start_date, view.end_ref))
    if not intervals:
        return 0.0
    intervals.sort(key=lambda pair: pair[0])
    total_days = 0
    cursor_start, cursor_end = intervals[0]
    for start, end in intervals[1:]:
        if start <= cursor_end:
            if end > cursor_end:
                cursor_end = end
        else:
            total_days += (cursor_end - cursor_start).days
            cursor_start, cursor_end = start, end
    total_days += (cursor_end - cursor_start).days
    return float(total_days) / 365.25 * 12.0


# --------------------------------------------------------------------------- #
# Evidence minting from real RawCandidate paths.                              #
# --------------------------------------------------------------------------- #


def _position_scalar(pos: RawPosition, field: str) -> str | int | float | bool:
    if field == "title":
        return pos.title
    if field == "description":
        return pos.description
    if field == "company":
        return pos.company
    if field == "industry":
        return pos.industry
    if field == "company_size":
        return pos.company_size.value
    if field == "duration_months":
        return int(pos.duration_months)
    if field == "is_current":
        return pos.is_current
    if field == "start_date":
        return pos.start_date.isoformat()
    raise KeyError(f"unsupported career evidence field: {field!r}")


def _ev_position(raw: RawCandidate, index: int, field: str) -> EvidenceRef:
    pos = raw.career_history[index]
    return make_evidence(
        _CAREER,
        f"career_history[{index}].{field}",
        _position_scalar(pos, field),
        raw=raw,
    )


def _ev_years(raw: RawCandidate) -> EvidenceRef:
    return make_evidence(
        _PROFILE,
        "profile.years_of_experience",
        float(raw.profile.years_of_experience),
        raw=raw,
    )


# --------------------------------------------------------------------------- #
# Career group ids (layout order).                                            #
# --------------------------------------------------------------------------- #

_C_PROGRESSION = "career.progression_quality"
_C_STABILITY = "career.stability"
_C_PROMOTION = "career.promotion_velocity"
_C_TITLE_INFLATION = "career.title_inflation"
_C_ROLE_CONSISTENCY = "career.role_consistency"
_C_AUTHENTICITY = "career.experience_authenticity"
_C_COMPANY_PROGRESSION = "career.company_progression"
_C_PRODUCT_DENSITY = "career.product_company_density"
_C_CONSULTING_DENSITY = "career.consulting_density"
_C_TECHNICAL_DEPTH = "career.technical_depth"
_C_HANDS_ON = "career.hands_on_engineering"
_C_RESEARCH_ONLY = "career.research_only"
_C_MANAGEMENT_ONLY = "career.management_only"
_C_PRODUCTION_EXPOSURE = "career.production_exposure"

_PVS_PRODUCT_DENSITY = "pvs.product_density"
_PVS_CONSULTING_DENSITY = "pvs.consulting_density"
_PVS_PRODUCT_RECENT = "pvs.product_recent"


def _confidences(views: tuple[_PosView, ...]) -> tuple[float, float]:
    """(structural_confidence, textual_confidence) for the candidate."""
    count = len(views)
    count_factor = bounded_log_scale(float(count), saturation=4.0)
    described = sum(1 for v in views if v.description_present)
    desc_coverage = described / count if count else 0.0
    conf_struct = clamp_unit(0.55 + 0.40 * count_factor)
    conf_text = clamp_unit(0.30 + 0.55 * desc_coverage + 0.10 * count_factor)
    return conf_struct, conf_text


def extract_career(raw: RawCandidate, *, as_of: date) -> CellEmission:
    """Emit the 14 ``career.*`` cells in layout order.

    Pure and total: ``career_history`` has ≥1 position (schema), so single-role
    candidates yield neutral trajectory values at reduced confidence rather than
    errors. All weighting is against ``as_of``; no wall clock is read.
    """
    views = tuple(
        _build_pos_view(i, pos, as_of) for i, pos in enumerate(raw.career_history)
    )
    chronological = tuple(sorted(views, key=lambda v: v.start_date))
    n = len(views)

    tenure_w = tuple(v.tenure_weight for v in views)
    combined_w = tuple(v.combined_weight for v in views)
    recency_w = tuple(v.recency_weight for v in views)

    conf_struct, conf_text = _confidences(views)
    conf_traj = clamp_unit(0.5 * (conf_struct + conf_text))
    conf_mixed = clamp_unit(max(conf_text, 0.5 * conf_struct))

    emissions: list[tuple[FeatureId, FeatureCell]] = []

    # 1. progression_quality — corroborated-level trajectory (tenure-weighted).
    if n == 1:
        progression = 0.5
    else:
        scores: list[float] = []
        weights: list[float] = []
        for a, b in zip(chronological, chronological[1:]):
            delta = b.level - a.level
            step = 1.0 if delta > 0.25 else (0.6 if delta >= -0.25 else 0.2)
            scores.append(step)
            weights.append(a.tenure_weight + b.tenure_weight)
        progression = _weighted(tuple(scores), tuple(weights))
    emissions.append(
        (
            _C_PROGRESSION,
            cell(
                clamp_unit(progression),
                conf_traj,
                (
                    _ev_position(raw, chronological[0].index, "title"),
                    _ev_position(raw, chronological[-1].index, "title"),
                ),
            ),
        )
    )

    # 2. stability — normalized mean tenure blended with inverse sub-18m hop rate.
    completed = tuple(v for v in views if v.duration_months > 0.0)
    mean_tenure = (
        math.fsum(v.duration_months for v in completed) / len(completed)
        if completed
        else 0.0
    )
    tenure_norm = bounded_log_scale(mean_tenure, saturation=DURATION_SATURATION_MONTHS)
    hoppy = tuple(v for v in views if 0.0 < v.duration_months < 18.0)
    hop_rate = len(hoppy) / n if n else 0.0
    stability = clamp_unit(0.5 * tenure_norm + 0.5 * (1.0 - hop_rate))
    shortest = min(views, key=lambda v: (v.duration_months, v.index))
    emissions.append(
        (
            _C_STABILITY,
            cell(
                stability,
                conf_struct,
                (
                    _ev_position(raw, shortest.index, "duration_months"),
                    _ev_position(raw, shortest.index, "title"),
                ),
            ),
        )
    )

    # 3. promotion_velocity — level gained per year (chronological).
    if n == 1:
        velocity = 0.3
    else:
        gain = chronological[-1].level - chronological[0].level
        span_days = max(
            (chronological[-1].start_date - chronological[0].start_date).days, 0
        )
        span_years = max(span_days / 365.25, 0.5)
        rate_per_year = gain / span_years
        velocity = clamp_unit(rate_per_year / 0.5)  # 0.5 level/yr ⇒ saturates
    emissions.append(
        (
            _C_PROMOTION,
            cell(
                velocity,
                conf_struct,
                (
                    _ev_position(raw, chronological[0].index, "title"),
                    _ev_position(raw, chronological[-1].index, "title"),
                ),
            ),
        )
    )

    # 4. title_inflation — claimed title above corroborated scope (tenure-weighted).
    inflation_scores = tuple(
        clamp_unit(max(0, v.title_rank - v.scope_rank) / 3.0)
        if v.description_present
        else clamp_unit(0.6 * v.title_rank / _MAX_RANK)
        for v in views
    )
    title_inflation = clamp_unit(_weighted(inflation_scores, tenure_w))
    infl_idx = _argmax_contribution(views, inflation_scores, tenure_w)
    emissions.append(
        (
            _C_TITLE_INFLATION,
            cell(
                title_inflation,
                conf_text,
                (
                    _ev_position(raw, infl_idx, "title"),
                    _ev_position(raw, infl_idx, "description"),
                ),
            ),
        )
    )

    # 5. role_consistency — low level dispersion + track homogeneity.
    if n == 1:
        role_consistency = 0.7
    else:
        levels = tuple(v.level for v in views)
        dispersion = (max(levels) - min(levels)) / _MAX_RANK
        product_majority = sum(1 for v in views if v.product_score >= 0.5)
        consulting_majority = sum(1 for v in views if v.consulting_score >= 0.5)
        homogeneity = (
            max(
                product_majority,
                consulting_majority,
                n - product_majority - consulting_majority,
            )
            / n
        )
        role_consistency = clamp_unit(0.6 * (1.0 - dispersion) + 0.4 * homogeneity)
    emissions.append(
        (
            _C_ROLE_CONSISTENCY,
            cell(
                role_consistency,
                conf_mixed,
                (
                    _ev_position(raw, chronological[0].index, "title"),
                    _ev_position(raw, chronological[-1].index, "title"),
                ),
            ),
        )
    )

    # 6. experience_authenticity — derived (union) tenure vs stated years.
    derived_years = min(_union_months(views) / 12.0, 50.0)
    stated_years = float(raw.profile.years_of_experience)
    gap = abs(stated_years - derived_years) / max(stated_years, derived_years, 1.0)
    authenticity = clamp_unit(1.0 - gap)
    longest = max(views, key=lambda v: (v.duration_months, -v.index))
    emissions.append(
        (
            _C_AUTHENTICITY,
            cell(
                authenticity,
                conf_struct,
                (_ev_years(raw), _ev_position(raw, longest.index, "duration_months")),
            ),
        )
    )

    # 7. company_progression — company-size trajectory (tenure-weighted).
    if n == 1:
        company_progression = 0.5
    else:
        c_scores: list[float] = []
        c_weights: list[float] = []
        for a, b in zip(chronological, chronological[1:]):
            delta = b.size_ordinal - a.size_ordinal
            step = 1.0 if delta > 0 else (0.6 if delta == 0 else 0.3)
            c_scores.append(step)
            c_weights.append(a.tenure_weight + b.tenure_weight)
        company_progression = _weighted(tuple(c_scores), tuple(c_weights))
    emissions.append(
        (
            _C_COMPANY_PROGRESSION,
            cell(
                clamp_unit(company_progression),
                conf_struct,
                (
                    _ev_position(raw, chronological[0].index, "company_size"),
                    _ev_position(raw, chronological[-1].index, "company_size"),
                ),
            ),
        )
    )

    # 8. product_company_density — tenure×recency weighted product fraction.
    product_scores = tuple(v.product_score for v in views)
    product_density = clamp_unit(_weighted(product_scores, combined_w))
    prod_idx = _argmax_contribution(views, product_scores, combined_w)
    emissions.append(
        (
            _C_PRODUCT_DENSITY,
            cell(
                product_density,
                conf_mixed,
                (
                    _ev_position(raw, prod_idx, "company"),
                    _ev_position(raw, prod_idx, "description"),
                ),
            ),
        )
    )

    # 9. consulting_density — tenure×recency weighted consulting fraction.
    consulting_scores = tuple(v.consulting_score for v in views)
    consulting_density = clamp_unit(_weighted(consulting_scores, combined_w))
    cons_idx = _argmax_contribution(views, consulting_scores, combined_w)
    emissions.append(
        (
            _C_CONSULTING_DENSITY,
            cell(
                consulting_density,
                conf_mixed,
                (
                    _ev_position(raw, cons_idx, "company"),
                    _ev_position(raw, cons_idx, "description"),
                ),
            ),
        )
    )

    # 10. technical_depth — coding density + distinct-tech breadth.
    coding_scores = tuple(v.coding_score for v in views)
    coding_density = _weighted(coding_scores, tenure_w)
    distinct_tech = (
        len(frozenset().union(*(v.tech_tokens for v in views))) if views else 0
    )
    tech_breadth = bounded_log_scale(float(distinct_tech), saturation=8.0)
    technical_depth = clamp_unit(0.6 * coding_density + 0.4 * tech_breadth)
    depth_idx = _argmax_contribution(views, coding_scores, tenure_w)
    emissions.append(
        (
            _C_TECHNICAL_DEPTH,
            cell(
                technical_depth,
                conf_text,
                (
                    _ev_position(raw, depth_idx, "description"),
                    _ev_position(raw, depth_idx, "title"),
                ),
            ),
        )
    )

    # 11. hands_on_engineering — recency-weighted hands-on (coding net of arch-only).
    hands_on_scores = tuple(v.hands_on_score for v in views)
    hands_on = clamp_unit(_weighted(hands_on_scores, recency_w))
    hands_idx = _argmax_contribution(views, hands_on_scores, recency_w)
    emissions.append(
        (
            _C_HANDS_ON,
            cell(hands_on, conf_text, (_ev_position(raw, hands_idx, "description"),)),
        )
    )

    # 12. research_only — research density gated by absence of production.
    research_scores = tuple(v.research_score for v in views)
    research_density = _weighted(research_scores, tenure_w)
    production_scores = tuple(v.production_score for v in views)
    production_density = _weighted(production_scores, combined_w)
    research_only = clamp_unit(research_density * (1.0 - production_density))
    res_idx = _argmax_contribution(views, research_scores, tenure_w)
    emissions.append(
        (
            _C_RESEARCH_ONLY,
            cell(
                research_only, conf_text, (_ev_position(raw, res_idx, "description"),)
            ),
        )
    )

    # 13. management_only — management density gated by absence of hands-on IC work.
    mgmt_scores = tuple(v.mgmt_score for v in views)
    mgmt_density = _weighted(mgmt_scores, tenure_w)
    management_only = clamp_unit(mgmt_density * (1.0 - max(coding_density, hands_on)))
    mgmt_idx = _argmax_contribution(views, mgmt_scores, tenure_w)
    emissions.append(
        (
            _C_MANAGEMENT_ONLY,
            cell(
                management_only,
                conf_text,
                (
                    _ev_position(raw, mgmt_idx, "description"),
                    _ev_position(raw, mgmt_idx, "title"),
                ),
            ),
        )
    )

    # 14. production_exposure — tenure×recency weighted production density.
    production_exposure = clamp_unit(production_density)
    prodexp_idx = _argmax_contribution(views, production_scores, combined_w)
    emissions.append(
        (
            _C_PRODUCTION_EXPOSURE,
            cell(
                production_exposure,
                conf_mixed,
                (_ev_position(raw, prodexp_idx, "description"),),
            ),
        )
    )

    return tuple(emissions)


def extract_pvs(raw: RawCandidate, *, as_of: date) -> CellEmission:
    """Emit the Group 7 ``pvs.*`` cells (product-vs-service densities).

    ``pvs.product_density`` is tenure×recency weighted; ``pvs.product_recent``
    re-weights by recency alone (emphasising the latest roles); both pair with
    ``pvs.consulting_density``. Shares the Part-3 classification so the two
    groups never disagree on what counts as a product company.
    """
    views = tuple(
        _build_pos_view(i, pos, as_of) for i, pos in enumerate(raw.career_history)
    )
    combined_w = tuple(v.combined_weight for v in views)
    recency_w = tuple(v.recency_weight for v in views)
    conf_struct, conf_text = _confidences(views)
    conf_mixed = clamp_unit(max(conf_text, 0.5 * conf_struct))

    product_scores = tuple(v.product_score for v in views)
    consulting_scores = tuple(v.consulting_score for v in views)

    product_density = clamp_unit(_weighted(product_scores, combined_w))
    product_recent = clamp_unit(_weighted(product_scores, recency_w))
    consulting_density = clamp_unit(_weighted(consulting_scores, combined_w))

    prod_idx = _argmax_contribution(views, product_scores, combined_w)
    recent_idx = _argmax_contribution(views, product_scores, recency_w)
    cons_idx = _argmax_contribution(views, consulting_scores, combined_w)

    return (
        (
            _PVS_PRODUCT_DENSITY,
            cell(
                product_density,
                conf_mixed,
                (
                    _ev_position(raw, prod_idx, "company"),
                    _ev_position(raw, prod_idx, "description"),
                ),
            ),
        ),
        (
            _PVS_CONSULTING_DENSITY,
            cell(
                consulting_density,
                conf_mixed,
                (
                    _ev_position(raw, cons_idx, "company"),
                    _ev_position(raw, cons_idx, "description"),
                ),
            ),
        ),
        (
            _PVS_PRODUCT_RECENT,
            cell(
                product_recent,
                conf_mixed,
                (
                    _ev_position(raw, recent_idx, "company"),
                    _ev_position(raw, recent_idx, "description"),
                ),
            ),
        ),
    )


__all__: tuple[str, ...] = ("extract_career", "extract_pvs")