Overlay leftover services and LLM factory into periscope/src (2.44.0).

Copy api_logs, findings normalize/dedupe, billing_hook, datasheet_store,
cost_estimator, storage, purple_parts, email, digikey, and llm factory/types/base
without deleting the inherited dependency copies.

periscope/dependency/frontend/content/changelog.md

@@ -2,6 +2,13 @@
 
 What's new in Periscope.
 
+## 2.44.0 — 2026-09-20 — Overlay leftover services src still imported
+
+Pipeline helpers still pulled from `dependency/` (`api_logs`, `normalize_findings`, `dedupe_findings`, `billing_hook`, `datasheet_store`, `cost_estimator`, `storage`, `purple_parts`, `email`, `digikey`) plus LLM `factory` / `types` / `base` now resolve from `periscope/src`. Inherited copies stay on disk.
+
+- [New] src overlay of leftover `backend/services` modules the live pipeline/review path still imports.
+- [New] src overlay of `services/llm/{factory,types,base}.py` (DeepSeek provider stays native; Anthropic/Gemini files stay in dependency).
+
 ## 2.43.0 — 2026-09-20 — Overlay leftover PinScope modules src still imported
 
 `utils`, `resolve_passives`, `derating`, `bom_summary`, `pin_mux_check`, `led_current_check`, `pin_function_tokens`, plus live analysis `validate` / `validation_tools` / `pipeline` / `validation` / `extraction` (fallback), resolve from `periscope/src`. Skills (`extract-*`) and taxonomy JSON are copied to `periscope/src/{skills,taxonomy}`; Docker overlays them after `dependency/`. Inherited files stay on disk.

periscope/src/backend/services/api_logs.py

@@ -0,0 +1,135 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Per-project API call logging.
+
+Captures metadata for every LLM API call made during a pipeline run and
+serialises to JSONL for storage alongside other project artefacts. Pricing
+lives in ``backend.services.llm.pricing`` and is provider-aware.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from dataclasses import dataclass, field
+
+from pydantic import BaseModel
+
+# Re-exported for callers (pipeline.total_cost) — provider-aware now
+from backend.services.llm.pricing import cost_for_entry, total_cost  # noqa: F401
+
+
+class ApiLogEntry(BaseModel):
+    timestamp: str
+    stage: str  # pintable | rules | pattern | validation | ...
+    identifier: str  # MPN or component designator
+    model: str
+    provider: str = "deepseek"  # deepseek | anthropic | gemini
+    input_tokens: int
+    output_tokens: int
+    cache_creation_input_tokens: int = 0
+    cache_read_input_tokens: int = 0
+    duration_ms: int
+    stop_reason: str
+    skill_id: str | None = None
+    turns: int | None = None
+    error: str | None = None
+    cost_usd: float | None = None
+    credits_charged: float | None = None
+    # True when the call ran in an admin-initiated free context (e.g. regen)
+    # — the raw USD cost is still recorded for accounting, but no credits
+    # are charged to the user.
+    free: bool = False
+
+
+@dataclass
+class CallMeta:
+    """Metadata returned alongside every LLM API call result."""
+    input_tokens: int
+    output_tokens: int
+    cache_creation_input_tokens: int
+    cache_read_input_tokens: int
+    duration_ms: int
+    stop_reason: str
+    turns: int = 1
+
+
+# ---------------------------------------------------------------------------
+# Logger
+# ---------------------------------------------------------------------------
+
+@dataclass
+class ApiLogger:
+    """Collects API call log entries during a pipeline run.
+
+    ``free=True`` marks every entry as admin-initiated and zeros the
+    ``credits_charged`` field so downstream charging / reporting treats the
+    run as free to the user. The underlying USD cost is still recorded.
+    """
+    entries: list[dict] = field(default_factory=list)
+    free: bool = False
+
+    def log(self, **kwargs: object) -> None:
+        kwargs.setdefault("timestamp", datetime.now(timezone.utc).isoformat())
+        entry = ApiLogEntry(**kwargs)  # type: ignore[arg-type]
+        d = entry.model_dump()
+        d["cost_usd"] = round(cost_for_entry(d), 6)
+        if self.free:
+            d["credits_charged"] = 0.0
+            d["free"] = True
+        else:
+            # Attribute credits to this call using the same margin used by
+            # the credit service.  Local import to avoid a module-load cycle.
+            from backend.services.billing_hook import get_billing
+
+            d["credits_charged"] = get_billing().credits_for_api_cost(d["cost_usd"])
+        self.entries.append(d)
+
+    def to_jsonl(self) -> str:
+        if not self.entries:
+            return ""
+        return "\n".join(json.dumps(e) for e in self.entries) + "\n"
+
+    def flush(self, storage, user_id: str, project_id: str) -> None:
+        """Write the current entries to ``api_logs.jsonl`` in storage.
+
+        Called periodically during a pipeline run so a preempted worker
+        doesn't lose billing data. Idempotent — safe to call repeatedly;
+        each flush overwrites the prior copy with the latest entries.
+        """
+        text = self.to_jsonl()
+        if not text:
+            return
+        # Local import avoids a cycle with services.projects (which imports
+        # from services.storage which imports from here transitively).
+        from backend.services.projects import project_prefix
+
+        key = f"{project_prefix(user_id, project_id)}/api_logs.jsonl"
+        storage.write_text(key, text)
+
+
+def cache_stats_by_stage(entries: list[dict]) -> dict[str, dict]:
+    """Roll up prompt-cache hit rate per pipeline stage.
+
+    Returns ``{stage: {calls, input_tokens, cache_read_tokens, hit_ratio}}``.
+    ``hit_ratio`` is cache_read / input when input > 0, else 0.
+    """
+    out: dict[str, dict] = {}
+    for e in entries:
+        stage = str(e.get("stage") or "unknown")
+        bucket = out.setdefault(
+            stage,
+            {"calls": 0, "input_tokens": 0, "cache_read_tokens": 0, "hit_ratio": 0.0},
+        )
+        bucket["calls"] += 1
+        bucket["input_tokens"] += int(e.get("input_tokens") or 0)
+        bucket["cache_read_tokens"] += int(e.get("cache_read_input_tokens") or 0)
+    for bucket in out.values():
+        inp = bucket["input_tokens"]
+        bucket["hit_ratio"] = (
+            round(bucket["cache_read_tokens"] / inp, 4) if inp else 0.0
+        )
+    return out

periscope/src/backend/services/billing_hook.py

@@ -0,0 +1,197 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Open-core billing seam.
+
+Everything outside the billing modules (``credits``, ``credit_grants``,
+``stripe_billing``, ``stripe_customer_map``, ``auto_topup`` and the
+``billing``/``credits`` routers) talks to billing exclusively through
+:func:`get_billing`. With ``BILLING_ENABLED=false`` the returned
+:class:`NullBilling` makes every pipeline run free — the same shape as the
+existing admin ``free=True`` path — so the core can run with no credits
+ledger, no Stripe, and no billing routes mounted.
+
+This module must stay a leaf: no billing module is imported at module
+level (``CreditsBilling`` lazy-imports inside each method), so the core
+never touches the Stripe SDK when billing is disabled.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol
+
+from backend.config import settings
+
+if TYPE_CHECKING:
+    from backend.services.storage import StorageBackend
+
+
+class InsufficientCredits(RuntimeError):
+    """Raised when a charge would drop the balance below zero."""
+
+    def __init__(self, required: float, available: float) -> None:
+        super().__init__(
+            f"Insufficient credits: required {required}, available {available}"
+        )
+        self.required = required
+        self.available = available
+
+
+class BillingHook(Protocol):
+    """The full billing surface the core is allowed to depend on."""
+
+    def credits_for_api_cost(self, cost_usd: float) -> float: ...
+
+    def get_balance(self, storage: "StorageBackend", user_id: str) -> float: ...
+
+    def charge(
+        self,
+        storage: "StorageBackend",
+        user_id: str,
+        amount: float,
+        *,
+        reason: str = "pipeline_charge",
+        run_id: str | None = None,
+        unit_id: str | None = None,
+        allow_overdraft: bool = False,
+    ) -> None: ...
+
+    def ensure_trial_grant(self, storage: "StorageBackend", user_id: str) -> bool: ...
+
+    def list_user_ids(self, storage: "StorageBackend") -> list[str]: ...
+
+    async def maybe_auto_topup(
+        self, storage: "StorageBackend", user_id: str
+    ) -> dict | None: ...
+
+
+class NullBilling:
+    """Billing disabled: everything is free and nothing is written.
+
+    ``credits_for_api_cost`` returning 0.0 is the linchpin — every
+    ``ApiLogger`` entry gets ``credits_charged=0``, so the pipeline's
+    charge path early-returns and the credit gate always allows.
+    """
+
+    def credits_for_api_cost(self, cost_usd: float) -> float:
+        return 0.0
+
+    def get_balance(self, storage: "StorageBackend", user_id: str) -> float:
+        return 0.0
+
+    def charge(
+        self,
+        storage: "StorageBackend",
+        user_id: str,
+        amount: float,
+        *,
+        reason: str = "pipeline_charge",
+        run_id: str | None = None,
+        unit_id: str | None = None,
+        allow_overdraft: bool = False,
+    ) -> None:
+        return None
+
+    def ensure_trial_grant(self, storage: "StorageBackend", user_id: str) -> bool:
+        return False
+
+    def list_user_ids(self, storage: "StorageBackend") -> list[str]:
+        return []
+
+    async def maybe_auto_topup(
+        self, storage: "StorageBackend", user_id: str
+    ) -> dict | None:
+        return None
+
+
+class CreditsBilling:
+    """Production billing: delegates to the credits ledger + auto top-up."""
+
+    def credits_for_api_cost(self, cost_usd: float) -> float:
+        from backend.services import credits as credits_svc
+
+        return credits_svc.credits_for_api_cost(cost_usd)
+
+    def get_balance(self, storage: "StorageBackend", user_id: str) -> float:
+        from backend.services import credits as credits_svc
+
+        return credits_svc.get_balance(storage, user_id)
+
+    def charge(
+        self,
+        storage: "StorageBackend",
+        user_id: str,
+        amount: float,
+        *,
+        reason: str = "pipeline_charge",
+        run_id: str | None = None,
+        unit_id: str | None = None,
+        allow_overdraft: bool = False,
+    ) -> None:
+        from backend.services import credits as credits_svc
+
+        credits_svc.charge(
+            storage, user_id, amount,
+            reason=reason,
+            run_id=run_id,
+            unit_id=unit_id,
+            allow_overdraft=allow_overdraft,
+        )
+
+    def ensure_trial_grant(self, storage: "StorageBackend", user_id: str) -> bool:
+        from backend.services import credits as credits_svc
+
+        return credits_svc.ensure_trial_grant(storage, user_id)
+
+    def list_user_ids(self, storage: "StorageBackend") -> list[str]:
+        from backend.services import credits as credits_svc
+
+        return credits_svc.list_user_ids(storage)
+
+    async def maybe_auto_topup(
+        self, storage: "StorageBackend", user_id: str
+    ) -> dict | None:
+        """Run an auto top-up attempt if configured.
+
+        Returns ``{"reason", "amount_usd"}`` when this call produced a NEW
+        failed attempt (so the caller can notify the user), else None.
+        """
+        from backend.services.auto_topup import get_config, maybe_trigger
+
+        before = get_config(storage, user_id).last_attempt_ts
+        try:
+            await maybe_trigger(storage, user_id)
+        except Exception:
+            return None
+        after = get_config(storage, user_id)
+        if (
+            after.last_attempt_status == "failed"
+            and after.last_attempt_ts
+            and after.last_attempt_ts != before
+        ):
+            return {
+                "reason": after.last_failure_reason or "unknown",
+                "amount_usd": after.amount_usd,
+            }
+        return None
+
+
+_NULL = NullBilling()
+_credits_billing: CreditsBilling | None = None
+
+
+def get_billing() -> BillingHook:
+    """Return the active billing implementation.
+
+    Selected per call (not at import) so the ``billing_enabled`` setting
+    can be monkeypatched in tests and so importing this module never pulls
+    in billing code.
+    """
+    if not settings.billing_enabled:
+        return _NULL
+    global _credits_billing
+    if _credits_billing is None:
+        _credits_billing = CreditsBilling()
+    return _credits_billing

periscope/src/backend/services/cost_estimator.py

