Rewrite leftover LLM factory (DeepSeek-only), storage, validation.

Anthropic/Gemini names resolve to DeepSeek. Extraction import path re-exports datasheet_extract. Inherited copies stay on disk. storage_gcs unused, skipped.

periscope/src/backend/periscopex/utils.py

@@ -0,0 +1,19 @@
+"""Filename-safe MPNs and natural designator sort (R1, R2, R10)."""
+
+from __future__ import annotations
+
+import re
+
+
+def safe_mpn(mpn: str) -> str:
+    return mpn.replace("/", "_").replace(":", "_")
+
+
+def natural_sort_key(s: str) -> tuple:
+    parts: list[int | str] = []
+    for chunk in re.split(r"(\d+)", s):
+        if chunk.isdigit():
+            parts.append(int(chunk))
+        else:
+            parts.append(chunk.lower())
+    return tuple(parts)

periscope/src/backend/services/admin_settings.py

@@ -0,0 +1,36 @@
+"""Admin settings blob at admin/settings.json (min extraction model version)."""
+
+from __future__ import annotations
+
+from packaging.version import Version
+
+from backend.services.storage import StorageBackend
+
+_SETTINGS_KEY = "admin/settings.json"
+_DEFAULTS = {"min_model_version": "0.0.0"}
+
+
+def get_admin_settings(storage: StorageBackend) -> dict:
+    if storage.exists(_SETTINGS_KEY):
+        return {**_DEFAULTS, **storage.read_json(_SETTINGS_KEY)}
+    return dict(_DEFAULTS)
+
+
+def get_min_model_version(storage: StorageBackend) -> str:
+    return get_admin_settings(storage).get("min_model_version", "0.0.0")
+
+
+def set_min_model_version(storage: StorageBackend, version: str) -> None:
+    Version(version)
+    data = get_admin_settings(storage)
+    data["min_model_version"] = version
+    storage.write_json(_SETTINGS_KEY, data)
+
+
+def version_is_stale(component_version: str, min_version: str) -> bool:
+    if min_version == "0.0.0":
+        return False
+    try:
+        return Version(component_version) < Version(min_version)
+    except Exception:
+        return True

periscope/src/backend/services/api_logs.py

