Rewrite normalize, cross-IC dedupe, and billing seam (2.49.0).

Downgrade-only clamp and fail-soft coverage stay in src. Inherited copies remain on disk. Self-host billing is NullBilling.

periscope/dependency/frontend/content/changelog.md

@@ -2,6 +2,14 @@
 
 What's new in Periscope.
 
+## 2.49.0 — 2026-09-20 — Native normalize, cross-IC dedupe, billing seam
+
+`normalize_findings.py`, `dedupe_findings.py`, and `billing_hook.py` are original Periscope code in `periscope/src`. Inherited copies stay on disk. Self-host `get_billing()` is NullBilling (no Stripe at import). Downgrade-only clamp unchanged (U2-001 / Emmaforo UART).
+
+- [New] Per-IC normalize builder + tool schema rewrite; fail-soft to originals.
+- [New] Cross-IC dedupe rewrite (CH340E↔MCU interface collapse).
+- [New] Billing seam rewrite; credits stay lazy behind `billing_enabled`.
+
 ## 2.48.0 — 2026-09-20 — Native models, schematic parsers, taxonomy loader
 
 `models.py`, `parsers.py`, `parsers_edif.py`, and `taxonomy.py` are original Periscope code in `periscope/src` (not overlay stamps). Inherited copies stay on disk. Taxonomy JSON is still `repo_paths.taxonomy_dir()` → `periscope/dependency/taxonomy`. Native `parsers_kicad.py` / `parsers_kicad_pcb.py` (via≠pad) unchanged.

periscope/src/backend/services/billing_hook.py

@@ -0,0 +1,165 @@
+"""Periscope billing seam: self-host is free; hosted credits stay behind a flag.
+
+Pipeline, API logs, and routers must call :func:`get_billing` only. This file
+does not import Stripe or a credits ledger at module load.
+"""
+
+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):
+    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):
+    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:
+    """Self-host default: zero credits, no writes, no trial grant."""
+
+    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:
+    """Optional hosted path. Imports credits only when a method runs."""
+
+    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:
+        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: CreditsBilling | None = None
+
+
+def get_billing() -> BillingHook:
+    if not settings.billing_enabled:
+        return _NULL
+    global _credits
+    if _credits is None:
+        _credits = CreditsBilling()
+    return _credits

periscope/src/backend/services/dedupe_findings.py