@@ -0,0 +1,403 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Pre-flight cost estimator for pipeline runs.
+
+Walks the uploaded BOM + library cache and returns a low/high credit
+range the user will see *before* they start a run.  Pure read-only:
+no storage writes, no API calls.
+
+The estimator is intentionally conservative.  Low/high bounds are
+bracketed around a central estimate (0.7× / 1.4×) so the user always
+sees a plausible range rather than a false-precision single number.
+
+Per-call USD is computed from per-stage **token baselines**
+(``STAGE_TOKEN_BASELINES``) multiplied by the runtime-resolved
+provider+model rate from ``backend.services.llm.pricing.PRICING``.
+This means a change to ``PROVIDER_VALIDATION`` / ``MODEL_VALIDATION``
+(or any other per-stage routing env var) automatically updates the
+estimate — no constant-bumping required. The baselines themselves
+are hand-tuned from historical ``api_logs.jsonl`` aggregates and
+should be recalibrated periodically.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Literal
+
+from pydantic import BaseModel
+
+from backend.config import settings
+from backend.periscopex.parsers import parse_bom
+from backend.periscopex.resolve_passives import resolve_mpn
+from backend.periscopex.taxonomy import SIMPLE_TYPES, type_for_ref
+from backend.periscopex.utils import safe_mpn
+from backend.services import projects as proj_svc
+from backend.services.billing_hook import get_billing
+from backend.services.llm.pricing import CACHE_RATES, PRICING
+from backend.services.storage import StorageBackend
+
+
+# ---------------------------------------------------------------------------
+# Per-stage token baselines (model-aware estimator)
+# ---------------------------------------------------------------------------
+
+# Average tokens per call for a single sub-unit of each stage. Values
+# come from aggregating recent ``api_logs.jsonl`` runs across staging +
+# prod (see ``scripts/recalibrate_estimator_baselines.py`` follow-up;
+# until that lands, eyeball + paste from gcloud-mined stats).
+#
+# ``settings_stage`` is the key passed to ``settings.model_for_stage`` /
+# ``settings.provider_for_stage``. The estimator-stage names mirror the
+# ``CostItem.kind`` Literal so the breakdown stays self-consistent.
+STAGE_TOKEN_BASELINES: dict[str, dict[str, int | str]] = {
+    "ic_extraction": {
+        "settings_stage": "pintable",
+        "input": 100, "output": 2000,
+        "cache_create": 80_000, "cache_read": 170_000,
+    },
+    "simple_extraction": {
+        "settings_stage": "specs",
+        "input": 100, "output": 1000,
+        "cache_create": 20_000, "cache_read": 20_000,
+    },
+    "passive_pattern": {
+        "settings_stage": "pattern",
+        "input": 100, "output": 7000,
+        "cache_create": 60_000, "cache_read": 330_000,
+    },
+    "digikey_resolve": {
+        "settings_stage": "auto_resolve",
+        "input": 2000, "output": 200,
+        "cache_create": 0, "cache_read": 0,
+    },
+    # Validation review per IC. The multi-turn validator pulls the
+    # cached system + graph + datasheet on each turn (~4-5 turns/IC),
+    # so cache_read dominates. Page-aware scaling was abandoned — token
+    # counts already encompass the PDF via cache reuse, and IC
+    # complexity correlates more weakly with raw page count than the
+    # old per-page heuristic assumed.
+    "review": {
+        "settings_stage": "validation",
+        "input": 13_500, "output": 2000,
+        "cache_create": 110_000, "cache_read": 300_000,
+    },
+    # Per-IC normalize pass — dedup + severity re-grade. Runs on the
+    # already-structured findings (no PDF, no graph tools), one turn,
+    # validation-class model (Sonnet). A few hundred input tokens for the
+    # rubric, a few hundred output tokens for the normalized list.
+    "normalize": {
+        "settings_stage": "normalize",
+        "input": 1500, "output": 600,
+        "cache_create": 0, "cache_read": 0,
+    },
+    # Cross-IC dedup — one call per run over all findings (no PDF/graph),
+    # validation-class model (Sonnet). Slightly larger input than per-IC
+    # normalize since it sees every IC's findings at once.
+    "cross_ic_dedup": {
+        "settings_stage": "normalize",
+        "input": 2500, "output": 700,
+        "cache_create": 0, "cache_read": 0,
+    },
+}
+
+LOW_MULT: float = 0.7
+HIGH_MULT: float = 1.4
+
+
+def estimate_stage_cost_usd(stage: str) -> float:
+    """Per-call USD for one sub-unit of ``stage``, model-aware.
+
+    Resolves provider+model from ``settings`` and computes
+    ``(input_tokens × rate) + ...`` using the same ``PRICING`` /
+    ``CACHE_RATES`` tables that real billing in
+    ``services.llm.pricing.cost_for_entry`` reads. Changing a
+    ``MODEL_*`` / ``PROVIDER_*`` env var therefore updates the estimate
+    automatically.
+
+    Falls through to ``PRICING[provider]["default"]`` when the resolved
+    model is missing from the table — same fallback semantics as the
+    real billing code, so a missing pricing entry surfaces uniformly
+    everywhere instead of crashing the estimator.
+
+    Raises ``KeyError`` only when ``stage`` itself is unknown.
+    """
+    base = STAGE_TOKEN_BASELINES[stage]
+    settings_stage = str(base["settings_stage"])
+    provider = settings.provider_for_stage(settings_stage)
+    model = settings.model_for_stage(settings_stage)
+    table = PRICING.get(provider) or PRICING["deepseek"]
+    rates = table.get(model, table["default"])
+    cache = CACHE_RATES.get(provider, CACHE_RATES["deepseek"])
+    return (
+        int(base["input"]) * rates["input"]
+        + int(base["output"]) * rates["output"]
+        + int(base["cache_create"]) * rates["input"] * cache["create"]
+        + int(base["cache_read"]) * rates["input"] * cache["read"]
+    ) / 1_000_000
+
+
+# ---------------------------------------------------------------------------
+# Data classes
+# ---------------------------------------------------------------------------
+
+
+UnitKind = Literal[
+    "ic_extraction",
+    "simple_extraction",
+    "passive_pattern",
+    "digikey_resolve",
+    "review",
+]
+
+
+class CostItem(BaseModel):
+    identifier: str   # MPN, ref, or a fixed token like a stage name
+    kind: UnitKind
+    api_cost_usd: float
+    source: Literal["cache_hit", "api_call", "api_call_estimated"]
+    note: str | None = None
+
+
+class CostEstimate(BaseModel):
+    """What the pipeline will likely cost this run."""
+    api_cost_low: float
+    api_cost_high: float
+    api_cost_mid: float
+    credits_low: float
+    credits_high: float
+    credits_mid: float
+    breakdown: list[CostItem]
+    ic_count: int
+    simple_count: int
+    passive_count: int
+    cached_ic_count: int
+    cached_simple_count: int
+    cached_passive_count: int
+    review_ic_count: int
+
+
+# ---------------------------------------------------------------------------
+# Internals
+# ---------------------------------------------------------------------------
+
+
+def _load_library_patterns(storage: StorageBackend):
+    """Load passive patterns from the library to test cache hits.
+
+    This mirrors the pipeline's own seeding behaviour but keeps the
+    estimator synchronous and side-effect free (downloads to a local
+    tempdir only if the backend is remote).
+    """
+    try:
+        return proj_svc.load_library_patterns(storage)
+    except Exception:
+        return []
+
+
+# ---------------------------------------------------------------------------
+# Public entry point
+# ---------------------------------------------------------------------------
+
+
+def estimate_pipeline_cost(
+    storage: StorageBackend,
+    user_id: str,
+    project_id: str,
+) -> CostEstimate:
+    """Produce a CostEstimate for the given project without running anything.
+
+    The BOM must already be uploaded; the netlist may or may not be.  If
+    the BOM is missing, raises FileNotFoundError.
+    """
+    bom_key = proj_svc.get_bom_key(storage, user_id, project_id)
+    if not bom_key:
+        raise FileNotFoundError("BOM not uploaded for this project")
+
+    meta = proj_svc.get_project(storage, user_id, project_id)
+    col_map = (meta.bom_columns if meta else None) or {}
+    ref_col = col_map.get("reference", "Reference")
+    mpn_col = col_map.get("mpn", "Manufacturer Part Number")
+
+    # Download BOM to a local temp path so parse_bom can read it.
+    import tempfile
+
+    with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as tmp:
+        tmp.write(storage.read_bytes(bom_key))
+        bom_local_path = Path(tmp.name)
+
+    try:
+        bom = parse_bom(str(bom_local_path), reference_col=ref_col, mpn_col=mpn_col)
+    finally:
+        bom_local_path.unlink(missing_ok=True)
+
+    # Classify unique MPNs by type.
+    ic_mpns: set[str] = set()
+    simple_mpns: set[str] = set()
+    passive_mpns: set[str] = set()
+    for ref, info in bom.items():
+        mpn = info.get("mpn")
+        if not mpn:
+            continue
+        typ = type_for_ref(ref)
+        if typ == "ic":
+            ic_mpns.add(mpn)
+        elif typ == "passive":
+            passive_mpns.add(mpn)
+        elif typ in SIMPLE_TYPES:
+            simple_mpns.add(mpn)
+
+    # Load library patterns once so we can resolve passives against cache.
+    patterns = _load_library_patterns(storage)
+
+    breakdown: list[CostItem] = []
+    cached_ic = 0
+    cached_simple = 0
+    cached_passive = 0
+
+    # IC extraction
+    for mpn in sorted(ic_mpns):
+        if proj_svc.library_has_extraction(storage, mpn):
+            breakdown.append(CostItem(
+                identifier=mpn, kind="ic_extraction",
+                api_cost_usd=0.0, source="cache_hit",
+                note="library hit",
+            ))
+            cached_ic += 1
+        else:
+            breakdown.append(CostItem(
+                identifier=mpn, kind="ic_extraction",
+                api_cost_usd=round(estimate_stage_cost_usd("ic_extraction"), 4),
+                source="api_call_estimated",
+            ))
+
+    # Simple component specs
+    for mpn in sorted(simple_mpns):
+        if proj_svc.library_has_model(storage, mpn):
+            breakdown.append(CostItem(
+                identifier=mpn, kind="simple_extraction",
+                api_cost_usd=0.0, source="cache_hit",
+            ))
+            cached_simple += 1
+        else:
+            breakdown.append(CostItem(
+                identifier=mpn, kind="simple_extraction",
+                api_cost_usd=round(estimate_stage_cost_usd("simple_extraction"), 4),
+                source="api_call_estimated",
+            ))
+
+    # Passives — pattern resolution covers many MPNs with one pattern
+    unresolved_passives: list[str] = []
+    for mpn in sorted(passive_mpns):
+        if patterns and resolve_mpn(mpn, patterns) is not None:
+            breakdown.append(CostItem(
+                identifier=mpn, kind="passive_pattern",
+                api_cost_usd=0.0, source="cache_hit",
+                note="pattern match",
+            ))
+            cached_passive += 1
+            continue
+        if proj_svc.library_has_passive_model(storage, mpn):
+            breakdown.append(CostItem(
+                identifier=mpn, kind="passive_pattern",
+                api_cost_usd=0.0, source="cache_hit",
+                note="cached passive model",
+            ))
+            cached_passive += 1
+            continue
+        unresolved_passives.append(mpn)
+
+    # Each unresolved passive MPN may contribute one pattern extraction.
+    # Heuristic: N unique first-7-char prefixes = N new patterns.
+    prefixes = {m[:7] for m in unresolved_passives}
+    for prefix in sorted(prefixes):
+        sample_mpn = next(m for m in unresolved_passives if m.startswith(prefix))
+        breakdown.append(CostItem(
+            identifier=sample_mpn, kind="passive_pattern",
+            api_cost_usd=round(estimate_stage_cost_usd("passive_pattern"), 4),
+            source="api_call_estimated",
+            note=f"may cover {sum(1 for m in unresolved_passives if m.startswith(prefix))} MPNs",
+        ))
+
+    # Direct datasheet review — per IC that has a datasheet available.
+    # Cost is the flat per-IC observed average; multi-turn cache reuse
+    # makes this less page-sensitive than the old per-page heuristic
+    # implied.
+    review_per_ic = estimate_stage_cost_usd("review")
+    if settings.normalize_findings_enabled:
+        review_per_ic += estimate_stage_cost_usd("normalize")
+    review_ic_count = 0
+    for mpn in sorted(ic_mpns):
+        pdf_path = _locate_datasheet_local(storage, user_id, project_id, mpn)
+        has_pdf = pdf_path is not None
+        has_library_pdf = (
+            proj_svc.library_has_datasheet(storage, mpn) is not None
+            if not has_pdf else False
+        )
+        if not (has_pdf or has_library_pdf):
+            continue  # Skipped in pipeline — no datasheet, no review
+        breakdown.append(CostItem(
+            identifier=mpn, kind="review",
+            api_cost_usd=round(review_per_ic, 4),
+            source="api_call_estimated",
+        ))
+        review_ic_count += 1
+
+    # One cross-IC dedup call per run, only when ≥2 ICs get reviewed (a
+    # single-IC run has no cross-IC pair to merge — see _maybe_dedupe_cross_ic).
+    if settings.cross_ic_dedup_enabled and review_ic_count > 1:
+        breakdown.append(CostItem(
+            identifier="cross-IC dedup", kind="review",
+            api_cost_usd=round(estimate_stage_cost_usd("cross_ic_dedup"), 4),
+            source="api_call_estimated",
+            note="collapses one interface defect reported from both ICs",
+        ))
+
+    api_total = sum(item.api_cost_usd for item in breakdown)
+    api_low = round(api_total * LOW_MULT, 4)
+    api_high = round(api_total * HIGH_MULT, 4)
+
+    billing = get_billing()
+    return CostEstimate(
+        api_cost_low=api_low,
+        api_cost_high=api_high,
+        api_cost_mid=round(api_total, 4),
+        credits_low=billing.credits_for_api_cost(api_low),
+        credits_high=billing.credits_for_api_cost(api_high),
+        credits_mid=billing.credits_for_api_cost(api_total),
+        breakdown=breakdown,
+        ic_count=len(ic_mpns),
+        simple_count=len(simple_mpns),
+        passive_count=len(passive_mpns),
+        cached_ic_count=cached_ic,
+        cached_simple_count=cached_simple,
+        cached_passive_count=cached_passive,
+        review_ic_count=review_ic_count,
+    )
+
+
+def _locate_datasheet_local(
+    storage: StorageBackend, user_id: str, project_id: str, mpn: str,
+) -> Path | None:
+    """Return a local Path to the datasheet PDF if it can be read quickly.
+
+    For LocalStorageBackend, reads directly from disk.  For remote backends
+    we skip the page-count read (returns None) — estimator will fall back
+    to the mid-cap heuristic rather than downloading the PDF during pre-flight.
+    """
+    from backend.services.storage import LocalStorageBackend
+
+    if not isinstance(storage, LocalStorageBackend):
+        return None
+    safe = safe_mpn(mpn)
+    key = f"users/{user_id}/projects/{project_id}/uploads/datasheets/{safe}.pdf"
+    if storage.exists(key):
+        return storage._path(key)  # type: ignore[attr-defined]
+    legacy = f"library/datasheets/{safe}.pdf"
+    if storage.exists(legacy):
+        return storage._path(legacy)  # type: ignore[attr-defined]
+    return None

periscope/src/backend/services/datasheet_store.py

