diff --git a/apps/api/app/ai_assistance.py b/apps/api/app/ai_assistance.py index 0afec45..528a1dc 100644 --- a/apps/api/app/ai_assistance.py +++ b/apps/api/app/ai_assistance.py @@ -1,8 +1,8 @@ -"""Evidence-bounded AI assistance primitives for Phase 13. +"""Evidence-bounded AI enrichment with a deterministic, network-free provider. -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. +Remote providers are intentionally only a status/configuration concept here. No +network client is present, so an unconfigured or unreviewed remote provider fails +closed and cannot accidentally perform outreach or exfiltrate tenant data. """ from __future__ import annotations @@ -15,68 +15,70 @@ 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,;]+") +_SECRET_KEY_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)") 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] + return _SECRET_RE.sub(r"\1: [REDACTED]", "" if value is None else str(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 {str(k)[:80]: ("[REDACTED]" if _SECRET_KEY_RE.search(str(k)) else redact(v)) for k, v in list(value.items())[:MAX_INPUT_ITEMS]} + if isinstance(value, list): return [redact(v) for v in value[:MAX_INPUT_ITEMS]] + if isinstance(value, str): return _text(value) return value +def _hash(item: Any) -> str: + return hashlib.sha256(json.dumps(redact(item), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() + + 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] + return [_hash(item) for item in evidence[:MAX_INPUT_ITEMS]] -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 input_fingerprint(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], max_items: int = MAX_INPUT_ITEMS) -> str: + payload = {"business": redact(business), "scans": [redact(x) for x in scans[:max_items]], "contacts": [redact(x) for x in contacts[:max_items]], "evidence": [redact(x) for x in evidence[:max_items]]} + return _hash(payload) -def _claim(item: dict[str, Any]) -> str: - return _text(item.get("claim", ""), MAX_FIELD_CHARS).strip() +def _citation(item: dict[str, Any], source_type: str, provenance: str) -> dict[str, Any]: + return {"source_type": source_type, "source_id": int(item["id"]) if str(item.get("id", "")).isdigit() else None, + "kind": _text(item.get("kind", source_type), 80), "url": _text(item.get("url", item.get("input_url", item.get("source_url", ""))), 500), + "provenance": provenance, "hash": _hash(item), "policy": "stored_evidence_only"} 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]] = [] + evidence = [redact(x) for x in evidence[:MAX_INPUT_ITEMS] if _text(x.get("claim", "")).strip()] + scans = [redact(x) for x in scans[:MAX_INPUT_ITEMS]] + contacts = [redact(x) for x in contacts[:MAX_INPUT_ITEMS]] + citations = [_citation(x, "evidence", "discovery_evidence") for x in evidence] + citations += [_citation(x, "website_scan", "website_scanner") for x in scans] + citations += [_citation(x, "contact_extraction", "contact_extractor") for x in contacts] + claims = [_text(x.get("claim", "")).strip() for x in evidence] 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"} + score = business.get("score") + priority = ("high" if score is not None and int(score) >= 70 else "medium" if score is not None and int(score) >= 40 else "low") + uncertainty = [] + if not claims: uncertainty.append("no_claim_bearing_evidence") + if not contacts: uncertainty.append("no_public_contact_extraction") + conflicts = [] + classifications = [str(x.get("classification", "")).lower() for x in scans if x.get("classification")] + if len(set(classifications)) > 1: conflicts.append("website_scan_classifications_disagree") + classification = "business_prospect" if claims or business.get("website") else "insufficient_evidence" + summary = f"Stored evidence for {name}: " + (" ".join(f"{claim} [evidence:{item['source_id']}]" for claim, item in zip(claims[:5], citations[:5])) if claims else "insufficient claim-bearing evidence") + output = { + "classification": classification, "summary": _text(summary, 2000), + "priority_recommendation": priority, "confidence": round(min(0.95, 0.45 + 0.1 * len(claims) + (0.1 if scans else 0)), 2), + "uncertainty": uncertainty, "conflicts": conflicts, "citations": citations[:MAX_INPUT_ITEMS], + "policy": {"claim_policy": "stored_evidence_only", "no_outreach": True, "redacted_inputs": True}, + "suggestions": ([{"type": "summary", "text": _text(summary), "citations": citations[:5]}] if claims else []), + "grounded": True, "provider": "local", "version": "deterministic-v2", + } 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"} + return output if len(encoded) <= MAX_OUTPUT_CHARS else {**output, "suggestions": output["suggestions"][:1], "citations": citations[:20]} def provider_name() -> str | None: @@ -84,10 +86,14 @@ def provider_name() -> str | None: return value or None +def provider_status(configured: str | None = None) -> dict[str, Any]: + provider = configured if configured is not None else provider_name() + local = provider in {"local", "deterministic"} + return {"provider": provider or "", "status": "ready" if local else "not_configured", "network_enabled": False, "reviewed": local, "outbound_calls": False} + + 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)} + 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": evidence_hashes(evidence), "input_fingerprint": input_fingerprint(business, scans, contacts, evidence)} + if provider not in {"local", "deterministic"}: return "not_configured", provider or "", "", metadata + return "succeeded", "local", "deterministic-v2", {**metadata, "output": build_local_suggestions(business, scans, contacts, evidence, score_history)} diff --git a/apps/api/app/main.py b/apps/api/app/main.py index ee6f38b..3ac21b6 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -13,7 +13,7 @@ if __package__ in (None, ""): from app.website_scanner import scan_website, validate_url from app.contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS from app.scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION - from app.ai_assistance import generate as generate_ai, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS + from app.ai_assistance import generate as generate_ai, input_fingerprint, provider_status, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS from app.discovery import discover as scoped_discover from app.config import load_config else: @@ -23,7 +23,7 @@ else: from .website_scanner import scan_website, validate_url from .contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS from .scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION - from .ai_assistance import generate as generate_ai, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS + from .ai_assistance import generate as generate_ai, input_fingerprint, provider_status, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS from .discovery import discover as scoped_discover from .config import load_config ORGANIZATION_ID = "demo-tenant" @@ -472,6 +472,40 @@ class ApiHandler(BaseHTTPRequestHandler): reasons.append("network_send_disabled"); self.audit(db, user, "outreach_draft.send_blocked", f"{did}:network_send_disabled"); db.commit() return self.send_json(409, {"status": "blocked", "blocked_reasons": reasons, "network_send": False, "id": did}) + def ai_provider_config(self, db, user, payload=None): + org = user["organization_id"] + row = db.execute("SELECT * FROM ai_provider_configs WHERE organization_id=?", (org,)).fetchone() + if payload is not None: + provider = str(payload.get("provider", "local")).strip().lower() + if not provider or len(provider) > 80: return self.send_json(400, {"error": "invalid_provider"}) + reviewed = bool(payload.get("reviewed", False)) + enabled = bool(payload.get("enabled", True)) + # A remote provider cannot be enabled by configuration alone: this + # service has no reviewed transport abstraction and never sends data. + if provider not in {"local", "deterministic"} and (enabled or reviewed): + return self.send_json(409, {"error": "provider_not_reviewed", "network_enabled": False}) + db.execute("INSERT INTO ai_provider_configs(organization_id,provider,enabled,reviewed) VALUES(?,?,?,?) ON CONFLICT(organization_id) DO UPDATE SET provider=excluded.provider,enabled=excluded.enabled,reviewed=excluded.reviewed,updated_at=CURRENT_TIMESTAMP", (org, provider, int(enabled), int(reviewed))) + self.audit(db, user, "ai_provider.updated", provider); db.commit() + row = db.execute("SELECT * FROM ai_provider_configs WHERE organization_id=?", (org,)).fetchone() + configured = row["provider"] if row and row["enabled"] else None + result = provider_status(configured if configured is not None else None) + result.update({"organization_id": org, "configured": bool(row), "enabled": bool(row and row["enabled"]), "reviewed": bool(row and row["reviewed"])}) + return self.send_json(200, result) + + def _ai_current_fingerprint(self, db, row): + if not row["business_id"]: return "" + try: limit = int(json.loads(row["prompt_metadata_json"] or "{}").get("request", {}).get("max_items", MAX_INPUT_ITEMS)) + except (TypeError, ValueError): limit = MAX_INPUT_ITEMS + limit = max(1, min(limit, MAX_INPUT_ITEMS)) + org, bid = row["organization_id"], row["business_id"] + business = self.business(db, bid, org) + if not business: return "" + scans = [dict(x) for x in db.execute("SELECT id,input_url,classification,result_json,scanned_at FROM website_scans WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))] + evidence = [dict(x) for x in db.execute("SELECT id,kind,url,claim,created_at FROM evidence WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))] + contacts = [dict(x) for x in db.execute("SELECT * FROM contacts WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))] + extracted = [dict(x) for x in db.execute("SELECT * FROM contact_extractions WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))] + return input_fingerprint(row_json(business), scans, contacts + extracted, evidence, limit) + def suggest_ai(self, bid, payload, db, user): org = user["organization_id"] business = self.business(db, bid, org) @@ -489,7 +523,7 @@ class ApiHandler(BaseHTTPRequestHandler): scans = [dict(r) for r in db.execute("SELECT id,input_url,classification,result_json,scanned_at FROM website_scans WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, requested)).fetchall()] evidence = [dict(r) for r in db.execute("SELECT id,kind,url,claim,created_at FROM evidence WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, requested)).fetchall()] history = [dict(r) for r in db.execute("SELECT score,eligible,priority_band,score_version,explanations_json,created_at FROM score_history WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, requested)).fetchall()] - status, provider, version, metadata = generate_ai(business_dict, scans[:requested], contacts[:requested], evidence[:requested], history[:requested]) + status, provider, version, metadata = generate_ai(business_dict, scans[:requested], contacts[:requested] + extracted[:requested], evidence[:requested], history[:requested]) output = metadata.pop("output", {}) if status == "succeeded" else {"suggestions": [], "grounded": True, "claim_policy": "stored_evidence_only"} hashes = metadata.get("evidence_hashes", []) cur = db.execute("INSERT INTO ai_runs(organization_id,business_id,input_evidence_hashes_json,model,provider,version,prompt_metadata_json,data_minimization_json,status,approval_state,output_json,actor_user_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", (org, bid, json.dumps(hashes, sort_keys=True), "local-deterministic" if provider else "", provider, version, json.dumps({"request": {"max_items": requested}, "output_limit": MAX_OUTPUT_CHARS}, sort_keys=True), json.dumps(metadata, sort_keys=True), status, "pending", json.dumps(output, sort_keys=True), user["id"])) @@ -517,6 +551,12 @@ class ApiHandler(BaseHTTPRequestHandler): if not row: return self.send_json(404, {"error": "not_found"}) if decision == "approve" and row["status"] != "succeeded": return self.send_json(409, {"error": "run_not_approvable"}) if row["approval_state"] != "pending": return self.send_json(409, {"error": "already_decided"}) + if decision == "approve": + try: metadata = json.loads(row["data_minimization_json"] or "{}") + except (TypeError, ValueError): metadata = {} + expected = metadata.get("input_fingerprint") + if expected and expected != self._ai_current_fingerprint(db, row): + return self.send_json(409, {"error": "ai_run_stale", "approval_state": "pending", "reason": "source_hash_changed"}) now = datetime.now(timezone.utc).replace(microsecond=0).isoformat() if decision == "approve": db.execute("UPDATE ai_runs SET approval_state='approved',approved_at=? WHERE id=? AND organization_id=?", (now, run_id, user["organization_id"])) else: db.execute("UPDATE ai_runs SET approval_state='rejected',rejected_at=? WHERE id=? AND organization_id=?", (now, run_id, user["organization_id"])) @@ -566,6 +606,7 @@ class ApiHandler(BaseHTTPRequestHandler): if path=="/api/v1/interactions": return self.list_interactions(db,org,parse_qs(parsed.query)) if path=="/api/v1/suppressions": return self.list_suppressions(db,org,parse_qs(parsed.query)) if path=="/api/v1/ai-runs": return self.list_ai_runs(db,user,parse_qs(parsed.query)) + if path=="/api/v1/ai/provider-config": return self.ai_provider_config(db,user) if path=="/api/v1/outreach/drafts": return self.list_outreach_drafts(db,user,parse_qs(parsed.query)) if path=="/api/v1/outreach/provider-config": return self.provider_config(db,user) if path in ("/api/v1/reports/pipeline","/api/v1/reports/outcomes","/api/v1/reports/activity"): return self.report(db,org,path.rsplit('/',1)[1],parse_qs(parsed.query)) @@ -968,6 +1009,7 @@ class ApiHandler(BaseHTTPRequestHandler): payload=self.read_json(); org=user["organization_id"] if path=="/api/v1/saved-filters": return self.save_filter(payload,db,user) if path=="/api/v1/outreach/provider-config": return self.provider_config(db,user,payload) + if path=="/api/v1/ai/provider-config": return self.ai_provider_config(db,user,payload) if path=="/api/v1/businesses/bulk-review": return self.bulk_review(payload,db,user) bits_ai=path.split("/") if len(bits_ai)==7 and bits_ai[:4]==["","api","v1","businesses"] and bits_ai[5]=="ai" and bits_ai[6]=="suggest": return self.suggest_ai(int(bits_ai[4]) if bits_ai[4].isdigit() else -1,payload,db,user) diff --git a/apps/api/schema.sql b/apps/api/schema.sql index 5002454..c1913ca 100644 --- a/apps/api/schema.sql +++ b/apps/api/schema.sql @@ -281,6 +281,17 @@ CREATE TABLE IF NOT EXISTS ai_suggestions ( ); CREATE INDEX IF NOT EXISTS idx_ai_suggestions_run ON ai_suggestions(organization_id,ai_run_id,id); +-- AI provider settings contain no secrets. Remote execution remains unavailable +-- unless a separately reviewed provider abstraction is added. +CREATE TABLE IF NOT EXISTS ai_provider_configs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, + provider TEXT NOT NULL DEFAULT 'local', enabled INTEGER NOT NULL DEFAULT 1, + reviewed INTEGER NOT NULL DEFAULT 0, policy_json TEXT NOT NULL DEFAULT '{"network_enabled":false}', + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE(organization_id) +); + -- Phase 14 outreach preparation. Provider configuration is metadata plus a -- one-way secret fingerprint; outbound transport is intentionally disabled. CREATE TABLE IF NOT EXISTS outreach_provider_configs ( diff --git a/apps/api/tests/test_phase13_api.py b/apps/api/tests/test_phase13_api.py index 52a4af3..b39fc88 100644 --- a/apps/api/tests/test_phase13_api.py +++ b/apps/api/tests/test_phase13_api.py @@ -49,6 +49,36 @@ class Phase13ApiTests(unittest.TestCase): self.assertEqual(status, "not_configured"); self.assertEqual(provider, ""); self.assertEqual(version, "") self.assertNotIn("output", metadata) + def test_structured_enrichment_contract_has_classification_priority_confidence_uncertainty_and_provenance(self): + status, business = self.request("POST", "/api/v1/businesses", {"name": "Structured Co", "website": "https://structured.test"}); self.assertEqual(status, 201) + bid = business["id"] + self.request("POST", f"/api/v1/businesses/{bid}/evidence", {"kind": "source", "url": "https://source.test", "claim": "Provides solar installation"}) + status, run = self.request("POST", f"/api/v1/businesses/{bid}/ai/suggest", {}); self.assertEqual(status, 201) + output = run["output"] + for key in ("classification", "summary", "priority_recommendation", "confidence", "uncertainty", "conflicts", "citations", "policy"): + self.assertIn(key, output) + self.assertTrue(output["citations"][0]["provenance"]) + self.assertTrue(output["citations"][0]["hash"]) + self.assertEqual(output["policy"]["claim_policy"], "stored_evidence_only") + + def test_approval_rejects_stale_source_hashes(self): + _, business = self.request("POST", "/api/v1/businesses", {"name": "Stale Co"}) + bid = business["id"] + self.request("POST", f"/api/v1/businesses/{bid}/evidence", {"kind": "source", "url": "https://source.test", "claim": "Original claim"}) + _, run = self.request("POST", f"/api/v1/businesses/{bid}/ai/suggest", {}) + self.request("POST", f"/api/v1/businesses/{bid}/evidence", {"kind": "source", "url": "https://source.test", "claim": "Changed claim"}) + status, result = self.request("POST", f"/api/v1/ai-runs/{run['id']}/approve", {}) + self.assertEqual(status, 409); self.assertEqual(result["error"], "ai_run_stale") + + def test_remote_provider_is_not_configured_and_status_never_makes_network_call(self): + os.environ["AI_PROVIDER"] = "openai" + status, provider, version, metadata = generate({"name": "Remote"}, [], [], []) + self.assertEqual(status, "not_configured"); self.assertEqual(provider, "openai"); self.assertNotIn("output", metadata) + status, result = self.request("GET", "/api/v1/ai/provider-config") + self.assertEqual(status, 200); self.assertEqual(result["status"], "not_configured"); self.assertFalse(result["network_enabled"]) + status, result = self.request("POST", "/api/v1/ai/provider-config", {"provider": "openai", "enabled": True, "reviewed": True}) + self.assertEqual(status, 409); self.assertFalse(result["network_enabled"]) + def test_endpoint_persists_citations_and_approval_without_crm_write(self): status, business = self.request("POST", "/api/v1/businesses", {"name": "Evidence Co", "website": "https://evidence.test"}); self.assertEqual(status, 201) bid = business["id"]