@@ -0,0 +1,305 @@
+"""Cross-IC dedup: one physical interface defect, one finding.
+
+Per-IC review cannot see the other endpoint. This pass groups concatenated
+findings. It never drops (normalize does that) and never raises severity
+above the strongest member. Fail-soft on bad coverage.
+"""
+
+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__)
+
+_RANK = {"INFO": 0, "WARNING": 1, "ERROR": 2}
+_FROM_RANK = {0: "INFO", 1: "WARNING", 2: "ERROR"}
+
+
+def _unverified(why: str | None) -> bool:
+    return (why or "").lstrip().lower().startswith("unverified:")
+
+
+def _clamp_status(proposed: str, members: list[Finding], fallback: str) -> str:
+    cap = max(_RANK.get(m.status, 2) for m in members)
+    if any(_unverified(m.why) for m in members):
+        cap = min(cap, 1)
+    key = proposed if proposed in _RANK else fallback
+    return _FROM_RANK[min(_RANK.get(key, cap), cap)]
+
+
+SYSTEM_PROMPT = """\
+Deduplicate Periscope review findings across ICs.
+
+Each IC was reviewed alone, so one UART/USB/SPI defect often appears twice —
+once from each chip. Group findings that name the same physical defect (same
+net, same interface, same shared part and same fix). Do not invent, drop, or
+rewrite the engineering claim.
+
+Keep findings separate when they need different fixes (different pins, a
+decoupling issue vs a voltage issue, unrelated problems on one part). Prefer
+a visible duplicate over a wrong merge.
+
+Severity of a merge is the highest among members, never above that. If any
+member `why` starts with `Unverified:`, keep the prefix and stay ≤ WARNING.
+
+Call `submit_deduped` once. Every original index 1..N is in exactly one
+group's `member_indices`. Length 1 is passthrough (do not restyle the text).
+Length > 1 is a merge: provide finding/why/status/recommendation and a
+`primary_index` (a member) whose datasheet citation and designator win.
+"""
+
+
+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 originals. Length 1 = passthrough; "
+                                "length > 1 = merge."
+                            ),
+                        },
+                        "primary_index": {
+                            "type": ["integer", "null"],
+                            "description": (
+                                "Required for merges: member that supplies "
+                                "designator and datasheet citation."
+                            ),
+                        },
+                        "finding": {"type": "string"},
+                        "why": {"type": "string"},
+                        "status": {
+                            "type": "string",
+                            "enum": ["ERROR", "WARNING", "INFO"],
+                        },
+                        "recommendation": {"type": "string"},
+                        "change_rationale": {
+                            "type": "string",
+                            "description": "passthrough, or merged N+M: shared interface.",
+                        },
+                    },
+                    "required": ["member_indices", "change_rationale"],
+                },
+            },
+        },
+        "required": ["groups"],
+    },
+)
+
+
+def _serialize_findings_for_prompt(findings: list[Finding]) -> str:
+    rows = [
+        {
+            "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,
+        }
+        for i, f in enumerate(findings, start=1)
+    ]
+    return json.dumps(rows, indent=2)
+
+
+def _build_deduped(
+    raw_groups: list[dict],
+    originals: list[Finding],
+) -> list[Finding] | None:
+    n = len(originals)
+    seen: set[int] = set()
+    result: list[Finding] = []
+
+    for group in raw_groups:
+        if not isinstance(group, dict):
+            return None
+        raw = group.get("member_indices") or []
+        if not isinstance(raw, list) or not raw:
+            return None
+        try:
+            indices = [int(x) for x in raw]
+        except (TypeError, ValueError):
+            return None
+        for idx in indices:
+            if idx < 1 or idx > n or idx in seen:
+                return None
+            seen.add(idx)
+
+        if len(indices) == 1:
+            result.append(originals[indices[0] - 1])
+            continue
+
+        try:
+            primary = int(group.get("primary_index"))
+        except (TypeError, ValueError):
+            primary = None
+        if primary not in indices:
+            log.warning(
+                "dedupe: merge %s bad primary_index %r — un-merge",
+                indices,
+                group.get("primary_index"),
+            )
+            result.extend(originals[i - 1] for i in indices)
+            continue
+
+        members = [originals[i - 1] for i in indices]
+        canon = originals[primary - 1]
+        final_status = _clamp_status(
+            str(group.get("status") or ""), members, canon.status
+        )
+        new_why = str(group.get("why") or canon.why)
+        if any(_unverified(m.why) for m in members) and not _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: Finding rebuild failed")
+            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]:
+    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,
+    }
+    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:
+        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 — keep originals")
+        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/normalize_findings.py