@@ -0,0 +1,242 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Content-addressed datasheet storage for the shared library.
+
+Stores PDF blobs by their MD5 hash and creates lightweight JSON ref files
+that map MPN names to blob keys.  This deduplicates identical PDFs that
+were previously stored under different human-readable names.
+
+Layout::
+
+    library/datasheets/
+      blobs/{md5hash}.pdf          -- unique PDF content, stored once
+      refs/{safe_mpn}.json         -- per-MPN pointer: {"hash": "...", "blob_key": "..."}
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+from backend.periscopex.utils import safe_mpn
+from backend.services.storage import StorageBackend
+
+BLOB_PREFIX = "library/datasheets/blobs/"
+REF_PREFIX = "library/datasheets/refs/"
+ALIAS_KEY = "library/datasheets/aliases.json"
+
+
+# ---------------------------------------------------------------------------
+# Hashing helpers
+# ---------------------------------------------------------------------------
+
+def compute_md5_from_path(local_path: Path) -> str:
+    """Return the hex MD5 digest of a local file (chunked read)."""
+    h = hashlib.md5()
+    with open(local_path, "rb") as f:
+        for chunk in iter(lambda: f.read(8192), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def compute_md5_from_bytes(data: bytes) -> str:
+    """Return the hex MD5 digest of in-memory bytes."""
+    return hashlib.md5(data).hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# Key construction
+# ---------------------------------------------------------------------------
+
+def blob_key(md5: str) -> str:
+    """Storage key for a content-addressed PDF blob."""
+    return f"{BLOB_PREFIX}{md5}.pdf"
+
+
+def ref_key(mpn: str) -> str:
+    """Storage key for an MPN → blob ref file."""
+    return f"{REF_PREFIX}{safe_mpn(mpn)}.json"
+
+
+# ---------------------------------------------------------------------------
+# Store / resolve / delete
+# ---------------------------------------------------------------------------
+
+def store_datasheet(
+    storage: StorageBackend,
+    local_path: Path,
+    mpn: str,
+) -> str:
+    """Store a datasheet PDF by content hash and create an MPN ref.
+
+    Idempotent: skips blob upload if it already exists, always writes the ref.
+    Returns the blob storage key.
+    """
+    md5 = compute_md5_from_path(local_path)
+    bk = blob_key(md5)
+    if not storage.exists(bk):
+        storage.upload_from_local(local_path, bk)
+    storage.write_json(ref_key(mpn), {"hash": md5, "blob_key": bk, "mpn": mpn})
+    return bk
+
+
+def store_datasheet_bytes(
+    storage: StorageBackend,
+    data: bytes,
+    mpn: str,
+    extra_mpns: list[str] | None = None,
+) -> str:
+    """Same as :func:`store_datasheet` but from in-memory bytes.
+
+    ``extra_mpns`` are additional catalog/orderable codes that should point
+    at the same blob (family MPN vs ``…-N16R16V``).
+    """
+    md5 = compute_md5_from_bytes(data)
+    bk = blob_key(md5)
+    if not storage.exists(bk):
+        storage.write_bytes(bk, data)
+    names = [mpn, *(extra_mpns or [])]
+    seen: set[str] = set()
+    for name in names:
+        name = (name or "").strip()
+        if not name:
+            continue
+        key = name.upper()
+        if key in seen:
+            continue
+        seen.add(key)
+        storage.write_json(ref_key(name), {"hash": md5, "blob_key": bk, "mpn": name})
+    _record_aliases(storage, mpn, extra_mpns or [])
+    return bk
+
+
+def _record_aliases(storage: StorageBackend, mpn: str, extra_mpns: list[str]) -> None:
+    from backend.services.datasheet_finder import _alnum, _MIN_FAMILY_LEN
+
+    names = [mpn, *extra_mpns]
+    compact = {n: _alnum(n) for n in names if n and n.strip()}
+    if len(set(compact.values())) < 2 and not extra_mpns:
+        return
+    table: dict[str, str] = {}
+    if storage.exists(ALIAS_KEY):
+        raw = storage.read_json(ALIAS_KEY)
+        table = dict(raw.get("aliases") or {})
+    canonical = extra_mpns[0].strip() if extra_mpns else mpn
+    for name, key in compact.items():
+        if len(key) >= _MIN_FAMILY_LEN:
+            table[key] = canonical
+    storage.write_json(ALIAS_KEY, {"aliases": table})
+
+
+def resolve_datasheet(storage: StorageBackend, mpn: str) -> str | None:
+    """Look up the blob key for an MPN via its ref file.
+
+    Tries spelling variants, then the shared alias table (family MPN →
+    orderable code stored in the library).
+    """
+    from backend.services.datasheet_finder import (
+        _MIN_FAMILY_LEN,
+        _alnum,
+        mpn_query_variants,
+    )
+
+    def _from_ref(name: str) -> str | None:
+        rk = ref_key(name)
+        if not storage.exists(rk):
+            return None
+        ref = storage.read_json(rk)
+        bk = ref.get("blob_key")
+        if bk and storage.exists(bk):
+            return bk
+        return None
+
+    for name in mpn_query_variants(mpn) or [mpn]:
+        hit = _from_ref(name)
+        if hit:
+            return hit
+
+    if not storage.exists(ALIAS_KEY):
+        return None
+    table = (storage.read_json(ALIAS_KEY) or {}).get("aliases") or {}
+    want = _alnum(mpn)
+    if not want:
+        return None
+    target = table.get(want)
+    if target:
+        hit = _from_ref(target)
+        if hit:
+            return hit
+    if len(want) >= _MIN_FAMILY_LEN:
+        for key, target in table.items():
+            if key.startswith(want) or (
+                want.startswith(key) and len(key) >= _MIN_FAMILY_LEN
+            ):
+                hit = _from_ref(target)
+                if hit:
+                    return hit
+    return None
+
+
+def delete_datasheet_ref(storage: StorageBackend, mpn: str) -> str | None:
+    """Delete the ref for an MPN.  Returns the blob key if a ref existed.
+
+    Does **not** delete the blob — other refs may point to it.  Use
+    :func:`gc_orphan_blobs` to clean up unreferenced blobs.
+    """
+    rk = ref_key(mpn)
+    if not storage.exists(rk):
+        return None
+    ref = storage.read_json(rk)
+    bk = ref.get("blob_key")
+    storage.delete_key(rk)
+    return bk
+
+
+# ---------------------------------------------------------------------------
+# Maintenance
+# ---------------------------------------------------------------------------
+
+def gc_orphan_blobs(
+    storage: StorageBackend, *, dry_run: bool = True,
+) -> list[str]:
+    """Find blobs not referenced by any ref file.  Optionally delete them.
+
+    Also checks pattern ``datasheet_key`` values so blobs referenced only
+    by patterns (not MPN refs) are kept.
+
+    Intended for maintenance scripts, not hot paths.
+    """
+    # Collect all hashes referenced by ref files
+    referenced_hashes: set[str] = set()
+    for rk in storage.list_recursive(REF_PREFIX):
+        if rk.endswith(".json"):
+            ref = storage.read_json(rk)
+            h = ref.get("hash")
+            if h:
+                referenced_hashes.add(h)
+
+    # Also collect hashes from pattern datasheet_key values
+    for pk in storage.list_recursive("library/patterns/"):
+        if pk.endswith(".json"):
+            pat = storage.read_json(pk)
+            ds_key = pat.get("datasheet_key", "")
+            if ds_key.startswith(BLOB_PREFIX) and ds_key.endswith(".pdf"):
+                h = ds_key.removeprefix(BLOB_PREFIX).removesuffix(".pdf")
+                referenced_hashes.add(h)
+
+    # Find orphan blobs
+    orphans: list[str] = []
+    for bk in storage.list_recursive(BLOB_PREFIX):
+        if not bk.endswith(".pdf"):
+            continue
+        filename = bk.rsplit("/", 1)[-1]
+        h = filename.removesuffix(".pdf")
+        if h not in referenced_hashes:
+            orphans.append(bk)
+            if not dry_run:
+                storage.delete_key(bk)
+
+    return orphans

periscope/src/backend/services/dedupe_findings.py

@@ -0,0 +1,396 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Cross-IC dedup pass — collapse one physical defect reported from both ends.
+
+Direct datasheet review runs once per IC, in isolation. An interface defect
+(e.g. a 5V driver into a non-5V-tolerant input on the U2↔U3 UART) is therefore
+discovered twice — once when reviewing U2, once when reviewing U3 — and the
+per-IC normalize pass cannot collapse them because it only sees one IC's
+findings at a time. The two copies land in the report as separate findings,
+double-counting the same problem and (worse) sometimes disagreeing with each
+other.
+
+This module runs a single small LLM call over the *concatenated* findings from
+all ICs (no PDF, no graph tools) to merge findings that describe the same
+physical defect on the same net/interface/component. It is the cross-IC analog
+of ``normalize_findings`` and follows the same fail-soft contract: on any LLM
+error, schema violation, or coverage gap, the original findings are returned
+unchanged. It never drops findings (that is normalize's job) and never raises a
+merged finding's severity above the highest of its members.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Awaitable, Callable
+
+from backend.periscopex.models import Finding
+from backend.services.api_logs import ApiLogger
+from backend.services.llm import Message, TextBlock
+from backend.services.llm.factory import call_with_fallback
+from backend.services.llm.types import ToolSchema
+
+log = logging.getLogger(__name__)
+
+# Severity ordering — a merged group's severity is capped at the highest
+# severity among its members (downgrade-only, same principle as normalize).
+_INFO, _WARN, _ERR = 0, 1, 2
+_SEVERITY_RANK = {"INFO": _INFO, "WARNING": _WARN, "ERROR": _ERR}
+_RANK_TO_SEV = {_INFO: "INFO", _WARN: "WARNING", _ERR: "ERROR"}
+
+
+def _is_unverified(why: str | None) -> bool:
+    return (why or "").lstrip().lower().startswith("unverified:")
+
+
+SYSTEM_PROMPT = """\
+You deduplicate hardware-review findings across multiple ICs.
+
+Each finding was produced by reviewing one IC in isolation, so a defect on
+the interface *between* two ICs is reported twice — once from each side. Your
+job is to group findings that describe the SAME physical defect and merge each
+group into one finding. You may NOT invent findings, drop findings, or change
+the engineering substance.
+
+### When two findings are the same defect (merge)
+
+Merge when they describe the same physical problem at the same place:
+- the same net or signal (e.g. both flag over-voltage on `/UART0.NCTS`),
+- the same component pair / interface (e.g. "U2 RTS# drives U3 PA14" and
+  "U3 PA14 is driven by U2's 5V output" are one interface defect seen from
+  each end),
+- the same shared part with the same fix.
+
+A merged group is resolved by ONE change to the design. Name that interface or
+root cause once in the merged `finding`; restate each side's consequence in
+`why`.
+
+### When findings are NOT the same defect (keep separate)
+
+Do NOT merge findings that need different fixes, even if they touch the same
+component or net:
+- different pins / different signals on the same IC,
+- a decoupling issue and a voltage issue on the same supply,
+- two unrelated problems that happen to involve the same part.
+
+When in doubt, keep them separate. Over-merging hides distinct problems and is
+worse than a visible duplicate.
+
+### Severity
+
+Use the HIGHEST severity among a group's members. Never grade a merged finding
+above its strongest member. If any member's `why` begins with `Unverified:`,
+keep that prefix and do not grade the merged finding above WARNING.
+
+### Output
+
+Call `submit_deduped` exactly once. Provide a `groups` array. Every original
+finding (numbered 1..N) must appear in exactly one group's `member_indices`,
+and no index may appear twice.
+- A group of ONE index is a passthrough — it is kept unchanged (you do not
+  need to restate its text).
+- A group of MORE THAN ONE index is a merge — supply the merged `finding`,
+  `why`, `status`, `recommendation`, and a `primary_index` (one of the group's
+  members) whose datasheet citation/page is the strongest evidence; that
+  member supplies the finding's component attribution and source reference.
+"""
+
+
+SUBMIT_DEDUPED_SCHEMA = ToolSchema(
+    name="submit_deduped",
+    description=(
+        "Submit the cross-IC deduplicated findings. Every original finding "
+        "(1..N) must appear in exactly one group's `member_indices`."
+    ),
+    input_schema={
+        "type": "object",
+        "properties": {
+            "groups": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "member_indices": {
+                            "type": "array",
+                            "items": {"type": "integer"},
+                            "minItems": 1,
+                            "description": (
+                                "1-indexed positions in the original findings "
+                                "list this group represents. Length 1 = "
+                                "passthrough; length > 1 = merge."
+                            ),
+                        },
+                        "primary_index": {
+                            "type": ["integer", "null"],
+                            "description": (
+                                "REQUIRED when member_indices has length > 1: "
+                                "the member whose datasheet citation/source is "
+                                "the strongest. Supplies the merged finding's "
+                                "component attribution and source reference. "
+                                "Must be one of member_indices."
+                            ),
+                        },
+                        "finding": {"type": "string"},
+                        "why": {"type": "string"},
+                        "status": {
+                            "type": "string",
+                            "enum": ["ERROR", "WARNING", "INFO"],
+                        },
+                        "recommendation": {"type": "string"},
+                        "change_rationale": {
+                            "type": "string",
+                            "description": (
+                                "≤1 line: 'passthrough', or 'merged N+M: "
+                                "<shared interface/root cause>'."
+                            ),
+                        },
+                    },
+                    "required": ["member_indices", "change_rationale"],
+                },
+            },
+        },
+        "required": ["groups"],
+    },
+)
+
+
+def _serialize_findings_for_prompt(findings: list[Finding]) -> str:
+    """Number findings 1..N with their IC, severity, and text.
+
+    Unlike the per-IC normalize pass, the designator IS included — it is the
+    primary signal for spotting that two findings sit on opposite ends of one
+    interface.
+    """
+    rows: list[dict] = []
+    for i, f in enumerate(findings, start=1):
+        rows.append({
+            "index": i,
+            "ic": f.designator,
+            "mpn": f.mpn,
+            "reviewer_severity": f.status,
+            "finding": f.finding,
+            "why": f.why,
+            "recommendation": f.recommendation,
+            "source_page": f.source_page,
+            "reference": f.reference,
+        })
+    return json.dumps(rows, indent=2)
+
+
+def _build_deduped(
+    raw_groups: list[dict],
+    originals: list[Finding],
+) -> list[Finding] | None:
+    """Validate the tool output and reconstruct the deduped finding list.
+
+    Returns the kept/merged findings, or ``None`` if coverage/schema
+    validation fails (caller falls back to originals). A merge that omits a
+    valid ``primary_index`` is not a hard failure — that group falls back to
+    its per-index originals (un-merged), preserving coverage and severities.
+    """
+    n = len(originals)
+    seen: set[int] = set()
+    result: list[Finding] = []
+
+    for group in raw_groups:
+        if not isinstance(group, dict):
+            return None
+        member_indices = group.get("member_indices") or []
+        if not isinstance(member_indices, list) or not member_indices:
+            return None
+        try:
+            indices = [int(x) for x in member_indices]
+        except (TypeError, ValueError):
+            return None
+        for idx in indices:
+            if idx < 1 or idx > n or idx in seen:
+                return None
+            seen.add(idx)
+
+        # Passthrough — keep the original verbatim. No laundering of text or
+        # severity for a finding the model chose not to merge.
+        if len(indices) == 1:
+            result.append(originals[indices[0] - 1])
+            continue
+
+        # Merge — needs a valid primary_index naming the canonical member.
+        # Missing/invalid → un-merge to per-index originals (coverage kept).
+        primary_raw = group.get("primary_index")
+        try:
+            primary = int(primary_raw)
+        except (TypeError, ValueError):
+            primary = None
+        if primary not in indices:
+            log.warning(
+                "dedupe: merge of %s has invalid primary_index %r — "
+                "falling back to per-index originals (un-merging)",
+                indices, primary_raw,
+            )
+            for idx in indices:
+                result.append(originals[idx - 1])
+            continue
+
+        canon = originals[primary - 1]
+        members = [originals[i - 1] for i in indices]
+        ceiling = max(_SEVERITY_RANK.get(m.status, _ERR) for m in members)
+        unverified = any(_is_unverified(m.why) for m in members)
+        if unverified:
+            ceiling = min(ceiling, _WARN)
+        proposed = str(group.get("status") or canon.status)
+        final_status = _RANK_TO_SEV[
+            min(_SEVERITY_RANK.get(proposed, ceiling), ceiling)
+        ]
+
+        new_why = str(group.get("why") or canon.why)
+        if unverified and not _is_unverified(new_why):
+            new_why = "Unverified: " + new_why
+
+        try:
+            result.append(canon.model_copy(update={
+                "finding": str(group.get("finding") or canon.finding),
+                "why": new_why,
+                "source_page": group.get("source_page", canon.source_page),
+                "status": final_status,
+                "recommendation": str(
+                    group.get("recommendation") or canon.recommendation
+                ),
+                "reference": str(group.get("reference") or canon.reference),
+            }))
+        except Exception:
+            log.exception("dedupe: failed to build merged Finding")
+            return None
+
+    if seen != set(range(1, n + 1)):
+        return None
+    return result
+
+
+async def dedupe_cross_ic_findings_async(
+    findings: list[Finding],
+    *,
+    api_logger: ApiLogger | None = None,
+    on_progress: Callable[[str, int, str, str], Awaitable[None]] | None = None,
+) -> tuple[list[Finding], dict]:
+    """Run the cross-IC dedup pass over findings from every IC.
+
+    Returns ``(deduped_findings, trace)``. On any failure (LLM error, schema
+    violation, coverage gap) returns the original findings unchanged with an
+    ``error`` field set in the trace.
+    """
+    trace: dict = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "input_findings": [f.model_dump(mode="json") for f in findings],
+        "output_findings": None,
+        "submission": None,
+        "model": None,
+        "provider": None,
+        "duration_ms": None,
+        "error": None,
+    }
+
+    # Nothing to merge across fewer than two findings.
+    if len(findings) < 2:
+        trace["output_findings"] = trace["input_findings"]
+        trace["error"] = "skipped: <2 findings"
+        return findings, trace
+
+    user_text = (
+        f"There are {len(findings)} findings across all reviewed ICs. "
+        f"Indices are 1-based. Group findings that describe the same physical "
+        f"defect (especially the same interface seen from both ICs) and call "
+        f"submit_deduped.\n\n{_serialize_findings_for_prompt(findings)}"
+    )
+
+    t0 = time.monotonic()
+
+    async def _run(provider, model):
+        trace["model"] = model
+        trace["provider"] = provider.name
+        session = await provider.create_session(
+            model=model,
+            system=SYSTEM_PROMPT,
+            max_tokens=4096,
+            temperature=0.0,
+        )
+        try:
+            completion = await session.complete(
+                messages=[Message(
+                    role="user",
+                    content=[TextBlock(text=user_text, cacheable=False)],
+                )],
+                tools=[SUBMIT_DEDUPED_SCHEMA],
+                tool_choice={"name": "submit_deduped"},
+            )
+            if api_logger:
+                api_logger.log(
+                    stage="cross_ic_dedupe",
+                    identifier="all",
+                    model=model,
+                    provider=provider.name,
+                    input_tokens=completion.usage.input_tokens,
+                    output_tokens=completion.usage.output_tokens,
+                    cache_creation_input_tokens=completion.usage.cache_creation_tokens,
+                    cache_read_input_tokens=completion.usage.cache_read_tokens,
+                    duration_ms=int((time.monotonic() - t0) * 1000),
+                    stop_reason="submit_deduped",
+                    turns=1,
+                )
+            for tc in completion.tool_calls:
+                if tc.name == "submit_deduped":
+                    return tc.input
+            return None
+        finally:
+            await session.close()
+
+    try:
+        # Reuse the "normalize" stage config (validation-class Sonnet model +
+        # any configured fallback); the log entry above is stamped
+        # "cross_ic_dedupe" so cost accounting still distinguishes it.
+        submission = await call_with_fallback("normalize", _run)
+    except Exception as exc:
+        log.exception("dedupe: call failed")
+        trace["error"] = f"{type(exc).__name__}: {exc}"
+        trace["duration_ms"] = int((time.monotonic() - t0) * 1000)
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    trace["duration_ms"] = int((time.monotonic() - t0) * 1000)
+    trace["submission"] = submission
+
+    if not submission or not isinstance(submission, dict):
+        trace["error"] = "no submission"
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    raw_groups = submission.get("groups") or []
+    if not isinstance(raw_groups, list):
+        trace["error"] = "submission.groups not a list"
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    built = _build_deduped(raw_groups, findings)
+    if built is None:
+        trace["error"] = "invalid index coverage or schema"
+        trace["output_findings"] = trace["input_findings"]
+        log.warning(
+            "dedupe: invalid output (%d originals, %d groups) — "
+            "falling back to originals", len(findings), len(raw_groups),
+        )
+        return findings, trace
+
+    trace["output_findings"] = [f.model_dump(mode="json") for f in built]
+    if on_progress:
+        try:
+            await on_progress(
+                "cross_ic_dedupe", 0, "deduped",
+                f"{len(findings)} → {len(built)} findings",
+            )
+        except Exception:
+            pass
+    return built, trace