@@ -0,0 +1,92 @@
+"""JSONL LLM call log for a pipeline run. Cost from DeepSeek pricing table."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+from pydantic import BaseModel
+
+from backend.services.llm.pricing import cost_for_entry, total_cost  # noqa: F401
+
+
+class ApiLogEntry(BaseModel):
+    timestamp: str
+    stage: str
+    identifier: str
+    model: str
+    provider: str = "deepseek"
+    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
+    free: bool = False
+
+
+@dataclass
+class CallMeta:
+    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
+
+
+@dataclass
+class ApiLogger:
+    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:
+            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:
+        text = self.to_jsonl()
+        if not text:
+            return
+        from backend.services.projects import project_prefix
+
+        storage.write_text(f"{project_prefix(user_id, project_id)}/api_logs.jsonl", text)
+
+
+def cache_stats_by_stage(entries: list[dict]) -> dict[str, dict]:
+    out: dict[str, dict] = {}
+    for e in entries:
+        stage = str(e.get("stage") or "unknown")
+        bucket = out.setdefault(
+            stage,
+            {"calls": 0, "input_tokens": 0, "cache_read_tokens": 0, "hit_ratio": 0.0},
+        )
+        bucket["calls"] += 1
+        bucket["input_tokens"] += int(e.get("input_tokens") or 0)
+        bucket["cache_read_tokens"] += int(e.get("cache_read_input_tokens") or 0)
+    for bucket in out.values():
+        inp = bucket["input_tokens"]
+        bucket["hit_ratio"] = round(bucket["cache_read_tokens"] / inp, 4) if inp else 0.0
+    return out

periscope/src/backend/services/datasheet_store.py

@@ -0,0 +1,157 @@
+"""Content-addressed datasheet PDFs: blobs by MD5, JSON refs by MPN."""
+
+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"
+
+
+def compute_md5_from_path(local_path: Path) -> str:
+    digest = hashlib.md5()
+    with open(local_path, "rb") as fh:
+        for chunk in iter(lambda: fh.read(8192), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def compute_md5_from_bytes(data: bytes) -> str:
+    return hashlib.md5(data).hexdigest()
+
+
+def blob_key(md5: str) -> str:
+    return f"{BLOB_PREFIX}{md5}.pdf"
+
+
+def ref_key(mpn: str) -> str:
+    return f"{REF_PREFIX}{safe_mpn(mpn)}.json"
+
+
+def store_datasheet(storage: StorageBackend, local_path: Path, mpn: str) -> str:
+    md5 = compute_md5_from_path(local_path)
+    key = blob_key(md5)
+    if not storage.exists(key):
+        storage.upload_from_local(local_path, key)
+    storage.write_json(ref_key(mpn), {"hash": md5, "blob_key": key, "mpn": mpn})
+    return key
+
+
+def store_datasheet_bytes(
+    storage: StorageBackend,
+    data: bytes,
+    mpn: str,
+    extra_mpns: list[str] | None = None,
+) -> str:
+    md5 = compute_md5_from_bytes(data)
+    key = blob_key(md5)
+    if not storage.exists(key):
+        storage.write_bytes(key, data)
+    seen: set[str] = set()
+    for name in [mpn, *(extra_mpns or [])]:
+        name = (name or "").strip()
+        if not name or name.upper() in seen:
+            continue
+        seen.add(name.upper())
+        storage.write_json(ref_key(name), {"hash": md5, "blob_key": key, "mpn": name})
+    _record_aliases(storage, mpn, extra_mpns or [])
+    return key
+
+
+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):
+        table = dict(storage.read_json(ALIAS_KEY).get("aliases") or {})
+    canonical = extra_mpns[0].strip() if extra_mpns else mpn
+    for name, compact_key in compact.items():
+        if len(compact_key) >= _MIN_FAMILY_LEN:
+            table[compact_key] = canonical
+    storage.write_json(ALIAS_KEY, {"aliases": table})
+
+
+def resolve_datasheet(storage: StorageBackend, mpn: str) -> str | None:
+    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
+        bk = storage.read_json(rk).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 alias_key, target in table.items():
+            if alias_key.startswith(want) or (
+                want.startswith(alias_key) and len(alias_key) >= _MIN_FAMILY_LEN
+            ):
+                hit = from_ref(target)
+                if hit:
+                    return hit
+    return None
+
+
+def delete_datasheet_ref(storage: StorageBackend, mpn: str) -> str | None:
+    rk = ref_key(mpn)
+    if not storage.exists(rk):
+        return None
+    bk = storage.read_json(rk).get("blob_key")
+    storage.delete_key(rk)
+    return bk
+
+
+def gc_orphan_blobs(storage: StorageBackend, *, dry_run: bool = True) -> list[str]:
+    referenced: set[str] = set()
+    for rk in storage.list_recursive(REF_PREFIX):
+        if rk.endswith(".json"):
+            h = storage.read_json(rk).get("hash")
+            if h:
+                referenced.add(h)
+    for pk in storage.list_recursive("library/patterns/"):
+        if not pk.endswith(".json"):
+            continue
+        ds_key = storage.read_json(pk).get("datasheet_key", "")
+        if ds_key.startswith(BLOB_PREFIX) and ds_key.endswith(".pdf"):
+            referenced.add(ds_key.removeprefix(BLOB_PREFIX).removesuffix(".pdf"))
+    orphans: list[str] = []
+    for bk in storage.list_recursive(BLOB_PREFIX):
+        if not bk.endswith(".pdf"):
+            continue
+        h = bk.rsplit("/", 1)[-1].removesuffix(".pdf")
+        if h in referenced:
+            continue
+        orphans.append(bk)
+        if not dry_run:
+            storage.delete_key(bk)
+    return orphans

periscope/src/backend/services/extraction.py

@@ -0,0 +1,27 @@
+"""Datasheet extraction import path. Implementation lives in datasheet_extract."""
+
+from __future__ import annotations
+
+from backend.services.datasheet_extract import (
+    CatalogResolveMiss,
+    auto_resolve_specs,
+    extract_pattern,
+    extract_pintable,
+    extract_specs,
+    resolve_from_value,
+    _coerce_abs_max,
+    _coerce_internal_features,
+    _coerce_layout_rules,
+)
+
+__all__ = [
+    "CatalogResolveMiss",
+    "auto_resolve_specs",
+    "extract_pattern",
+    "extract_pintable",
+    "extract_specs",
+    "resolve_from_value",
+    "_coerce_abs_max",
+    "_coerce_internal_features",
+    "_coerce_layout_rules",
+]

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

@@ -1,4 +1,4 @@
-"""Native LLM pieces (DeepSeek, local skills, PDF ingest) merged with PinScope providers."""
+"""Periscope LLM facade: DeepSeek provider, local skills, shared types."""
 from pkgutil import extend_path
 
 __path__ = extend_path(__path__, __name__)

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

@@ -0,0 +1,55 @@
+"""LLMProvider / LLMSession contracts used by DeepSeek and local skills."""
+
+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):
+    """One multi-turn conversation. Always close() in a finally block."""
+
+    provider_name: str
+    model: str
+
+    @abstractmethod
+    async def complete(
+        self,
+        *,
+        messages: list[Message],
+        tools: list[ToolSchema] | None = None,
+        tool_choice: ToolChoice = "auto",
+    ) -> Completion:
+        ...
+
+    @abstractmethod
+    async def close(self) -> None:
+        ...
+
+
+class LLMProvider(Protocol):
+    name: str
+
+    async def create_session(
+        self,
+        *,
+        model: str,
+        system: str,
+        max_tokens: int = 4096,
+        temperature: float | None = None,
+    ) -> LLMSession:
+        ...
+
+    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]:
+        ...

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

@@ -0,0 +1,54 @@
+"""Stage routing: DeepSeek only. Anthropic/Gemini names still resolve to DeepSeek."""
+
+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:
+    if name not in ("deepseek", "anthropic", "gemini"):
+        raise ValueError(f"Unknown LLM provider: {name!r}")
+    if name != "deepseek":
+        log.warning("%s routing is disabled — using DeepSeek", name)
+    from backend.services.llm.deepseek_provider import DeepSeekProvider
+
+    return DeepSeekProvider()
+
+
+_get_provider_by_name = get_provider_by_name
+
+
+def get_provider(stage: str) -> LLMProvider:
+    return get_provider_by_name(settings.provider_for_stage(stage))
+
+
+async def call_with_fallback(
+    stage: str,
+    body: Callable[[LLMProvider, str], Awaitable[T]],
+) -> T:
+    primary = get_provider(stage)
+    model = settings.model_for_stage(stage)
+    try:
+        return await body(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) — fallback %s/%s",
+            stage, primary.name, model, exc, fb[0], fb[1],
+        )
+        return await body(get_provider_by_name(fb[0]), fb[1])

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

@@ -0,0 +1,46 @@
+"""DeepSeek USD rates; leftover log rows may still name other vendors."""
+
+from __future__ import annotations
+
+_FLASH = {"input": 0.30, "output": 1.20}
+_PRO = {"input": 1.32, "output": 3.96}
+
+PRICING: dict[str, dict[str, dict[str, float]]] = {
+    "deepseek": {
+        "deepseek-flash": _FLASH,
+        "deepseek-v4-flash": _FLASH,
+        "deepseek-v4-flash-vision-exp": _FLASH,
+        "deepseek-v4-pro": _PRO,
+        "default": _FLASH,
+    },
+    "anthropic": {
+        "default": {"input": 3.00, "output": 15.00},
+    },
+    "gemini": {
+        "default": {"input": 0.30, "output": 2.50},
+    },
+}
+
+CACHE_RATES: dict[str, dict[str, float]] = {
+    "deepseek": {"create": 1.00, "read": 0.02},
+    "anthropic": {"create": 1.25, "read": 0.10},
+    "gemini": {"create": 1.00, "read": 0.25},
+}
+
+
+def cost_for_entry(entry: dict) -> float:
+    provider = entry.get("provider") or "deepseek"
+    table = PRICING.get(provider) or PRICING["deepseek"]
+    rates = table.get(entry.get("model", ""), table["default"])
+    cache = CACHE_RATES.get(provider, CACHE_RATES["deepseek"])
+    inp = rates["input"]
+    return (
+        entry.get("input_tokens", 0) * inp
+        + entry.get("cache_creation_input_tokens", 0) * inp * cache["create"]
+        + entry.get("cache_read_input_tokens", 0) * inp * cache["read"]
+        + entry.get("output_tokens", 0) * rates["output"]
+    ) / 1_000_000
+
+
+def total_cost(entries: list[dict]) -> float:
+    return round(sum(cost_for_entry(e) for e in entries), 6)

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

@@ -0,0 +1,78 @@
+"""Shared message / tool / usage types for Periscope LLM sessions.
+
+DeepSeek is the live provider. Fields that other SDKs used to round-trip
+(thought signatures) stay optional so history blobs remain loadable.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Literal
+
+
+@dataclass
+class TextBlock:
+    text: str
+    cacheable: bool = False
+    thought_signature: bytes | None = None
+    reasoning_content: str | None = None
+
+
+@dataclass
+class PdfBlock:
+    """PDF on disk. DeepSeek ingest turns this into text (and page images)."""
+    path: Path
+    cacheable: bool = False
+
+
+@dataclass
+class ToolCall:
+    id: str
+    name: str
+    input: dict[str, Any]
+    thought_signature: bytes | None = None
+    reasoning_content: str | None = None
+
+
+@dataclass
+class ToolResultBlock:
+    tool_use_id: str
+    name: str
+    content: str
+
+
+ContentBlock = TextBlock | PdfBlock | ToolCall | ToolResultBlock
+
+
+@dataclass
+class Message:
+    role: Literal["user", "assistant"]
+    content: list[ContentBlock]
+
+
+@dataclass
+class ToolSchema:
+    name: str
+    description: str
+    input_schema: dict[str, Any]
+
+
+ToolChoice = Literal["auto", "none"] | dict
+
+
+@dataclass
+class Usage:
+    input_tokens: int = 0
+    output_tokens: int = 0
+    cache_creation_tokens: int = 0
+    cache_read_tokens: int = 0
+
+
+@dataclass
+class Completion:
+    text: str
+    tool_calls: list[ToolCall]
+    usage: Usage
+    stop_reason: str
+    raw_assistant_blocks: list[ContentBlock] = field(default_factory=list)