@@ -0,0 +1,385 @@
+"""Per-IC finding normalize: drop, merge, re-grade — downgrade only.
+
+The reviewer already saw the datasheet and graph. This pass sees finding text
+only, so it may lower severity, never raise it. Fail-soft: bad tool output
+returns the originals.
+"""
+
+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__)
+
+_RANK = {"INFO": 0, "WARNING": 1, "ERROR": 2}
+_FROM_RANK = {0: "INFO", 1: "WARNING", 2: "ERROR"}
+
+
+def _unverified(why: str | None) -> bool:
+    return (why or "").lstrip().lower().startswith("unverified:")
+
+
+def _ceiling(members: list[Finding]) -> int:
+    top = max(_RANK.get(m.status, 2) for m in members)
+    if any(_unverified(m.why) for m in members):
+        top = min(top, 1)
+    return top
+
+
+def _clamp_status(proposed: str, members: list[Finding], fallback: str) -> str:
+    cap = _ceiling(members)
+    want = _RANK.get(proposed, cap)
+    return _FROM_RANK[min(want, cap)] if proposed else _FROM_RANK[min(_RANK.get(fallback, cap), cap)]
+
+
+SYSTEM_PROMPT = """\
+Normalize one IC's review findings for Periscope.
+
+You do not invent findings. You do not invent facts. Each original finding
+numbered 1..N is either kept/merged in `findings` (`merged_from`) or removed
+in `dropped` (`index`). Partition is exact: every index once, nowhere twice.
+
+Drop when the finding's own `why` already proves the design meets the spec
+(self-cancelling). Do not keep those as INFO. Keep a finding if it is a real
+issue that is only partially mitigated.
+
+Merge only when one atomic hardware change fixes every member (remove X,
+replace X, rewire X, or add X — one action, no "and"/"also"). Put that action
+in `single_fix`. If you cannot, do not merge.
+
+Severity: grade against the rubric, but only downward from `reviewer_severity`.
+A merge cannot exceed the strongest member. If `why` starts with `Unverified:`,
+keep that prefix and stay at WARNING or below.
+
+Rubric:
+- ERROR — as drawn, the circuit fails (won't regulate, won't meet the output,
+  abs-max strictly exceeded, required pin undriven).
+- WARNING — works with thin margin, load/temp/firmware conditions, missing
+  recommended part, or incomplete evidence (`Unverified:`).
+- INFO — optional feature off by topology, or a note that a netlist cannot
+  prove (layout/docs).
+
+Call `submit_normalized` once.
+"""
+
+
+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 original positions. Length 1 = "
+                                "passthrough; length > 1 = merge."
+                            ),
+                        },
+                        "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. "
+                                "One atomic change that fixes every member."
+                            ),
+                        },
+                        "change_rationale": {
+                            "type": "string",
+                            "description": "One line: unchanged, merged, or re-grade reason.",
+                        },
+                    },
+                    "required": [
+                        "merged_from",
+                        "finding",
+                        "why",
+                        "status",
+                        "recommendation",
+                        "change_rationale",
+                    ],
+                },
+            },
+            "dropped": {
+                "type": "array",
+                "description": "Self-cancelling originals removed from the report.",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "index": {"type": "integer"},
+                        "reason": {"type": "string"},
+                    },
+                    "required": ["index", "reason"],
+                },
+            },
+        },
+        "required": ["findings"],
+    },
+)
+
+
+def _serialize_findings_for_prompt(findings: list[Finding]) -> str:
+    rows = [
+        {
+            "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,
+        }
+        for i, f in enumerate(findings, start=1)
+    ]
+    return json.dumps(rows, indent=2)
+
+
+def _parse_indices(raw: object, n: int, seen: set[int]) -> list[int] | None:
+    if not isinstance(raw, list) or not raw:
+        return None
+    try:
+        indices = [int(x) for x in raw]
+    except (TypeError, ValueError):
+        return None
+    for idx in indices:
+        if idx < 1 or idx > n or idx in seen:
+            return None
+        seen.add(idx)
+    return indices
+
+
+def _build_normalized(
+    raw_findings: list[dict],
+    raw_dropped: list[dict],
+    originals: list[Finding],
+) -> tuple[list[Finding], list[dict]] | None:
+    n = len(originals)
+    seen: set[int] = set()
+    kept: list[Finding] = []
+    dropped_records: list[dict] = []
+
+    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
+        indices = _parse_indices(entry.get("merged_from"), n, seen)
+        if indices is None:
+            return None
+
+        if len(indices) > 1 and not str(entry.get("single_fix") or "").strip():
+            log.warning("normalize: merge %s missing single_fix — un-merge", indices)
+            kept.extend(originals[i - 1] for i in indices)
+            continue
+
+        members = [originals[i - 1] for i in indices]
+        canon = members[0]
+        final_status = _clamp_status(
+            str(entry.get("status") or ""), members, canon.status
+        )
+        new_why = str(entry.get("why") or canon.why)
+        if any(_unverified(m.why) for m in members) and not _unverified(new_why):
+            new_why = "Unverified: " + new_why
+        try:
+            kept.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: Finding rebuild failed")
+            return None
+
+    if seen != set(range(1, n + 1)):
+        return None
+    return kept, 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]:
+    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,
+    }
+    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 — keep originals",
+            ic_ref,
+        )
+        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/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md

