Native DeepSeek review loop in periscope/src (Fase C2).
Live review_ic_async comes from review_session; PinScope validate.py and validation_tools.py stay in dependency as _inherited_review_ic_async fallback.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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: <which of (1)-(4) you could not establish>`. \
|
||||
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 <n>` 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 <part> on pin <n> 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: <what you could not \
|
||||
establish in three queries>` 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 <net> 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: <the assumption you had to make>`. 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 <ref>'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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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 | |
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user