"""Evidence-bounded AI assistance primitives for Phase 13. This module deliberately has no network or model dependency. The local provider is an auditable formatter over stored records; other providers are reported as not_configured rather than guessed at. """ from __future__ import annotations import hashlib import json import os import re from typing import Any MAX_INPUT_ITEMS = 100 MAX_FIELD_CHARS = 500 MAX_OUTPUT_CHARS = 12_000 SUPPORTED_KINDS = ("summary", "qualification_explanation", "missing_data_questions", "research_note") _SECRET_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)\s*[:=]\s*[^\s,;]+") def _text(value: Any, limit: int = MAX_FIELD_CHARS) -> str: value = "" if value is None else str(value) value = _SECRET_RE.sub(r"\1: [REDACTED]", value) return value[:limit] def redact(value: Any) -> Any: if isinstance(value, dict): return {str(k)[:80]: ("[REDACTED]" if re.search(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)", str(k)) else redact(v)) for k, v in list(value.items())[:100]} if isinstance(value, list): return [redact(v) for v in value[:MAX_INPUT_ITEMS]] if isinstance(value, str): return _text(value) return value def evidence_hashes(evidence: list[dict[str, Any]]) -> list[str]: return [hashlib.sha256(json.dumps(redact(item), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() for item in evidence] def _citation(item: dict[str, Any]) -> dict[str, Any]: return {"evidence_id": int(item["id"]), "kind": _text(item.get("kind", "evidence"), 80), "url": _text(item.get("url", ""), 500)} def _claim(item: dict[str, Any]) -> str: return _text(item.get("claim", ""), MAX_FIELD_CHARS).strip() def build_local_suggestions(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> dict[str, Any]: """Create deterministic suggestions using only supplied stored data. Every claim-bearing item cites one or more rows from ``evidence``. No contact details are emitted, and contacts are used only as aggregate counts. """ evidence = [redact(x) for x in evidence[:MAX_INPUT_ITEMS] if _claim(x)] citations = [_citation(x) for x in evidence] claims = [_claim(x) for x in evidence] suggestions: list[dict[str, Any]] = [] name = _text(business.get("name", "this business"), 200) if claims: joined = " ".join(f"{claim} [evidence:{item['id']}]" for claim, item in zip(claims[:5], evidence[:5])) suggestions.append({"type": "summary", "text": f"Stored evidence for {name}: {joined}", "citations": citations[:5]}) score = business.get("score") if score is not None: suggestions.append({"type": "qualification_explanation", "text": f"The stored qualification score is {_text(score, 30)}; review the cited evidence before relying on it. [evidence:{evidence[0]['id']}]", "citations": citations[:1]}) missing = [] if not _text(business.get("website", "")).strip(): missing.append("official website") if not contacts: missing.append("public contact evidence") if missing: suggestions.append({"type": "missing_data_questions", "text": "Confirm whether the following data is available: " + ", ".join(missing) + f". [evidence:{evidence[0]['id']}]", "citations": citations[:1]}) suggestions.append({"type": "research_note", "text": f"Draft note: independently verify the stored claims for {name}; do not infer facts beyond the cited records. [evidence:{evidence[0]['id']}]", "citations": citations[:1]}) else: # No claim is fabricated. A question is safe but has no citation, so # return no suggestions and let the caller expose the missing-data state. suggestions = [] output = {"provider": "local", "version": "deterministic-v1", "suggestions": suggestions, "grounded": True, "claim_policy": "stored_evidence_only"} encoded = json.dumps(output, sort_keys=True, ensure_ascii=False) return json.loads(encoded[:MAX_OUTPUT_CHARS]) if len(encoded) <= MAX_OUTPUT_CHARS else {"provider": "local", "version": "deterministic-v1", "suggestions": suggestions[:1], "grounded": True, "claim_policy": "stored_evidence_only"} def provider_name() -> str | None: value = os.environ.get("AI_PROVIDER", "").strip().lower() return value or None def generate(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> tuple[str, str, str, dict[str, Any]]: provider = provider_name() hashes = evidence_hashes(evidence) metadata = {"input_counts": {"business": 1, "scans": min(len(scans), MAX_INPUT_ITEMS), "contacts": min(len(contacts), MAX_INPUT_ITEMS), "evidence": min(len(evidence), MAX_INPUT_ITEMS)}, "redacted": True, "max_input_items": MAX_INPUT_ITEMS, "max_field_chars": MAX_FIELD_CHARS, "evidence_hashes": hashes} if provider not in {"local", "deterministic"}: return "not_configured", provider or "", "", metadata return "succeeded", "local", "deterministic-v1", {**metadata, "output": build_local_suggestions(business, scans, contacts, evidence, score_history)}