@@ -1,6 +1,6 @@
 # Piano — indipendenza architettonica e di licenza da PinScope
 
-**Stato:** split **2.38.0**. Overlay copies **reverted 2.46.0**. **2.47.0** native `graph.py`. **2.48.0** native `models.py` / `parsers.py` / `parsers_edif.py` / `taxonomy.py` (JSON still `dependency/taxonomy` via `repo_paths`). Originals **not** deleted. Fork non staccato. Next: leftover services rewrite.  
+**Stato:** split **2.38.0**. Overlay copies **reverted 2.46.0**. **2.47.0** native `graph.py`. **2.48.0** native `models.py` / `parsers.py` / `parsers_edif.py` / `taxonomy.py`. **2.49.0** native normalize / cross-IC dedupe / `billing_hook`. Originals **not** deleted. Fork non staccato. Next: leftover services (`storage`, `pipeline`, `validation`, …).  
 **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.
 

tests/test_periscope_postpass_rewrite.py

@@ -0,0 +1,90 @@
+"""Post-review services load from src; Emmaforo clamp contracts stay downgrade-only."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from backend.periscopex.models import Finding
+from backend.services.billing_hook import NullBilling, get_billing
+from backend.services.dedupe_findings import _build_deduped
+from backend.services.normalize_findings import _build_normalized
+import backend.services.billing_hook as billing_hook
+import backend.services.dedupe_findings as dedupe_findings
+import backend.services.normalize_findings as normalize_findings
+
+
+def test_postpass_modules_are_src():
+    for mod, name in (
+        (normalize_findings, "normalize_findings.py"),
+        (dedupe_findings, "dedupe_findings.py"),
+        (billing_hook, "billing_hook.py"),
+    ):
+        path = Path(mod.__file__).resolve()
+        assert path.name == name
+        assert "src" in path.parts
+        assert "Native Periscope overlay" not in path.read_text(encoding="utf-8")[:500]
+
+
+def test_self_host_billing_is_null():
+    hook = get_billing()
+    assert isinstance(hook, NullBilling)
+    assert hook.credits_for_api_cost(12.0) == 0.0
+
+
+def test_emmaforo_uart_normalize_cannot_promote_warning():
+    """U2-001 class: hedged UART abs-max must stay WARNING."""
+    originals = [
+        Finding(
+            designator="U2",
+            mpn="CH340E",
+            finding="5V UART into MCU",
+            why="Unverified: abs-max for PA14 not confirmed",
+            status="WARNING",
+            recommendation="level shift",
+            source_page=11,
+            source_quote="",
+            reference="",
+        )
+    ]
+    raw = [{
+        "merged_from": [1],
+        "finding": "CH340E 5V will damage the MCU",
+        "why": "will damage the MCU",
+        "status": "ERROR",
+        "recommendation": "level shift",
+        "change_rationale": "graded ERROR",
+    }]
+    built = _build_normalized(raw, [], originals)
+    assert built is not None
+    kept, _ = built
+    assert kept[0].status == "WARNING"
+    assert kept[0].why.lower().startswith("unverified:")
+
+
+def test_emmaforo_uart_dedupe_collapses_both_endpoints():
+    originals = [
+        Finding(
+            designator="U2", mpn="CH340E", finding="5V out on UART",
+            why="w", status="ERROR", recommendation="", source_page=11,
+            source_quote="q2", reference="r2",
+        ),
+        Finding(
+            designator="U3", mpn="ESP32", finding="PA14 overvoltage",
+            why="w", status="ERROR", recommendation="", source_page=25,
+            source_quote="q3", reference="r3",
+        ),
+    ]
+    groups = [{
+        "member_indices": [1, 2],
+        "primary_index": 2,
+        "finding": "CH340E 5V into MCU UART pin",
+        "why": "same UART interface",
+        "status": "ERROR",
+        "recommendation": "level shift",
+        "change_rationale": "merged 1+2",
+    }]
+    built = _build_deduped(groups, originals)
+    assert built is not None
+    assert len(built) == 1
+    assert built[0].designator == "U3"
+    assert built[0].source_page == 25