periscope/src/backend/services/digikey.py

@@ -0,0 +1,377 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""DigiKey API integration — fetch datasheet PDFs and product parameters by MPN.
+
+Uses DigiKey Product Information API v4 with OAuth2 client credentials.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass, field
+
+import httpx
+
+from backend.config import settings
+from backend.services.datasheet_finder import (
+    _alnum,
+    mpn_catalog_match,
+    mpn_matches,
+    mpn_query_variants,
+)
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# OAuth2 token cache
+# ---------------------------------------------------------------------------
+
+_token_cache: dict[str, str | float] = {"access_token": "", "expires_at": 0.0}
+
+_BASE_URLS = {
+    "production": "https://api.digikey.com",
+    "sandbox": "https://sandbox-api.digikey.com",
+}
+
+
+async def _get_access_token() -> str:
+    """Get a DigiKey OAuth2 access token, refreshing if expired."""
+    now = time.time()
+    if _token_cache["access_token"] and float(_token_cache["expires_at"]) > now + 60:
+        return str(_token_cache["access_token"])
+
+    base = _BASE_URLS.get(settings.digikey_environment, _BASE_URLS["production"])
+    async with httpx.AsyncClient(timeout=15) as client:
+        resp = await client.post(
+            f"{base}/v1/oauth2/token",
+            data={
+                "client_id": settings.digikey_client_id,
+                "client_secret": settings.digikey_client_secret,
+                "grant_type": "client_credentials",
+            },
+        )
+        resp.raise_for_status()
+        data = resp.json()
+
+    _token_cache["access_token"] = data["access_token"]
+    _token_cache["expires_at"] = now + data.get("expires_in", 3600)
+    logger.info("DigiKey OAuth token refreshed (expires in %ds)", data.get("expires_in", 3600))
+    return str(_token_cache["access_token"])
+
+
+# ---------------------------------------------------------------------------
+# Product search
+# ---------------------------------------------------------------------------
+
+
+def _get_mpn(product: dict) -> str:
+    return product.get("ManufacturerProductNumber") or product.get("ManufacturerPartNumber") or ""
+
+
+def _get_ds_url(product: dict) -> str:
+    url = product.get("DatasheetUrl") or product.get("PrimaryDatasheet") or ""
+    # DigiKey sometimes returns protocol-relative URLs
+    if url.startswith("//"):
+        url = "https:" + url
+    return url
+
+
+async def _keyword_search(mpn: str) -> list[dict]:
+    """Run a DigiKey keyword search and return the raw products list."""
+    base = _BASE_URLS.get(settings.digikey_environment, _BASE_URLS["production"])
+    token = await _get_access_token()
+
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "X-DIGIKEY-Client-Id": settings.digikey_client_id,
+        "X-DIGIKEY-Locale-Site": settings.digikey_locale_site,
+        "X-DIGIKEY-Locale-Language": settings.digikey_locale_language,
+        "X-DIGIKEY-Locale-Currency": settings.digikey_locale_currency,
+        "Content-Type": "application/json",
+    }
+
+    body = {
+        "Keywords": mpn,
+        "Limit": 5,
+        "Offset": 0,
+        "ExcludeMarketPlaceProducts": True,
+    }
+
+    async with httpx.AsyncClient(timeout=20) as client:
+        resp = await client.post(
+            f"{base}/products/v4/search/keyword",
+            headers=headers,
+            json=body,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+
+    return data.get("Products") or data.get("products") or []
+
+
+def _find_product(mpn: str, products: list[dict]) -> dict | None:
+    """Pick a DigiKey product for ``mpn``.
+
+    Prefers punctuation-insensitive equality, then packing suffixes, then a
+    longer orderable code that starts with the BOM MPN. Does not fall back
+    to ``products[0]``. Tries BOM spelling variants (underscore, reel, extra
+    description after an em dash).
+    """
+    if not products:
+        return None
+    for query in mpn_query_variants(mpn):
+        hit = _find_product_one(query, products)
+        if hit:
+            return hit
+    return None
+
+
+def _find_product_one(mpn: str, products: list[dict]) -> dict | None:
+    exact = None
+    loose = None
+    family = None
+    want = _alnum(mpn)
+    for product in products:
+        cand = _get_mpn(product)
+        if not cand:
+            continue
+        got = _alnum(cand)
+        if got == want:
+            exact = product
+            break
+        if loose is None and mpn_matches(mpn, cand):
+            loose = product
+        elif family is None and mpn_catalog_match(mpn, cand):
+            family = product
+    return exact or loose or family
+
+
+async def _search_mpn(mpn: str) -> tuple[str | None, str | None]:
+    """Search DigiKey; return (datasheet_url, catalog_mpn)."""
+    tried: set[str] = set()
+    for keyword in mpn_query_variants(mpn):
+        key = keyword.upper()
+        if key in tried:
+            continue
+        tried.add(key)
+        products = await _keyword_search(keyword)
+        product = _find_product(mpn, products)
+        if not product:
+            continue
+        url = _get_ds_url(product)
+        if url:
+            return url, _get_mpn(product) or None
+    return None, None
+
+
+# ---------------------------------------------------------------------------
+# PDF download + validation
+# ---------------------------------------------------------------------------
+
+_PDF_MAGIC = b"%PDF-"
+_MIN_PDF_SIZE = 5_000  # 5 KB — anything smaller is probably an error page
+
+
+async def _download_pdf(url: str) -> bytes:
+    """Download a PDF from a URL and validate it.
+
+    Raises ValueError if the file isn't a valid PDF or is too small.
+    Raises httpx.HTTPStatusError on 4xx/5xx responses.
+    """
+    async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
+        resp = await client.get(url)
+        resp.raise_for_status()
+        data = resp.content
+
+    if not data.startswith(_PDF_MAGIC):
+        raise ValueError("Downloaded file is not a valid PDF (bad magic bytes)")
+
+    if len(data) < _MIN_PDF_SIZE:
+        raise ValueError(f"PDF too small ({len(data)} bytes) — likely an error page")
+
+    return data
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+class DatasheetFetchResult:
+    """Result of a datasheet fetch attempt."""
+
+    def __init__(
+        self,
+        mpn: str,
+        pdf_bytes: bytes | None = None,
+        error: str | None = None,
+        url: str | None = None,
+        catalog_mpn: str | None = None,
+    ):
+        self.mpn = mpn
+        self.pdf_bytes = pdf_bytes
+        self.error = error
+        self.url = url
+        self.catalog_mpn = catalog_mpn
+
+    @property
+    def ok(self) -> bool:
+        return self.pdf_bytes is not None
+
+
+async def fetch_datasheet(mpn: str) -> DatasheetFetchResult:
+    """Fetch a datasheet PDF for the given MPN from DigiKey.
+
+    Returns a DatasheetFetchResult with either pdf_bytes or an error message.
+    The `url` field is set whenever DigiKey returns a datasheet link, even if
+    the PDF download itself fails.
+    Never raises — all errors are captured in the result.
+    """
+    if not settings.use_digikey:
+        return DatasheetFetchResult(mpn, error="DigiKey API not configured")
+
+    try:
+        url, catalog_mpn = await _search_mpn(mpn)
+    except httpx.HTTPStatusError as e:
+        logger.warning("DigiKey search failed for %s: %s", mpn, e)
+        return DatasheetFetchResult(mpn, error=f"DigiKey search failed ({e.response.status_code})")
+    except Exception as e:
+        msg = str(e) or type(e).__name__
+        logger.warning("DigiKey search error for %s: %s", mpn, msg)
+        return DatasheetFetchResult(mpn, error=f"DigiKey search error: {msg}")
+
+    if not url:
+        return DatasheetFetchResult(mpn, error="No datasheet found on DigiKey")
+
+    try:
+        pdf_bytes = await _download_pdf(url)
+    except httpx.HTTPStatusError as e:
+        logger.warning("Datasheet download blocked for %s (%s): %s", mpn, url, e)
+        return DatasheetFetchResult(
+            mpn, error=f"Download blocked ({e.response.status_code})", url=url,
+            catalog_mpn=catalog_mpn,
+        )
+    except ValueError as e:
+        logger.warning("Invalid PDF for %s (%s): %s", mpn, url, e)
+        return DatasheetFetchResult(mpn, error=str(e), url=url, catalog_mpn=catalog_mpn)
+    except httpx.TimeoutException:
+        logger.warning("Datasheet download timed out for %s (%s)", mpn, url)
+        return DatasheetFetchResult(mpn, error="Download timed out", url=url, catalog_mpn=catalog_mpn)
+    except Exception as e:
+        msg = str(e) or type(e).__name__
+        logger.warning("Datasheet download failed for %s (%s): %s", mpn, url, msg)
+        return DatasheetFetchResult(
+            mpn, error=f"Download failed: {msg}", url=url, catalog_mpn=catalog_mpn,
+        )
+
+    logger.info("Fetched datasheet for %s (%d KB)", mpn, len(pdf_bytes) // 1024)
+    return DatasheetFetchResult(
+        mpn, pdf_bytes=pdf_bytes, url=url, catalog_mpn=catalog_mpn,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Product parameters
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class ProductParams:
+    """Structured product parameters from a DigiKey search result."""
+
+    mpn: str
+    parameters: list[dict[str, str]] = field(default_factory=list)  # [{"name": ..., "value": ...}]
+    category: str = ""
+    description: str = ""
+
+
+class ParamsFetchResult:
+    """Result of a product parameters fetch attempt."""
+
+    def __init__(self, mpn: str, params: ProductParams | None = None, error: str | None = None):
+        self.mpn = mpn
+        self.params = params
+        self.error = error
+
+    @property
+    def ok(self) -> bool:
+        return self.params is not None
+
+
+def _parse_product_params(mpn: str, product: dict) -> ProductParams:
+    """Extract structured parameters from a DigiKey product dict."""
+    raw_params = product.get("Parameters") or product.get("parameters") or []
+    parameters = []
+    for p in raw_params:
+        name = p.get("ParameterText") or p.get("parameterText") or ""
+        value = p.get("ValueText") or p.get("valueText") or ""
+        if name and value and value != "-":
+            parameters.append({"name": name, "value": value})
+
+    # Category
+    cat = product.get("Category") or product.get("category") or {}
+    category = cat.get("Name") or cat.get("name") or ""
+
+    # Description
+    desc_obj = product.get("Description") or product.get("description") or {}
+    if isinstance(desc_obj, str):
+        description = desc_obj
+    else:
+        description = (
+            desc_obj.get("ProductDescription")
+            or desc_obj.get("productDescription")
+            or desc_obj.get("DetailedDescription")
+            or desc_obj.get("detailedDescription")
+            or ""
+        )
+
+    return ProductParams(
+        mpn=mpn,
+        parameters=parameters,
+        category=category,
+        description=description,
+    )
+
+
+async def fetch_params(mpn: str) -> ParamsFetchResult:
+    """Fetch DigiKey product parameters for the given MPN.
+
+    Returns structured parameter data (no PDF download needed).
+    Never raises — all errors are captured in the result.
+    """
+    if not settings.use_digikey:
+        return ParamsFetchResult(mpn, error="DigiKey API not configured")
+
+    try:
+        products: list[dict] = []
+        tried: set[str] = set()
+        for keyword in mpn_query_variants(mpn):
+            key = keyword.upper()
+            if key in tried:
+                continue
+            tried.add(key)
+            products = await _keyword_search(keyword)
+            if _find_product(mpn, products):
+                break
+    except httpx.HTTPStatusError as e:
+        logger.warning("DigiKey search failed for %s: %s", mpn, e)
+        return ParamsFetchResult(mpn, error=f"DigiKey search failed ({e.response.status_code})")
+    except Exception as e:
+        msg = str(e) or type(e).__name__
+        logger.warning("DigiKey search error for %s: %s", mpn, msg)
+        return ParamsFetchResult(mpn, error=f"DigiKey search error: {msg}")
+
+    product = _find_product(mpn, products)
+    if not product:
+        return ParamsFetchResult(mpn, error="No results found on DigiKey")
+
+    params = _parse_product_params(mpn, product)
+    if not params.parameters:
+        return ParamsFetchResult(mpn, error="No parameters available on DigiKey")
+
+    logger.info("Fetched %d params for %s (category: %s)", len(params.parameters), mpn, params.category)
+    return ParamsFetchResult(mpn, params=params)

periscope/src/backend/services/llm/base.py

@@ -0,0 +1,103 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Abstract LLMProvider + LLMSession interfaces."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Protocol
+
+from backend.services.llm.types import (
+    Completion,
+    Message,
+    ToolChoice,
+    ToolSchema,
+)
+
+
+class LLMSession(ABC):
+    """A multi-turn conversation with provider-specific cache lifecycle.
+
+    Lifecycle::
+
+        session = await provider.create_session(model=..., system=...)
+        try:
+            messages = [Message("user", [
+                PdfBlock(path, cacheable=True),
+                TextBlock(context, cacheable=True),
+            ])]
+            for turn in range(N):
+                completion = await session.complete(
+                    messages=messages, tools=..., tool_choice=...,
+                )
+                # process tool_calls, append to messages, repeat
+        finally:
+            await session.close()
+
+    Caching: blocks with ``cacheable=True`` participate in provider caching.
+    Anthropic stamps ``cache_control: ephemeral`` on each cacheable block on
+    every call. Gemini collects all cacheable blocks (plus the system prompt)
+    on the first ``complete()`` call into a ``CachedContent`` object and
+    references it on subsequent calls. The system prompt is always cached.
+    """
+
+    provider_name: str
+    """Provider identifier ("anthropic", "gemini") — used for api_logs."""
+    model: str
+
+    @abstractmethod
+    async def complete(
+        self,
+        *,
+        messages: list[Message],
+        tools: list[ToolSchema] | None = None,
+        tool_choice: ToolChoice = "auto",
+    ) -> Completion:
+        """Run one inference turn."""
+
+    @abstractmethod
+    async def close(self) -> None:
+        """Release any provider-side resources (e.g. delete a cache object).
+        Safe to call multiple times."""
+
+
+class LLMProvider(Protocol):
+    """Top-level provider interface."""
+
+    name: str
+
+    async def create_session(
+        self,
+        *,
+        model: str,
+        system: str,
+        max_tokens: int = 4096,
+        temperature: float | None = None,
+    ) -> LLMSession:
+        """Construct a session. ``system`` is always cached by the session.
+
+        ``temperature`` — if not None, applied to every ``complete()`` call on
+        this session. ``None`` means use the provider's default. Set to 0.0
+        for deterministic-as-possible behavior in agentic loops where the same
+        inputs should produce the same outputs."""
+        ...
+
+    async def run_skill(
+        self,
+        *,
+        skill_name: str,
+        model: str,
+        system: str,
+        user_text: str,
+        pdf_path: str | None,
+        output_tool: ToolSchema,
+    ) -> tuple[dict, "Completion"]:
+        """Execute a managed Skill and return (forced-tool input, Completion).
+
+        DeepSeek and Gemini inline ``skills/<name>/SKILL.md`` and run
+        ``validate.py`` locally. Anthropic uses Console Skills when a
+        skill_id is configured, otherwise the same local path."""
+        ...

periscope/src/backend/services/llm/factory.py

@@ -0,0 +1,83 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Provider factory + per-stage routing."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from functools import lru_cache
+from typing import Awaitable, Callable, TypeVar
+
+from backend.config import settings
+from backend.services.llm.base import LLMProvider
+
+log = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
+
+@lru_cache(maxsize=8)
+def get_provider_by_name(name: str) -> LLMProvider:
+    """Return a singleton provider instance for ``name`` ("deepseek" |
+    "anthropic" | "gemini"). Used by :func:`get_provider` and
+    :func:`call_with_fallback`."""
+    if name == "deepseek":
+        from backend.services.llm.deepseek_provider import DeepSeekProvider
+        return DeepSeekProvider()
+    if name == "anthropic":
+        log.warning("Anthropic is disabled — using DeepSeek instead")
+        from backend.services.llm.deepseek_provider import DeepSeekProvider
+        return DeepSeekProvider()
+    if name == "gemini":
+        from backend.services.llm.gemini_provider import GeminiProvider
+        return GeminiProvider()
+    raise ValueError(f"Unknown LLM provider: {name!r}")
+
+
+# Backwards-compatible alias
+_get_provider_by_name = get_provider_by_name
+
+
+def get_provider(stage: str) -> LLMProvider:
+    """Return the provider configured for ``stage``.
+
+    Falls back to ``settings.provider_default`` if no per-stage override.
+    Providers are cached per-name, so repeated calls return the same
+    instance (and share the underlying SDK client)."""
+    name = settings.provider_for_stage(stage)
+    return get_provider_by_name(name)
+
+
+async def call_with_fallback(
+    stage: str,
+    body: Callable[[LLMProvider, str], Awaitable[T]],
+) -> T:
+    """Run ``body(provider, model)`` for ``stage``; on any exception,
+    retry once with the fallback provider/model if one is configured via
+    ``FALLBACK_PROVIDER_<STAGE>`` / ``FALLBACK_MODEL_<STAGE>``.
+
+    The fallback runs ``body`` from scratch — any tokens spent in the
+    primary attempt are lost (and not logged). ``asyncio.CancelledError``
+    is always re-raised so cancellation still works.
+    """
+    primary_provider = get_provider(stage)
+    primary_model = settings.model_for_stage(stage)
+    try:
+        return await body(primary_provider, primary_model)
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:
+        fb = settings.fallback_for_stage(stage)
+        if fb is None:
+            raise
+        log.warning(
+            "[%s] primary %s/%s failed (%s) — falling back to %s/%s",
+            stage, primary_provider.name, primary_model,
+            exc, fb[0], fb[1],
+        )
+        fallback_provider = get_provider_by_name(fb[0])
+        return await body(fallback_provider, fb[1])

periscope/src/backend/services/llm/types.py

@@ -0,0 +1,127 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Provider-agnostic message and completion types.
+
+These dataclasses are the lingua franca between calling code and providers.
+Each provider implementation translates these into its native shape on the
+way out and back into these on the way in.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Literal
+
+
+# ---------------------------------------------------------------------------
+# Content blocks — what goes inside a Message
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class TextBlock:
+    text: str
+    cacheable: bool = False
+    # Gemini 3 / thinking-mode: opaque bytes the model returns alongside text
+    # parts that came from internal reasoning. Must be replayed verbatim when
+    # this turn is fed back into the conversation, or the next call 400s.
+    # Anthropic: always None.
+    thought_signature: bytes | None = None
+    # DeepSeek thinking-mode: assistant ``reasoning_content`` that must be
+    # replayed on the next turn or the API returns 400.
+    reasoning_content: str | None = None
+
+
+@dataclass
+class PdfBlock:
+    """Inline PDF document. Anthropic encodes as base64, Gemini as
+    inline_data. DeepSeek does not accept PDFs natively — the provider
+    converts the file to extracted text (and page images on a vision
+    model) before sending."""
+    path: Path
+    cacheable: bool = False
+
+
+@dataclass
+class ToolCall:
+    """Assistant turn: model called a tool."""
+    id: str
+    name: str
+    input: dict[str, Any]
+    # Same purpose as TextBlock.thought_signature — Gemini 3 attaches one
+    # to every function_call part when thinking is on. Round-trip required.
+    thought_signature: bytes | None = None
+    reasoning_content: str | None = None
+
+
+@dataclass
+class ToolResultBlock:
+    """User turn: result fed back from a tool the model invoked previously."""
+    tool_use_id: str
+    name: str
+    content: str
+
+
+ContentBlock = TextBlock | PdfBlock | ToolCall | ToolResultBlock
+
+
+@dataclass
+class Message:
+    role: Literal["user", "assistant"]
+    content: list[ContentBlock]
+
+
+# ---------------------------------------------------------------------------
+# Tools
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class ToolSchema:
+    """JSON-schema tool definition. Both providers accept the same shape."""
+    name: str
+    description: str
+    input_schema: dict[str, Any]
+
+
+# Tool choice: "auto" (model picks), "none" (no tools), or a forced name
+ToolChoice = Literal["auto", "none"] | dict  # {"name": "save_xyz"}
+
+
+# ---------------------------------------------------------------------------
+# Completion / usage
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class Usage:
+    """Token usage normalised across providers.
+
+    Anthropic exposes cache_creation_input_tokens (write) and
+    cache_read_input_tokens (hit). Gemini only exposes a cache hit count
+    (cached_content_token_count) — its cache writes don't bill as input.
+
+    For Gemini, ``cache_creation_tokens`` is always 0; ``cache_read_tokens``
+    holds the cached hit count when a cache was used.
+    """
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cache_creation_tokens: int = 0
+    cache_read_tokens: int = 0
+
+
+@dataclass
+class Completion:
+    """Result of a single provider.complete() / session.complete() call."""
+    text: str  # any text block(s) concatenated
+    tool_calls: list[ToolCall]
+    usage: Usage
+    stop_reason: str
+    raw_assistant_blocks: list[ContentBlock] = field(default_factory=list)
+    """The full assistant message, in our normalised content-block form, so
+    callers can append it back to the conversation history when continuing
+    the loop."""

periscope/src/backend/services/normalize_findings.py

@@ -0,0 +1,565 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Per-IC normalize pass — dedup findings with a shared root cause and
+re-grade severity against a fixed rubric.
+
+Two runs of the reviewer on identical inputs can produce different *judgments*
+(severity choices, finding-splitting) even when they reach the same underlying
+observations. This module runs a single small LLM call against the structured
+findings (no PDF, no graph tools) to:
+
+  - merge findings that describe the same defect from different angles, and
+  - re-grade each remaining finding's severity using an anchored rubric.
+
+It is intentionally conservative: if the call fails, the schema is malformed,
+or the index coverage is invalid, the original findings are returned
+unchanged. A normalize failure must never break the per-IC review.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Awaitable, Callable
+
+from backend.config import settings
+from backend.periscopex.models import Finding
+from backend.services.api_logs import ApiLogger
+from backend.services.llm import Message, TextBlock
+from backend.services.llm.factory import call_with_fallback
+from backend.services.llm.types import ToolSchema
+
+log = logging.getLogger(__name__)
+
+# Severity ordering for the downgrade-only clamp. Normalize may lower a
+# finding's severity but never raise it above what the reviewer chose — the
+# reviewer had the datasheet + graph; this pass sees only text.
+_INFO, _WARN, _ERR = 0, 1, 2
+_SEVERITY_RANK = {"INFO": _INFO, "WARNING": _WARN, "ERROR": _ERR}
+_RANK_TO_SEV = {_INFO: "INFO", _WARN: "WARNING", _ERR: "ERROR"}
+
+
+def _is_unverified(why: str | None) -> bool:
+    """True when a finding's ``why`` is flagged ``Unverified:`` — the reviewer
+    could not confirm the spec from the datasheet and deliberately hedged."""
+    return (why or "").lstrip().lower().startswith("unverified:")
+
+
+SYSTEM_PROMPT = """\
+You normalize a single IC's review findings for a hardware design review tool.
+
+Three operations:
+1. **Drop** self-cancelling findings whose own analysis confirms the \
+design is correct.
+2. **Merge** findings that share a single-fix root cause (atomic-fix test).
+3. **Re-grade severity** independently against the rubric below.
+
+You CANNOT invent new findings or new facts. Every original finding \
+(numbered 1..N) must end up in exactly one of:
+- a kept/merged entry in `findings` (referenced by `merged_from`), or
+- a dropped entry in `dropped` (referenced by `index`).
+
+You ARE shown the reviewer's original severity. The reviewer graded each \
+finding with the datasheet PDF and the design graph in front of it; you \
+see only the finding text. You may **lower** a severity when the rubric \
+clearly supports a milder grade — over-stated, conditional, or the \
+`why` itself flags incomplete evidence — but you must **never raise** a \
+finding above the reviewer's grade. Upgrading is where you have the \
+least evidence and do the most damage: a normalize pass that promotes a \
+hedged WARNING into a confident ERROR is the exact failure this rule \
+exists to prevent.
+
+### Drop rule (self-cancelling findings)
+
+A finding is self-cancelling when its own `why` confirms the requirement \
+is met or no issue actually exists. The surface reading suggested a \
+problem; the analysis itself proved otherwise. Examples:
+
+- "Output cap C1 (100 nF) is below the 1 µF minimum, but C24 (1 µF) in \
+parallel satisfies the spec." → drop. Total Cout meets spec; no issue.
+- "No dedicated input decoupling cap directly at VIN — but C3 (1 µF) is \
+on the VIN net and satisfies the requirement." → drop. C3 IS the input \
+cap, in the correct place.
+- "Pin X appears unconnected, however net Y shows it is grounded." → drop.
+
+Drop these via the `dropped` array with a short `reason`. Do NOT keep \
+them as INFO — they dilute the signal of real issues. If the `why` \
+contains "satisfies", "meets the requirement", "is in the correct \
+place", "no issue", or equivalent language confirming the design is \
+correct, the finding is almost certainly self-cancelling.
+
+A finding that flags a real concern but acknowledges *partial* \
+mitigation or *conditional* validity ("works at low load only", "meets \
+spec only at room temperature") is NOT self-cancelling — keep it.
+
+### Root-cause merge rule (atomic-fix test)
+
+Two findings share a root cause if a SINGLE atomic change resolves both. \
+The atomic-fix test: can you describe the fix in `single_fix` as ONE \
+action — remove X, replace X with Y, rewire X to Z, or add X — without \
+using "and", "also", or describing multiple steps?
+
+If yes: merge. Write the combined `finding` title naming the root cause \
+once. Restate downstream consequences inside `why`. Keep `source_page`, \
+`source_quote`, and `reference` from the original with the strongest \
+evidence.
+
+If no: do NOT merge. Two defects involving the same component, the same \
+net, or the same fix-area are still separate root causes when they \
+require separate changes.
+
+**Invalid merge example**: combining "R1 (17.8Ω) in series with VIN \
+causes dropout" with "EN tied to VIN — no independent enable" into a \
+single ERROR with `single_fix` = "Remove R1 AND route EN from a \
+separate GPIO." That is TWO changes (remove R1; rewire EN). Keep these \
+as two separate findings — the dropout finding alone may be ERROR or \
+WARNING; the EN finding is INFO.
+
+When you merge, you MUST populate `single_fix` with the one atomic \
+action. If you cannot, do not merge.
+
+### Severity rubric (grade independently)
+
+- **ERROR**: The circuit, as wired, will not function correctly. The \
+output won't reach spec, the regulator won't regulate, the signal \
+won't reach the destination, abs-max is exceeded with a strict \
+inequality (actual > limit), or a required pin is left undriven. A \
+concrete failure mode is reachable from the design as drawn.
+
+- **WARNING**: The circuit functions but has reduced margin, degraded \
+performance, or conditional malfunction (depends on load, \
+temperature, or firmware state). A recommended-but-not-required \
+component is missing. The finding is "unverified" because evidence \
+was incomplete.
+
+- **INFO**: A valid topology choice that disables an optional \
+feature, or a documentation/layout observation that cannot be \
+verified from a netlist. Examples: EN tied to VIN to use the LDO's \
+always-on mode (firmware shutdown unavailable but the chip works), \
+an optional bypass cap omitted on a non-critical pin.
+
+Grade each kept finding against this rubric, but only ever *downward* \
+from the reviewer's original severity (shown to you). A merged \
+finding's severity may not exceed the highest original severity among \
+its members. If a finding's `why` begins with `Unverified:`, the \
+reviewer could not confirm the spec from the datasheet — keep the \
+`Unverified:` prefix and never grade it above WARNING.
+
+### Output
+
+Call the `submit_normalized` tool exactly once with:
+- `findings`: kept and merged entries (each with `merged_from` indices, \
+`single_fix` if merged, and a re-graded `status`).
+- `dropped`: self-cancelling entries (each with `index` and `reason`).
+
+Every original index 1..N must appear in exactly one location across \
+both arrays. No index may appear twice.
+"""
+
+
+SUBMIT_NORMALIZED_SCHEMA = ToolSchema(
+    name="submit_normalized",
+    description=(
+        "Submit the normalized findings. Every original finding (1..N) "
+        "must appear in exactly one location across `findings.merged_from` "
+        "or `dropped.index`."
+    ),
+    input_schema={
+        "type": "object",
+        "properties": {
+            "findings": {
+                "type": "array",
+                "description": (
+                    "Kept and merged findings. Re-graded severity; merged "
+                    "entries must include `single_fix`."
+                ),
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "merged_from": {
+                            "type": "array",
+                            "items": {"type": "integer"},
+                            "minItems": 1,
+                            "description": (
+                                "1-indexed positions in the original "
+                                "findings list this output entry "
+                                "represents. Length 1 = passed through; "
+                                "length > 1 = merged."
+                            ),
+                        },
+                        "finding": {"type": "string"},
+                        "why": {"type": "string"},
+                        "status": {
+                            "type": "string",
+                            "enum": ["ERROR", "WARNING", "INFO"],
+                        },
+                        "recommendation": {"type": "string"},
+                        "source_page": {"type": ["integer", "null"]},
+                        "source_quote": {"type": "string"},
+                        "reference": {"type": "string"},
+                        "single_fix": {
+                            "type": "string",
+                            "description": (
+                                "REQUIRED when merged_from has length > 1. "
+                                "The single atomic component or net change "
+                                "that resolves ALL members of the merge "
+                                "(remove X, replace X with Y, rewire X to "
+                                "Z, or add X). If you cannot write the fix "
+                                "in one sentence without 'and' / 'also' / "
+                                "multiple steps, do NOT merge."
+                            ),
+                        },
+                        "change_rationale": {
+                            "type": "string",
+                            "description": (
+                                "≤1 line: 'unchanged', or what changed "
+                                "and why (merged X+Y, graded <sev> per "
+                                "rubric because <reason>, ...)."
+                            ),
+                        },
+                    },
+                    "required": [
+                        "merged_from",
+                        "finding",
+                        "why",
+                        "status",
+                        "recommendation",
+                        "change_rationale",
+                    ],
+                },
+            },
+            "dropped": {
+                "type": "array",
+                "description": (
+                    "Self-cancelling findings whose own `why` confirms "
+                    "the design is correct. These should NOT appear in "
+                    "`findings` — they are removed entirely from the "
+                    "report. Use this rather than demoting to INFO."
+                ),
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "index": {
+                            "type": "integer",
+                            "description": (
+                                "1-indexed position of the original "
+                                "finding being dropped."
+                            ),
+                        },
+                        "reason": {
+                            "type": "string",
+                            "description": (
+                                "Short explanation of why the finding is "
+                                "self-cancelling (e.g., 'C1<1µF but C24 "
+                                "in parallel meets spec', 'C3 is the "
+                                "input cap, already in the correct "
+                                "place')."
+                            ),
+                        },
+                    },
+                    "required": ["index", "reason"],
+                },
+            },
+        },
+        "required": ["findings"],
+    },
+)
+
+
+def _serialize_findings_for_prompt(findings: list[Finding]) -> str:
+    """Number the original findings 1..N and emit a compact JSON block.
+
+    The reviewer's `status` IS included: normalize re-grades only *downward*
+    from it (the reviewer had the datasheet + graph; this pass sees only
+    text). A deterministic clamp in ``_build_normalized`` enforces the
+    downgrade-only invariant regardless of what the model returns.
+    """
+    rows: list[dict] = []
+    for i, f in enumerate(findings, start=1):
+        rows.append({
+            "index": i,
+            "reviewer_severity": f.status,
+            "finding": f.finding,
+            "why": f.why,
+            "recommendation": f.recommendation,
+            "source_page": f.source_page,
+            "source_quote": f.source_quote,
+            "reference": f.reference,
+        })
+    return json.dumps(rows, indent=2)
+
+
+def _build_normalized(
+    raw_findings: list[dict],
+    raw_dropped: list[dict],
+    originals: list[Finding],
+) -> tuple[list[Finding], list[dict]] | None:
+    """Validate the tool output and reconstruct Finding objects.
+
+    Returns ``(kept_findings, dropped_records)`` or ``None`` if coverage /
+    schema validation fails (caller falls back to originals).
+
+    A merge with ``len(merged_from) > 1`` that omits ``single_fix`` is not
+    a hard failure — the merge is rejected and its members fall back to
+    their per-index originals. Self-cancelling drops require a non-empty
+    `reason`; missing reason = treat as ungrouped and fail coverage.
+
+    The `change_rationale` and `single_fix` fields are informational and
+    are not carried onto Finding objects; the full normalize trace keeps
+    them for forensics.
+    """
+    n = len(originals)
+    seen: set[int] = set()
+    result: list[Finding] = []
+    dropped_records: list[dict] = []
+
+    # Process explicit drops first so indices are reserved before any
+    # accidental double-coverage from a merge.
+    for d in raw_dropped or []:
+        if not isinstance(d, dict):
+            return None
+        try:
+            idx = int(d.get("index"))
+        except (TypeError, ValueError):
+            return None
+        if idx < 1 or idx > n or idx in seen:
+            return None
+        reason = str(d.get("reason") or "").strip()
+        if not reason:
+            return None
+        seen.add(idx)
+        dropped_records.append({
+            "index": idx,
+            "reason": reason,
+            "original_finding": originals[idx - 1].model_dump(mode="json"),
+        })
+
+    for entry in raw_findings:
+        if not isinstance(entry, dict):
+            return None
+        merged_from = entry.get("merged_from") or []
+        if not isinstance(merged_from, list) or not merged_from:
+            return None
+        try:
+            indices = [int(x) for x in merged_from]
+        except (TypeError, ValueError):
+            return None
+        for idx in indices:
+            if idx < 1 or idx > n or idx in seen:
+                return None
+            seen.add(idx)
+
+        # Atomic-fix test: a merge (len > 1) must populate `single_fix`.
+        # If missing, reject the merge and fall back to the per-index
+        # originals — preserves coverage but un-merges. The reviewer's
+        # original severity is preserved on the fallback path because we
+        # construct each Finding directly from `originals[i-1]`.
+        if len(indices) > 1:
+            single_fix = str(entry.get("single_fix") or "").strip()
+            if not single_fix:
+                log.warning(
+                    "normalize: merge of %s lacks single_fix — falling "
+                    "back to per-index originals (un-merging)",
+                    indices,
+                )
+                for idx in indices:
+                    result.append(originals[idx - 1])
+                continue
+
+        # Use the first original in the group as the canonical source for
+        # fields the normalize layer doesn't own (designator, mpn, aspect,
+        # finding_id). These are identical across an IC's findings anyway
+        # since normalize is per-IC.
+        canon = originals[indices[0] - 1]
+
+        # Severity safety net — downgrade-only. Normalize may lower a
+        # finding's severity but never raise it above the reviewer's
+        # calibrated grade (the reviewer had the datasheet + graph; this
+        # pass sees only text). Cap at the highest original severity among
+        # merged members; findings the reviewer marked "Unverified:" are
+        # capped at WARNING and keep that prefix. This deterministic clamp
+        # holds even when the model ignores the prompt instruction.
+        members = [originals[i - 1] for i in indices]
+        ceiling = max(_SEVERITY_RANK.get(m.status, _ERR) for m in members)
+        unverified = any(_is_unverified(m.why) for m in members)
+        if unverified:
+            ceiling = min(ceiling, _WARN)
+        proposed = str(entry.get("status") or canon.status)
+        final_status = _RANK_TO_SEV[
+            min(_SEVERITY_RANK.get(proposed, ceiling), ceiling)
+        ]
+
+        new_why = str(entry.get("why") or canon.why)
+        if unverified and not _is_unverified(new_why):
+            new_why = "Unverified: " + new_why
+
+        try:
+            result.append(canon.model_copy(update={
+                "finding": str(entry.get("finding") or canon.finding),
+                "why": new_why,
+                "source_page": entry.get("source_page", canon.source_page),
+                "source_quote": str(entry.get("source_quote") or canon.source_quote),
+                "status": final_status,
+                "recommendation": str(entry.get("recommendation") or canon.recommendation),
+                "reference": str(entry.get("reference") or canon.reference),
+            }))
+        except Exception:
+            log.exception("normalize: failed to build merged Finding")
+            return None
+    if seen != set(range(1, n + 1)):
+        return None
+    return result, dropped_records
+
+
+async def normalize_findings_async(
+    ic_ref: str,
+    mpn: str,
+    findings: list[Finding],
+    *,
+    api_logger: ApiLogger | None = None,
+    on_progress: Callable[[str, int, str, str], Awaitable[None]] | None = None,
+) -> tuple[list[Finding], dict]:
+    """Run the per-IC normalize pass.
+
+    Returns ``(normalized_findings, trace)``. On any failure (LLM error,
+    schema violation, index coverage gap), returns the original findings
+    unchanged with an ``error`` field set in the trace.
+    """
+    trace: dict = {
+        "ic_ref": ic_ref,
+        "mpn": mpn,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "input_findings": [f.model_dump(mode="json") for f in findings],
+        "output_findings": None,
+        "dropped_findings": None,
+        "submission": None,
+        "model": None,
+        "provider": None,
+        "duration_ms": None,
+        "error": None,
+    }
+
+    # Nothing to do for 0 findings. With 1 finding there is no merge to
+    # consider but the drop and re-grade rules still apply — let it
+    # through to the LLM call.
+    if not findings:
+        trace["output_findings"] = []
+        trace["dropped_findings"] = []
+        trace["error"] = "skipped: 0 findings"
+        return findings, trace
+
+    user_text = (
+        f"Original findings for IC {ic_ref} ({mpn}). "
+        f"There are {len(findings)} findings. "
+        f"Indices are 1-based.\n\n"
+        f"{_serialize_findings_for_prompt(findings)}\n\n"
+        f"Normalize them per the rubric and call submit_normalized."
+    )
+
+    t0 = time.monotonic()
+
+    async def _run(provider, model):
+        trace["model"] = model
+        trace["provider"] = provider.name
+        session = await provider.create_session(
+            model=model,
+            system=SYSTEM_PROMPT,
+            max_tokens=4096,
+            temperature=0.0,
+        )
+        try:
+            completion = await session.complete(
+                messages=[Message(
+                    role="user",
+                    content=[TextBlock(text=user_text, cacheable=False)],
+                )],
+                tools=[SUBMIT_NORMALIZED_SCHEMA],
+                tool_choice={"name": "submit_normalized"},
+            )
+            if api_logger:
+                api_logger.log(
+                    stage="normalize",
+                    identifier=ic_ref,
+                    model=model,
+                    provider=provider.name,
+                    input_tokens=completion.usage.input_tokens,
+                    output_tokens=completion.usage.output_tokens,
+                    cache_creation_input_tokens=completion.usage.cache_creation_tokens,
+                    cache_read_input_tokens=completion.usage.cache_read_tokens,
+                    duration_ms=int((time.monotonic() - t0) * 1000),
+                    stop_reason="submit_normalized",
+                    turns=1,
+                )
+            for tc in completion.tool_calls:
+                if tc.name == "submit_normalized":
+                    return tc.input
+            return None
+        finally:
+            await session.close()
+
+    try:
+        submission = await call_with_fallback("normalize", _run)
+    except Exception as exc:
+        log.exception("normalize: call failed for %s", ic_ref)
+        trace["error"] = f"{type(exc).__name__}: {exc}"
+        trace["duration_ms"] = int((time.monotonic() - t0) * 1000)
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    trace["duration_ms"] = int((time.monotonic() - t0) * 1000)
+    trace["submission"] = submission
+
+    if not submission or not isinstance(submission, dict):
+        trace["error"] = "no submission"
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    raw_findings = submission.get("findings") or []
+    if not isinstance(raw_findings, list):
+        trace["error"] = "submission.findings not a list"
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    raw_dropped = submission.get("dropped") or []
+    if not isinstance(raw_dropped, list):
+        trace["error"] = "submission.dropped not a list"
+        trace["output_findings"] = trace["input_findings"]
+        return findings, trace
+
+    built = _build_normalized(raw_findings, raw_dropped, findings)
+    if built is None:
+        trace["error"] = "invalid index coverage or schema"
+        trace["output_findings"] = trace["input_findings"]
+        log.warning(
+            "normalize: invalid output for %s (%d originals, %d kept, "
+            "%d dropped) — falling back to originals",
+            ic_ref, len(findings), len(raw_findings), len(raw_dropped),
+        )
+        if on_progress:
+            try:
+                await on_progress(
+                    ic_ref, 0, "normalize_skipped",
+                    f"invalid output, kept {len(findings)} originals",
+                )
+            except Exception:
+                pass
+        return findings, trace
+
+    normalized, dropped_records = built
+    trace["output_findings"] = [f.model_dump(mode="json") for f in normalized]
+    trace["dropped_findings"] = dropped_records
+    if on_progress:
+        try:
+            await on_progress(
+                ic_ref, 0, "normalized",
+                f"{len(findings)} → {len(normalized)} kept, "
+                f"{len(dropped_records)} dropped",
+            )
+        except Exception:
+            pass
+    return normalized, trace

