diff --git a/periscope/dependency/frontend/content/changelog.md b/periscope/dependency/frontend/content/changelog.md index f65eefc..db22c39 100644 --- a/periscope/dependency/frontend/content/changelog.md +++ b/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. diff --git a/periscope/src/backend/services/billing_hook.py b/periscope/src/backend/services/billing_hook.py new file mode 100644 index 0000000..f77d0eb --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/dedupe_findings.py b/periscope/src/backend/services/dedupe_findings.py new file mode 100644 index 0000000..11e8dee --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/normalize_findings.py b/periscope/src/backend/services/normalize_findings.py new file mode 100644 index 0000000..72e3647 --- /dev/null +++ b/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 diff --git a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md b/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md index 6facee9..1bffa42 100644 --- a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md +++ b/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. diff --git a/tests/test_periscope_postpass_rewrite.py b/tests/test_periscope_postpass_rewrite.py new file mode 100644 index 0000000..728e087 --- /dev/null +++ b/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