diff --git a/periscope/dependency/frontend/content/changelog.md b/periscope/dependency/frontend/content/changelog.md index 3528ec3..178ffc7 100644 --- a/periscope/dependency/frontend/content/changelog.md +++ b/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. diff --git a/periscope/src/backend/services/api_logs.py b/periscope/src/backend/services/api_logs.py new file mode 100644 index 0000000..9c1bff8 --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/billing_hook.py b/periscope/src/backend/services/billing_hook.py new file mode 100644 index 0000000..eaa8237 --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/cost_estimator.py b/periscope/src/backend/services/cost_estimator.py new file mode 100644 index 0000000..756702d --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/datasheet_store.py b/periscope/src/backend/services/datasheet_store.py new file mode 100644 index 0000000..5895dd4 --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/dedupe_findings.py b/periscope/src/backend/services/dedupe_findings.py new file mode 100644 index 0000000..507f642 --- /dev/null +++ b/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: " + "'." + ), + }, + }, + "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 diff --git a/periscope/src/backend/services/digikey.py b/periscope/src/backend/services/digikey.py new file mode 100644 index 0000000..8ebb12b --- /dev/null +++ b/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) diff --git a/periscope/src/backend/services/email.py b/periscope/src/backend/services/email.py new file mode 100644 index 0000000..a483456 --- /dev/null +++ b/periscope/src/backend/services/email.py @@ -0,0 +1,1270 @@ +"""Native Periscope overlay: leftover service still imported from src. + +PinScope original remains in dependency/. +""" + +"""Email notification service using Gmail API with domain-wide delegation.""" + +from __future__ import annotations + +import asyncio +import base64 +import logging +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText + +import httpx + +from backend.config import settings + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Clerk user resolution +# --------------------------------------------------------------------------- + + +async def _resolve_clerk_user(user_id: str) -> dict | None: + """Fetch user profile from Clerk Backend API. Returns None on failure.""" + if not settings.use_auth: + return None + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.get( + f"https://api.clerk.com/v1/users/{user_id}", + headers={"Authorization": f"Bearer {settings.clerk_secret_key}"}, + ) + if resp.status_code == 200: + return resp.json() + except Exception: + logger.warning("Failed to fetch Clerk user %s for email notification", user_id) + return None + + +# --------------------------------------------------------------------------- +# Gmail API +# --------------------------------------------------------------------------- + + +def _build_gmail_service(): + """Build an authenticated Gmail API service using domain-wide delegation. + + On Cloud Run, google.auth.default() returns compute engine credentials + which don't support .with_subject() for domain-wide delegation. We use + the IAM signBlob API to create proper service account credentials that + can impersonate the sender via domain-wide delegation. + + Returns None if credentials cannot be built. + """ + try: + import google.auth + import google.auth.transport.requests + from google.auth import iam + from google.oauth2 import service_account + from googleapiclient.discovery import build + except ImportError: + logger.warning("google-api-python-client not installed; email disabled") + return None + + scopes = ["https://www.googleapis.com/auth/gmail.send"] + + try: + source_credentials, _ = google.auth.default() + logger.debug("Gmail: got default credentials type=%s", type(source_credentials).__name__) + + # Check if these credentials already support with_subject (e.g. key-file) + if hasattr(source_credentials, "_signer"): + logger.debug("Gmail: using service account key-file path (with_subject)") + delegated = source_credentials.with_subject(settings.email_sender) + return build("gmail", "v1", credentials=delegated, cache_discovery=False) + + # Cloud Run path: use IAM signBlob to create credentials that support + # the `subject` claim needed for domain-wide delegation. + logger.debug("Gmail: using IAM signBlob path (Cloud Run / Compute Engine)") + request = google.auth.transport.requests.Request() + source_credentials.refresh(request) + sa_email = source_credentials.service_account_email + logger.debug("Gmail: resolved service account email=%s", sa_email) + + signer = iam.Signer( + request=request, + credentials=source_credentials, + service_account_email=sa_email, + ) + + credentials = service_account.Credentials( + signer=signer, + service_account_email=sa_email, + token_uri="https://oauth2.googleapis.com/token", + scopes=scopes, + subject=settings.email_sender, + ) + + svc = build("gmail", "v1", credentials=credentials, cache_discovery=False) + logger.debug("Gmail: service built successfully, sender=%s", settings.email_sender) + return svc + + except Exception: + logger.warning("Could not obtain credentials for Gmail API", exc_info=True) + return None + + +# --------------------------------------------------------------------------- +# HTML email template +# --------------------------------------------------------------------------- + +_STATUS_COLORS = { + "ERROR": "#ef4444", + "WARNING": "#f59e0b", + "INFO": "#3b82f6", +} + + +def _render_report_email( + recipient_name: str, + project_name: str, + project_id: str, + summary: dict[str, int], + total_cost_usd: float | None, +) -> str: + """Render the HTML email body with inline CSS.""" + report_url = f"{settings.email_frontend_url}/project/{project_id}/report" + + total = summary.get("total", 0) + errors = summary.get("ERROR", 0) + warnings = summary.get("WARNING", 0) + infos = summary.get("INFO", 0) + + # Summary rows + summary_rows = "" + for label, count, color in [ + ("Errors", errors, _STATUS_COLORS["ERROR"]), + ("Warnings", warnings, _STATUS_COLORS["WARNING"]), + ("Info", infos, _STATUS_COLORS["INFO"]), + ]: + if count > 0: + summary_rows += f""" + + + + {count} {label} + + """ + + # Headline color based on worst finding + if errors > 0: + headline_color = _STATUS_COLORS["ERROR"] + headline_text = f"{errors} error{'s' if errors != 1 else ''} found" + elif warnings > 0: + headline_color = _STATUS_COLORS["WARNING"] + headline_text = f"{warnings} warning{'s' if warnings != 1 else ''} found" + elif total == 0: + headline_color = "#10b981" + headline_text = "No issues found" + else: + headline_color = _STATUS_COLORS["INFO"] + headline_text = f"{infos} note{'s' if infos != 1 else ''}" + + return f"""\ + + + + + + +
+ + + + + + + + + + + +
+ + + + + +
+ Periscope + + Report Ready +
+
+ + + + + + + + + + + + + + + +
+ Hi {recipient_name}, +
+ Your validation report for {project_name} is ready. +
+ + +
+ {headline_text} +
+
+ + +
+ + + + +
+ Validation Summary +
+ {total} findings +
+ + {summary_rows} +
+
+
+
+ + + + View Report → + + +
+
+ + +
+ Periscope · Agentic schematic validation +
+
+
+ +""" + + +# --------------------------------------------------------------------------- +# Pipeline-started email template (admin notification) +# --------------------------------------------------------------------------- + + +def _render_pipeline_started_email( + creator_name: str, + creator_email: str, + project_name: str, + project_id: str, + num_components: int, + num_nets: int, + num_ics: int, + num_passives: int, + num_simple: int, +) -> str: + """Render the pipeline-started HTML email for admin notification.""" + project_url = f"{settings.email_frontend_url}/project/{project_id}" + + return f"""\ + + + + + + +
+ + + + + + + + + + + +
+ + + + + +
+ Periscope + + Pipeline Started +
+
+ + + + + + + + + + + + + + +
+ A new pipeline has been triggered for {project_name}. +
+ + +
+ + + + +
+ Created by +
+ {creator_name} +
+ {creator_email} +
+
+
+ + +
+ + + + + +
+ Design Overview +
+ + + + + + +
+ {num_components} + Components + + {num_nets} + Nets +
+
+ + + + + + +
+ + {num_ics} IC{"s" if num_ics != 1 else ""} + + + {num_passives} Passive{"s" if num_passives != 1 else ""} + + + {num_simple} Discrete +
+
+
+
+ + + + View Project → + + +
+
+ + +
+ Periscope · Agentic schematic validation +
+
+
+ +""" + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def _encode_message(msg: MIMEMultipart) -> dict: + """Encode a MIME message as a Gmail API payload.""" + raw = base64.urlsafe_b64encode(msg.as_bytes()).decode("ascii") + return {"raw": raw} + + +async def _send_raw(to_email: str, msg: MIMEMultipart, label: str) -> None: + """Send a MIME message via Gmail API. Logs but never raises.""" + try: + service = _build_gmail_service() + if not service: + logger.warning("Gmail service unavailable; skipping %s", label) + return + await asyncio.to_thread( + service.users().messages().send( + userId="me", body=_encode_message(msg) + ).execute + ) + logger.info("%s sent to %s", label, to_email) + except Exception: + logger.exception("Failed to send %s to %s", label, to_email) + + +def _build_report_message( + to_email: str, + recipient_name: str, + project_name: str, + project_id: str, + summary: dict[str, int], + total_cost_usd: float | None, +) -> MIMEMultipart: + """Build the report-ready email message.""" + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = f"Report ready: {project_name}" + + # Plain text fallback + report_url = f"{settings.email_frontend_url}/project/{project_id}/report" + total = summary.get("total", 0) + errors = summary.get("ERROR", 0) + warnings = summary.get("WARNING", 0) + infos = summary.get("INFO", 0) + text_body = ( + f"Hi {recipient_name},\n\n" + f"Your Periscope validation report for \"{project_name}\" is ready.\n\n" + f"Summary: {total} findings — {errors} errors, {warnings} warnings, {infos} info\n\n" + f"View the report: {report_url}\n" + ) + msg.attach(MIMEText(text_body, "plain")) + + html_body = _render_report_email( + recipient_name, project_name, project_id, + summary, total_cost_usd, + ) + msg.attach(MIMEText(html_body, "html")) + return msg + + +def _build_paused_message( + to_email: str, + recipient_name: str, + project_name: str, + project_id: str, + last_completed: str | None, + stage: str | None, + balance: float, + credits_needed_low: float, +) -> MIMEMultipart: + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = f"Paused: {project_name} is waiting for credits" + + project_url = f"{settings.email_frontend_url}/project/{project_id}" + last_line = f"Last completed: {last_completed}." if last_completed else "" + stage_line = f"Paused during: {stage}." if stage else "" + + text_body = ( + f"Hi {recipient_name},\n\n" + f"Your Periscope run for \"{project_name}\" paused because you're low on credits.\n\n" + f"{last_line}\n{stage_line}\n\n" + f"Current balance: {balance:.2f} credits\n" + f"Credits needed to finish (est): {credits_needed_low:.2f}+\n\n" + f"Top up and resume here: {project_url}\n" + ) + msg.attach(MIMEText(text_body, "plain")) + return msg + + +def _build_topup_failed_message( + to_email: str, recipient_name: str, + amount_usd: float, reason: str, +) -> MIMEMultipart: + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = "Periscope: auto top-up failed" + manage_url = f"{settings.email_frontend_url}/credits" + text_body = ( + f"Hi {recipient_name},\n\n" + f"We tried to auto top-up your Periscope balance with " + f"${amount_usd:.2f} but the charge failed.\n\n" + f"Reason: {reason}\n\n" + f"Auto top-up has been disabled until you update your payment method. " + f"Update your card here: {manage_url}\n" + ) + msg.attach(MIMEText(text_body, "plain")) + return msg + + +async def send_topup_failed_email( + user_id: str, *, amount_usd: float, reason: str, +) -> None: + if not settings.use_email: + return + clerk_user = await _resolve_clerk_user(user_id) + if not clerk_user: + return + emails = clerk_user.get("email_addresses", []) + to_email = emails[0].get("email_address") if emails else None + if not to_email: + return + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + name = f"{first} {last}".strip() or "there" + msg = _build_topup_failed_message(to_email, name, amount_usd, reason) + await _send_raw(to_email, msg, "Top-up-failed email") + + +def _build_low_balance_message( + to_email: str, recipient_name: str, balance: float, threshold: float, +) -> MIMEMultipart: + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = "Periscope: low credit balance" + credits_url = f"{settings.email_frontend_url}/credits" + text_body = ( + f"Hi {recipient_name},\n\n" + f"Your Periscope credit balance has dropped to " + f"{balance:.2f} credits (below your threshold of {threshold:.2f}).\n\n" + f"Top up here so your pipelines don't pause mid-run: {credits_url}\n" + ) + msg.attach(MIMEText(text_body, "plain")) + return msg + + +async def send_low_balance_email( + user_id: str, *, balance: float, threshold: float, +) -> None: + if not settings.use_email: + return + clerk_user = await _resolve_clerk_user(user_id) + if not clerk_user: + return + emails = clerk_user.get("email_addresses", []) + to_email = emails[0].get("email_address") if emails else None + if not to_email: + return + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + name = f"{first} {last}".strip() or "there" + msg = _build_low_balance_message(to_email, name, balance, threshold) + await _send_raw(to_email, msg, "Low-balance email") + + +async def send_pipeline_paused_email( + user_id: str, + project_name: str, + project_id: str, + *, + last_completed: str | None, + stage: str | None, + balance: float, + credits_needed_low: float, +) -> None: + """Send a 'pipeline paused, awaiting credits' email. Fire-and-forget.""" + if not settings.use_email: + return + clerk_user = await _resolve_clerk_user(user_id) + if not clerk_user: + return + emails = clerk_user.get("email_addresses", []) + to_email = emails[0].get("email_address") if emails else None + if not to_email: + return + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + recipient_name = f"{first} {last}".strip() or "there" + + msg = _build_paused_message( + to_email, recipient_name, project_name, project_id, + last_completed, stage, balance, credits_needed_low, + ) + await _send_raw(to_email, msg, "Pipeline-paused email") + + +async def send_report_ready_email( + user_id: str, + project_name: str, + project_id: str, + summary: dict[str, int], + total_cost_usd: float | None = None, +) -> None: + """Send a 'report ready' email to the project creator. Fire-and-forget.""" + if not settings.use_email: + return + + clerk_user = await _resolve_clerk_user(user_id) + if not clerk_user: + logger.warning("Cannot send report email: Clerk user %s not found", user_id) + return + + emails = clerk_user.get("email_addresses", []) + to_email = emails[0].get("email_address") if emails else None + if not to_email: + logger.warning("Cannot send report email: no email for Clerk user %s", user_id) + return + + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + recipient_name = f"{first} {last}".strip() or "there" + + msg = _build_report_message( + to_email, recipient_name, project_name, project_id, + summary, total_cost_usd, + ) + await _send_raw(to_email, msg, "Report-ready email") + + +async def send_test_email(to_email: str) -> dict: + """Send a test email directly to the given address. Returns a status dict.""" + result: dict = {"ok": False, "step": "", "error": ""} + + if not settings.use_email: + result["step"] = "config" + result["error"] = f"use_email=False (email_sender={settings.email_sender!r}, email_frontend_url={settings.email_frontend_url!r})" + return result + + result["step"] = "build_service" + try: + import google.auth + import google.auth.transport.requests + from google.auth import iam + from google.oauth2 import service_account + from googleapiclient.discovery import build + except ImportError as e: + result["error"] = f"Import failed: {e}" + return result + + scopes = ["https://www.googleapis.com/auth/gmail.send"] + try: + source_credentials, _ = google.auth.default() + cred_type = type(source_credentials).__name__ + + if hasattr(source_credentials, "_signer"): + delegated = source_credentials.with_subject(settings.email_sender) + service = build("gmail", "v1", credentials=delegated, cache_discovery=False) + else: + req = google.auth.transport.requests.Request() + source_credentials.refresh(req) + sa_email = source_credentials.service_account_email + signer = iam.Signer(request=req, credentials=source_credentials, service_account_email=sa_email) + credentials = service_account.Credentials( + signer=signer, + service_account_email=sa_email, + token_uri="https://oauth2.googleapis.com/token", + scopes=scopes, + subject=settings.email_sender, + ) + service = build("gmail", "v1", credentials=credentials, cache_discovery=False) + cred_type = f"{cred_type} → IAM signer sa={sa_email}" + + result["step"] = "send" + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = "Periscope email test" + msg.attach(MIMEText(f"Test email from Periscope. Sender: {settings.email_sender}. Creds: {cred_type}", "plain")) + + import asyncio as _asyncio + raw = base64.urlsafe_b64encode(msg.as_bytes()).decode("ascii") + await _asyncio.to_thread( + service.users().messages().send(userId="me", body={"raw": raw}).execute + ) + result["ok"] = True + result["step"] = "sent" + result["error"] = "" + logger.info("Test email sent to %s via %s", to_email, cred_type) + except Exception as exc: + result["error"] = str(exc) + logger.exception("Test email failed at step=%s", result["step"]) + + return result + + +async def send_pipeline_started_email( + user_id: str, + project_name: str, + project_id: str, + num_components: int, + num_nets: int, + num_ics: int, + num_passives: int, + num_simple: int, +) -> None: + """Send a 'pipeline started' email to the admin. Fire-and-forget.""" + if not settings.use_email or not settings.email_admin_notify: + return + + # Resolve creator info from Clerk + creator_name = "Unknown" + creator_email = "unknown" + clerk_user = await _resolve_clerk_user(user_id) + if clerk_user: + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + creator_name = f"{first} {last}".strip() or "Unknown" + emails = clerk_user.get("email_addresses", []) + creator_email = emails[0].get("email_address", "unknown") if emails else "unknown" + + to_email = settings.email_admin_notify + + # Build message + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = f"Pipeline started: {project_name} ({num_components} components)" + + text_body = ( + f"Pipeline started for \"{project_name}\"\n\n" + f"Created by: {creator_name} ({creator_email})\n" + f"Components: {num_components} ({num_ics} ICs, {num_passives} passives, {num_simple} discrete)\n" + f"Nets: {num_nets}\n\n" + f"View project: {settings.email_frontend_url}/project/{project_id}\n" + ) + msg.attach(MIMEText(text_body, "plain")) + + html_body = _render_pipeline_started_email( + creator_name, creator_email, project_name, project_id, + num_components, num_nets, num_ics, num_passives, num_simple, + ) + msg.attach(MIMEText(html_body, "html")) + + await _send_raw(to_email, msg, "Pipeline-started email") + + +# --------------------------------------------------------------------------- +# Feedback received (admin notification) +# --------------------------------------------------------------------------- + + +_FEEDBACK_TYPE_LABELS = { + "bug": "Bug report", + "rule_feedback": "Finding feedback", + "feature_request": "Feature request", +} + +_FEEDBACK_TYPE_COLORS = { + "bug": "#ef4444", + "rule_feedback": "#f59e0b", + "feature_request": "#3b82f6", +} + + +def _esc(s: str | None) -> str: + """Minimal HTML escape so user text can't break the template.""" + if s is None: + return "" + return ( + s.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + ) + + +def _render_feedback_email( + ticket_id: str, + feedback_type: str, + type_label: str, + type_color: str, + submitter_name: str, + submitter_email: str, + project_name: str | None, + project_id: str | None, + finding_designator: str | None, + finding_mpn: str | None, + finding_status: str | None, + finding_text: str | None, + message: str, +) -> str: + admin_url = f"{settings.email_frontend_url}/admin?tab=feedback" + + project_row = "" + if project_name: + project_link = ( + f"{settings.email_frontend_url}/project/{project_id}" + if project_id else "" + ) + project_value = ( + f'{_esc(project_name)}' + if project_link else _esc(project_name) + ) + project_row = f""" + + Project + {project_value} + """ + + finding_rows = "" + if finding_designator or finding_mpn or finding_status: + bits = [] + if finding_designator: + bits.append(f'{_esc(finding_designator)}') + if finding_mpn: + bits.append(f'{_esc(finding_mpn)}') + if finding_status: + bits.append(f'{_esc(finding_status)}') + finding_rows = f""" + + Finding + {' · '.join(bits)} + """ + + finding_text_block = "" + if finding_text: + finding_text_block = f""" + + + +
+ Finding text + {_esc(finding_text)} +
+ """ + + return f"""\ + + + + + + +
+ + + + + + + + + + + +
+ + + + + +
+ Periscope + + Feedback Received +
+
+ + + + + + + + + + + + {finding_text_block} + + + + + + + + + + +
+ + +
+ {type_label} +
+
+ + +
+ + + + +
+ Submitted by +
+ {_esc(submitter_name)} +
+ {_esc(submitter_email)} +
+
+
+ + {project_row} + {finding_rows} +
+
+ + +
+ {_esc(message)} +
+
+ + + + Open in admin → + + +
+ Ticket {_esc(ticket_id)} +
+
+ + +
+ Periscope · Agentic schematic validation +
+
+
+ +""" + + +async def send_feedback_received_email( + ticket_id: str, + user_id: str, + feedback_type: str, + message: str, + *, + submitter_name: str | None = None, + submitter_email: str | None = None, + project_name: str | None = None, + project_id: str | None = None, + finding_designator: str | None = None, + finding_mpn: str | None = None, + finding_status: str | None = None, + finding_text: str | None = None, +) -> None: + """Notify the admin inbox that a new feedback ticket landed. Fire-and-forget.""" + if not settings.use_email or not settings.email_admin_notify: + return + + # Fill in submitter info from Clerk when the client didn't pass it. + name = (submitter_name or "").strip() + email = (submitter_email or "").strip() + if not name or not email: + clerk_user = await _resolve_clerk_user(user_id) + if clerk_user: + if not name: + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + name = f"{first} {last}".strip() + if not email: + emails = clerk_user.get("email_addresses", []) + email = emails[0].get("email_address", "") if emails else "" + name = name or "Unknown user" + email = email or user_id + + type_label = _FEEDBACK_TYPE_LABELS.get(feedback_type, feedback_type) + type_color = _FEEDBACK_TYPE_COLORS.get(feedback_type, "#6b7280") + + to_email = settings.email_admin_notify + subject_ctx = project_name or "general" + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = f"Feedback ({type_label}): {subject_ctx}" + + # Plain text fallback + lines = [ + f"{type_label} from {name} <{email}>", + ] + if project_name: + lines.append(f"Project: {project_name}") + if finding_designator or finding_mpn or finding_status: + finding_bits = " · ".join( + x for x in (finding_designator, finding_mpn, finding_status) if x + ) + lines.append(f"Finding: {finding_bits}") + if finding_text: + lines.append(f"Finding text: {finding_text}") + lines.append("") + lines.append(message) + lines.append("") + lines.append(f"Open in admin: {settings.email_frontend_url}/admin?tab=feedback") + lines.append(f"Ticket: {ticket_id}") + msg.attach(MIMEText("\n".join(lines), "plain")) + + html_body = _render_feedback_email( + ticket_id=ticket_id, + feedback_type=feedback_type, + type_label=type_label, + type_color=type_color, + submitter_name=name, + submitter_email=email, + project_name=project_name, + project_id=project_id, + finding_designator=finding_designator, + finding_mpn=finding_mpn, + finding_status=finding_status, + finding_text=finding_text, + message=message, + ) + msg.attach(MIMEText(html_body, "html")) + + await _send_raw(to_email, msg, "Feedback-received email") + + +# --------------------------------------------------------------------------- +# Feedback reply (notify the original submitter) +# --------------------------------------------------------------------------- + + +def _render_feedback_reply_email( + recipient_first_name: str, + project_name: str | None, + finding_designator: str | None, + finding_mpn: str | None, + original_message: str, + reply_text: str, +) -> str: + feedback_url = f"{settings.email_frontend_url}/feedback" + + context_line = "" + if project_name: + finding_bits = " · ".join( + x for x in (finding_designator, finding_mpn) if x + ) + context_suffix = f" on {_esc(finding_bits)}" if finding_bits else "" + context_line = f""" + + In response to your feedback on {_esc(project_name)}{context_suffix}. + """ + else: + context_line = """ + + In response to the feedback you shared. + """ + + return f"""\ + + + + + + +
+ + + + + + + + + + + +
+ + + + + +
+ Periscope + + New Reply +
+
+ + + + + + + {context_line} + + + + + + + + + + + + + + +
+ Hi {_esc(recipient_first_name)}, +
+ The Periscope team just replied to your feedback. +
+ + +
+ + + +
+ Periscope team +
+ {_esc(reply_text)} +
+
+
+ + +
+ + + +
+ Your original message +
+ {_esc(original_message)} +
+
+
+ + + + View in Periscope → + + +
+ Thank you so much for taking the time to share your feedback — we truly value it. +
+ — The Periscope team +
+
+ + +
+ Periscope · Agentic schematic validation +
+
+
+ +""" + + +async def send_feedback_reply_email( + user_id: str, + reply_text: str, + original_message: str, + *, + recipient_name: str | None = None, + recipient_email: str | None = None, + project_name: str | None = None, + finding_designator: str | None = None, + finding_mpn: str | None = None, +) -> None: + """Notify the original submitter that the Periscope team replied. Fire-and-forget.""" + if not settings.use_email: + return + + full_name = (recipient_name or "").strip() + to_email = (recipient_email or "").strip() + if not full_name or not to_email: + clerk_user = await _resolve_clerk_user(user_id) + if clerk_user: + if not full_name: + first = clerk_user.get("first_name") or "" + last = clerk_user.get("last_name") or "" + full_name = f"{first} {last}".strip() + if not to_email: + emails = clerk_user.get("email_addresses", []) + to_email = emails[0].get("email_address", "") if emails else "" + + if not to_email: + logger.warning( + "Cannot send feedback-reply email: no email for user %s", user_id + ) + return + + first_name = full_name.split()[0] if full_name else "there" + + msg = MIMEMultipart("alternative") + msg["From"] = f"Periscope <{settings.email_sender}>" + msg["To"] = to_email + msg["Subject"] = "The Periscope team replied to your feedback" + + # Plain text fallback + text_lines = [ + f"Hi {first_name},", + "", + "The Periscope team just replied to your feedback.", + "", + "— Reply —", + reply_text, + "", + "— Your original message —", + original_message, + "", + f"View in Periscope: {settings.email_frontend_url}/feedback", + "", + "Thank you so much for taking the time to share your feedback — we truly value it.", + "— The Periscope team", + ] + msg.attach(MIMEText("\n".join(text_lines), "plain")) + + html_body = _render_feedback_reply_email( + recipient_first_name=first_name, + project_name=project_name, + finding_designator=finding_designator, + finding_mpn=finding_mpn, + original_message=original_message, + reply_text=reply_text, + ) + msg.attach(MIMEText(html_body, "html")) + + await _send_raw(to_email, msg, "Feedback-reply email") diff --git a/periscope/src/backend/services/llm/base.py b/periscope/src/backend/services/llm/base.py new file mode 100644 index 0000000..ef09860 --- /dev/null +++ b/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//SKILL.md`` and run + ``validate.py`` locally. Anthropic uses Console Skills when a + skill_id is configured, otherwise the same local path.""" + ... diff --git a/periscope/src/backend/services/llm/factory.py b/periscope/src/backend/services/llm/factory.py new file mode 100644 index 0000000..67b4fb7 --- /dev/null +++ b/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_`` / ``FALLBACK_MODEL_``. + + 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]) diff --git a/periscope/src/backend/services/llm/types.py b/periscope/src/backend/services/llm/types.py new file mode 100644 index 0000000..d8b86f2 --- /dev/null +++ b/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.""" diff --git a/periscope/src/backend/services/normalize_findings.py b/periscope/src/backend/services/normalize_findings.py new file mode 100644 index 0000000..606d354 --- /dev/null +++ b/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 per " + "rubric because , ...)." + ), + }, + }, + "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 diff --git a/periscope/src/backend/services/purple_parts.py b/periscope/src/backend/services/purple_parts.py new file mode 100644 index 0000000..a8b25e9 --- /dev/null +++ b/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 diff --git a/periscope/src/backend/services/storage.py b/periscope/src/backend/services/storage.py new file mode 100644 index 0000000..e3049f3 --- /dev/null +++ b/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}" diff --git a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md b/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md index a0a139b..0546968 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**. 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 | diff --git a/tests/test_native_leftover_services_overlay.py b/tests/test_native_leftover_services_overlay.py new file mode 100644 index 0000000..2d937bf --- /dev/null +++ b/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"}