periscope/src/backend/services/purple_parts.py

@@ -0,0 +1,339 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Purple Parts API client — LCSC code → MPN resolution.
+
+Wraps the external `purple-parts` HTTP service (a read-only API over the
+jlcparts/LCSC catalogue, deployed at the URL in `settings.purple_parts_url`).
+Used by the BOM-parse stage to convert LCSC codes (e.g. "C12345") into
+manufacturer part numbers before the DigiKey resolver runs.
+
+The remote service is Cloud Run with IAM auth, so calls send a Google
+identity token (audience = purple_parts_url) plus an X-API-Key header. In
+Cloud Run the identity token is minted automatically via ADC + the
+metadata server; locally `fetch_id_token` only works if
+GOOGLE_APPLICATION_CREDENTIALS points at a service-account key file. On
+local dev with user creds the helper logs a debug line and the call is
+skipped (returns an empty result), which the caller treats as a no-op.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+import time
+from typing import Optional
+
+import httpx
+
+from backend.config import settings
+
+logger = logging.getLogger(__name__)
+
+_LCSC_RE = re.compile(r"^C\d+$", re.IGNORECASE)
+
+# Identity tokens are valid for ~1h; refresh ~10 min early.
+_TOKEN_TTL_SECONDS = 50 * 60
+_token_cache: dict[str, float | str] = {"token": "", "expires_at": 0.0}
+_token_lock = asyncio.Lock()
+
+# Conservative batch size — purple-parts accepts up to 500 per request.
+_BATCH_SIZE = 400
+
+
+def is_lcsc_code(value: str | None) -> bool:
+    """Return True if `value` looks like an LCSC part number (e.g. C12345)."""
+    if not value:
+        return False
+    return bool(_LCSC_RE.match(value.strip()))
+
+
+async def _get_identity_token() -> str | None:
+    """Mint a Google ID token for the purple-parts audience, cached.
+
+    Returns None when credentials don't support identity-token minting
+    (typical for local dev with `gcloud auth application-default login` user
+    creds). Caller should treat None as "skip the purple-parts call."
+    """
+    now = time.time()
+    cached = _token_cache.get("token", "")
+    if cached and float(_token_cache.get("expires_at", 0.0)) > now:
+        return str(cached)
+
+    async with _token_lock:
+        cached = _token_cache.get("token", "")
+        if cached and float(_token_cache.get("expires_at", 0.0)) > now:
+            return str(cached)
+
+        try:
+            from google.auth.transport.requests import Request
+            from google.oauth2 import id_token as gid_token
+        except ImportError:
+            logger.warning("google-auth not installed; purple-parts disabled")
+            return None
+
+        loop = asyncio.get_running_loop()
+        try:
+            token = await loop.run_in_executor(
+                None,
+                lambda: gid_token.fetch_id_token(Request(), settings.purple_parts_url),
+            )
+        except Exception as e:
+            logger.debug(
+                "purple-parts: identity-token mint failed (%s: %s) — "
+                "expected for local user creds, skipping",
+                type(e).__name__, e,
+            )
+            return None
+
+        _token_cache["token"] = token
+        _token_cache["expires_at"] = now + _TOKEN_TTL_SECONDS
+        return token
+
+
+def detect_lcsc_column(csv_bytes: bytes, mpn_col: str) -> bool:
+    """Return True when every non-empty value in `mpn_col` matches `^C\\d+$`.
+
+    Used by the upload endpoint to auto-detect when the user's chosen MPN
+    column is actually an LCSC column (i.e. the user pasted LCSC ids into
+    the MPN slot, or labeled their LCSC column as "Manufacturer Part Number").
+    Column-level — a single non-LCSC entry disqualifies the column so that
+    BOMs mixing real MPNs with LCSC ids aren't silently mangled.
+    """
+    import csv as csv_mod
+    import io
+
+    text = csv_bytes.decode("utf-8", errors="replace")
+    reader = csv_mod.DictReader(io.StringIO(text))
+    if not reader.fieldnames or mpn_col not in reader.fieldnames:
+        return False
+
+    seen_any = False
+    for row in reader:
+        val = (row.get(mpn_col) or "").strip()
+        if not val:
+            continue
+        if not is_lcsc_code(val):
+            return False
+        seen_any = True
+    return seen_any
+
+
+async def resolve_lcsc_column_bytes(
+    csv_bytes: bytes,
+    *,
+    mpn_col: str = "Manufacturer Part Number",
+) -> tuple[bytes, int, dict[str, str], dict[str, dict]]:
+    """Replace every value in `mpn_col` with the manufacturer part number
+    resolved via purple-parts.
+
+    Returns `(new_csv_bytes, rows_updated, lcsc_to_mpn_map, lcsc_payloads_map)`.
+    The first map is keyed by LCSC id (e.g. "C12044") → resolved MPN string,
+    so the caller can surface "C12044 → STM32F103C8T6" in the UI. The second
+    map is keyed by the same LCSC id → the full purple-parts payload (mpn,
+    manufacturer, package, description, category, subcategory, ...) so the
+    caller can cache it on the project for the wizard's per-row resolve
+    endpoint. Preserves column order, headers, and untouched cells. No-op
+    when purple-parts isn't configured.
+    """
+    import csv as csv_mod
+    import io
+
+    if not settings.use_purple_parts:
+        return csv_bytes, 0, {}, {}
+
+    text = csv_bytes.decode("utf-8", errors="replace")
+    reader = csv_mod.DictReader(io.StringIO(text))
+    fieldnames = reader.fieldnames or []
+    rows = list(reader)
+
+    if not rows or mpn_col not in fieldnames:
+        return csv_bytes, 0, {}, {}
+
+    todo: list[tuple[int, str]] = []
+    for i, row in enumerate(rows):
+        code = (row.get(mpn_col) or "").strip()
+        if is_lcsc_code(code):
+            todo.append((i, code))
+
+    if not todo:
+        return csv_bytes, 0, {}, {}
+
+    unique_codes = sorted({c for _, c in todo})
+    resolved = await lookup_lcsc_batch(unique_codes)
+
+    updated = 0
+    lcsc_to_mpn: dict[str, str] = {}
+    lcsc_payloads: dict[str, dict] = {}
+    for i, code in todo:
+        part = resolved.get(code)
+        if part and part.get("mpn"):
+            rows[i][mpn_col] = part["mpn"]
+            lcsc_to_mpn[code] = part["mpn"]
+            lcsc_payloads[code] = dict(part)
+            updated += 1
+
+    if updated == 0:
+        return csv_bytes, 0, {}, {}
+
+    out = io.StringIO()
+    writer = csv_mod.DictWriter(out, fieldnames=fieldnames, extrasaction="ignore")
+    writer.writeheader()
+    writer.writerows(rows)
+    return out.getvalue().encode("utf-8"), updated, lcsc_to_mpn, lcsc_payloads
+
+
+async def lookup_lcsc_batch(lcsc_codes: list[str]) -> dict[str, Optional[dict]]:
+    """Batch LCSC → MPN lookup.
+
+    Returns `{lcsc_code: part_dict_or_None}` for every code in input. Misses,
+    invalid codes, and (after warning) total failures all return None values
+    so the caller can treat the result as a uniform per-code map. The pipeline
+    never aborts on a purple-parts miss; the row simply stays unresolved and
+    the existing DigiKey/Haiku paths handle it.
+
+    Part dict shape: {lcsc, mpn, manufacturer, package, description, stock,
+    basic, preferred}.
+    """
+    if not settings.use_purple_parts:
+        return {c: None for c in lcsc_codes}
+
+    codes = [c for c in (raw.strip() for raw in lcsc_codes) if c]
+    if not codes:
+        return {}
+
+    token = await _get_identity_token()
+    if token is None:
+        logger.info("purple-parts: no identity token, skipping batch of %d", len(codes))
+        return {c: None for c in codes}
+
+    base_url = settings.purple_parts_url.rstrip("/")
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "X-API-Key": settings.purple_parts_api_key,
+        "Content-Type": "application/json",
+    }
+
+    results: dict[str, Optional[dict]] = {c: None for c in codes}
+
+    async with httpx.AsyncClient(timeout=15) as client:
+        for i in range(0, len(codes), _BATCH_SIZE):
+            chunk = codes[i:i + _BATCH_SIZE]
+            try:
+                resp = await client.post(
+                    f"{base_url}/v1/parts/by-lcsc/batch",
+                    headers=headers,
+                    json={"ids": chunk},
+                )
+                resp.raise_for_status()
+                body = resp.json()
+            except httpx.HTTPStatusError as e:
+                logger.warning(
+                    "purple-parts: batch call failed %s for chunk of %d",
+                    e.response.status_code, len(chunk),
+                )
+                continue
+            except Exception as e:
+                logger.warning("purple-parts: batch call error: %s", e)
+                continue
+
+            for code, part in (body.get("results") or {}).items():
+                results[code] = part
+
+    return results
+
+
+def _norm_mpn(value: str | None) -> str:
+    """Normalize an MPN for comparison: drop whitespace, uppercase."""
+    return "".join((value or "").split()).upper()
+
+
+def _pick_exact(query: str, candidates: list[dict]) -> Optional[dict]:
+    """Return the candidate whose ``mpn`` exactly matches ``query``.
+
+    Match is case- and whitespace-insensitive. purple-parts' ``by-mpn``
+    endpoint returns exact matches first and then prefix matches, but we
+    re-check rather than trust ordering — a prefix-only hit (e.g. a series
+    family for a more specific MPN) must be treated as a miss so it can't
+    pollute the shared passive library. Mirrors the exact-MPN discipline of
+    ``services.digikey._find_product``.
+    """
+    q = _norm_mpn(query)
+    for part in candidates:
+        if part and _norm_mpn(part.get("mpn")) == q:
+            return part
+    return None
+
+
+async def lookup_mpn_batch(mpns: list[str]) -> dict[str, Optional[dict]]:
+    """Reverse lookup: manufacturer part number → LCSC catalogue record.
+
+    Fans the unique MPNs out to purple-parts' batch endpoint
+    (``POST /v1/parts/by-mpn/batch``) in chunks of ``_BATCH_SIZE`` — one indexed
+    query per chunk instead of a GET per MPN, which is what stalled huge-BOM
+    uploads when the by-mpn query was seq-scanning. Returns
+    ``{mpn: part_dict_or_None}`` keyed by the *input* MPN string.
+
+    The endpoint is exact-match only, and we additionally run :func:`_pick_exact`
+    over each MPN's candidate list (case/whitespace-insensitive) to keep the
+    exact-MPN discipline — a prefix / family hit can carry the wrong
+    voltage / dielectric / package and must never reach the shared
+    ``library/passives``. Misses, missing creds (no identity token), and per-chunk
+    failures all come back as ``None`` so the caller can treat the map uniformly.
+    No-op (all ``None``) when purple-parts isn't configured.
+
+    Part dict shape matches :func:`lookup_lcsc_batch`: {lcsc, mpn, manufacturer,
+    package, description, category, subcategory, stock, basic, preferred}.
+    """
+    if not settings.use_purple_parts:
+        return {m: None for m in mpns}
+
+    # Preserve input keys but query each unique, non-empty MPN once.
+    names = list(dict.fromkeys(m.strip() for m in mpns if m and m.strip()))
+    if not names:
+        return {}
+
+    token = await _get_identity_token()
+    if token is None:
+        logger.info("purple-parts: no identity token, skipping by-mpn batch of %d", len(names))
+        return {m: None for m in names}
+
+    base_url = settings.purple_parts_url.rstrip("/")
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "X-API-Key": settings.purple_parts_api_key,
+        "Content-Type": "application/json",
+    }
+
+    results: dict[str, Optional[dict]] = {m: None for m in names}
+
+    async with httpx.AsyncClient(timeout=15) as client:
+        for i in range(0, len(names), _BATCH_SIZE):
+            chunk = names[i:i + _BATCH_SIZE]
+            try:
+                resp = await client.post(
+                    f"{base_url}/v1/parts/by-mpn/batch",
+                    headers=headers,
+                    json={"mpns": chunk},
+                )
+                resp.raise_for_status()
+                body = resp.json()
+            except httpx.HTTPStatusError as e:
+                logger.warning(
+                    "purple-parts: by-mpn batch call failed %s for chunk of %d",
+                    e.response.status_code, len(chunk),
+                )
+                continue
+            except Exception as e:
+                msg = str(e) or type(e).__name__
+                logger.warning("purple-parts: by-mpn batch call error: %s", msg)
+                continue
+
+            # Each MPN maps to a candidate list; keep only the exact match.
+            for mpn, candidates in (body.get("results") or {}).items():
+                results[mpn] = _pick_exact(mpn, candidates or [])
+
+    return results

