diff --git a/periscope/dependency/backend/services/validation.py b/periscope/dependency/backend/services/validation.py index fe4b138..33bc868 100644 --- a/periscope/dependency/backend/services/validation.py +++ b/periscope/dependency/backend/services/validation.py @@ -64,7 +64,13 @@ from backend.periscopex.internal_features_check import check_internal_features from backend.periscopex.crystal_cl_check import check_crystal_cl from backend.periscopex.nc_pin_check import check_nc_pins -TRACE_VERSION = 1 +from backend.services.review_session import ( + TRACE_VERSION, + review_ic_async, + _signal_neighbors, + _select_review_pages, + _assistant_text, +) def _is_deterministic(f: Finding) -> bool: @@ -285,7 +291,9 @@ def _select_review_pages(pdf_path: str) -> str: # --------------------------------------------------------------------------- -async def review_ic_async( +# PinScope loop kept for rollback if live native smoke fails. Live path is +# review_ic_async imported from backend.services.review_session. +async def _inherited_review_ic_async( graph: DesignGraph, constraints_map: ConstraintsMap, ic_ref: str, diff --git a/periscope/dependency/frontend/content/changelog.md b/periscope/dependency/frontend/content/changelog.md index 1f444b0..861bf06 100644 --- a/periscope/dependency/frontend/content/changelog.md +++ b/periscope/dependency/frontend/content/changelog.md @@ -2,6 +2,14 @@ What's new in Periscope. +## 2.39.0 — 2026-09-20 — Fase C2: native DeepSeek review loop + +Live per-IC review runs from `periscope/src` (`review_session.py`, parse, graph tools). PinScope `validate.py` / `validation_tools.py` stay in `dependency/` as fallback; they are not deleted. LLM findings stay REVIEW; recommended never ERROR. AGPL unchanged. No parser/graph rewrite. No auto-place. + +- [Changed] `review_ic_async` is native (`backend.services.review_session`); inherited copy kept as `_inherited_review_ic_async`. +- [New] `review_parse.py` (FACT/REQUIREMENT/INFERENCE + `complete_findings` ids), `review_tools.py`, `review_context.py`, `constraints_lookup.py`. +- [Changed] `finding_engine.complete_finding` remains the clamp after submit_review. + ## 2.38.0 — 2026-09-20 — Physical split: periscope/src vs periscope/dependency Split the checkout so native Periscope and inherited PinScope are separate trees. No finding-engine rewrite. AGPL `LICENSE` stays at the repo root. The GitHub fork is not detached. diff --git a/periscope/src/backend/periscopex/constraints_lookup.py b/periscope/src/backend/periscopex/constraints_lookup.py new file mode 100644 index 0000000..cbb8852 --- /dev/null +++ b/periscope/src/backend/periscopex/constraints_lookup.py @@ -0,0 +1,51 @@ +"""MPN → extracted ComponentConstraints lookup (native).""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +from backend.periscopex.models import ComponentConstraints + +ConstraintsMap = dict[str, ComponentConstraints] + + +def load_datasheets(directory: str | Path) -> dict[str, ComponentConstraints]: + """Load all extracted datasheet JSONs, keyed by MPN.""" + result: dict[str, ComponentConstraints] = {} + dirpath = Path(directory) + if not dirpath.is_dir(): + return result + for f in dirpath.glob("*.json"): + raw = json.loads(f.read_text()) + c = ComponentConstraints.model_validate(raw) + result[c.mpn] = c + return result + + +def match_constraints( + mpn: str | None, + datasheets: dict[str, ComponentConstraints], +) -> ComponentConstraints | None: + """Match a component MPN to extracted constraints (exact then normalized).""" + if not mpn: + return None + if mpn in datasheets: + return datasheets[mpn] + norm = re.sub(r"[/_\-\s]", "", mpn).upper() + for ds_mpn, constraints in datasheets.items(): + if re.sub(r"[/_\-\s]", "", ds_mpn).upper() == norm: + return constraints + return None + + +def build_constraints_map(datasheets: dict[str, ComponentConstraints]) -> ConstraintsMap: + """Build MPN -> constraints map for tool lookups.""" + return dict(datasheets) + + +# PinScope-style aliases used by inherited pipeline / pin_mux_check. +_load_datasheets = load_datasheets +_match_constraints = match_constraints +_build_constraints_map = build_constraints_map diff --git a/periscope/src/backend/periscopex/review_context.py b/periscope/src/backend/periscopex/review_context.py new file mode 100644 index 0000000..7e293ff --- /dev/null +++ b/periscope/src/backend/periscopex/review_context.py @@ -0,0 +1,292 @@ +"""IC neighborhood text for the native datasheet review loop.""" + +from __future__ import annotations + +from backend.periscopex.models import ComponentType, DesignGraph, NetType +from backend.periscopex.pin_function_tokens import parse_net_token +from backend.periscopex.review_tools import ( + ConstraintsMap, + _format_specs, + _is_thermal_pad_pin, + _pin_sort_key, + _reviewer_voltage_str, +) + +_GROUND_NET_MAX_COMPONENTS = 5 + +def build_component_context( + graph: DesignGraph, + constraints_map: ConstraintsMap, + ref: str, +) -> str: + """Build a text summary of an IC's full circuit neighborhood. + + Shows every pin, its net, and every component connected to that net + (with values and specs). Ground/power nets with many connections are + summarized to avoid noise. + """ + comp = graph.components.get(ref) + if not comp: + return f"Component '{ref}' not found in design graph." + + constraints = constraints_map.get(comp.mpn or "") + lines: list[str] = [] + + # Header + lines.append(f"Component: {ref} ({comp.mpn or comp.value})") + if comp.component_subtype: + lines.append(f"Type: {comp.component_subtype}") + if constraints and constraints.package_info: + pi = constraints.package_info + lines.append(f"Package: {pi.package}, {pi.pin_count} pins") + if constraints and constraints.absolute_maximum_ratings: + lines.append("Extracted ratings (abs-max, plus Vrwm/polarity for ESD):") + for r in constraints.absolute_maximum_ratings: + bits = [] + if r.min is not None: + bits.append(f"min {r.min:g}") + if r.max is not None: + bits.append(f"max {r.max:g}") + span = " ".join(bits) if bits else "?" + lines.append( + f" {r.parameter}: {span} {r.unit} (datasheet p.{r.source_page})" + ) + lines.append("") + + # Build pin list — prefer extracted pintable order, fall back to netlist. + # (pin_num, pin_name, net_name, note, functions) + pin_entries: list[ + tuple[str, str | None, str | None, str | None, list[str] | None] + ] = [] + matched_schematic_pins: set[str] = set() + unmatched_ep_entries: list = [] # pintable EP rows whose number isn't in schematic + + if constraints and constraints.pintable: + for p in sorted(constraints.pintable, key=lambda x: _pin_sort_key(str(x.number))): + net_name = comp.pins.get(str(p.number)) + if net_name is not None: + matched_schematic_pins.add(str(p.number)) + pin_entries.append((str(p.number), p.name, net_name, None, p.functions)) + elif _is_thermal_pad_pin(p): + unmatched_ep_entries.append(p) + else: + pin_entries.append((str(p.number), p.name, None, None, p.functions)) + else: + for pn in sorted(comp.pins.keys(), key=_pin_sort_key): + matched_schematic_pins.add(pn) + pin_entries.append((pn, None, comp.pins[pn], None, None)) + + # Orphan schematic pins: present in netlist but not matched to any + # pintable entry. Commonly this is the EP/thermal pad under a user- + # chosen pin number (e.g. pin_count+1). + orphan_pins = [pn for pn in comp.pins if pn not in matched_schematic_pins] + orphan_pins.sort(key=_pin_sort_key) + + # If there's exactly one unmatched EP pintable row and one orphan + # schematic pin, map them together in the main list rather than + # listing both separately. + fused_ep_note = ( + "exposed pad / thermal pad — datasheet pintable lists this " + "without a usable pin number; matched to orphan schematic pin" + ) + if len(unmatched_ep_entries) == 1 and len(orphan_pins) == 1: + ep_row = unmatched_ep_entries[0] + orphan_pin = orphan_pins[0] + pin_entries.append(( + orphan_pin, + ep_row.name, + comp.pins[orphan_pin], + fused_ep_note, + ep_row.functions, + )) + unmatched_ep_entries = [] + orphan_pins = [] + + # Track nets already shown to avoid repetition + seen_nets: set[str] = set() + + for pin_num, pin_name, net_name, note, functions in pin_entries: + name_str = f" ({pin_name})" if pin_name else "" + note_str = f" [{note}]" if note else "" + # Render the datasheet alternate-function list inline only for pins whose + # net name asserts a peripheral role (UART5_TX, I2C1_SDA, ...), so the + # reviewer can check the asserted function against what the pin actually + # supports — without bloating the context for every GPIO. + alt_str = "" + if functions and net_name and parse_net_token(net_name): + alt_str = f" [alt: {', '.join(functions)}]" + + if not net_name: + lines.append(f"Pin {pin_num}{name_str} → [unconnected]{note_str}") + lines.append("") + continue + + net = graph.nets.get(net_name) + if not net: + lines.append(f"Pin {pin_num}{name_str} → {net_name}{alt_str}") + lines.append("") + continue + + voltage_str = _reviewer_voltage_str(net) + lines.append( + f"Pin {pin_num}{name_str} → {net_name} " + f"[{net.net_type.value}{voltage_str}]{alt_str}{note_str}" + ) + + # If we already showed this net's components, just note it + if net_name in seen_nets: + lines.append(f" (same net as above)") + lines.append("") + continue + seen_nets.add(net_name) + + # Collect neighbors on this net (excluding self) + neighbors = [ + pc for pc in net.pins + if pc.component_ref != ref and pc.component_ref in graph.components + ] + + # For large ground/power nets, summarize + if len(neighbors) > _GROUND_NET_MAX_COMPONENTS and net.net_type in (NetType.GROUND, NetType.POWER): + # Group by type + by_type: dict[str, list[str]] = {} + for pc in neighbors: + nb = graph.components[pc.component_ref] + ctype = nb.component_type.value + by_type.setdefault(ctype, []).append(pc.component_ref) + parts = [f"{len(refs)} {ctype}{'s' if len(refs) > 1 else ''}" for ctype, refs in sorted(by_type.items())] + lines.append(f" {len(neighbors)} components on this net: {', '.join(parts)}") + # Still list ICs specifically since they're important + for pc in neighbors: + nb = graph.components[pc.component_ref] + if nb.component_type == ComponentType.IC: + pin_name_str = "" + nb_constraints = constraints_map.get(nb.mpn or "") + if nb_constraints: + p = nb_constraints.pin_by_number(pc.pin_number) + if p: + pin_name_str = f" ({p.name})" + lines.append(f" {nb.reference}: {nb.mpn or nb.value} [pin {pc.pin_number}{pin_name_str}]") + else: + for pc in neighbors: + nb = graph.components[pc.component_ref] + mpn_str = f", {nb.mpn}" if nb.mpn else "" + specs_str = _format_specs(nb.specs) + if specs_str: + specs_str = f" ({specs_str})" + + # Pin name on the neighbor + pin_name_str = "" + nb_constraints = constraints_map.get(nb.mpn or "") + if nb_constraints: + p = nb_constraints.pin_by_number(pc.pin_number) + if p: + pin_name_str = f" ({p.name})" + + lines.append( + f" {nb.reference}: {nb.value}{mpn_str}{specs_str}" + f" [pin {pc.pin_number}{pin_name_str}]" + ) + + lines.append("") + + # Bridges: components whose pins land on two or more of this IC's nets. + # Captures Rsense / feedback dividers / decoupling caps / snubbers / + # protection resistors that span two IC pins — easy to miss when each + # endpoint is on a different net section, especially when one endpoint + # is on a power/ground net that gets summarized. + pin_name_by_num = {pn: pname for pn, pname, _, _, _ in pin_entries if pname} + ic_net_to_pins: dict[str, list[str]] = {} + for ic_pin, ic_net in comp.pins.items(): + if ic_net: + ic_net_to_pins.setdefault(ic_net, []).append(ic_pin) + ic_nets_set = set(ic_net_to_pins.keys()) + + def _label_endpoint(net: str) -> str: + pins = sorted(ic_net_to_pins.get(net, []), key=_pin_sort_key) + prefix = "pins" if len(pins) > 1 else "pin" + names = [pin_name_by_num.get(p) for p in pins] + named = [n for n in names if n] + if named: + unique = list(dict.fromkeys(named)) # preserve order, dedupe + name_str = "/".join(unique) + return f"{prefix} {'/'.join(pins)} ({name_str}, {net})" + return f"{prefix} {'/'.join(pins)} ({net})" + + # A bridge is "interesting" only if at least one endpoint is a signal + # net. Pure VCC↔GND bridges (bypass caps, every IC sharing the rail) + # would otherwise drown out the signal-bearing topology like Rsense or + # MCU pulldowns. Decoupling-cap counts are already visible in the + # per-pin listing's "N capacitors on this net" summary. + def _is_signal_net(name: str) -> bool: + n = graph.nets.get(name) + return bool(n and n.net_type == NetType.SIGNAL) + + bridge_lines: list[str] = [] + skipped_power_only = 0 + for nb_ref, nb in graph.components.items(): + if nb_ref == ref: + continue + nets_touched = {n for n in nb.pins.values() if n in ic_nets_set} + if len(nets_touched) < 2: + continue + if not any(_is_signal_net(n) for n in nets_touched): + skipped_power_only += 1 + continue + nets_sorted = sorted(nets_touched) + endpoints = " ↔ ".join(_label_endpoint(n) for n in nets_sorted) + mpn_str = f", {nb.mpn}" if nb.mpn else "" + specs_str = _format_specs(nb.specs) + if specs_str: + specs_str = f" ({specs_str})" + value_str = nb.value if nb.value else nb.component_type.value + bridge_lines.append( + f" {nb.reference}: {value_str}{mpn_str}{specs_str} — bridges {endpoints}" + ) + + if bridge_lines or skipped_power_only: + lines.append(f"Bridges between {ref}'s pins (signal-bearing only):") + if bridge_lines: + bridge_lines.sort() + lines.extend(bridge_lines) + if skipped_power_only: + lines.append( + f" ({skipped_power_only} additional bypass/rail-sharing " + f"bridges between power & ground nets, omitted — see per-pin listing for counts)" + ) + lines.append("") + + # Orphan schematic pins — not matched to any pintable entry. These are + # frequently the EP/thermal pad (schematic symbols commonly assign a + # custom pin number to the exposed pad). + if orphan_pins or unmatched_ep_entries: + lines.append("Additional schematic pins (not in datasheet pintable):") + if unmatched_ep_entries: + ep_names = ", ".join( + f"{p.name} (pintable #{p.number})" for p in unmatched_ep_entries + ) + lines.append( + f" (datasheet pintable lists these without a schematic-matched " + f"pin number — likely the exposed pad: {ep_names})" + ) + for pn in orphan_pins: + net_name = comp.pins.get(pn) + if not net_name: + continue + net = graph.nets.get(net_name) + if net is None: + lines.append(f" Pin {pn} → {net_name}") + continue + voltage_str = _reviewer_voltage_str(net) + lines.append( + f" Pin {pn} → {net_name} [{net.net_type.value}{voltage_str}]" + ) + if not orphan_pins and unmatched_ep_entries: + lines.append( + " (no matching orphan schematic pin found — the EP may be " + "genuinely unconnected in the schematic)" + ) + lines.append("") + + return "\n".join(lines) + diff --git a/periscope/src/backend/periscopex/review_parse.py b/periscope/src/backend/periscopex/review_parse.py new file mode 100644 index 0000000..fdaa0f3 --- /dev/null +++ b/periscope/src/backend/periscopex/review_parse.py @@ -0,0 +1,555 @@ +"""Native submit_review parser — FACT / REQUIREMENT / INFERENCE. + +DeepSeek (and other providers) submit findings via the submit_review tool. +This module owns the system prompt, ReviewResult, and parse → Finding mapping. +LLM output is source=review / finding_class=REVIEW. complete_finding clamps +recommended ≠ ERROR and LLM ≠ RULE. +""" + +from __future__ import annotations + +import json +import sys +from collections import Counter + +from backend.periscopex.finding_engine import complete_findings +from backend.periscopex.models import Finding + +SYSTEM_PROMPT = """\ +You are an electrical engineer reviewing how a component is used in a \ +hardware design. You have the component's datasheet and a description of \ +how it's wired in the actual circuit. + +### Review approach +Treat this IC as a COVERAGE CHECKLIST, not a single investigation. Before \ +hunting for problems, enumerate every focus area this IC has — derive them \ +from its pins, nets, neighbors, and subtype. A typical checklist: +- Power & decoupling on each supply pin — recommended Cin/Cout values, \ +ESR, and placement notes, not just "a cap is present". +- Each signal interface to each connected component — voltage \ +compatibility, direction, and correct cross-connection (e.g. TX↔RX). +- Absolute-maximum ratings on each pin vs. the actual rail driving it. \ +Use the extracted abs-max table in the component context when present; \ +confirm against the datasheet page if a number is missing or ambiguous. +- Recommended operating conditions and electrical characteristics \ +(VIH/VIL, VOL/VOH, input leakage, drive strength) where they change \ +whether the interface actually works. +- Reset / enable / boot / mode-strap / configuration pins. +- Clock or crystal circuit, if present — load capacitors and the \ +datasheet's recommended values. +- Required external components named by the datasheet (bootstrap, \ +compensation, feedback divider, sense resistor). +- Unused / no-connect pins. + +Then work the areas one at a time. For EACH area, don't just confirm a \ +part is present — ask what specific failure mode would make it wrong \ +(missing part, wrong value, over-voltage, swapped pair, wrong topology) \ +and check the datasheet and the actual netlist topology against that \ +failure mode. + +Every area must end up accounted for: either as a finding, or listed in \ +`checked_areas` as reviewed-and-correct. After you resolve one area, move \ +on to the NEXT area — do NOT stop and submit just because you found or \ +cleared the first issue. You have a generous turn budget; the goal is to \ +cover the whole IC, not to finish fast. + +### Reference designators — datasheet vs. schematic +The datasheet's reference/application circuit uses its OWN example \ +designators (e.g. "R2", "C1", "L1"). These are NOT the designators in \ +this project's schematic. The project's real designators are the ones \ +shown in the component context (e.g. "U1", "R5", "C12"). + +Before citing any passive or discrete in a finding, resolve its role to \ +the actual schematic designator: +1. Identify the component's *role* from the datasheet (e.g. "the resistor \ +between the VIN pin and the SW pin", "the feedback divider top resistor", \ +"the bootstrap capacitor between SW and BOOT"). +2. Use the component context — or `find_connected_components` / \ +`get_net_for_pin` — to find which schematic designator plays that role \ +in this design. +3. Cite ONLY the schematic designator (and its value/MPN) in your \ +finding. Never cite the datasheet's example designator. + +If no schematic component plays that role, say so explicitly ("no \ +component is connected between pin 3 (VIN) and pin 5 (SW)") rather than \ +naming a datasheet-example part. If you cannot resolve the role to a \ +schematic designator with confidence, demote the finding to WARNING or \ +INFO and describe the role instead of naming a part. + +### What to report +Only report issues. Do not report things that are correct. + +If your investigation concludes the design is correct — even when the \ +surface reading suggested otherwise (e.g., "C1 (100 nF) is below the \ +1 µF minimum, but C24 (1 µF) in parallel satisfies the spec", or "no \ +dedicated input cap is shown, but C3 is on the VIN net and satisfies \ +the requirement") — do NOT submit it as a finding. Add the topic to \ +`checked_areas` instead. A finding whose own `why` field confirms the \ +requirement is met dilutes the signal of real issues. If you write \ +"satisfies", "meets the requirement", "is in the correct place", or \ +"no issue" in your reasoning, the result belongs in `checked_areas`, \ +not `findings`. + +For each issue: +- **finding**: A concise one-line title of the issue (the rule title). \ +Keep it to a single line — cite the key component refs, values, net names, \ +or pin numbers, but do not elaborate. No multi-sentence descriptions here. +- **why**: The explanation — what the datasheet says and what could go \ +wrong. Keep this to **2 lines at most** (roughly 2 short sentences). This \ +is the most important field — explain the engineering consequence, not \ +just the rule, but stay terse. +- **status**: ERROR (will cause malfunction or violate abs max), \ +WARNING (may degrade reliability or is conditionally wrong), \ +INFO (worth noting but unlikely to cause problems). +- **source_page**: The datasheet page where the requirement is stated. +- **source_quote**: The exact verbatim sentence or clause from the datasheet \ +that states the requirement. Copy it precisely, character-for-character (a \ +short span, ~200 chars max) so it can be located and highlighted in the PDF. \ +ERROR and WARNING findings **must** include this field. Periscope checks the \ +quote against the extracted text of the cited page (±1); invented or \ +paraphrased quotes are demoted to Unverified WARNING. Omit the field only \ +when the requirement is shown solely in a figure or a rasterized table with \ +no selectable text — then status is WARNING at most and `why` must start \ +with `Unverified:`. +- **source_designator**: Leave unset when `source_page`/`source_quote` come \ +from THIS component's datasheet (the default). Set it to a connected \ +component's designator (e.g. `U3`) only when the page/quote come from that \ +neighbor's datasheet that you fetched via `get_datasheet_excerpt` — this \ +links the page number to the right datasheet. +- **recommendation**: What to change (for ERROR/WARNING only). + +This is a **design review**, not a design rule. Put observations in \ +`finding` (FACT), datasheet text in `why` (REQUIREMENT), and judgment \ +only there — do not invent millimetres, IEC numbers, or typical values. \ +Recommended datasheet notes are never ERROR. If evidence is missing, say \ +so (Unverified) instead of guessing. + +### Calibration +ERROR only for clear violations: required pin floating, voltage exceeding \ +absolute max, required external component completely missing, wrong \ +connection topology. + +WARNING when: component value differs from recommended but might be \ +adequate, rule is conditional on firmware/mode, concern is real but not \ +certain to cause failure. + +INFO when: design uses a valid but non-standard approach, optional feature \ +is unused, or a layout-level concern exists that cannot be verified from \ +the netlist. + +### ERROR requires a concrete harm pathway +Every ERROR that alleges damage, abs-max violation, or out-of-spec \ +stress must state the harm pathway with concrete numbers, not \ +speculation. Before submitting an ERROR, the `why` field must answer: +1. **Which pin or component takes the stress** (this IC's pin, an \ +internal node named by the datasheet, or an external part). +2. **What the actual voltage / current / temperature on it is**, derived \ +from the topology (the rail it ties to, the divider ratio, the regulator \ +output, the bias current). Numbers, not net labels. +3. **What the datasheet's limit is**, quoted from an abs-max table, \ +recommended-operating range, or pin description. +4. **Why (1) exceeds (3)** — the inequality, in numbers. + +If you cannot produce all four, downgrade to WARNING and write the \ +`why` as `Unverified: `. \ +Hedged language alone — "may damage", "could degrade", "might cause" — \ +is not enough for ERROR; replace it with the inequality or demote the \ +finding. This applies especially when the alleged damage is to an \ +*internal* component (internal DC-block cap, ESD diode, on-die clamp): \ +those are designed against the same package abs-max ratings as the \ +external pin, so an external stress within the pin's abs-max does not \ +damage the part inside. + +Two additional constraints on the inequality: +- **Pin-matched limit.** The abs-max number in (3) must be from the \ +abs-max row for *the same pin or signal* that takes the stress in \ +(1). Vdd's abs-max does not apply to an RF, signal, or I/O pin — \ +those pins have their own abs-max rows (commonly `V_RFIN`, `V_pin`, \ +`V_in` ranges, or are governed by the recommended-operating range). \ +If the datasheet does not list an abs-max for the specific pin under \ +stress, write `Unverified: no abs-max listed for pin ` and demote \ +to WARNING — do not borrow a different pin's number. +- **Strict inequality.** Abs-max is the don't-exceed line. The \ +inequality in (4) must be strict (`actual > limit`). "Equal to \ +abs-max" is not a violation — it may stress lifetime but does not \ +qualify as damage. If the math comes out to `=` rather than `>`, the \ +finding is at most a WARNING. + +### Decoupling capacitors +Larger caps satisfy smaller specs: 470nF satisfies "0.1uF", 10uF satisfies \ +"1uF minimum". Only flag if actual value is below the minimum specified. + +### Netlist limitations +You are reviewing a NETLIST, not PCB layout. You cannot verify component \ +proximity, trace routing, or thermal management. If a functional \ +requirement is met at the netlist level, do not flag it as an issue. + +### Identify the role of each external part before judging it +For each external part on this IC's pins (R, C, L, FB, diodes, \ +transistors), derive its role in the design from first principles \ +before concluding whether the connection is correct. The role is the \ +answer to "what does this part do in this circuit?" — not the answer \ +to "does this pattern have a name I recognize?". Reason from: +1. **What the pin does** (from the datasheet pin description in your \ +context — e.g. "DC blocked", "AC coupled", "internally biased", \ +"open-drain", "high impedance", "reference output"). +2. **What the part is** (its value class and approximate value — an \ +inductor at RF frequencies is a choke; a small cap to GND is shunt \ +decoupling; a series cap is AC coupling or DC blocking; a divider \ +sets a sense ratio). +3. **Where the other end of the part goes** — trace it with \ +`find_connected_components` / `get_net_for_pin` / the bridges list. \ +A part terminated on a power rail does something different from one \ +terminated at a connector or another IC pin. +4. **What (1) + (2) + (3) imply about the part's purpose.** + +When a documented characteristic of the pin would *prevent* the \ +surface-reading interaction (e.g. a DC rail tied through an inductor \ +to a pin that is documented as DC-blocked), the part is almost always \ +serving the rest of the circuit, not the chip — its role is found by \ +asking what the remaining circuit needs the part for, including \ +loads reached through a connector or coax further down the net. Do \ +not raise a finding against the chip for a part that does not stress \ +the chip. + +If you cannot articulate the role after a brief look, submit the \ +concern as `status="WARNING"` with `why` starting `Unverified: role \ +of on pin not determined` — never ERROR on a component \ +whose purpose you have not identified. + +### Budget per concern: cap ONE concern, not the whole review +A single concern (one potential finding under investigation) gets at \ +most three follow-up tool calls beyond what was already in your initial \ +context. If the concern is not resolved within that budget, submit it \ +as WARNING with `why` starting `Unverified: ` and move on to the next area. This per-concern \ +cap exists so one concern cannot swallow the whole review — NOT so you \ +finish early. Your total budget across all concerns is generous: spend it \ +on breadth. The failure mode to avoid is leaving focus areas of this IC \ +uninvestigated, not spending too many turns. Do not call submit_review \ +while any enumerated focus area is still uninvestigated. + +### Net names are not voltage labels +Net names are user-chosen labels — they describe a signal's *role*, not its \ +actual voltage. A net named `VBAT_SENSE`, `8S_LiPo`, or `HV_FB` may carry \ +only a low-voltage MCU control line, a divided-down sense voltage, or be \ +misnamed entirely. + +Before flagging any absolute-max violation, supply-mismatch, or "pin driven \ +beyond rated input" issue, use `find_connected_components` to identify the \ +*actual* driver of the net (power rail, regulator output, MCU pin, voltage \ +divider, connector, etc.). Only flag when the topology confirms the \ +voltage. If the driver is ambiguous, demote to WARNING and describe what \ +would need to be verified. + +### Voltage tags in tool output: trusted but sparse +The `(power, X.X V)` annotation in net-info lines and `[power, X.X V]` tag \ +on pin lines only appear when the voltage is sourced from the netlist \ +itself — either the net name encodes it (`+5V`, `+3V3`, `1V5`) or the \ +user declared it via a power-source hint. Power-tree-derived voltages \ +(deterministic propagation through passthroughs, regulator-output back- \ +annotation, model inferences) are deliberately suppressed from your tool \ +output — they are too lossy to trust at review time, and trusting them \ +has produced false-positive findings in the past. + +When a pin's net has no voltage tag, the netlist does not establish what \ +voltage flows there. Trace topology (find_connected_components, walking \ +back through passthroughs and regulators) to discover the source, or \ +treat the rail as unknown. + +### Rail voltages and VREF: do not guess +When you cannot establish an IC's supply or signal voltage from any of: + +- the net name (e.g., `+5V`, `+3V3`, `GND`), +- a `(power, X.X V)` tag in tool output, +- a connected source / regulator output whose voltage IS visible by the \ +rules above (reached by walking topology through find_connected_components), + +you must NOT reconstruct it by assuming a VREF on an upstream regulator's \ +feedback divider. VREF varies by part (1.20V LDO, 1.25V LDO, 0.6V buck, \ +0.8V buck, 0.925V buck-boost, 1.205V LDO, ...). A guessed VREF cascades \ +into a wrong rail voltage and false-positive out-of-spec findings — this \ +has happened (assumed VREF=0.5V → Vdd=1.48V → bogus 'below operating \ +range' WARNING). + +If a finding hinges on knowing the rail voltage and you cannot establish \ +it from the rules above, downgrade to WARNING with `why` starting \ +`Unverified: rail voltage at could not be established without \ +guessing a regulator's VREF`. Do not raise ERROR on guessed rails. + +### Cross-IC interface checks and uncertainty +When a finding hinges on a *connected* IC's spec (5V-tolerance, abs-max, \ +VIH/VIL, drive strength), that spec lives in the neighbor's datasheet, \ +not yours. Before raising ERROR on such a finding, call \ +`get_datasheet_excerpt(designator, topic)` on the neighbor (e.g. \ +`topic="pin_voltage_levels"` for 5V-tolerance, `"absolute_max"` for \ +stress ratings) and read the returned pages. When a finding then cites a \ +page or quote you read from that neighbor's excerpt, set the finding's \ +`source_designator` to the neighbor's designator and put the neighbor's \ +page number in `source_page` — the citation must point at the datasheet the \ +evidence actually lives in, not yours. + +If the excerpt does not resolve the spec, call `submit_review` with \ +`status="WARNING"` (not ERROR) for that finding, and start its `why` \ +field with `Unverified: `. Reserve ERROR \ +for cases where the violation is established from both sides of the \ +interface — a false ERROR is the single biggest trust-killer for this \ +review. + +### Alternate-function feasibility vs. direction +Pins on peripheral-named nets show their datasheet alternate-function list \ +inline as `[alt: ...]` (and `get_pintable` shows it for any pin on demand). \ +That list is datasheet-extracted ground truth for what the pin can be muxed \ +to. Use it for a FEASIBILITY check — never a direction check: + +- FEASIBILITY (hard ERROR): if a net name asserts a peripheral function — \ +e.g. a net `...UART5-TX...` on a pin whose `[alt: ...]` exposes UART5 only \ +as `UART5_RX` — the silicon cannot route that function to that pin. It is \ +physically unrealizable regardless of anything downstream. Raise ERROR and \ +name the functions the pin actually exposes for that peripheral. +- DIRECTION (context-dependent — do NOT auto-flag): a TX wired to the other \ +device's RX is normal. A direct UART link crosses TX→RX; a transceiver, \ +isolator, or level-shifter is often straight-through (MCU TX → transceiver \ +TXD/DI). Whether a TX/RX (or SDA/SCL) connection is correct depends on the \ +role of the part on the other end, which you must reason about from the \ +circuit — never flag a TX-on-an-RX-named-net (or vice versa) on naming \ +alone. Only raise a direction ERROR when topology forces it (e.g. two \ +push-pull outputs on one net). Otherwise WARNING/INFO, stating the \ +downstream role you'd need to confirm. + +### Pin labels in your context can be wrong +The `Pin N (NAME)` labels in the component context come from a separate \ +datasheet-extraction pass. For image-only PDFs, small or dense pin \ +tables, and non-standard parts, that pass can mis-label individual pins \ +(D+/D− swaps, TX/RX, CC1/CC2, IN+/IN−, anode/cathode, A/K, +/−). Before \ +raising an ERROR whose logic turns on the polarity or identity of a \ +specific pin pair on THIS IC (differential-pair swap, supply polarity, \ +input/output orientation), re-read the pin-mapping page of the datasheet \ +PDF already in your initial context and verify each pin label against \ +it. If the datasheet contradicts the in-context label, trust the \ +datasheet — demote the finding to WARNING and state explicitly which \ +pin label in the context appears mis-extracted (e.g. \ +`Pin A6 labeled "D−" in context, datasheet shows "D+"`). The `[alt: ...]` \ +alternate-function list shown for peripheral-named-net pins is taken \ +verbatim from the datasheet pin table and is reliable even when the short \ +`(NAME)` label is not — prefer it when judging what a pin can be muxed to. + +### ESD / TVS arrays — do not invent the diode topology +An IO pin whose neighbor is GND (or whose pin name is IO/I/O) does NOT \ +mean a single steering diode from IO to GND that conducts at ~0.7 V. \ +Many 2-channel ESD arrays (audio, RS-232, RS-485) are *bidirectional \ +back-to-back* with a signed working voltage (Vrwm, often ±12 V or \ +±13 V). In that topology a 1 Vrms AC-coupled audio swing is inside the \ +standoff range and is not clipped. + +Before claiming clipping, forward conduction, or "unidirectional clamp": +1. Quote the datasheet topology (block diagram or "bidirectional" / \ +"unidirectional" / "back-to-back" wording) in `source_quote`. +2. Quote Vrwm (or equivalent working-voltage row) with sign. Use that \ +number as the standoff, not a generic silicon Vf. +3. If the block diagram or electrical table is unreadable, status is \ +WARNING at most and `why` must start with `Unverified:` — never ERROR \ +from "typical for this part" or from pin names alone. + +A replacement recommendation must name a part whose topology matches \ +the signal (do not suggest a unidirectional array for a bipolar \ +AC-coupled audio net). + +### Direction-control and transceiver function tables +Bidirectional transceivers, level shifters, mux/demux, bus switches, and \ +analog switches (74xx245, 74xx125, 74xx157, TS3A-family, etc.) often \ +print their function/truth table in a column-segmented layout where two \ +adjacent cells read as a single English phrase ("input B = A", \ +"high-Z input"). Scanning left-to-right inverts the meaning and \ +invalidates every downstream finding. Before raising any ERROR \ +involving bus contention, "two outputs on one net", or direction-control \ +polarity, re-read the function table from THIS IC's datasheet PDF and \ +quote each cell of the relevant row separately. State the direction \ +explicitly ("DIR=H → A is input, B is output, A→B") before claiming \ +output contention. + +### One root cause = one finding +If two ERRORs you're about to submit collapse to the same underlying \ +mistake — e.g. a single mis-configured DIR pin produces both "bus \ +contention on TXD" AND "device is unidirectional only" — submit ONE \ +combined finding that names the root cause. Restate the downstream \ +consequences inside the `why` field instead of as separate findings. \ +Two ERRORs that share a premise read as independent problems, double \ +the review's apparent severity, and dilute trust if the shared premise \ +turns out to be wrong. + +### Bridges between IC pins +The component context includes a `Bridges between 's pins:` section \ +listing 2-or-more-terminal components that connect two of this IC's nets \ +(decoupling caps, feedback dividers, sense resistors, snubbers, etc.). This \ +is the most direct view of external passives associated with the IC. \ +Before claiming a required external part is missing, scan this section — \ +the part may be there under a different role label. Required passives \ +must always be cited from the bridges list (or via a graph tool query) — \ +never inferred from a single-pin listing alone. + +### Exposed pad / thermal pad (EP, DAP, ePAD) +The datasheet pintable's EP/DAP pin number often does not match the number \ +the schematic symbol uses. Schematic symbols commonly assign the exposed \ +pad a custom number (frequently pin_count+1, or a unique name). \ +Unmatched schematic pins that aren't in the datasheet pintable are listed \ +as "Additional schematic pins (not in datasheet pintable)" at the end of \ +the component context — these are almost always the EP/thermal pad. \ +Before flagging an EP-unconnected error, check that no additional \ +schematic pin is tied to GND. If any additional pin is on a GND net, \ +treat the EP requirement as satisfied and do not flag it. + +### Output — submit_review is the ONLY way findings reach the report +You MUST call the `submit_review` tool to record findings. Writing \ +findings as a JSON block in your text response does NOT save them — they \ +will be dropped. When you are ready to record findings (even just one), \ +call `submit_review` with the findings array and checked_areas list. If \ +the circuit matches the datasheet with no issues, still call \ +`submit_review` with an empty findings array. + +In **checked_areas**, list what you reviewed and confirmed correct — \ +short labels like "input decoupling", "output capacitor", "enable logic", \ +"voltage margins". This tells the engineer what was verified, not just \ +what failed. + +Before calling submit_review, confirm every focus area you enumerated at \ +the start is accounted for — present in either `findings` or \ +`checked_areas`. If any enumerated area is still uninvestigated, \ +investigate it before submitting. + +Use the graph query tools to investigate connections beyond the provided \ +context if needed. +""" + + +# Maximum turns for the review agentic loop +_MAX_REVIEW_TURNS = 16 + + +class ReviewResult: + """Findings + coverage from a single IC review.""" + __slots__ = ("findings", "checked_areas") + + def __init__(self, findings: list[Finding], checked_areas: list[str]): + self.findings = findings + self.checked_areas = checked_areas + + +def _coerce_str_list(value) -> list[str]: + """Coerce a tool-input value into a list of non-empty strings. + + Claude occasionally violates the tool schema (e.g. returns a stringified + list instead of a real array). Sanitize here so downstream Pydantic + validation of ValidationReport cannot fail on a single IC's output. + """ + if value is None: + return [] + if isinstance(value, list): + return [str(x).strip() for x in value if x is not None and str(x).strip()] + if isinstance(value, str): + s = value.strip() + if not s: + return [] + try: + parsed = json.loads(s) + if isinstance(parsed, list): + return [str(x).strip() for x in parsed if x is not None and str(x).strip()] + except (json.JSONDecodeError, ValueError): + pass + return [s] + return [str(value).strip()] + + +def parse_submit_review( + tool_input: dict, + ic_ref: str, + mpn: str, + *, + mpn_by_designator: dict[str, str] | None = None, + connected: set[str] | None = None, +) -> ReviewResult: + """Parse submit_review tool output into findings + coverage. + + A finding whose evidence came from a *connected* neighbor's datasheet + excerpt carries that neighbor's designator in ``source_designator`` — its + ``source_page`` is a page in the neighbor's PDF, not the IC under review. + ``mpn_by_designator`` resolves that designator to the MPN so ``reference`` + (and the frontend viewer) point at the correct datasheet; ``connected`` + restricts which neighbor designators are honored. When the map/neighbor is + absent or unresolvable, the citation falls back to this IC's own datasheet + so the cited page and the datasheet the viewer opens never disagree. + """ + mpn_by_designator = mpn_by_designator or {} + findings: list[Finding] = [] + raw_findings = tool_input.get("findings") or [] + if not isinstance(raw_findings, list): + raw_findings = [] + for item in raw_findings: + if not isinstance(item, dict): + continue + try: + page = item.get("source_page") + raw_src = str(item.get("source_designator") or "").strip() + if ( + raw_src + and raw_src != ic_ref + and raw_src in mpn_by_designator + and (connected is None or raw_src in connected) + ): + src_designator: str | None = raw_src + src_mpn = mpn_by_designator[raw_src] + else: + src_designator = None + src_mpn = mpn + status = item["status"] + why = str(item.get("why") or "") + quote = str(item.get("source_quote") or "").strip() + # ERROR/WARNING with no verbatim quote: demote before PDF check. + if status in ("ERROR", "WARNING") and not quote: + if status == "ERROR": + status = "WARNING" + if not why.startswith("Unverified:"): + why = ( + "Unverified: no verbatim datasheet quote. " + why + ).strip() + rec = str(item.get("recommendation") or item.get("action") or "").strip() + act = str(item.get("action") or rec).strip() + findings.append(Finding( + designator=ic_ref, + mpn=mpn, + source_designator=src_designator, + finding=item["finding"], + facts=str(item.get("finding") or ""), + requirement=why, + inference=str(item.get("inference") or ""), + why=why, + status=status, + source_page=page, + source_quote=item.get("source_quote", ""), + recommendation=rec, + action=act, + reference=f"{src_mpn} datasheet p.{page if page is not None else '?'}", + source="review", + finding_class="REVIEW", + evidence_status="SUFFICIENT" if quote else "INSUFFICIENT", + )) + except (KeyError, TypeError, ValueError) as exc: + print(f"Skipping malformed finding for {ic_ref}: {exc}", file=sys.stderr) + continue + checked_areas = _coerce_str_list(tool_input.get("checked_areas")) + return ReviewResult(findings, checked_areas) + + +def assign_finding_ids(findings: list[Finding]) -> None: + """Assign finding_id: {designator}-{001}, {002}, ... then run the finding engine.""" + counter: Counter[str] = Counter() + for f in findings: + counter[f.designator] += 1 + f.finding_id = f"{f.designator}-{counter[f.designator]:03d}" + complete_findings(findings) + + + +def _parse_review(*args, **kwargs) -> ReviewResult: + """Alias kept for tests and the inherited validate.py adapter.""" + return parse_submit_review(*args, **kwargs) diff --git a/periscope/src/backend/periscopex/review_tools.py b/periscope/src/backend/periscopex/review_tools.py new file mode 100644 index 0000000..3d8f3c2 --- /dev/null +++ b/periscope/src/backend/periscopex/review_tools.py @@ -0,0 +1,940 @@ +"""Native graph-query tools for the Periscope datasheet review loop. + +Lives in periscope/src. The PinScope validation_tools.py remains in +dependency/ and is not imported by this module or by the live loop. +""" + +from __future__ import annotations + +import logging +import re +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from backend.periscopex.models import ( + ComponentConstraints, + DesignGraph, +) +from backend.periscopex.utils import safe_mpn + +log = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _pin_sort_key(pin: str) -> tuple: + m = re.match(r"^(\d+)", pin) + if m: + return (0, int(m.group(1)), pin) + return (1, 0, pin) + + +_THERMAL_PAD_NAME_RE = re.compile( + r"\b(e[\s\-]?pad|epad|ep|dap|thermal\s*pad|exposed\s*(?:pad|paddle)|die[\s\-]?(?:attach\s*)?pad)\b", + re.IGNORECASE, +) + + +def _reviewer_voltage_str(net) -> str: + """Format a net's voltage for reviewer tool output.""" + if net is None or net.voltage is None: + return "" + return f", {net.voltage}V" + + +def _is_thermal_pad_pin(pin) -> bool: + """Heuristic: does a pintable entry describe the exposed/thermal pad? + + Users commonly assign the EP a custom pin number in their schematic + symbol (often pin_count+1) that doesn't match the datasheet pintable's + number for the same pad. Detecting EP pintable entries lets the + reviewer match them to orphan schematic pins instead of reporting them + as unconnected. + """ + for field in (getattr(pin, "name", None), getattr(pin, "description", None)): + if field and _THERMAL_PAD_NAME_RE.search(str(field)): + return True + number = str(getattr(pin, "number", "")).strip() + if number and not number.isdigit() and _THERMAL_PAD_NAME_RE.search(number): + return True + return False + + +def _format_specs(specs) -> str: + """Format component specs as a compact string.""" + if not specs: + return "" + d = specs.model_dump(exclude_none=True, exclude={"specs_type"}) + if not d: + return "" + parts = [] + for k, v in d.items(): + parts.append(f"{k}={v}") + return ", ".join(parts) + + +# Type alias for constraints lookup +ConstraintsMap = dict[str, ComponentConstraints] # MPN -> constraints + + +# --------------------------------------------------------------------------- +# Excerpt tool — per-review state, topic regexes, page selection +# --------------------------------------------------------------------------- + +# Each topic maps to a narrow keyword regex used to pick relevant pages from +# a neighbor IC's datasheet. Narrower than _REVIEW_KEYWORDS so an excerpt +# fetch returns a focused slice (~5-10 pages) rather than 30+. +EXCERPT_TOPICS: dict[str, re.Pattern] = { + "absolute_max": re.compile( + r"absolute\s+maximum|maximum\s+ratings?|stress\s+rating", + re.IGNORECASE, + ), + "recommended_operating": re.compile( + r"recommended\s+operating|operating\s+conditions?|operating\s+range", + re.IGNORECASE, + ), + "electrical_characteristics": re.compile( + r"electrical\s+characteristics?|DC\s+characteristics?|AC\s+characteristics?" + r"|V[IO][HL]\s*\(|input\s+(high|low)\s+voltage|output\s+(high|low)\s+voltage", + re.IGNORECASE, + ), + "pin_voltage_levels": re.compile( + r"5[\s\-]?V[\s\-]?tolerant|5V[\s\-]?tolerance|voltage\s+tolerance" + r"|input\s+voltage\s+range|pin\s+voltage|I/O\s+voltage" + r"|V[IO][HL]\b|VIO\b|VDDIO\b|tolerant\s+input", + re.IGNORECASE, + ), + "power_supply": re.compile( + r"power\s+supply|supply\s+voltage|VDD|VCC|VBAT|supply\s+current" + r"|quiescent\s+current", + re.IGNORECASE, + ), + "thermal": re.compile( + r"thermal\s+(resistance|shutdown|pad|characteristics)|junction\s+temperature" + r"|theta[\s\-]?J[AC]|θJ[AC]", + re.IGNORECASE, + ), + "application_circuit": re.compile( + r"application\s+(circuit|schematic|information|note)" + r"|typical\s+application|reference\s+design|recommended\s+circuit", + re.IGNORECASE, + ), +} + +_EXCERPT_MAX_PAGES_PER_FETCH = 10 # cap per single excerpt call + + +@dataclass +class ExcerptState: + """Per-review state threaded through ``execute_tool`` so the excerpt tool + can enforce neighbor-only access, run a fetch/page budget, and reuse + pypdf trim work across ICs in the same validation run. + + Created in ``review_ic_async``; carries the cross-IC ``cache`` from the + caller (``validate_design_async``). + """ + + current_ic: str + connected_designators: set[str] + graph: DesignGraph + pdf_dir: Path + storage: Any | None = None + # Cross-IC trimmed-PDF cache keyed by (designator, topic, ds_md5) + # -> (trimmed_pdf_path, [original_page_numbers]). Lives for the duration + # of one validate_design_async. + cache: dict[tuple[str, str, str], tuple[str, list[int]]] = field( + default_factory=dict + ) + # Per-review budget counters. ``page_budget`` is the global ceiling that + # bounds total fan-out on a hub IC; ``per_neighbor_page_budget`` is a + # sub-budget so that verifying ONE interface (which needs ~2-3 topic + # fetches from a single neighbor — e.g. pin_voltage_levels + absolute_max) + # is never blocked by pages already spent on a *different* neighbor. This + # is the fix for the U2-001 / U3-001 false positives, where a single + # 25-page global budget got exhausted before the abs-max table could be + # read, forcing the reviewer to guess. + fetch_count: int = 0 + page_count: int = 0 + fetch_budget: int = 8 + page_budget: int = 60 + per_neighbor_page_budget: int = 30 + pages_per_neighbor: dict[str, int] = field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# Tool implementations +# --------------------------------------------------------------------------- + + +def find_connected_components( + graph: DesignGraph, + constraints_map: ConstraintsMap, + designator: str, + pin: str, + designator_filter: str | None = None, +) -> str: + """Find all components on the net at designator.pin, with full specs.""" + comp = graph.components.get(designator) + if not comp: + return f"Component '{designator}' not found." + + net_name = comp.pins.get(str(pin)) + if not net_name: + return f"Pin {pin} on {designator} is not connected in the netlist." + + net = graph.nets[net_name] + voltage_str = _reviewer_voltage_str(net) + lines = [f"Net: {net_name} ({net.net_type.value}{voltage_str})"] + + count = 0 + for pc in net.pins: + if pc.component_ref == designator: + continue + if designator_filter and not pc.component_ref.upper().startswith(designator_filter.upper()): + continue + + neighbor = graph.components.get(pc.component_ref) + if not neighbor: + continue + count += 1 + + # Component header + mpn_str = f", MPN={neighbor.mpn}" if neighbor.mpn else "" + sub_str = f", {neighbor.component_subtype}" if neighbor.component_subtype else "" + specs_str = _format_specs(neighbor.specs) + if specs_str: + specs_str = f" ({specs_str})" + + lines.append( + f" {neighbor.reference}: {neighbor.value}{mpn_str}, " + f"{neighbor.component_type.value}{sub_str}{specs_str}" + ) + + # Pin map + pin_strs = [] + for pn, pnet in sorted(neighbor.pins.items(), key=lambda x: _pin_sort_key(x[0])): + pin_strs.append(f"{pn}->{pnet}") + lines.append(f" pins: {', '.join(pin_strs)}") + + if count == 0: + filter_note = f" matching '{designator_filter}*'" if designator_filter else "" + lines.append(f" (no components{filter_note} on this net)") + + return "\n".join(lines) + + +def get_net_for_pin( + graph: DesignGraph, + constraints_map: ConstraintsMap, + designator: str, + pin: str, +) -> str: + """Get net info for a specific pin — lightweight, no component listing.""" + comp = graph.components.get(designator) + if not comp: + return f"Component '{designator}' not found." + + net_name = comp.pins.get(str(pin)) + if not net_name: + return f"Pin {pin} on {designator} is not connected in the netlist." + + net = graph.nets[net_name] + voltage_str = _reviewer_voltage_str(net) + + # Get pin name from constraints + pin_name = "" + constraints = constraints_map.get(comp.mpn or "") + if constraints: + p = constraints.pin_by_number(pin) + if p: + pin_name = f" ({p.name})" + + return f"Pin {pin}{pin_name} on {designator} -> {net_name} [{net.net_type.value}{voltage_str}]" + + +def shortest_path( + graph: DesignGraph, + constraints_map: ConstraintsMap, + designator_a: str, + pin_a: str, + designator_b: str, + pin_b: str, + *, + max_hops: int = 12, +) -> str: + """BFS through the bipartite graph from A.pin to B.pin. + + Hops alternate component→net→component. Returns the hop list or a + clear miss message. Caps depth so the reviewer cannot explode memory + on dense power nets. + """ + a = graph.components.get(designator_a) + b = graph.components.get(designator_b) + if not a: + return f"Component '{designator_a}' not found." + if not b: + return f"Component '{designator_b}' not found." + + net_a = a.pins.get(str(pin_a)) + net_b = b.pins.get(str(pin_b)) + if not net_a: + return f"Pin {pin_a} on {designator_a} is not connected in the netlist." + if not net_b: + return f"Pin {pin_b} on {designator_b} is not connected in the netlist." + + if designator_a == designator_b and str(pin_a) == str(pin_b): + return f"Same endpoint: {designator_a}.{pin_a} on {net_a}." + + if net_a == net_b: + return ( + f"Direct (same net): {designator_a}.{pin_a} —[{net_a}]— " + f"{designator_b}.{pin_b}" + ) + + # BFS on component nodes; edges are nets shared between components. + from collections import deque + + start = designator_a + goal = designator_b + queue: deque[str] = deque([start]) + # prev[ref] = (previous_ref, via_net) + prev: dict[str, tuple[str, str] | None] = {start: None} + hops = 0 + found = False + while queue and hops < max_hops: + hops += 1 + for _ in range(len(queue)): + cur = queue.popleft() + for net_name, others in graph.neighbors(cur).items(): + for other in others: + if other in prev: + continue + prev[other] = (cur, net_name) + if other == goal: + found = True + queue.clear() + break + queue.append(other) + if found: + break + if found: + break + + if not found or goal not in prev: + return ( + f"No path within {max_hops} hops from " + f"{designator_a}.{pin_a} ({net_a}) to " + f"{designator_b}.{pin_b} ({net_b})." + ) + + # Reconstruct component chain, then decorate endpoints with pins. + chain_refs: list[str] = [] + via_nets: list[str] = [] + node = goal + while node != start: + chain_refs.append(node) + parent, via = prev[node] # type: ignore[misc] + via_nets.append(via) + node = parent + chain_refs.append(start) + chain_refs.reverse() + via_nets.reverse() + + parts: list[str] = [f"{designator_a}.{pin_a}"] + for i, via in enumerate(via_nets): + nxt = chain_refs[i + 1] + if nxt == designator_b: + parts.append(f"—[{via}]— {designator_b}.{pin_b}") + else: + parts.append(f"—[{via}]— {nxt}") + return f"Path ({len(via_nets)} hop(s)): " + " ".join(parts) + + +def get_pintable( + graph: DesignGraph, + constraints_map: ConstraintsMap, + designator: str, +) -> str: + """Get full pintable with connection status.""" + comp = graph.components.get(designator) + if not comp: + return f"Component '{designator}' not found." + + constraints = constraints_map.get(comp.mpn or "") + if not constraints: + # Fall back to just showing netlist pins + lines = [f"Pintable for {designator} ({comp.mpn or comp.value}) — no extracted pintable:"] + for pn, pnet in sorted(comp.pins.items(), key=lambda x: _pin_sort_key(x[0])): + net = graph.nets.get(pnet) + ntype = f" [{net.net_type.value}]" if net else "" + lines.append(f" Pin {pn}: -> {pnet}{ntype} [connected]") + return "\n".join(lines) + + lines = [f"Pintable for {designator} ({comp.mpn}):"] + matched: set[str] = set() + for p in sorted(constraints.pintable, key=lambda x: _pin_sort_key(str(x.number))): + net_name = comp.pins.get(str(p.number)) + func_str = f" [alt: {', '.join(p.functions)}]" if p.functions else "" + if net_name: + matched.add(str(p.number)) + net = graph.nets.get(net_name) + voltage_str = _reviewer_voltage_str(net) + ntype = net.net_type.value if net else "?" + lines.append(f" Pin {p.number} ({p.name}): -> {net_name} [{ntype}{voltage_str}]{func_str} [connected]") + else: + tp_note = " [likely exposed pad — check orphan schematic pins below]" if _is_thermal_pad_pin(p) else "" + lines.append(f" Pin {p.number} ({p.name}){func_str}: [unconnected]{tp_note}") + + orphans = [pn for pn in comp.pins if pn not in matched] + if orphans: + lines.append("") + lines.append( + "Additional schematic pins (not in datasheet pintable — " + "commonly the EP/thermal pad under a user-chosen pin number):" + ) + for pn in sorted(orphans, key=_pin_sort_key): + net_name = comp.pins.get(pn) or "" + net = graph.nets.get(net_name) + voltage_str = _reviewer_voltage_str(net) + ntype = net.net_type.value if net else "?" + lines.append(f" Pin {pn}: -> {net_name} [{ntype}{voltage_str}]") + + return "\n".join(lines) + + +def _resolve_neighbor_pdf( + state: ExcerptState, + mpn: str, +) -> Path | None: + """Resolve a neighbor IC's MPN to a local PDF path. + + Mirrors validation._find_pdf's local-then-library lookup so neighbor + datasheets follow the same resolution rules as the IC under review. + """ + from backend.services.datasheet_finder import find_local_pdf + + local = find_local_pdf(state.pdf_dir, mpn) + if local is not None and local.is_file(): + wanted = state.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 state.storage is not None: + try: + from backend.services import projects as proj_svc + lib_key = proj_svc.library_has_datasheet(state.storage, mpn) + if lib_key: + wanted = state.pdf_dir / f"{safe_mpn(mpn)}.pdf" + state.storage.download_to_local(lib_key, wanted) + if wanted.is_file(): + return wanted + except Exception: + log.exception("excerpt: library lookup failed for %s", mpn) + return None + + +def _trim_pdf_by_keywords( + pdf_path: Path, + keyword_re: re.Pattern, + max_pages: int, +) -> tuple[str, list[int]]: + """Pypdf-trim a PDF to pages matching a keyword regex (+/-1 neighbors). + + Returns ``(trimmed_pdf_path, kept_page_numbers_1indexed)``. The trimmed + path is a temp file the caller is responsible for cleaning up *eventually* + — in practice we keep these for the lifetime of the validation run so the + same excerpt can be reused across ICs. + + Page numbers in the return list are 1-indexed and refer to the *original* + PDF, so the model can cite them as ``source_page`` consistent with the + no-remap convention used everywhere else in the reviewer. + """ + from pypdf import PdfReader, PdfWriter + + reader = PdfReader(str(pdf_path)) + total = len(reader.pages) + if total == 0: + return str(pdf_path), [] + + keep: set[int] = set() + for i, page in enumerate(reader.pages): + try: + text = page.extract_text() or "" + except Exception: + text = "" + if keyword_re.search(text): + for n in (i - 1, i, i + 1): + if 0 <= n < total: + keep.add(n) + if len(keep) >= max_pages: + break + + if not keep: + # Fall back: first few pages so the model gets *something* it can + # decline to use, rather than an empty excerpt. + keep = set(range(min(3, total))) + + selected = sorted(keep)[:max_pages] + writer = PdfWriter() + for i in selected: + writer.add_page(reader.pages[i]) + tmp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) + writer.write(tmp) + tmp.close() + return tmp.name, [i + 1 for i in selected] + + +def get_datasheet_excerpt( + graph: DesignGraph, + constraints_map: ConstraintsMap, + designator: str, + topic: str, + state: ExcerptState | None, +): + """Return pages from a *connected* neighbor IC's datasheet for a topic. + + Returns ``(text_summary, pdf_block_or_none)`` — the caller treats the text + as the tool's ``content`` and attaches the PdfBlock (if present) to the + same user message so the model can read the pages on the next turn. + + Restricted to neighbors of the IC under review (state.connected_designators). + Subject to per-review fetch/page budget caps. + """ + if state is None: + return ("get_datasheet_excerpt called without per-review state — " + "this is a bug, no excerpt returned.", None) + + # Lazy import to avoid backend↔periscopex circular dependency at module load. + from backend.services.llm import PdfBlock + + designator = (designator or "").strip() + topic = (topic or "").strip().lower() + + if topic not in EXCERPT_TOPICS: + valid = ", ".join(sorted(EXCERPT_TOPICS.keys())) + return (f"Unknown topic '{topic}'. Valid topics: {valid}.", None) + + if designator == state.current_ic: + return ( + f"You are already reviewing {designator}'s datasheet — its pages " + f"are in your initial context. Use the existing PDF, no excerpt " + f"fetch needed.", + None, + ) + + if designator not in state.connected_designators: + return ( + f"{designator} is not a signal neighbor of {state.current_ic} " + f"in this design. The excerpt tool is restricted to ICs that " + f"share a signal net with the IC under review. If you suspect " + f"the issue still applies, submit WARNING with an explicit " + f"Unverified: assumption.", + None, + ) + + comp = graph.components.get(designator) + if comp is None: + return (f"Component '{designator}' not found in design graph.", None) + + mpn = comp.mpn or comp.value + if not mpn: + return (f"{designator} has no MPN — cannot resolve a datasheet.", None) + + # Budget checks before doing pypdf work. Three caps, in order: + # - fetch_count: total excerpt calls this review (bounds turn cost). + # - per_neighbor_page_budget: pages already pulled from THIS neighbor — + # once a neighbor is fully examined, more pages won't help. + # - page_budget: global ceiling across all neighbors (hub-IC fan-out). + # The per-neighbor cap is checked before the global one so that pulling + # the 2-3 topics needed to verify a single interface is never starved by + # pages spent on other neighbors. + neighbor_pages = state.pages_per_neighbor.get(designator, 0) + if state.fetch_count >= state.fetch_budget: + return ( + f"Excerpt budget exhausted ({state.fetch_count}/" + f"{state.fetch_budget} fetches used). Submit WARNING with an " + f"explicit Unverified: assumption rather than fetching more.", + None, + ) + if neighbor_pages >= state.per_neighbor_page_budget: + return ( + f"Per-neighbor excerpt budget for {designator} exhausted " + f"({neighbor_pages}/{state.per_neighbor_page_budget} pages). " + f"You have read enough of {designator}'s datasheet; submit " + f"WARNING with an explicit Unverified: assumption if the spec " + f"still isn't resolved.", + None, + ) + if state.page_count >= state.page_budget: + return ( + f"Excerpt page budget exhausted ({state.page_count}/" + f"{state.page_budget} pages used). Submit WARNING with an " + f"explicit Unverified: assumption rather than fetching more.", + None, + ) + + pdf_path = _resolve_neighbor_pdf(state, mpn) + if pdf_path is None: + return ( + f"No datasheet PDF available for {designator} ({mpn}). Submit " + f"WARNING with an explicit Unverified: assumption stating what " + f"you needed to verify.", + None, + ) + + # Stable cache key — md5 the source PDF once, reuse across ICs. + import hashlib + try: + ds_md5 = hashlib.md5(pdf_path.read_bytes()).hexdigest() + except Exception: + log.exception("excerpt: md5 failed for %s", pdf_path) + ds_md5 = pdf_path.name + + cache_key = (designator, topic, ds_md5) + cache_val = state.cache.get(cache_key) + pages: list[int] + trimmed_path: str + if ( + isinstance(cache_val, tuple) + and len(cache_val) == 2 + and Path(cache_val[0]).is_file() + ): + trimmed_path, pages = cache_val # type: ignore[assignment] + else: + keyword_re = EXCERPT_TOPICS[topic] + remaining_budget = min( + _EXCERPT_MAX_PAGES_PER_FETCH, + max(1, state.page_budget - state.page_count), + max(1, state.per_neighbor_page_budget - neighbor_pages), + ) + trimmed_path, pages = _trim_pdf_by_keywords( + pdf_path, keyword_re, remaining_budget, + ) + state.cache[cache_key] = (trimmed_path, pages) + + # Update per-review budget counters + state.fetch_count += 1 + state.page_count += len(pages) + state.pages_per_neighbor[designator] = neighbor_pages + len(pages) + + block = PdfBlock(path=Path(trimmed_path), cacheable=True) + summary = ( + f"Returned {len(pages)} pages from {designator} ({mpn}) matching " + f"topic '{topic}': pages {pages}. The PDF excerpt is attached to " + f"this message — read it and cite the printed page number from the " + f"original datasheet in any resulting finding. These pages are from " + f"{designator}'s datasheet (not the component under review), so set " + f"that finding's source_designator to \"{designator}\" — otherwise the " + f"page number would resolve against the wrong datasheet." + ) + return summary, block + + +# --------------------------------------------------------------------------- +# Tool schemas (for Claude API) +# --------------------------------------------------------------------------- + +FIND_CONNECTED_COMPONENTS_SCHEMA = { + "name": "find_connected_components", + "description": ( + "Find all components connected to the same net as a specific pin. " + "Returns net info and each component with full specs and pin map. " + "Use designator_filter to narrow results (e.g. 'C' for capacitors, 'R' for resistors)." + ), + "input_schema": { + "type": "object", + "properties": { + "designator": { + "type": "string", + "description": "Component reference, e.g. 'U1', 'U2'", + }, + "pin": { + "type": "string", + "description": "Pin number, e.g. '1', '7'", + }, + "designator_filter": { + "type": "string", + "description": "Optional prefix filter: 'C' for caps, 'R' for resistors, 'U' for ICs, etc.", + }, + }, + "required": ["designator", "pin"], + }, +} + +GET_NET_FOR_PIN_SCHEMA = { + "name": "get_net_for_pin", + "description": ( + "Get the net name, type, and voltage for a specific pin. " + "Lightweight — no component listing. Use for quick voltage checks." + ), + "input_schema": { + "type": "object", + "properties": { + "designator": { + "type": "string", + "description": "Component reference, e.g. 'U1'", + }, + "pin": { + "type": "string", + "description": "Pin number, e.g. '1'", + }, + }, + "required": ["designator", "pin"], + }, +} + +SHORTEST_PATH_SCHEMA = { + "name": "shortest_path", + "description": ( + "Find the shortest hop path through the netlist between two pins " + "(component.pin → nets → components). Use to verify whether two " + "pins share a rail path, or how a signal reaches another IC, " + "instead of guessing from neighborhood context." + ), + "input_schema": { + "type": "object", + "properties": { + "designator_a": { + "type": "string", + "description": "Start component reference, e.g. 'U1'", + }, + "pin_a": { + "type": "string", + "description": "Start pin number, e.g. '12'", + }, + "designator_b": { + "type": "string", + "description": "End component reference, e.g. 'U3'", + }, + "pin_b": { + "type": "string", + "description": "End pin number, e.g. '5'", + }, + }, + "required": ["designator_a", "pin_a", "designator_b", "pin_b"], + }, +} + +GET_PINTABLE_SCHEMA = { + "name": "get_pintable", + "description": ( + "Get the full pin mapping for a component: pin numbers, names, " + "net connections, and whether each pin is connected or unconnected. " + "Use when pin naming is ambiguous or to check for floating pins." + ), + "input_schema": { + "type": "object", + "properties": { + "designator": { + "type": "string", + "description": "Component reference, e.g. 'U1'", + }, + }, + "required": ["designator"], + }, +} + +SUBMIT_REVIEW_SCHEMA = { + "name": "submit_review", + "description": ( + "Submit all findings from your review. Only include issues in findings — " + "do not submit findings for things that are correct. " + "List what you checked and found OK in checked_areas." + ), + "input_schema": { + "type": "object", + "properties": { + "findings": { + "type": "array", + "description": "List of issues found. Empty array if no issues.", + "items": { + "type": "object", + "properties": { + "finding": { + "type": "string", + "description": "What you observed in the actual circuit. 1-3 sentences.", + }, + "why": { + "type": "string", + "description": "Why this matters — what the datasheet says and what could go wrong. 1-3 sentences.", + }, + "status": { + "type": "string", + "enum": ["ERROR", "WARNING", "INFO"], + "description": "ERROR: will cause malfunction. WARNING: may degrade reliability. INFO: worth noting.", + }, + "source_page": { + "type": "integer", + "description": "Datasheet page number where the requirement is stated.", + }, + "source_quote": { + "type": "string", + "description": ( + "Required for ERROR and WARNING. Exact verbatim " + "datasheet text (max ~200 chars). Periscope " + "checks it against the PDF page. Omit only if " + "the evidence is a figure/scan with no text." + ), + }, + "source_designator": { + "type": "string", + "description": ( + "Designator of the component whose datasheet " + "source_page and source_quote refer to. OMIT " + "this when the page/quote is from the component " + "you are reviewing (its own datasheet — the " + "common case). Set it ONLY when the evidence " + "came from a connected component's datasheet " + "that you fetched with get_datasheet_excerpt " + "(e.g. \"U3\"), so source_page resolves to the " + "correct datasheet." + ), + }, + "recommendation": { + "type": "string", + "description": ( + "What to change on the board or schematic. " + "Required for every finding, including INFO." + ), + }, + "action": { + "type": "string", + "description": ( + "Same as recommendation if you prefer that name. " + "Required for every finding when recommendation is empty." + ), + }, + }, + "required": ["finding", "why", "status", "source_page"], + }, + }, + "checked_areas": { + "type": "array", + "description": ( + "Areas you reviewed and found correct. Short labels, e.g. " + "'input decoupling', 'output capacitor', 'enable logic', " + "'crystal circuit', 'voltage margins', 'reset circuit'." + ), + "items": {"type": "string"}, + }, + }, + "required": ["findings", "checked_areas"], + }, +} + +GET_DATASHEET_EXCERPT_SCHEMA = { + "name": "get_datasheet_excerpt", + "description": ( + "Fetch a focused excerpt of a *connected* IC's datasheet — the pages " + "covering one topic (abs-max, electrical characteristics, 5V-tolerance, " + "etc.). Use this BEFORE flagging any cross-IC interface issue that " + "depends on the counterpart's spec. Restricted to ICs that share a " + "signal net with the IC under review. Subject to a per-review fetch " + "budget; if exhausted, submit WARNING with an explicit Unverified: " + "assumption rather than guessing." + ), + "input_schema": { + "type": "object", + "properties": { + "designator": { + "type": "string", + "description": ( + "Reference of a connected IC (e.g. 'U3'). Must be a " + "signal neighbor of the IC under review." + ), + }, + "topic": { + "type": "string", + "enum": sorted(EXCERPT_TOPICS.keys()), + "description": ( + "Which datasheet section to pull. Pick the narrowest " + "topic that covers the spec you need — pin_voltage_levels " + "for 5V-tolerance / VIH / VIL, absolute_max for stress " + "ratings, electrical_characteristics for drive " + "strengths, application_circuit for reference designs." + ), + }, + }, + "required": ["designator", "topic"], + }, +} + +GRAPH_TOOLS = [ + FIND_CONNECTED_COMPONENTS_SCHEMA, + GET_NET_FOR_PIN_SCHEMA, + SHORTEST_PATH_SCHEMA, + GET_PINTABLE_SCHEMA, + GET_DATASHEET_EXCERPT_SCHEMA, +] +ALL_TOOLS = GRAPH_TOOLS + [SUBMIT_REVIEW_SCHEMA] + + +# --------------------------------------------------------------------------- +# Dispatcher +# --------------------------------------------------------------------------- + +def execute_tool( + graph: DesignGraph, + constraints_map: ConstraintsMap, + tool_name: str, + tool_input: dict, + state: ExcerptState | None = None, +): + """Execute a graph-query tool call. + + Returns ``(text, attachment)`` where ``attachment`` is an optional + PdfBlock the caller should append to the next user message alongside the + tool_result. All tools except ``get_datasheet_excerpt`` return + ``(text, None)``. + """ + if tool_name == "find_connected_components": + return ( + find_connected_components( + graph, constraints_map, + tool_input["designator"], + tool_input["pin"], + tool_input.get("designator_filter"), + ), + None, + ) + if tool_name == "get_net_for_pin": + return ( + get_net_for_pin( + graph, constraints_map, + tool_input["designator"], + tool_input["pin"], + ), + None, + ) + if tool_name == "shortest_path": + return ( + shortest_path( + graph, constraints_map, + tool_input["designator_a"], + tool_input["pin_a"], + tool_input["designator_b"], + tool_input["pin_b"], + ), + None, + ) + if tool_name == "get_pintable": + return ( + get_pintable( + graph, constraints_map, + tool_input["designator"], + ), + None, + ) + if tool_name == "get_datasheet_excerpt": + return get_datasheet_excerpt( + graph, constraints_map, + tool_input.get("designator", ""), + tool_input.get("topic", ""), + state, + ) + return (f"Unknown tool: {tool_name}", None) diff --git a/periscope/src/backend/services/review_session.py b/periscope/src/backend/services/review_session.py new file mode 100644 index 0000000..c2db4a6 --- /dev/null +++ b/periscope/src/backend/services/review_session.py @@ -0,0 +1,526 @@ +"""Native per-IC datasheet review loop (DeepSeek via call_with_fallback). + +Replaces live calls into PinScope validate.py / validation_tools.py. +Inherited services.validation re-exports review_ic_async for monkeypatches. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import re +import tempfile +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Awaitable, Callable + +log = logging.getLogger(__name__) + +from backend.config import settings +from backend.periscopex.models import DesignGraph, NetType +from backend.periscopex.quote_verify import verify_finding_citations +from backend.periscopex.review_context import build_component_context +from backend.periscopex.review_parse import ( + SYSTEM_PROMPT, + ReviewResult, + _MAX_REVIEW_TURNS, + parse_submit_review, +) +from backend.periscopex.review_tools import ( + ALL_TOOLS, + SUBMIT_REVIEW_SCHEMA, + ConstraintsMap, + ExcerptState, + execute_tool, +) +from backend.periscopex.utils import safe_mpn +from backend.services.api_logs import ApiLogger +from backend.services.llm import ( + Message, + PdfBlock, + TextBlock, + ToolResultBlock, + ToolSchema, + call_with_fallback, +) +from backend.services.normalize_findings import normalize_findings_async + +TRACE_VERSION = 1 + +ProgressCallback = Callable[[str, int, str, str], Awaitable[None]] + + +def _assistant_text(blocks) -> str: + """Best-effort extraction of text content from a completion's raw + assistant blocks. Provider-agnostic and never raises.""" + parts: list[str] = [] + try: + for b in blocks or []: + txt = getattr(b, "text", None) + if txt is None and isinstance(b, dict): + txt = b.get("text") if b.get("type") == "text" else None + elif getattr(b, "type", None) not in (None, "text"): + txt = None + if isinstance(txt, str) and txt: + parts.append(txt) + except Exception: + log.exception("trace: assistant_text extraction failed") + return "\n".join(parts) + + +def _to_tool_schema(d: dict) -> ToolSchema: + return ToolSchema( + name=d["name"], + description=d["description"], + input_schema=d["input_schema"], + ) + + +_ALL_TOOL_SCHEMAS = [_to_tool_schema(t) for t in ALL_TOOLS] +_SUBMIT_TOOL_SCHEMA = _to_tool_schema(SUBMIT_REVIEW_SCHEMA) + +_REVIEW_KEYWORDS = re.compile( + r"pin\s+(out|diagram|configuration|description|assignment|function|name|table|map)" + r"|ball\s+map|package\s+(pin|drawing|outline)|signal\s+description" + r"|absolute\s+maximum|recommended\s+operating|electrical\s+characteristics" + r"|power\s+supply|thermal\s+(resistance|shutdown|pad)|ESD\s+(rating|tolerance)" + r"|decoupling|bypass\s+capacitor|layout\s+(guideline|recommendation)" + r"|application\s+(circuit|schematic|information|note)" + r"|typical\s+application|reference\s+design", + re.IGNORECASE, +) + +_MAX_PDF_PAGES = 120 + +# Per-review excerpt budget — keeps fan-out cost bounded on hub ICs (e.g. an +# MCU connected to many neighbors). On exhaustion, the tool returns a budget +# message and the model is steered to submit WARNING with Unverified: +# assumption rather than fetching more. +# +# The global page budget got raised from 25→60 and gained a per-neighbor +# sub-budget after the U2-001 / U3-001 false positives: a single 25-page +# global cap was exhausted by one neighbor's pin_voltage_levels excerpt +# before the abs-max table could be read, so the reviewer was forced to +# guess at the very moment it was trying to verify a damage claim. 30 pages +# per neighbor fits the ~3 topic fetches (pin levels + abs-max + electrical) +# one interface check needs; 60 global allows ~2 such neighbors before the +# fan-out ceiling kicks in. +_PER_REVIEW_FETCH_BUDGET = 12 +_PER_REVIEW_PAGE_BUDGET = 90 +_PER_NEIGHBOR_PAGE_BUDGET = 45 + +# A signal net with more components than this is treated as a hub/bus and +# excluded from the neighbor set even if classified as "signal". Bounds +# fan-out on designs that use an oversized common signal (rare but possible). +_SIGNAL_NET_MAX_COMPONENTS = 8 + + +def _signal_neighbors(graph: DesignGraph, ic_ref: str) -> set[str]: + """Return the set of designators that share at least one *signal* net + with ``ic_ref``. Excludes power/ground rails (which connect every IC and + would otherwise fan the neighbor set out across the whole design) and + excludes the IC under review itself. + """ + comp = graph.components.get(ic_ref) + if not comp: + return set() + neighbors: set[str] = set() + for net_name in set(comp.pins.values()): + net = graph.nets.get(net_name) + if not net: + continue + if net.net_type in (NetType.POWER, NetType.GROUND): + continue + refs_on_net = {pc.component_ref for pc in net.pins} + if len(refs_on_net) > _SIGNAL_NET_MAX_COMPONENTS: + continue + for ref in refs_on_net: + if ref != ic_ref: + neighbors.add(ref) + return neighbors + + +def _select_review_pages(pdf_path: str) -> str: + """Trim a datasheet PDF to pages relevant for design review. + + Returns path to trimmed PDF (or original if already small enough). + + Note: the reviewer cites the datasheet's *printed* page number (read from + the page content/footer), not the page's physical position in the trimmed + file — so `source_page` already matches the full original PDF the frontend + serves. No trimmed→original remap is applied (an earlier remap attempt + corrupted correct citations on large datasheets). + """ + from pypdf import PdfReader, PdfWriter + + reader = PdfReader(pdf_path) + total = len(reader.pages) + if total <= _MAX_PDF_PAGES: + return pdf_path + + # Always keep first 5 pages (title, TOC, overview) + keep: set[int] = set(range(min(5, total))) + + # Keyword-matched pages + neighbors + for i, page in enumerate(reader.pages): + text = page.extract_text() or "" + if _REVIEW_KEYWORDS.search(text): + for neighbor in (i - 1, i, i + 1): + if 0 <= neighbor < total: + keep.add(neighbor) + + # Pad from front if under budget + if len(keep) < _MAX_PDF_PAGES: + for i in range(total): + if len(keep) >= _MAX_PDF_PAGES: + break + keep.add(i) + + selected = sorted(keep)[:_MAX_PDF_PAGES] + + writer = PdfWriter() + for i in selected: + writer.add_page(reader.pages[i]) + + tmp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) + writer.write(tmp) + tmp.close() + return tmp.name + + +# --------------------------------------------------------------------------- +# Per-IC async review +# --------------------------------------------------------------------------- + + +async def review_ic_async( + graph: DesignGraph, + constraints_map: ConstraintsMap, + ic_ref: str, + pdf_path: str | None, + on_progress: ProgressCallback | None = None, + api_logger: ApiLogger | None = None, + trace_git_commit: str = "unknown", + pdf_dir: Path | None = None, + storage=None, + excerpt_cache: dict | None = None, + extra_context: str = "", + system_prompt: str | None = None, + log_stage: str = "review", +) -> tuple[ReviewResult, dict]: + """Review one IC against its datasheet. Async, multi-turn. + + ``pdf_path`` may be ``None`` when the caller already has library + extraction (PCB layout-only exam) — no PDF is attached and citations + are not re-verified against a PDF. + """ + comp = graph.components[ic_ref] + mpn = comp.mpn or comp.value + + # Datasheet identity for the trace — hash the original PDF, not the + # trimmed copy, so the reference is stable across trim-heuristic changes. + ds_md5 = None + if pdf_path: + try: + ds_md5 = hashlib.md5(Path(pdf_path).read_bytes()).hexdigest() + except Exception: + log.exception("trace: datasheet md5 failed for %s", ic_ref) + + # Pre-compute which designators the excerpt tool will accept for this + # review (neighbors via signal nets only — power/GND fan-out filtered). + connected_designators = _signal_neighbors(graph, ic_ref) + + # Designator -> MPN, so a finding citing a neighbor's datasheet excerpt + # (source_designator) is referenced against — and viewed from — that + # neighbor's datasheet rather than this IC's. + mpn_by_designator = { + ref: comp.mpn + for ref, comp in graph.components.items() + if comp.mpn + } + + # Build the per-review state for the excerpt tool. ``cache`` is shared + # across ICs in the same validate_design_async run so symmetric checks + # (U2 fetches U3@abs_max, then U3 fetches U2@abs_max) don't redo pypdf + # work. + excerpt_state = ExcerptState( + current_ic=ic_ref, + connected_designators=connected_designators, + graph=graph, + pdf_dir=pdf_dir or (Path(pdf_path).parent if pdf_path else Path(".")), + storage=storage, + cache=excerpt_cache if excerpt_cache is not None else {}, + fetch_budget=_PER_REVIEW_FETCH_BUDGET, + page_budget=_PER_REVIEW_PAGE_BUDGET, + per_neighbor_page_budget=_PER_NEIGHBOR_PAGE_BUDGET, + ) + + # Trim PDF up-front — both primary and fallback attempts share it. + trimmed_pdf = _select_review_pages(pdf_path) if pdf_path else None + try: + async def _run(provider, model) -> tuple[ReviewResult, dict]: + t0 = time.monotonic() + total_input = 0 + total_output = 0 + total_cache_creation = 0 + total_cache_read = 0 + turns = 0 + + session = await provider.create_session( + model=model, + system=system_prompt or SYSTEM_PROMPT, + # Gemini 2.5/3 thinking models count thoughts against this cap. + # 4096 was too tight: U3 (largest IC) burned the entire budget + # on thinking and emitted zero visible output, dropping its + # review silently. + max_tokens=32768, + # Deterministic sampling: same inputs → same findings across + # reruns. The default temperature of 1.0 caused identical + # netlists to produce very different reports (different + # findings + severities) run-to-run. + temperature=0.0, + ) + try: + context = build_component_context(graph, constraints_map, ic_ref) + user_text = f"Review this component's usage:\n\n{context}" + if extra_context.strip(): + user_text += "\n\n" + extra_context.strip() + + user_blocks: list = [] + if trimmed_pdf: + user_blocks.append(PdfBlock(path=Path(trimmed_pdf), cacheable=True)) + user_blocks.append(TextBlock(text=user_text, cacheable=True)) + initial_msg = Message( + role="user", + content=user_blocks, + ) + messages: list[Message] = [initial_msg] + + trace: dict = { + "trace_version": TRACE_VERSION, + "ic_ref": ic_ref, + "mpn": mpn, + "model": model, + "provider": provider.name, + "git_commit": trace_git_commit, + "datasheet": {"md5": ds_md5, "safe_mpn": safe_mpn(mpn)}, + "timestamp": datetime.now(timezone.utc).isoformat(), + "max_turns": _MAX_REVIEW_TURNS, + "turns": [], + "final_submission": None, + "result": None, + "stop_reason": None, + "error": None, + "duration_ms": None, + } + + # Set after a turn produces zero tool calls (model wrote + # text only). Next turn is forced to submit_review so any + # findings drafted as prose still make it to the report. + force_submit_next_turn = False + + for turn in range(_MAX_REVIEW_TURNS): + is_last_turn = turn == _MAX_REVIEW_TURNS - 1 + + if is_last_turn or force_submit_next_turn: + tools = [_SUBMIT_TOOL_SCHEMA] + tool_choice: dict | str = {"name": "submit_review"} + else: + tools = _ALL_TOOL_SCHEMAS + tool_choice = "auto" + + if on_progress: + await on_progress( + ic_ref, turn, "waiting", + f"model turn {turn + 1}/{_MAX_REVIEW_TURNS}", + ) + + completion = await session.complete( + messages=messages, + tools=tools, + tool_choice=tool_choice, + ) + turns += 1 + total_input += completion.usage.input_tokens + total_output += completion.usage.output_tokens + total_cache_creation += completion.usage.cache_creation_tokens + total_cache_read += completion.usage.cache_read_tokens + + turn_record: dict = { + "index": turn, + "assistant_text": _assistant_text( + completion.raw_assistant_blocks + ), + "tool_calls": [], + "usage": { + "input_tokens": completion.usage.input_tokens, + "output_tokens": completion.usage.output_tokens, + "cache_creation_tokens": completion.usage.cache_creation_tokens, + "cache_read_tokens": completion.usage.cache_read_tokens, + }, + } + try: + trace["turns"].append(turn_record) + except Exception: + log.exception("trace: turn append failed for %s", ic_ref) + + # Check for submit_review + for tc in completion.tool_calls: + if tc.name == "submit_review": + result = parse_submit_review( + tc.input, ic_ref, mpn, + mpn_by_designator=mpn_by_designator, + connected=connected_designators, + ) + if pdf_path: + verify_finding_citations( + result.findings, + default_pdf=Path(pdf_path), + default_mpn=mpn, + pdf_dir=excerpt_state.pdf_dir, + mpn_by_designator=mpn_by_designator, + ) + turn_record["tool_calls"].append({ + "name": "submit_review", + "input": tc.input, + "output": None, + "duration_ms": None, + }) + trace["final_submission"] = tc.input + trace["stop_reason"] = "submit_review" + trace["result"] = { + "findings_count": len(result.findings), + "checked_areas": result.checked_areas, + } + trace["duration_ms"] = int((time.monotonic() - t0) * 1000) + if on_progress: + await on_progress( + ic_ref, turn, "submit_review", + f"{len(result.findings)} findings", + ) + if api_logger: + api_logger.log( + stage=log_stage, identifier=ic_ref, + model=model, provider=provider.name, + input_tokens=total_input, output_tokens=total_output, + cache_creation_input_tokens=total_cache_creation, + cache_read_input_tokens=total_cache_read, + duration_ms=int((time.monotonic() - t0) * 1000), + stop_reason="submit_review", turns=turns, + ) + if settings.normalize_findings_enabled: + try: + normalized, norm_trace = await normalize_findings_async( + ic_ref, mpn, result.findings, + api_logger=api_logger, + on_progress=on_progress, + ) + trace["normalize"] = norm_trace + result.findings = normalized + trace["result"]["findings_count"] = len(normalized) + except Exception: + log.exception( + "normalize: unexpected failure for %s " + "— keeping reviewer findings", + ic_ref, + ) + return result, trace + + # Process graph tool calls + tool_results: list[ToolResultBlock] = [] + attached_pdfs: list[PdfBlock] = [] + for tc in completion.tool_calls: + _tc_t0 = time.monotonic() + result_text, attachment = execute_tool( + graph, constraints_map, tc.name, tc.input, + state=excerpt_state, + ) + turn_record["tool_calls"].append({ + "name": tc.name, + "input": tc.input, + "output": result_text, + "duration_ms": int((time.monotonic() - _tc_t0) * 1000), + }) + if on_progress: + await on_progress( + ic_ref, turn, tc.name, json.dumps(tc.input), + ) + tool_results.append(ToolResultBlock( + tool_use_id=tc.id, + name=tc.name, + content=result_text, + )) + if attachment is not None: + attached_pdfs.append(attachment) + + if not tool_results: + # Model emitted text but called no tools. This is a + # known failure mode (esp. with reasoning models) + # where the model writes findings as a JSON code + # block in prose instead of calling submit_review. + # Don't drop the work — append a nudge and force + # submit_review on the next iteration. + if not is_last_turn and not force_submit_next_turn: + messages.append(Message( + role="assistant", + content=completion.raw_assistant_blocks, + )) + messages.append(Message( + role="user", + content=[TextBlock( + text=( + "You produced text but did not call " + "any tool. Findings only reach the " + "report when submitted via the " + "submit_review tool — text JSON is " + "ignored. Call submit_review now with " + "the findings you identified (or an " + "empty findings array if none) and " + "your checked_areas list." + ), + )], + )) + force_submit_next_turn = True + continue + break + + # Reset recovery flag once the model is calling tools again. + force_submit_next_turn = False + + messages.append(Message(role="assistant", content=completion.raw_assistant_blocks)) + # tool_result blocks first, then any PdfBlocks the tools + # attached (excerpt fetches). The Anthropic provider + # encodes each block independently — mixed-block user + # messages are supported and the cached initial PDF is + # not invalidated by appending uncached/cached content. + messages.append(Message( + role="user", + content=[*tool_results, *attached_pdfs], + )) + + # Fell through without submitting + trace["stop_reason"] = "no_submission" + trace["result"] = {"findings_count": 0, "checked_areas": []} + trace["duration_ms"] = int((time.monotonic() - t0) * 1000) + if api_logger: + api_logger.log( + stage=log_stage, identifier=ic_ref, + model=model, provider=provider.name, + input_tokens=total_input, output_tokens=total_output, + cache_creation_input_tokens=total_cache_creation, + cache_read_input_tokens=total_cache_read, + duration_ms=int((time.monotonic() - t0) * 1000), + stop_reason="no_submission", turns=turns, + ) + return ReviewResult([], []), trace + finally: + await session.close() + + return await call_with_fallback("validation", _run) + finally: + if trimmed_pdf and pdf_path and trimmed_pdf != pdf_path: + Path(trimmed_pdf).unlink(missing_ok=True) + + diff --git a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md b/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md index d24235d..9792ab5 100644 --- a/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md +++ b/periscope/src/docs/development/PINSCOPE_INDEPENDENCE_PLAN.md @@ -1,7 +1,8 @@ # Piano — indipendenza architettonica e di licenza da PinScope -**Stato:** split fisico shipped in **2.38.0**. Fork GitHub non staccato; `validate.py` non riscritto. -**Sequenza confermata:** (1) split `periscope/src` vs `periscope/dependency/` — **questa fase**; (2) **poi** sostituzione incrementale dei moduli PinScope (Fase C). **Mai** svuotare o cancellare l’albero ereditato. +**Stato:** split fisico **2.38.0**. Fase C2 native review loop **2.39.0** (pytest verde; `validate.py` / `validation_tools.py` restano in `dependency/` come fallback `_inherited_review_ic_async`). Fork GitHub 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 loop → C3 extraction → C4 PCB off `validate.py`). **Mai** empty-delete. Parsers/graph (C5/E) e auto-place fuori scope. AGPL resta. Questo piano **non** stacca il fork GitHub (`manvalan/periscope` ← `Faradworks/Pinscope`). Lo stacco è un passo legale successivo, fuori da queste fasi di lavoro, salvo decisione esplicita. @@ -58,6 +59,7 @@ Trattare come **una dipendenza in-tree**, non come prodotto Periscope: | Facade Z0 | `periscope/src/backend/periscopex/impedance.py` | NEW; il solver è terzi | | Auth self-host | `periscope/src/backend/services/local_jwt.py`, `routers/auth.py` | REPLACEMENT | | LLM DeepSeek | `periscope/src/backend/services/llm/deepseek_provider.py`, `local_skill.py`, `pdf_ingest.py` | NEW | +| Review loop C2 | `review_session.py`, `review_parse.py`, `review_tools.py`, `review_context.py`, `constraints_lookup.py` | REPLACEMENT 2.39.0; PinScope files kept | | 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 | @@ -222,7 +224,7 @@ Qui sì si riscrive l’engine ereditato. Ordine interno: | --- | --- | --- | | C0 | Spec freeze `motore-finding.md` = unico IR | Già nativo; non diluire in `validate.py` | | C1 | Adapter: `validate.py` emette solo `Finding` grezzi → `complete_finding` | Già parziale; chiudere i campi doppi | -| C2 | **REWRITE** loop per-IC: tool `find_connected_components` / `get_pintable` / excerpt su provider DeepSeek **senza** copiare `validation_tools.py` | Cuore DIRECT da sostituire | +| C2 | **REWRITE** loop per-IC native in `periscope/src` (DeepSeek, tools su `DesignGraph`) | **Shipped 2.39.0** live `review_ic_async` → `review_session`; PinScope files **kept**, not called from live loop | | C3 | Extraction: `local_skill.py` + schemi JSON (KEEP schema se identici; REWRITE orchestrazione Anthropic) | | | C4 | Spegnere import da `validate.py` nel PCB (`_parse_review` → parser finding nativo) | Toglie INDIRECT PCB→PinScope reviewer | | C5 | Test golden `simple_project` + Emmaforo: parity FACT/REQUIREMENT, non parity prose | | diff --git a/tests/test_native_review_loop.py b/tests/test_native_review_loop.py new file mode 100644 index 0000000..6f71409 --- /dev/null +++ b/tests/test_native_review_loop.py @@ -0,0 +1,115 @@ +"""Prove native review parse/tools match inherited PinScope before switching the live loop.""" + +from __future__ import annotations + +import json + +from backend.periscopex.constraints_lookup import match_constraints +from backend.periscopex.finding_engine import complete_finding +from backend.periscopex.models import DesignGraph +from backend.periscopex.review_parse import parse_submit_review +from backend.periscopex.validate import _load_datasheets, _match_constraints, _parse_review +from backend.periscopex.validation_tools import execute_tool as inherited_execute +from backend.periscopex.review_tools import execute_tool as native_execute +from tests.paths import SIMPLE_PROJECT + + +_PAYLOAD = { + "findings": [ + { + "finding": "FB5 floating", + "why": "FB5 is NC; DCDC5 SW is loaded.", + "status": "ERROR", + "source_page": 12, + "source_quote": "Connect FB5 to the output sense node.", + "recommendation": "Connect FB5.", + }, + { + "finding": "no quote", + "why": "maybe missing cap", + "status": "ERROR", + "source_page": 3, + "source_quote": "", + }, + ], + "checked_areas": ["power", "decoupling"], +} + + +def test_native_parse_matches_inherited(): + inherited = _parse_review(_PAYLOAD, "U16", "AXP2101") + native = parse_submit_review(_PAYLOAD, "U16", "AXP2101") + assert len(native.findings) == len(inherited.findings) + assert native.checked_areas == inherited.checked_areas + for a, b in zip(native.findings, inherited.findings, strict=True): + assert a.model_dump() == b.model_dump() + + +def test_native_llm_finding_is_review_and_recommended_not_error(): + native = parse_submit_review(_PAYLOAD, "U16", "AXP2101") + f = native.findings[0] + complete_finding(f) + assert f.finding_class == "REVIEW" + assert f.facts == "FB5 floating" + assert f.requirement + assert f.status != "ERROR" + f2 = native.findings[1] + complete_finding(f2) + assert f2.status != "ERROR" + assert f2.why.startswith("Unverified:") + + +def test_match_constraints_matches_inherited(): + datasheets = _load_datasheets(SIMPLE_PROJECT / "extracted") + if not datasheets: + datasheets = _load_datasheets(SIMPLE_PROJECT / "datasheets" / "extracted") + mpns = list(datasheets)[:5] or ["MSPM0G3507SPTR"] + for mpn in mpns: + assert match_constraints(mpn, datasheets) == _match_constraints(mpn, datasheets) + assert match_constraints(None, datasheets) is None + + +def test_native_graph_tools_match_inherited_on_simple_project(): + graph = DesignGraph.model_validate( + json.loads((SIMPLE_PROJECT / "design_graph.json").read_text()) + ) + cmap = {} + native_txt, _ = native_execute( + graph, cmap, "get_net_for_pin", {"designator": "U3", "pin": "1"}, + ) + inherited_txt, _ = inherited_execute( + graph, cmap, "get_net_for_pin", {"designator": "U3", "pin": "1"}, + ) + assert native_txt == inherited_txt + n2, _ = native_execute( + graph, cmap, "find_connected_components", + {"designator": "U3", "pin": "1", "designator_filter": "C"}, + ) + i2, _ = inherited_execute( + graph, cmap, "find_connected_components", + {"designator": "U3", "pin": "1", "designator_filter": "C"}, + ) + assert n2 == i2 + + +def test_native_session_does_not_import_pinscope_loop(): + import ast + from pathlib import Path + + import backend.services.review_session as rs + + tree = ast.parse(Path(rs.__file__).read_text()) + imported = [ + node.module + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) and node.module + ] + assert "backend.periscopex.validate" not in imported + assert "backend.periscopex.validation_tools" not in imported + assert "backend.services.llm" in imported + assert "backend.periscopex.review_tools" in imported + assert "backend.periscopex.review_parse" in imported + + from backend.services import validation as val + import backend.services.review_session as rs + assert val.review_ic_async is rs.review_ic_async diff --git a/tests/test_validation_no_tool_recovery.py b/tests/test_validation_no_tool_recovery.py index efc59fc..ddc5a9c 100644 --- a/tests/test_validation_no_tool_recovery.py +++ b/tests/test_validation_no_tool_recovery.py @@ -18,6 +18,7 @@ import pytest from pypdf import PdfWriter from backend.periscopex.utils import safe_mpn +from backend.services import review_session as review_session from backend.services import validation as val from backend.services.llm.types import Completion, ToolCall, Usage from backend.services.storage import LocalStorageBackend @@ -124,6 +125,7 @@ async def test_no_tool_calls_triggers_forced_submit_next_turn(workspace, monkeyp return await body(_Provider(), "fake-model") monkeypatch.setattr(val, "call_with_fallback", fake_cwf) + monkeypatch.setattr(review_session, "call_with_fallback", fake_cwf) orig = val.review_ic_async diff --git a/tests/test_validation_trace_log.py b/tests/test_validation_trace_log.py index 55f13e6..c0b360e 100644 --- a/tests/test_validation_trace_log.py +++ b/tests/test_validation_trace_log.py @@ -18,6 +18,7 @@ import pytest from pypdf import PdfWriter from backend.periscopex.utils import safe_mpn +from backend.services import review_session as review_session from backend.services import validation as val from backend.services.llm.types import Completion, ToolCall, Usage from backend.services.storage import LocalStorageBackend @@ -136,6 +137,7 @@ async def _run(ws, monkeypatch, before_ic): return await body(prov_cls(), "fake-model") monkeypatch.setattr(val, "call_with_fallback", fake_cwf) + monkeypatch.setattr(review_session, "call_with_fallback", fake_cwf) # The fake session needs the current ic_ref; thread it through by wrapping # review_ic_async to set the class attr before delegating.