periscope/src/backend/services/storage.py

@@ -0,0 +1,154 @@
+"""Local key/value storage (GCS-shaped keys). GCS backend stays in dependency."""
+
+from __future__ import annotations
+
+import json
+import shutil
+import threading
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+GENERATION_NEW = 0
+
+
+class StaleGeneration(Exception):
+    """Conditional write lost the generation race."""
+
+
+@runtime_checkable
+class StorageBackend(Protocol):
+    def read_json(self, key: str) -> dict: ...
+    def write_json(self, key: str, data: dict) -> None: ...
+    def read_bytes(self, key: str) -> bytes: ...
+    def write_bytes(self, key: str, data: bytes) -> None: ...
+    def read_text(self, key: str) -> str: ...
+    def write_text(self, key: str, text: str) -> None: ...
+    def exists(self, key: str) -> bool: ...
+    def list_prefix(self, prefix: str) -> list[str]: ...
+    def list_recursive(self, prefix: str) -> list[str]: ...
+    def list_prefix_after(self, prefix: str, after_key: str | None = None) -> list[str]: ...
+    def read_json_with_generation(self, key: str) -> tuple[dict, int]: ...
+    def write_json_if_match(self, key: str, data: dict, generation: int) -> int: ...
+    def delete_key(self, key: str) -> None: ...
+    def delete_prefix(self, prefix: str) -> None: ...
+    def copy_object(self, src_key: str, dst_key: str) -> None: ...
+    def download_to_local(self, key: str, local_path: Path) -> Path: ...
+    def upload_from_local(self, local_path: Path, key: str) -> None: ...
+    def signed_url(self, key: str, expiration_minutes: int = 15) -> str: ...
+
+
+class LocalStorageBackend:
+    def __init__(self, base_dir: Path) -> None:
+        self._base = base_dir
+        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 []
+        return [str(child.relative_to(self._base)) for child in sorted(d.iterdir())]
+
+    def list_recursive(self, prefix: str) -> list[str]:
+        d = self._path(prefix)
+        if not d.is_dir():
+            return []
+        return [
+            str(child.relative_to(self._base))
+            for child in sorted(d.rglob("*"))
+            if child.is_file()
+        ]
+
+    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:
+        dst = self._path(dst_key)
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(self._path(src_key), dst)
+
+    def download_to_local(self, key: str, local_path: Path) -> Path:
+        local_path.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(self._path(key), 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:
+        return f"/api/datasheets/_local/{key}"

periscope/src/backend/services/survey.py

@@ -0,0 +1,83 @@
+"""Optional onboarding survey. Sheets append is best-effort; completion is local."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import datetime, timezone
+
+from backend.config import settings
+from backend.services.storage import StorageBackend
+
+logger = logging.getLogger(__name__)
+
+_SURVEY_PREFIX = "admin/survey/"
+
+
+def _status_key(user_id: str) -> str:
+    return f"{_SURVEY_PREFIX}{user_id}.json"
+
+
+def is_completed(storage: StorageBackend, user_id: str) -> bool:
+    return storage.exists(_status_key(user_id))
+
+
+def _mark_completed(storage: StorageBackend, user_id: str) -> None:
+    storage.write_json(
+        _status_key(user_id),
+        {"completed": True, "timestamp": datetime.now(timezone.utc).isoformat()},
+    )
+
+
+def _build_sheets_service():
+    try:
+        import google.auth
+        from googleapiclient.discovery import build
+    except ImportError:
+        logger.warning("Sheets client not installed; survey append disabled")
+        return None
+    try:
+        credentials, _project = google.auth.default()
+        if hasattr(credentials, "with_scopes"):
+            credentials = credentials.with_scopes(
+                ["https://www.googleapis.com/auth/spreadsheets"]
+            )
+        return build("sheets", "v4", credentials=credentials, cache_discovery=False)
+    except Exception:
+        logger.exception("Sheets service unavailable")
+        return None
+
+
+async def append_to_sheet(
+    user_id: str,
+    email: str,
+    name: str,
+    referral_source: str,
+    user_profile: str,
+) -> bool:
+    sheet_id = settings.survey_sheet_id
+    if not sheet_id:
+        logger.warning("SURVEY_SHEET_ID unset; skip append for %s", user_id)
+        return False
+    service = _build_sheets_service()
+    if not service:
+        return False
+    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
+    row = [timestamp, user_id, email, name, referral_source, user_profile]
+    try:
+        await asyncio.to_thread(
+            service.spreadsheets()
+            .values()
+            .append(
+                spreadsheetId=sheet_id,
+                range="Sheet1!A:F",
+                valueInputOption="RAW",
+                insertDataOption="INSERT_ROWS",
+                body={"values": [row]},
+            )
+            .execute
+        )
+        return True
+    except Exception:
+        logger.exception("survey append failed for %s", user_id)
+        return False

periscope/src/backend/services/validation.py

@@ -0,0 +1,437 @@
+"""Per-IC datasheet review orchestration (DeepSeek via review_session).
+
+Deterministic graph checks run first. Each IC is reviewed with its PDF.
+Fail-soft per IC. Cross-IC dedupe is downgrade-only and optional.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Awaitable, Callable
+
+from backend.config import settings
+from backend.periscopex.bom_match_check import check_bom_schematic_match
+from backend.periscopex.cad_bridge import annotate_findings_cad, build_cad_bridge, write_cad_bridge
+from backend.periscopex.constraints_lookup import build_constraints_map, load_datasheets
+from backend.periscopex.crystal_cl_check import check_crystal_cl
+from backend.periscopex.dnp_check import check_dnp_enables
+from backend.periscopex.errata_check import check_errata
+from backend.periscopex.filter_check import check_filters
+from backend.periscopex.finding_engine import apply_decisions
+from backend.periscopex.hf_coverage_check import check_hf_decoupling_coverage
+from backend.periscopex.internal_features_check import check_internal_features
+from backend.periscopex.led_current_check import check_led_current
+from backend.periscopex.lifecycle import check_lifecycle, load_lifecycle_dir
+from backend.periscopex.models import ComponentType, DesignGraph, Finding, ValidationReport
+from backend.periscopex.nc_pin_check import check_nc_pins
+from backend.periscopex.passive_rail_check import (
+    check_i2c_pullups,
+    check_reset_pullups,
+    check_supply_decoupling,
+)
+from backend.periscopex.pin_mux_check import check_pin_mux_feasibility
+from backend.periscopex.power_margin_check import check_power_margin
+from backend.periscopex.review_parse import ReviewResult, assign_finding_ids
+from backend.periscopex.sequencing_check import check_power_sequencing
+from backend.periscopex.thermal_check import check_thermal
+from backend.periscopex.utils import safe_mpn
+from backend.services.api_logs import ApiLogger
+from backend.services.dedupe_findings import dedupe_cross_ic_findings_async
+from backend.services.llm.factory import call_with_fallback  # tests monkeypatch this name
+from backend.services.review_session import TRACE_VERSION, review_ic_async
+
+log = logging.getLogger(__name__)
+
+ProgressCallback = Callable[[str, int, str, str], Awaitable[None]]
+BeforeIcCallback = Callable[[str], Awaitable[bool]]
+OnIcDoneCallback = Callable[[str, ReviewResult, ApiLogger | None], Awaitable[None]]
+OnIcErrorCallback = Callable[[str, BaseException], Awaitable[None]]
+OnDedupeDoneCallback = Callable[[ApiLogger | None], Awaitable[None]]
+
+
+def _is_deterministic(f: Finding) -> bool:
+    return bool(getattr(f, "source", None)) and f.source != "review"
+
+
+def _run_deterministic_checks(
+    graph: DesignGraph,
+    constraints_map: dict,
+    lifecycle_map: dict | None = None,
+) -> list[Finding]:
+    out: list[Finding] = []
+    for name, fn in (
+        ("pin_mux_check", lambda: check_pin_mux_feasibility(graph, constraints_map)),
+        ("led_current_check", lambda: check_led_current(graph)),
+        ("supply_decoupling_check", lambda: check_supply_decoupling(graph, constraints_map)),
+        ("i2c_pullup_check", lambda: check_i2c_pullups(graph, constraints_map)),
+        ("reset_pullup_check", lambda: check_reset_pullups(graph, constraints_map)),
+        ("bom_match_check", lambda: check_bom_schematic_match(
+            graph.schematic_fields, graph.bom_fields,
+        )),
+        ("hf_coverage_check", lambda: check_hf_decoupling_coverage(graph, constraints_map)),
+        ("filter_check", lambda: check_filters(graph, constraints_map)),
+        ("thermal_check", lambda: check_thermal(graph, constraints_map)),
+        ("power_margin_check", lambda: check_power_margin(graph, constraints_map)),
+        ("sequencing_check", lambda: check_power_sequencing(graph, constraints_map)),
+        ("dnp_check", lambda: check_dnp_enables(graph, constraints_map)),
+        ("lifecycle_check", lambda: check_lifecycle(graph, lifecycle_map)),
+        ("errata_check", lambda: check_errata(graph, constraints_map)),
+        ("internal_features_check", lambda: check_internal_features(graph, constraints_map)),
+        ("crystal_cl_check", lambda: check_crystal_cl(graph)),
+        ("nc_pin_check", lambda: check_nc_pins(graph, constraints_map)),
+    ):
+        try:
+            out.extend(fn())
+        except Exception:
+            log.exception("deterministic check %s failed — skipping", name)
+    return out
+
+
+def _find_pdf(mpn: str, pdf_dir: Path, storage=None) -> Path | None:
+    from backend.services.datasheet_finder import find_local_pdf
+
+    mpn = (mpn or "").strip()
+    if not mpn:
+        return None
+    local = find_local_pdf(pdf_dir, mpn)
+    if local is not None and local.is_file():
+        wanted = pdf_dir / f"{safe_mpn(mpn)}.pdf"
+        if local.resolve() != wanted.resolve() and not wanted.is_file():
+            wanted.write_bytes(local.read_bytes())
+            return wanted
+        return local
+    if storage:
+        from backend.services import projects as proj_svc
+
+        lib_key = proj_svc.library_has_datasheet(storage, mpn)
+        if lib_key:
+            wanted = pdf_dir / f"{safe_mpn(mpn)}.pdf"
+            storage.download_to_local(lib_key, wanted)
+            if wanted.is_file():
+                return wanted
+    return None
+
+
+async def validate_design_async(
+    graph_path: str,
+    output_path: str,
+    datasheets_dir: str = "datasheets/extracted",
+    pdf_dir: str = "uploads/datasheets",
+    on_progress: ProgressCallback | None = None,
+    api_logger: ApiLogger | None = None,
+    storage=None,
+    skip_refs: set[str] | None = None,
+    before_ic: BeforeIcCallback | None = None,
+    on_ic_done: OnIcDoneCallback | None = None,
+    on_ic_error: OnIcErrorCallback | None = None,
+    on_dedupe_done: OnDedupeDoneCallback | None = None,
+    project_prefix: str | None = None,
+    run_meta: dict | None = None,
+) -> ValidationReport:
+    skip_refs = skip_refs or set()
+    graph = DesignGraph.model_validate(json.loads(Path(graph_path).read_text()))
+    datasheets = load_datasheets(datasheets_dir)
+    constraints_map = build_constraints_map(datasheets)
+    lifecycle_map: dict = {}
+    for cand in (Path(datasheets_dir).parent / "lifecycle", Path(datasheets_dir) / "lifecycle"):
+        loaded = load_lifecycle_dir(cand)
+        if loaded:
+            lifecycle_map.update(loaded)
+    deterministic_findings = _run_deterministic_checks(graph, constraints_map, lifecycle_map)
+
+    pdf_dir_path = Path(pdf_dir)
+    ic_tasks: list[tuple[str, str]] = []
+    not_reviewed: list[dict] = []
+    for ref, comp in sorted(graph.components.items()):
+        if comp.component_type != ComponentType.IC:
+            continue
+        mpn = (comp.mpn or "").strip() or (comp.value or "").strip()
+        if not mpn:
+            not_reviewed.append({"designator": ref, "reason": "no MPN in BOM"})
+            if on_progress:
+                await on_progress(ref, 0, "skipped", "no MPN in BOM")
+            continue
+        pdf = _find_pdf(mpn, pdf_dir_path, storage=storage)
+        if pdf:
+            ic_tasks.append((ref, str(pdf)))
+        else:
+            not_reviewed.append({"designator": ref, "reason": "no datasheet PDF"})
+            if on_progress:
+                await on_progress(ref, 0, "skipped", "no datasheet PDF")
+
+    existing_path = Path(output_path)
+    preserved_findings: list[Finding] = []
+    preserved_coverage: dict[str, list[str]] = {}
+    preserved_comments = None
+    preserved_review_states = None
+    if existing_path.is_file():
+        try:
+            existing = json.loads(existing_path.read_text())
+            preserved_comments = existing.get("comments")
+            preserved_review_states = existing.get("review_states")
+            if before_ic is not None:
+                for f in existing.get("findings", []):
+                    ref = f.get("component_ref") or f.get("designator") or ""
+                    if ref in skip_refs:
+                        preserved_findings.append(Finding.model_validate(f))
+                for ref, areas in (existing.get("coverage") or {}).items():
+                    if ref in skip_refs:
+                        preserved_coverage[ref] = list(areas)
+        except (json.JSONDecodeError, OSError):
+            pass
+
+    preserved_review = [f for f in preserved_findings if not _is_deterministic(f)]
+    all_findings: list[Finding] = list(preserved_review) + list(deterministic_findings)
+    all_coverage: dict[str, list[str]] = dict(preserved_coverage)
+    review_errors: dict[str, str] = {}
+
+    def _sanitize_coverage(src: dict[str, list[str]]) -> dict[str, list[str]]:
+        clean: dict[str, list[str]] = {}
+        for ref, areas in src.items():
+            if isinstance(areas, list) and all(isinstance(a, str) for a in areas):
+                clean[ref] = areas
+        return clean
+
+    def _write_report(paused: bool = False) -> ValidationReport:
+        annotate_findings_cad(all_findings, graph.cad_index)
+        assign_finding_ids(all_findings)
+        dec_path = existing_path.with_name("decisions.json")
+        if dec_path.is_file():
+            try:
+                apply_decisions(all_findings, json.loads(dec_path.read_text()))
+            except Exception:
+                log.exception("decisions.json apply failed")
+        summary = {"total": len(all_findings), "ERROR": 0, "WARNING": 0, "INFO": 0}
+        for f in all_findings:
+            summary[f.status] = summary.get(f.status, 0) + 1
+        try:
+            report = ValidationReport(
+                project=Path(graph_path).stem,
+                timestamp=datetime.now(timezone.utc).isoformat(),
+                findings=all_findings,
+                summary=summary,
+                coverage=_sanitize_coverage(all_coverage),
+                review_errors=dict(review_errors),
+                not_reviewed=not_reviewed,
+            )
+        except Exception:
+            report = ValidationReport(
+                project=Path(graph_path).stem,
+                timestamp=datetime.now(timezone.utc).isoformat(),
+                findings=all_findings,
+                summary=summary,
+                coverage={},
+                review_errors=dict(review_errors),
+                not_reviewed=not_reviewed,
+            )
+        report_dict = json.loads(report.model_dump_json(indent=2))
+        if preserved_comments is not None:
+            report_dict["comments"] = preserved_comments
+        if preserved_review_states is not None:
+            report_dict["review_states"] = preserved_review_states
+        if paused:
+            report_dict["partial"] = True
+        existing_path.write_text(json.dumps(report_dict, indent=2))
+        try:
+            prefix_id = (project_prefix or "").rstrip("/").rsplit("/", 1)[-1]
+            bridge = build_cad_bridge(report, prefix_id or report.project)
+            write_cad_bridge(existing_path.with_name("periscope-findings.json"), bridge)
+        except Exception:
+            log.exception("cad bridge write failed")
+        return report
+
+    git_commit = (run_meta or {}).get("git_commit", "unknown")
+
+    def _write_trace(trace: dict, ref: str) -> None:
+        if not storage or not project_prefix or not trace:
+            return
+        try:
+            storage.write_json(f"{project_prefix}/review_traces/{safe_mpn(ref)}.json", trace)
+        except Exception:
+            log.exception("trace: write failed for %s", ref)
+
+    async def _maybe_dedupe_cross_ic() -> None:
+        if not settings.cross_ic_dedup_enabled:
+            return
+        review = [f for f in all_findings if not _is_deterministic(f)]
+        deterministic = [f for f in all_findings if _is_deterministic(f)]
+        if len({f.designator for f in review}) < 2:
+            return
+        private = (
+            ApiLogger(free=api_logger.free)
+            if (api_logger is not None and on_dedupe_done is not None)
+            else None
+        )
+        try:
+            deduped, dedupe_trace = await dedupe_cross_ic_findings_async(
+                review,
+                api_logger=private if private is not None else api_logger,
+                on_progress=on_progress,
+            )
+        except Exception:
+            log.exception("cross-IC dedupe failed — keeping per-IC findings")
+            return
+        all_findings[:] = deduped + deterministic
+        if storage and project_prefix and dedupe_trace:
+            try:
+                storage.write_json(
+                    f"{project_prefix}/review_traces/_cross_ic_dedupe.json",
+                    dedupe_trace,
+                )
+            except Exception:
+                log.exception("trace: cross-IC dedupe write failed")
+        if private is not None and on_dedupe_done is not None:
+            try:
+                await on_dedupe_done(private)
+            except Exception:
+                log.exception("on_dedupe_done callback failed")
+
+    def _stub_trace(ref: str, error: str) -> dict:
+        try:
+            comp = graph.components.get(ref)
+            mpn = (comp.mpn or comp.value) if comp else ref
+        except Exception:
+            mpn = ref
+        return {
+            "trace_version": TRACE_VERSION,
+            "ic_ref": ref,
+            "mpn": mpn,
+            "git_commit": git_commit,
+            "datasheet": {"md5": None, "safe_mpn": safe_mpn(str(mpn))},
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "turns": [],
+            "final_submission": None,
+            "result": None,
+            "stop_reason": "error",
+            "error": error,
+            "duration_ms": None,
+        }
+
+    excerpt_cache: dict = {}
+
+    def _cleanup_excerpt_cache() -> None:
+        for entry in excerpt_cache.values():
+            try:
+                if isinstance(entry, tuple) and len(entry) == 2:
+                    Path(entry[0]).unlink(missing_ok=True)
+            except Exception:
+                pass
+
+    if before_ic is None:
+        sem = asyncio.Semaphore(settings.ic_concurrency)
+
+        async def _review_one(ref: str, pdf_path: str) -> tuple[ReviewResult, dict]:
+            async with sem:
+                return await review_ic_async(
+                    graph, constraints_map, ref, pdf_path,
+                    on_progress=on_progress, api_logger=api_logger,
+                    trace_git_commit=git_commit,
+                    pdf_dir=pdf_dir_path, storage=storage,
+                    excerpt_cache=excerpt_cache,
+                )
+
+        results = await asyncio.gather(
+            *(_review_one(ref, pdf) for ref, pdf in ic_tasks if ref not in skip_refs),
+            return_exceptions=True,
+        )
+        remaining_tasks = [t for t in ic_tasks if t[0] not in skip_refs]
+        for i, result in enumerate(results):
+            ref = remaining_tasks[i][0]
+            if isinstance(result, BaseException):
+                msg = f"{type(result).__name__}: {result}"
+                log.exception("Review failed for %s", ref, exc_info=result)
+                review_errors[ref] = msg
+                _write_trace(_stub_trace(ref, msg), ref)
+                if on_progress:
+                    await on_progress(ref, 0, "error", msg)
+                if on_ic_error is not None:
+                    try:
+                        await on_ic_error(ref, result)
+                    except Exception:
+                        log.exception("on_ic_error callback failed for %s", ref)
+            elif isinstance(result, tuple):
+                rr, trace = result
+                _write_trace(trace, ref)
+                all_findings.extend(rr.findings)
+                if rr.checked_areas:
+                    all_coverage[ref] = rr.checked_areas
+        await _maybe_dedupe_cross_ic()
+        try:
+            return _write_report(paused=False)
+        finally:
+            _cleanup_excerpt_cache()
+
+    sem = asyncio.Semaphore(settings.ic_concurrency)
+    stop = False
+
+    async def _gated_review_one(ref: str, pdf_path: str) -> None:
+        nonlocal stop
+        async with sem:
+            if stop:
+                return
+            try:
+                ok = await before_ic(ref)
+            except Exception:
+                ok = True
+            if not ok:
+                stop = True
+                return
+            private = ApiLogger(free=api_logger.free) if api_logger is not None else None
+            try:
+                result, trace = await review_ic_async(
+                    graph, constraints_map, ref, pdf_path,
+                    on_progress=on_progress, api_logger=private,
+                    trace_git_commit=git_commit,
+                    pdf_dir=pdf_dir_path, storage=storage,
+                    excerpt_cache=excerpt_cache,
+                )
+            except Exception as exc:
+                msg = f"{type(exc).__name__}: {exc}"
+                log.exception("Review failed for %s", ref)
+                review_errors[ref] = msg
+                _write_trace(_stub_trace(ref, msg), ref)
+                if on_progress:
+                    await on_progress(ref, 0, "error", msg)
+                if on_ic_error is not None:
+                    try:
+                        await on_ic_error(ref, exc)
+                    except Exception:
+                        log.exception("on_ic_error callback failed for %s", ref)
+                try:
+                    _write_report(paused=False)
+                except Exception:
+                    log.exception("incremental report write failed after error on %s", ref)
+                return
+            all_findings.extend(result.findings)
+            if result.checked_areas:
+                all_coverage[ref] = result.checked_areas
+            try:
+                _write_report(paused=False)
+            except Exception as exc:
+                all_coverage.pop(ref, None)
+                if on_progress:
+                    await on_progress(ref, 0, "warning", f"report write failed: {exc}")
+            _write_trace(trace, ref)
+            if on_ic_done is not None:
+                try:
+                    await on_ic_done(ref, result, private)
+                except Exception:
+                    log.exception("on_ic_done callback failed for %s", ref)
+
+    results = await asyncio.gather(
+        *(_gated_review_one(ref, pdf) for ref, pdf in ic_tasks if ref not in skip_refs),
+        return_exceptions=True,
+    )
+    for r in results:
+        if isinstance(r, asyncio.CancelledError):
+            raise r
+    if not stop:
+        await _maybe_dedupe_cross_ic()
+    try:
+        return _write_report(paused=bool(stop))
+    finally:
+        _cleanup_excerpt_cache()

tests/test_periscope_services_core_rewrite.py

@@ -0,0 +1,75 @@
+"""Leftover service cores resolve from src; LLM factory stays on DeepSeek."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import backend.periscopex.utils as utils
+import backend.services.admin_settings as admin_settings
+import backend.services.api_logs as api_logs
+import backend.services.datasheet_store as datasheet_store
+import backend.services.llm.factory as factory
+import backend.services.storage as storage
+from backend.periscopex.utils import natural_sort_key, safe_mpn
+from backend.services.admin_settings import version_is_stale
+from backend.services.datasheet_store import (
+    compute_md5_from_bytes,
+    resolve_datasheet,
+    store_datasheet_bytes,
+)
+from backend.services.llm.factory import get_provider_by_name
+from backend.services.storage import GENERATION_NEW, LocalStorageBackend, StaleGeneration
+
+
+def _assert_src(mod, name: str) -> None:
+    path = Path(mod.__file__).resolve()
+    assert path.name == name
+    assert "src" in path.parts
+    assert "Native Periscope overlay" not in path.read_text(encoding="utf-8")[:400]
+
+
+def test_core_service_modules_are_src():
+    for mod, name in (
+        (factory, "factory.py"),
+        (storage, "storage.py"),
+        (api_logs, "api_logs.py"),
+        (datasheet_store, "datasheet_store.py"),
+        (admin_settings, "admin_settings.py"),
+        (utils, "utils.py"),
+    ):
+        _assert_src(mod, name)
+
+
+def test_factory_routes_anthropic_and_gemini_names_to_deepseek(monkeypatch):
+    from backend.config import settings
+
+    monkeypatch.setattr(settings, "deepseek_api_key", "sk-test")
+    get_provider_by_name.cache_clear()
+    try:
+        assert get_provider_by_name("deepseek").name == "deepseek"
+        assert get_provider_by_name("anthropic").name == "deepseek"
+        assert get_provider_by_name("gemini").name == "deepseek"
+    finally:
+        get_provider_by_name.cache_clear()
+
+
+def test_safe_mpn_and_natural_sort():
+    assert safe_mpn("A/B:C") == "A_B_C"
+    assert sorted(["R10", "R2", "R1"], key=natural_sort_key) == ["R1", "R2", "R10"]
+
+
+def test_local_storage_generation_and_datasheet_roundtrip(tmp_path: Path):
+    backend = LocalStorageBackend(tmp_path)
+    backend.write_json_if_match("k.json", {"a": 1}, GENERATION_NEW)
+    try:
+        backend.write_json_if_match("k.json", {"a": 2}, GENERATION_NEW)
+        raise AssertionError("expected StaleGeneration")
+    except StaleGeneration:
+        pass
+    pdf = b"%PDF-1.4 fake"
+    key = store_datasheet_bytes(backend, pdf, "ABC/1")
+    assert key.endswith(".pdf")
+    assert resolve_datasheet(backend, "ABC/1") == key
+    assert compute_md5_from_bytes(pdf)
+    assert version_is_stale("1.0.0", "2.0.0")
+    assert not version_is_stale("2.0.0", "0.0.0")