periscope/src/backend/services/storage.py

@@ -0,0 +1,269 @@
+"""Native Periscope overlay: leftover service still imported from src.
+
+PinScope original remains in dependency/.
+"""
+
+"""Storage abstraction layer.
+
+Provides a StorageBackend protocol with two implementations:
+  - LocalStorageBackend: maps GCS-style keys to local filesystem paths (dev/test)
+  - GCSStorageBackend: uses Google Cloud Storage (production)
+
+Keys use forward-slash-separated paths like GCS object names:
+  users/{user_id}/projects/{project_id}/project.json
+  library/extracted/{safe_mpn}.json
+  taxonomy/ic.json
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import threading
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+
+# Sentinel used by conditional writes to require that the object does not yet
+# exist (matches GCS ``if_generation_match=0`` semantics).
+GENERATION_NEW = 0
+
+
+class StaleGeneration(Exception):
+    """Raised when a conditional write loses an optimistic-concurrency race."""
+
+
+@runtime_checkable
+class StorageBackend(Protocol):
+    """Abstract storage interface used by all backend services."""
+
+    def read_json(self, key: str) -> dict:
+        """Read and parse a JSON object."""
+        ...
+
+    def write_json(self, key: str, data: dict) -> None:
+        """Serialize and write a JSON object."""
+        ...
+
+    def read_bytes(self, key: str) -> bytes:
+        """Read raw bytes."""
+        ...
+
+    def write_bytes(self, key: str, data: bytes) -> None:
+        """Write raw bytes."""
+        ...
+
+    def read_text(self, key: str) -> str:
+        """Read as UTF-8 text."""
+        ...
+
+    def write_text(self, key: str, text: str) -> None:
+        """Write UTF-8 text."""
+        ...
+
+    def exists(self, key: str) -> bool:
+        """Check if an object exists."""
+        ...
+
+    def list_prefix(self, prefix: str) -> list[str]:
+        """List all keys under a prefix (non-recursive by default).
+
+        Returns keys that are direct children of the prefix — i.e. one level
+        deep. For example, listing ``users/abc/projects/`` returns keys like
+        ``users/abc/projects/p1/project.json`` but NOT keys nested further.
+
+        To list all keys recursively, use list_recursive().
+        """
+        ...
+
+    def list_recursive(self, prefix: str) -> list[str]:
+        """List all keys under a prefix, recursively."""
+        ...
+
+    def list_prefix_after(self, prefix: str, after_key: str | None = None) -> list[str]:
+        """List keys under ``prefix`` whose name lexicographically follows
+        ``after_key``. Used by the GCS-backed event tail (worker writes one
+        object per event with a zero-padded sequence number; the SSE
+        consumer pages through new files only).
+        """
+        ...
+
+    def read_json_with_generation(self, key: str) -> tuple[dict, int]:
+        """Read JSON and return ``(data, generation)``.
+
+        ``generation`` is an opaque token that callers pass back to
+        ``write_json_if_match`` to detect lost-update races.
+        """
+        ...
+
+    def write_json_if_match(self, key: str, data: dict, generation: int) -> int:
+        """Write JSON only if the current generation equals ``generation``.
+
+        Pass ``GENERATION_NEW`` (0) to require that the key does not exist.
+        Returns the new generation. Raises :class:`StaleGeneration` when the
+        precondition fails (loser of a race).
+        """
+        ...
+
+    def delete_key(self, key: str) -> None:
+        """Delete a single object."""
+        ...
+
+    def delete_prefix(self, prefix: str) -> None:
+        """Delete all objects under a prefix (recursive)."""
+        ...
+
+    def copy_object(self, src_key: str, dst_key: str) -> None:
+        """Copy an object from src to dst."""
+        ...
+
+    def download_to_local(self, key: str, local_path: Path) -> Path:
+        """Download an object to a local file path. Returns the local path."""
+        ...
+
+    def upload_from_local(self, local_path: Path, key: str) -> None:
+        """Upload a local file to storage."""
+        ...
+
+    def signed_url(self, key: str, expiration_minutes: int = 15) -> str:
+        """Generate a time-limited URL for direct access to an object.
+
+        For LocalStorageBackend, returns a backend-proxied URL.
+        For GCS, returns a signed GCS URL.
+        """
+        ...
+
+
+class LocalStorageBackend:
+    """Maps GCS-style keys to local filesystem paths under a base directory.
+
+    Key ``users/abc/projects/p1/project.json`` becomes
+    ``{base_dir}/users/abc/projects/p1/project.json``.
+    """
+
+    def __init__(self, base_dir: Path) -> None:
+        self._base = base_dir
+        # In-memory generation counter for optimistic-concurrency parity with
+        # GCS. Single-process only; subprocess-based local workers run in a
+        # different process and will collide on the meta key. The local
+        # subprocess path is dev-only and rarely concurrent, so we accept it.
+        self._generations: dict[str, int] = {}
+        self._gen_lock = threading.Lock()
+
+    def _path(self, key: str) -> Path:
+        return self._base / key
+
+    def read_json(self, key: str) -> dict:
+        return json.loads(self._path(key).read_text())
+
+    def write_json(self, key: str, data: dict) -> None:
+        p = self._path(key)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(json.dumps(data, indent=2) + "\n")
+
+    def read_bytes(self, key: str) -> bytes:
+        return self._path(key).read_bytes()
+
+    def write_bytes(self, key: str, data: bytes) -> None:
+        p = self._path(key)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_bytes(data)
+
+    def read_text(self, key: str) -> str:
+        return self._path(key).read_text()
+
+    def write_text(self, key: str, text: str) -> None:
+        p = self._path(key)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(text)
+
+    def exists(self, key: str) -> bool:
+        return self._path(key).is_file()
+
+    def list_prefix(self, prefix: str) -> list[str]:
+        d = self._path(prefix)
+        if not d.is_dir():
+            return []
+        keys: list[str] = []
+        for child in sorted(d.iterdir()):
+            rel = child.relative_to(self._base)
+            keys.append(str(rel))
+        return keys
+
+    def list_recursive(self, prefix: str) -> list[str]:
+        d = self._path(prefix)
+        if not d.is_dir():
+            return []
+        keys: list[str] = []
+        for child in sorted(d.rglob("*")):
+            if child.is_file():
+                rel = child.relative_to(self._base)
+                keys.append(str(rel))
+        return keys
+
+    def list_prefix_after(self, prefix: str, after_key: str | None = None) -> list[str]:
+        d = self._path(prefix)
+        if not d.is_dir():
+            return []
+        keys: list[str] = []
+        for child in sorted(d.iterdir()):
+            if not child.is_file():
+                continue
+            rel = str(child.relative_to(self._base))
+            if after_key is not None and rel <= after_key:
+                continue
+            keys.append(rel)
+        return keys
+
+    def read_json_with_generation(self, key: str) -> tuple[dict, int]:
+        data = json.loads(self._path(key).read_text())
+        with self._gen_lock:
+            gen = self._generations.get(key, 1)
+        return data, gen
+
+    def write_json_if_match(self, key: str, data: dict, generation: int) -> int:
+        p = self._path(key)
+        with self._gen_lock:
+            current = self._generations.get(key, 0 if not p.is_file() else 1)
+            if generation != current:
+                raise StaleGeneration(
+                    f"generation mismatch on {key}: expected {generation}, current {current}"
+                )
+            new_gen = current + 1
+            p.parent.mkdir(parents=True, exist_ok=True)
+            p.write_text(json.dumps(data, indent=2) + "\n")
+            self._generations[key] = new_gen
+        return new_gen
+
+    def delete_key(self, key: str) -> None:
+        p = self._path(key)
+        if p.is_file():
+            p.unlink()
+            with self._gen_lock:
+                self._generations.pop(key, None)
+
+    def delete_prefix(self, prefix: str) -> None:
+        d = self._path(prefix)
+        if d.is_dir():
+            shutil.rmtree(d)
+
+    def copy_object(self, src_key: str, dst_key: str) -> None:
+        src = self._path(src_key)
+        dst = self._path(dst_key)
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(src, dst)
+
+    def download_to_local(self, key: str, local_path: Path) -> Path:
+        src = self._path(key)
+        local_path.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(src, local_path)
+        return local_path
+
+    def upload_from_local(self, local_path: Path, key: str) -> None:
+        dst = self._path(key)
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(local_path, dst)
+
+    def signed_url(self, key: str, expiration_minutes: int = 15) -> str:
+        # Local dev: return a path that the backend can serve directly
+        return f"/api/datasheets/_local/{key}"

periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md

@@ -1,6 +1,6 @@
 # Piano — indipendenza architettonica e di licenza da PinScope
 
-**Stato:** split **2.38.0**. C2–C4 as before. **2.41.0** job workspace. **2.42.0** graph/parsers/models/taxonomy overlay. **2.43.0** leftover helpers + live `pipeline`/`validation`/`extraction`/`validate` overlay; skills + taxonomy JSON copied to `periscope/src`. Originals **not** deleted. Fork non staccato.  
+**Stato:** split **2.38.0**. C2–C4 as before. **2.41.0** job workspace. **2.42.0** graph/parsers/models/taxonomy overlay. **2.43.0** leftover helpers + live `pipeline`/`validation`/`extraction`/`validate` overlay; skills + taxonomy JSON copied to `periscope/src`. **2.44.0** leftover services + LLM factory/types/base overlay. Originals **not** deleted. Fork non staccato.  
 **Gate Michele:** sostituire/smettere di chiamare un modulo `dependency/` solo dopo pytest + deploy smoke. Se la verifica fallisce, resta il path ereditato.  
 **Sequenza:** split → sostituzione incrementale (C2…C5 overlay). **Mai** empty-delete. Auto-place fuori scope. AGPL resta.
 
@@ -32,7 +32,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope:
 | Pacchetto logico | Path **dopo lo split** | Licenza da audit |
 | --- | --- | --- |
 | Core schematico PinScope | `periscope/dependency/backend/periscopex/…` — **overlay 2.42.0–2.43.0** in `periscope/src` (same names; keep inherited files) | AGPL-3.0 del fork (blob `LICENSE` identico a upstream) |
-| Orchestrazione review | `pipeline_worker.py` still dependency-only; `pipeline.py` / `validation.py` / `extraction.py` **overlay 2.43.0** in src | stessa |
+| Orchestrazione review | `pipeline_worker.py` still dependency-only; `pipeline.py` / `validation.py` / `extraction.py` **overlay 2.43.0**; leftover services **2.44.0** in src | stessa |
 | Skills Anthropic/Console | `periscope/dependency/skills/…` kept; **src copy 2.43.0** `periscope/src/skills/` + `src/backend/skills_manifest.json`. `upload_skills.py` still dependency | stessa + contratto Claude |
 | UI OSS / marketing shell | `periscope/dependency/frontend/` (UPSTREAM/DERIVED); file nativi in `periscope/src/frontend/` con symlink nel recinto | AGPL |
 | Gateway seams | `billing_hook.py`, `proxy.ts`, `clerk-theme-provider.tsx`, … sotto `periscope/dependency/` | stub open-core PinScope |
@@ -63,6 +63,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope:
 | Job workspace | `periscope/src/backend/services/job_workspace.py` | REPLACEMENT 2.41.0; PCB/placement off `pipeline.py` |
 | Graph / parsers / models / taxonomy | `periscope/src/backend/periscopex/{graph,parsers,parsers_edif,models,taxonomy}.py` | OVERLAY 2.42.0; inherited copies kept |
 | Leftover helpers + analysis overlay | `utils`, `resolve_passives`, `derating`, `bom_summary`, `pin_mux_check`, `led_current_check`, `pin_function_tokens`, `validate`, `validation_tools`, `services/{pipeline,validation,extraction}.py`, `src/taxonomy`, `src/skills` | OVERLAY 2.43.0 |
+| Leftover services overlay | `api_logs`, `normalize_findings`, `dedupe_findings`, `billing_hook`, `datasheet_store`, `cost_estimator`, `storage`, `purple_parts`, `email`, `digikey`, `llm/{factory,types,base}` | OVERLAY 2.44.0 |
 | Deploy | `scripts/update-periscope.sh`, `docker-compose.yml`, `periscope/src/backend/Dockerfile` | NEW (nomi `pinscope_*` ancora WEAK) |
 | Plugin KiCad | `periscope/src/plugins/kicad/` | NEW |
 | Check deterministici fork | `dnp_check`, `sequencing_check`, `layout_rules`, … under `periscope/src` | NEW ma **INDIRECT**: usano `models` / graph |

tests/test_native_leftover_services_overlay.py

@@ -0,0 +1,64 @@
+"""Native overlay: leftover services src still imports resolve from src."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import backend.services.api_logs as api_logs
+import backend.services.billing_hook as billing_hook
+import backend.services.cost_estimator as cost_estimator
+import backend.services.datasheet_store as datasheet_store
+import backend.services.dedupe_findings as dedupe_findings
+import backend.services.digikey as digikey
+import backend.services.email as email
+import backend.services.normalize_findings as normalize_findings
+import backend.services.purple_parts as purple_parts
+import backend.services.storage as storage
+import backend.services.llm.base as llm_base
+import backend.services.llm.factory as llm_factory
+import backend.services.llm.types as llm_types
+
+
+def _src(mod) -> Path:
+    return Path(mod.__file__).resolve()
+
+
+def test_leftover_services_load_from_src():
+    root = Path(__file__).resolve().parents[1]
+    svc = (root / "periscope" / "src" / "backend" / "services").resolve()
+    for mod, name in (
+        (api_logs, "api_logs.py"),
+        (normalize_findings, "normalize_findings.py"),
+        (dedupe_findings, "dedupe_findings.py"),
+        (billing_hook, "billing_hook.py"),
+        (datasheet_store, "datasheet_store.py"),
+        (cost_estimator, "cost_estimator.py"),
+        (storage, "storage.py"),
+        (purple_parts, "purple_parts.py"),
+        (email, "email.py"),
+        (digikey, "digikey.py"),
+    ):
+        path = _src(mod)
+        assert path == svc / name, path
+        assert "Native Periscope overlay" in path.read_text(encoding="utf-8")[:500]
+
+
+def test_leftover_llm_factory_types_base_load_from_src():
+    root = Path(__file__).resolve().parents[1]
+    llm = (root / "periscope" / "src" / "backend" / "services" / "llm").resolve()
+    for mod, name in (
+        (llm_factory, "factory.py"),
+        (llm_types, "types.py"),
+        (llm_base, "base.py"),
+    ):
+        path = _src(mod)
+        assert path == llm / name, path
+        assert "Native Periscope overlay" in path.read_text(encoding="utf-8")[:500]
+
+
+def test_billing_hook_still_null_without_stripe():
+    from backend.services.billing_hook import InsufficientCredits, get_billing
+
+    assert billing_hook.InsufficientCredits is InsufficientCredits
+    billing = get_billing()
+    assert billing.__class__.__name__ in {"NullBilling", "CreditsBilling"}