# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } import datetime import hashlib import json from genlayer import * # ===================================================================== # Constants # ===================================================================== CONTRACT_VERSION = "2.3.0" DECISION_APPROVED = "approved" DECISION_REJECTED = "rejected" DECISION_ESCALATED = "escalated" VALID_DECISIONS = (DECISION_APPROVED, DECISION_REJECTED, DECISION_ESCALATED) VERDICT_LEGITIMATE = "legitimate" VERDICT_SUSPICIOUS = "suspicious" VERDICT_UNVERIFIABLE = "unverifiable" VALID_VERDICTS = (VERDICT_LEGITIMATE, VERDICT_SUSPICIOUS, VERDICT_UNVERIFIABLE) # Truncation length for fetched vendor website text before it enters a # prompt - keeps prompt size and validator cost bounded regardless of # how large the fetched page is. MAX_WEBSITE_TEXT_CHARS = 4000 EVENT_SUBMITTED = "submitted" EVENT_VALIDATED = "validated" EVENT_REVALIDATION_REQUESTED = "revalidation_requested" EVENT_VENDOR_BLOCKED = "vendor_blocked" EVENT_VENDOR_UNBLOCKED = "vendor_unblocked" EVENT_ORG_CALLER_AUTHORIZED = "org_caller_authorized" EVENT_ORG_CALLER_REVOKED = "org_caller_revoked" EVENT_AUTO_REJECTED_BLOCKED_VENDOR = "auto_rejected_blocked_vendor" EVENT_AUTO_REJECTED_DUPLICATE = "auto_rejected_duplicate_invoice" EVENT_VENDOR_VERIFIED = "vendor_verified" EVENT_ARBITRATED = "arbitrated" # Default decision thresholds, used when an organization has not set # its own via set_organization_settings. These mirror the thresholds # that used to live only inside the LLM prompt, now applied # deterministically in Python after a numeric score is obtained. DEFAULT_APPROVE_BELOW = 50 DEFAULT_REJECT_AT_OR_ABOVE = 80 REQUIRED_SUBMISSION_FIELDS = ( "invoice_number", "amount", "vendor_id", "vendor_name", "file_hash", "evidence_hash", ) ROLE_OWNER = "owner" ROLE_ADMIN = "admin" ROLE_FINANCE_REVIEWER = "finance_reviewer" AUTHORIZED_ROLES = (ROLE_OWNER, ROLE_ADMIN, ROLE_FINANCE_REVIEWER) ORG_ADMIN_ROLES = (ROLE_OWNER, ROLE_ADMIN) ORG_REVIEWER_ROLES = (ROLE_OWNER, ROLE_ADMIN, ROLE_FINANCE_REVIEWER) # ===================================================================== # Nondet validator judgments (module-level, not contract methods) # ===================================================================== # Every one of these is called from inside a run_nondet_unsafe # validator_fn closure. If they were instance methods, calling them as # self._whatever(...) would capture `self` - the whole contract # instance, including its live TreeMap/DynArray storage bindings - in # the closure GenVM pickles to run the validator in its sandbox, which # GenVM does not support (surfaces as a "Detected pickling storage # class" UserWarning even though these particular helpers never read # storage). Keeping them as plain functions with no `self` in scope # makes that class of bug structurally impossible here, not just # accidentally avoided. def _clamp_score(value) -> int: """Coerce a value to an integer risk score within [0, 100].""" try: score = int(value) except Exception: score = 50 return max(0, min(100, score)) def _minimum_plausible_score(ctx: dict) -> int: """Lower bound validators use to reject impossible low-risk output.""" minimum = 0 flags = ctx.get("flags", []) if isinstance(flags, list): for flag in flags: if not isinstance(flag, dict): continue severity = str(flag.get("severity", "")).lower() if severity == "high": minimum = max(minimum, 60) elif severity == "medium": minimum = max(minimum, 40) elif severity == "low": minimum = max(minimum, 20) preliminary = int(ctx.get("preliminary_score", 0)) if preliminary >= 80: minimum = max(minimum, 55) elif preliminary >= 50: minimum = max(minimum, 35) verification = ctx.get("vendor_verification") or {} if isinstance(verification, dict) and verification.get("verdict") == VERDICT_SUSPICIOUS: minimum = max(minimum, 45) return minimum def _parse_ai_analysis(raw, fallback_score: int) -> dict: """ Parse and normalize the leader AI output without throwing. With response_format="json", the real GenVM runtime already decodes the LLM's output into a dict before the contract ever sees it (see gl.nondet.exec_prompt's json overload). Some test harnesses instead hand back a raw JSON string. Both shapes are accepted here defensively. """ if isinstance(raw, dict): result = raw else: try: result = json.loads(raw) except Exception: return { "valid": False, "final_risk_score": fallback_score, "risk_factors": [], "reasoning": ( "LLM_ERROR: The AI analysis output could not be " "parsed as JSON. The deterministic preliminary score " "was used as a fallback." ), } if not isinstance(result, dict): return { "valid": False, "final_risk_score": fallback_score, "risk_factors": [], "reasoning": "LLM_ERROR: AI analysis was not a JSON object.", } score = _clamp_score(result.get("final_risk_score", fallback_score)) raw_factors = result.get("risk_factors", []) factors = [] if isinstance(raw_factors, list): for item in raw_factors: if not isinstance(item, dict): continue factor = str(item.get("factor", "unknown"))[:80] detail = str(item.get("detail", ""))[:500] factors.append({ "factor": factor, "weight": _clamp_score(item.get("weight", score)), "detail": detail, }) reasoning = str(result.get("reasoning", ""))[:2000] return { "valid": True, "final_risk_score": score, "risk_factors": factors, "reasoning": reasoning, } def _significant_words(text) -> set: """ Lowercased alphanumeric tokens of length >= 4 from text. Used to detect whether a leader's reasoning shares real content with a piece of free-form evidence (e.g. submitted arbitration text) rather than being generic prose unrelated to it. Deliberately dependency-free (no re import) to keep this GenVM-safe. """ words = set() current = [] for ch in str(text or "").lower(): if ch.isalnum(): current.append(ch) else: if len(current) >= 4: words.add("".join(current)) current = [] if len(current) >= 4: words.add("".join(current)) return words def _reasoning_grounded_in_evidence(reasoning: str, evidence_snippets) -> bool: """ Deterministic check that the leader's reasoning actually engages with concrete, evidence-anchored facts rather than reading as generic prose that could describe any submission. Requires at least one real snippet (e.g. the vendor name or invoice number pulled from the same evidence_hash-anchored data used to build the prompt) to appear in the reasoning text, case-insensitively. This is intentionally lenient - it does not re-derive the leader's judgment, only confirms the leader engaged with the real evidence it was given. A stricter check risks validators disagreeing on borderline wording and reintroducing the UNDETERMINED-consensus problem this contract's design already fought to eliminate (see module docstring). """ reasoning_lower = str(reasoning or "").lower() for snippet in evidence_snippets: snippet = str(snippet or "").strip() if len(snippet) >= 3 and snippet.lower() in reasoning_lower: return True return False def _accept_ai_analysis(ctx: dict, raw) -> bool: """Deterministically validate the leader's AI output.""" parsed = _parse_ai_analysis(raw, ctx["preliminary_score"]) if not parsed.get("valid"): return False score = int(parsed["final_risk_score"]) expected_min = _minimum_plausible_score(ctx) if score < expected_min: return False if ctx.get("amount", 0) >= 100000 and score < 35: return False factors = parsed.get("risk_factors", []) if expected_min >= 45 and len(factors) == 0: return False reasoning = parsed.get("reasoning", "") if not (reasoning and len(reasoning) >= 20): return False # Validators must confirm the leader actually engaged with the # real, evidence-hash-anchored invoice facts, not just produce # structurally well-formed, numerically plausible output. evidence_snippets = [ctx.get("vendor_name", ""), str(ctx.get("invoice_number", ""))] return _reasoning_grounded_in_evidence(reasoning, evidence_snippets) def _parse_vendor_verification(raw) -> dict: """Parse and normalize the leader's verification output without throwing.""" if isinstance(raw, dict): result = raw else: try: result = json.loads(raw) except Exception: return { "valid": False, "verdict": VERDICT_UNVERIFIABLE, "confidence": 0, "reasoning": ( "LLM_ERROR: verification output could not be " "parsed as JSON." ), } if not isinstance(result, dict): return { "valid": False, "verdict": VERDICT_UNVERIFIABLE, "confidence": 0, "reasoning": "LLM_ERROR: verification output was not a JSON object.", } verdict = str(result.get("verdict", "")).strip().lower() confidence = _clamp_score(result.get("confidence", 0)) reasoning = str(result.get("reasoning", ""))[:1000] valid = verdict in VALID_VERDICTS and bool(reasoning) and len(reasoning) >= 15 if verdict not in VALID_VERDICTS: verdict = VERDICT_UNVERIFIABLE return {"valid": valid, "verdict": verdict, "confidence": confidence, "reasoning": reasoning} def _accept_vendor_verification(raw) -> bool: """Deterministically validate the leader's vendor-verification output.""" return bool(_parse_vendor_verification(raw).get("valid")) def _parse_arbitration(raw) -> dict: """Parse and normalize the leader's arbitration output without throwing.""" if isinstance(raw, dict): result = raw else: try: result = json.loads(raw) except Exception: return { "valid": False, "decision": DECISION_REJECTED, "reasoning": ( "LLM_ERROR: arbitration output could not be " "parsed as JSON. Defaulted to rejected pending " "manual review." ), } if not isinstance(result, dict): return { "valid": False, "decision": DECISION_REJECTED, "reasoning": ( "LLM_ERROR: arbitration output was not a JSON " "object. Defaulted to rejected pending manual " "review." ), } decision = str(result.get("decision", "")).strip().lower() reasoning = str(result.get("reasoning", ""))[:2000] valid = decision in (DECISION_APPROVED, DECISION_REJECTED) and bool(reasoning) and len(reasoning) >= 20 if decision not in (DECISION_APPROVED, DECISION_REJECTED): decision = DECISION_REJECTED if not reasoning: reasoning = ( "LLM_ERROR: arbitration returned an invalid " "decision value. Defaulted to rejected pending " "manual review." ) return {"valid": valid, "decision": decision, "reasoning": reasoning} def _accept_arbitration(ctx: dict, evidence_text: str, raw) -> bool: """ Deterministically validate the leader's arbitration output. Beyond basic shape/decision validity, the ruling must be grounded in real evidence: either it references a concrete fact from the original invoice (vendor name, invoice number), or it meaningfully engages with the newly submitted evidence text itself (shares real content, not just a generic restatement). A binding arbitration ruling that engages with neither is not distinguishable from one that ignored the evidence entirely. """ parsed = _parse_arbitration(raw) if not parsed.get("valid"): return False reasoning = parsed.get("reasoning", "") evidence_snippets = [ctx.get("vendor_name", ""), str(ctx.get("invoice_number", ""))] if _reasoning_grounded_in_evidence(reasoning, evidence_snippets): return True shared_words = _significant_words(reasoning) & _significant_words(evidence_text) return len(shared_words) > 0 # ===================================================================== # Contract # ===================================================================== class InvoiceValidator(gl.Contract): # ------------------------------------------------------------------ # Persistent storage # ------------------------------------------------------------------ # All collection fields below are zero-initialized by the GenLayer # runtime (TreeMap -> {}, DynArray -> []). They must not be # constructed directly - TreeMap() / DynArray() raise at runtime if # called from contract code, since instantiation is owned by the # storage layer itself. contract_owner: str """Wallet that deployed the contract and can bootstrap organization callers""" organization_callers: TreeMap[str, str] """key = "{org_id}:{caller_address}" -> JSON-encoded caller role binding""" submissions: TreeMap[str, str] """key = "{org_id}:{invoice_id}" -> JSON-encoded InvoiceSubmission""" validations: TreeMap[str, str] """key = "{org_id}:{invoice_id}" -> JSON-encoded ValidationResult""" vendor_history: TreeMap[str, str] """key = "{org_id}:{vendor_id}" -> JSON-encoded VendorSummary""" vendor_blocklist: TreeMap[str, str] """key = "{org_id}:{vendor_id}" -> JSON-encoded BlockRecord, present only when blocked""" invoice_number_index: TreeMap[str, str] """key = "{org_id}:{vendor_id}:{invoice_number}" -> invoice_id, for O(1) duplicate detection""" organization_settings: TreeMap[str, str] """key = "{org_id}" -> JSON-encoded OrganizationSettings (decision thresholds)""" organization_stats: TreeMap[str, str] """key = "{org_id}" -> JSON-encoded OrganizationStats (running counters)""" vendor_verifications: TreeMap[str, str] """key = "{org_id}:{vendor_id}" -> JSON-encoded VendorVerification (cached website legitimacy check)""" arbitrations: TreeMap[str, str] """key = "{org_id}:{invoice_id}" -> JSON-encoded ArbitrationRecord (evidence + binding decision)""" audit_trail: DynArray[str] """Append-only log of JSON-encoded AuditEntry records, across all organizations""" def __init__(self): # Collection fields default to empty storage maps/arrays. The # scalar owner is fixed at deployment and is the only account # that can bootstrap the first caller for each organization. self.contract_owner = self._current_sender() # ================================================================== # SECTION 1: Public write methods - organization caller registry # ================================================================== @gl.public.write def authorize_org_caller( self, org_id: str, caller_address: str, role: str, ) -> None: """ Bind an organization namespace to an on-chain caller wallet. The contract owner may bootstrap the first caller for any organization. After that, only existing owner/admin callers for that organization may add or change bindings. This prevents a backend from writing arbitrary org_id values unless the signing wallet is already authorized on-chain for that namespace. """ actor = self._current_sender() role = self._normalize_role(role) caller = self._normalize_address(caller_address) if not org_id: raise gl.vm.UserError("org_id is required") if not caller: raise gl.vm.UserError("caller_address is required") if role not in AUTHORIZED_ROLES: raise gl.vm.UserError("role must be owner, admin, or finance_reviewer") if not self._is_contract_owner(actor): self._require_org_role(org_id, ORG_ADMIN_ROLES) key = self._caller_key(org_id, caller) self.organization_callers[key] = json.dumps({ "org_id": org_id, "caller_address": caller, "role": role, "authorized_by": actor, "authorized_at": self._now_iso(), }) self._append_audit_entry( entry_id=f"{org_id}:caller:{caller}:authorized:{self._now_iso()}", org_id=org_id, invoice_id="", event=EVENT_ORG_CALLER_AUTHORIZED, actor_wallet=actor, decision=None, risk_score=None, evidence_hash="", ) @gl.public.write def revoke_org_caller(self, org_id: str, caller_address: str) -> None: """Remove a caller binding for an organization namespace.""" actor = self._current_sender() if not self._is_contract_owner(actor): self._require_org_role(org_id, ORG_ADMIN_ROLES) caller = self._normalize_address(caller_address) key = self._caller_key(org_id, caller) if key in self.organization_callers: del self.organization_callers[key] self._append_audit_entry( entry_id=f"{org_id}:caller:{caller}:revoked:{self._now_iso()}", org_id=org_id, invoice_id="", event=EVENT_ORG_CALLER_REVOKED, actor_wallet=actor, decision=None, risk_score=None, evidence_hash="", ) # ================================================================== # SECTION 2: Public write methods - invoice lifecycle # ================================================================== @gl.public.write def submit_invoice(self, org_id: str, invoice_id: str, payload: str) -> str: """ Record a new invoice submission and run (or deterministically skip) its fraud/compliance validation. Called by the submit-to-genlayer Supabase Edge Function once an invoice has passed off-chain deterministic pre-checks. Args: org_id: UUID of the submitting organization. invoice_id: UUID of the invoice in the off-chain database. payload: JSON string with invoice metadata, vendor info, deterministic anomaly flags, and the org's procurement policy text. See _parse_submission_payload for the expected shape. Returns: The composite storage key "{org_id}:{invoice_id}", used by the caller as a submission receipt. Raises: Exception: if the invoice was already submitted for this org, the payload is malformed, or a required field is missing. """ caller = self._require_org_role(org_id, ORG_REVIEWER_ROLES) key = self._invoice_key(org_id, invoice_id) if key in self.submissions: raise gl.vm.UserError( f"Invoice {invoice_id} already submitted for organization {org_id}" ) data = self._parse_submission_payload(payload) self._validate_required_fields(data) self._verify_submitter_wallet(data, caller) evidence_hash = self._verify_evidence_anchor(org_id, invoice_id, data) now = self._now_iso() data["org_id"] = org_id data["invoice_id"] = invoice_id data["submitter_wallet"] = caller data["evidence_hash"] = evidence_hash data["submitted_at"] = now self.submissions[key] = json.dumps(data) self._index_invoice_number(org_id, data) self._append_audit_entry( entry_id=f"{key}:submitted", org_id=org_id, invoice_id=invoice_id, event=EVENT_SUBMITTED, actor_wallet=caller, decision=None, risk_score=None, evidence_hash=evidence_hash, ) self._bump_org_stat(org_id, "total_submitted") # --- Deterministic fast paths: skip the LLM entirely when the # outcome is already certain. This both resolves faster and # removes opportunities for validator disagreement. blocked_reason = self._blocked_vendor_reason(org_id, data.get("vendor_id", "")) if blocked_reason is not None: self._finalize_validation( org_id=org_id, invoice_id=invoice_id, decision=DECISION_REJECTED, risk_score=100, risk_factors=[{ "factor": "blocked_vendor", "weight": 100, "detail": blocked_reason, }], reasoning=( f"This vendor is on the organization's blocklist " f"({blocked_reason}). The invoice was rejected " f"automatically without AI review." ), vendor_name=data.get("vendor_name", "Unknown"), vendor_id=data.get("vendor_id", ""), amount=float(data.get("amount", 0)), flagged=True, audit_event=EVENT_AUTO_REJECTED_BLOCKED_VENDOR, actor_wallet=caller, evidence_hash=evidence_hash, ) return key duplicate_invoice_id = self._find_duplicate_invoice_number( org_id, data.get("vendor_id", ""), data.get("invoice_number", "") ) if duplicate_invoice_id and duplicate_invoice_id != invoice_id: self._finalize_validation( org_id=org_id, invoice_id=invoice_id, decision=DECISION_REJECTED, risk_score=95, risk_factors=[{ "factor": "duplicate_invoice_number", "weight": 95, "detail": ( f"Invoice number {data.get('invoice_number')} was " f"already submitted for this vendor as invoice " f"{duplicate_invoice_id}." ), }], reasoning=( "Duplicate invoice numbers for the same vendor are a " "well-established indicator of double billing or " "fraud. This invoice was rejected automatically " "without AI review." ), vendor_name=data.get("vendor_name", "Unknown"), vendor_id=data.get("vendor_id", ""), amount=float(data.get("amount", 0)), flagged=True, audit_event=EVENT_AUTO_REJECTED_DUPLICATE, actor_wallet=caller, evidence_hash=evidence_hash, ) return key self._run_validation(org_id, invoice_id, data, caller, evidence_hash) return key @gl.public.write def request_revalidation(self, org_id: str, invoice_id: str) -> str: """ Re-run AI validation for an invoice that was previously escalated for manual review. Intended to be called when a finance reviewer wants a second opinion after, for example, the vendor's history has changed since the original submission. Args: org_id: UUID of the organization. invoice_id: UUID of the invoice to re-validate. Returns: The composite storage key, as a receipt. Raises: Exception: if no submission exists, or the existing decision is not "escalated" (approved/rejected invoices are considered final and are not re-opened through this method). """ caller = self._require_org_role(org_id, ORG_REVIEWER_ROLES) key = self._invoice_key(org_id, invoice_id) submission_raw = self.submissions.get(key) if not submission_raw: raise gl.vm.UserError(f"No submission found for invoice {invoice_id}") validation_raw = self.validations.get(key) if validation_raw: existing = json.loads(validation_raw) if existing.get("decision") != DECISION_ESCALATED: raise gl.vm.UserError( "Only escalated invoices can be requested for " "revalidation. Current decision: " f"{existing.get('decision')}" ) data = json.loads(submission_raw) self._append_audit_entry( entry_id=f"{key}:revalidation:{self._now_iso()}", org_id=org_id, invoice_id=invoice_id, event=EVENT_REVALIDATION_REQUESTED, actor_wallet=caller, decision=None, risk_score=None, evidence_hash=data.get("evidence_hash", ""), ) self._run_validation( org_id, invoice_id, data, caller, data.get("evidence_hash", "") ) return key @gl.public.write def arbitrate_invoice(self, org_id: str, invoice_id: str, evidence_text: str) -> str: """ Render a final, binding decision for an invoice that is stuck in "escalated", using supplementary evidence a finance reviewer supplies (e.g. vendor correspondence, PO clarification, or other context not available at first validation). Unlike request_revalidation - which re-runs the same risk analysis unchanged and can escalate again - arbitration always resolves to approved or rejected; the AI is never allowed to return "escalated" here, since the entire point is to stop the invoice from being stuck. Reuses _finalize_validation exactly like the normal validation path. Args: org_id: UUID of the organization. invoice_id: UUID of the invoice to arbitrate. evidence_text: Supplementary context/evidence to weigh alongside the original invoice data. Returns: The composite storage key, as a receipt. Raises: Exception: if no submission/validation exists, the current decision is not "escalated", or evidence_text is empty. """ caller = self._require_org_role(org_id, ORG_REVIEWER_ROLES) key = self._invoice_key(org_id, invoice_id) submission_raw = self.submissions.get(key) if not submission_raw: raise gl.vm.UserError(f"No submission found for invoice {invoice_id}") validation_raw = self.validations.get(key) if not validation_raw: raise gl.vm.UserError("Invoice has not been validated yet") existing = json.loads(validation_raw) if existing.get("decision") != DECISION_ESCALATED: raise gl.vm.UserError( "Only escalated invoices can be arbitrated. Current " f"decision: {existing.get('decision')}" ) evidence_text = str(evidence_text or "").strip() if not evidence_text: raise gl.vm.UserError("evidence_text is required") data = json.loads(submission_raw) context = self._build_analysis_context(org_id, data) decision, reasoning = self._arbitrate(context, evidence_text) self.arbitrations[key] = json.dumps({ "org_id": org_id, "invoice_id": invoice_id, "evidence_text": evidence_text, "decision": decision, "reasoning": reasoning, "arbitrated_by": caller, "arbitrated_at": self._now_iso(), }) self._finalize_validation( org_id=org_id, invoice_id=invoice_id, decision=decision, risk_score=int(existing.get("risk_score", 50)), risk_factors=existing.get("risk_factors", []), reasoning=reasoning, vendor_name=data.get("vendor_name", "Unknown"), vendor_id=data.get("vendor_id", ""), amount=float(data.get("amount", 0)), flagged=bool(data.get("anomaly_flags")), audit_event=EVENT_ARBITRATED, actor_wallet=caller, evidence_hash=data.get("evidence_hash", ""), ) return key # ================================================================== # SECTION 3: Public write methods - vendor management # ================================================================== @gl.public.write def block_vendor(self, org_id: str, vendor_id: str, reason: str) -> None: """ Add a vendor to an organization's blocklist. Future invoice submissions from this vendor are deterministically rejected in submit_invoice without any AI review. Args: org_id: UUID of the organization. vendor_id: UUID of the vendor to block. reason: Human-readable justification, surfaced verbatim in future rejection reasoning text. """ caller = self._require_org_role(org_id, ORG_ADMIN_ROLES) if not vendor_id: raise gl.vm.UserError("vendor_id is required") if not reason or not reason.strip(): raise gl.vm.UserError("A reason is required to block a vendor") vendor_key = self._vendor_key(org_id, vendor_id) self.vendor_blocklist[vendor_key] = json.dumps({ "org_id": org_id, "vendor_id": vendor_id, "reason": reason.strip(), "blocked_at": self._now_iso(), }) self._append_audit_entry( entry_id=f"{vendor_key}:blocked:{self._now_iso()}", org_id=org_id, invoice_id="", event=EVENT_VENDOR_BLOCKED, actor_wallet=caller, decision=None, risk_score=None, evidence_hash="", ) @gl.public.write def unblock_vendor(self, org_id: str, vendor_id: str) -> None: """ Remove a vendor from an organization's blocklist. Args: org_id: UUID of the organization. vendor_id: UUID of the vendor to unblock. """ caller = self._require_org_role(org_id, ORG_ADMIN_ROLES) vendor_key = self._vendor_key(org_id, vendor_id) if vendor_key in self.vendor_blocklist: del self.vendor_blocklist[vendor_key] self._append_audit_entry( entry_id=f"{vendor_key}:unblocked:{self._now_iso()}", org_id=org_id, invoice_id="", event=EVENT_VENDOR_UNBLOCKED, actor_wallet=caller, decision=None, risk_score=None, evidence_hash="", ) @gl.public.write def verify_vendor_legitimacy(self, org_id: str, vendor_id: str, website_url: str) -> str: """ Fetch a vendor's declared website live and have the AI judge whether it looks like a genuine, matching business - not proof of legal registration, but a real signal that catches shell vendors with no actual web footprint. The verdict is cached and fed back into future invoice risk analysis for this vendor (see _build_analysis_context), so a "suspicious" result directly influences fraud scoring rather than sitting unused. Args: org_id: UUID of the organization. vendor_id: UUID of the vendor to verify. website_url: The vendor's claimed website, must start with http:// or https://. Returns: The composite storage key "{org_id}:{vendor_id}". """ caller = self._require_org_role(org_id, ORG_REVIEWER_ROLES) if not vendor_id: raise gl.vm.UserError("vendor_id is required") website_url = str(website_url or "").strip() if not (website_url.startswith("http://") or website_url.startswith("https://")): raise gl.vm.UserError("website_url must start with http:// or https://") vendor_name = vendor_id hist = self._get_vendor_history_dict(org_id, vendor_id) if hist and hist.get("vendor_name"): vendor_name = hist.get("vendor_name") verdict, confidence, reasoning = self._verify_vendor_website( vendor_name, vendor_id, website_url ) vendor_key = self._vendor_key(org_id, vendor_id) self.vendor_verifications[vendor_key] = json.dumps({ "org_id": org_id, "vendor_id": vendor_id, "website": website_url, "verdict": verdict, "confidence": confidence, "reasoning": reasoning, "checked_by": caller, "checked_at": self._now_iso(), }) self._append_audit_entry( entry_id=f"{vendor_key}:verified:{self._now_iso()}", org_id=org_id, invoice_id="", event=EVENT_VENDOR_VERIFIED, actor_wallet=caller, decision=verdict, risk_score=confidence, evidence_hash="", ) return vendor_key # ================================================================== # SECTION 4: Public write methods - organization configuration # ================================================================== @gl.public.write def set_organization_settings( self, org_id: str, approve_below: int, reject_at_or_above: int, ) -> None: """ Configure the deterministic risk-score thresholds used to turn a numeric AI risk score into a decision for this organization. Args: org_id: UUID of the organization. approve_below: Risk scores strictly below this value are approved automatically (subject to escalated range, see reject_at_or_above). reject_at_or_above: Risk scores at or above this value are rejected automatically. Scores between approve_below and reject_at_or_above are escalated for manual review. Raises: Exception: if the thresholds are out of the valid 0-100 range or are not in increasing order. """ self._require_org_role(org_id, ORG_ADMIN_ROLES) if not (0 <= approve_below <= 100) or not (0 <= reject_at_or_above <= 100): raise gl.vm.UserError("Thresholds must be between 0 and 100") if approve_below >= reject_at_or_above: raise gl.vm.UserError("approve_below must be less than reject_at_or_above") self.organization_settings[org_id] = json.dumps({ "org_id": org_id, "approve_below": approve_below, "reject_at_or_above": reject_at_or_above, "updated_at": self._now_iso(), }) # ================================================================== # SECTION 5: Internal - nondeterministic AI validation # ================================================================== def _run_validation( self, org_id: str, invoice_id: str, data: dict, actor_wallet: str, evidence_hash: str, ) -> None: """ Run AI fraud/compliance analysis to obtain a single numeric risk score, then derive the decision deterministically. See module docstring for why the decision itself is computed in Python rather than asked of the LLM. This used to be five sequential gl.eq_principle.prompt_comparative calls (vendor credibility, pricing, fraud, compliance, then an aggregation step). Each prompt_comparative call requires the leader AND every validator to *independently* run their own LLM and then agree with each other - and different validator nodes in this contract's committee run different model providers (e.g. deepseek, qwen, gpt, claude), which routinely produced genuinely different scores and caused the whole transaction to end UNDETERMINED, even after widening the numeric tolerance bands. This version uses an explicit GenVM-supported nondeterministic block: the leader calls the LLM once, then validators run a deterministic validation function over the leader's JSON. The validators check shape, bounds, and consistency with anchored invoice evidence instead of trying to reproduce the same prose. """ vendor_id = data.get("vendor_id", "") vendor_name = data.get("vendor_name", "Unknown") amount = float(data.get("amount", 0)) preliminary_score = int(data.get("preliminary_risk_score", 10)) context = self._build_analysis_context(org_id, data) final_risk_score, risk_factors, reasoning = self._analyze_invoice_risk(context) decision = self._derive_decision(org_id, final_risk_score) flags = data.get("anomaly_flags", []) self._finalize_validation( org_id=org_id, invoice_id=invoice_id, decision=decision, risk_score=final_risk_score, risk_factors=risk_factors, reasoning=reasoning, vendor_name=vendor_name, vendor_id=vendor_id, amount=amount, flagged=len(flags) > 0, audit_event=EVENT_VALIDATED, actor_wallet=actor_wallet, evidence_hash=evidence_hash, ) def _build_analysis_context(self, org_id: str, data: dict) -> dict: """Gather and format every input the risk-analysis prompt needs.""" vendor_id = data.get("vendor_id", "") vendor_hist = self._get_vendor_history_dict(org_id, vendor_id) vendor_verification = self._get_vendor_verification_dict(org_id, vendor_id) return { "vendor_name": data.get("vendor_name", "Unknown"), "vendor_status": data.get("vendor_status", "active"), "vendor_hist_text": self._format_vendor_history(vendor_hist), "vendor_verification": vendor_verification or {}, "vendor_verification_text": self._format_vendor_verification(vendor_verification), "evidence_hash": data.get("evidence_hash", ""), "file_hash": data.get("file_hash", ""), "flags": data.get("anomaly_flags", []), "flags_text": self._format_anomaly_flags(data.get("anomaly_flags", [])), "line_items_text": self._format_line_items(data.get("line_items", [])), "policy_text": data.get("procurement_policy") or ( "No specific procurement policy provided. Use general " "accounts-payable best practices as the standard." ), "amount": float(data.get("amount", 0)), "currency": data.get("currency", "USD"), "preliminary_score": int(data.get("preliminary_risk_score", 10)), "invoice_number": data.get("invoice_number"), "issue_date": data.get("issue_date", "-"), "due_date": data.get("due_date", "-"), } def _build_analysis_input_text(self, ctx: dict) -> str: """ Render the full invoice/vendor/policy context as plain text. This is the `fn` passed to prompt_non_comparative - it runs on every node (leader and validators alike) and must be a purely deterministic function of already-agreed-upon data, since it is not itself covered by any equivalence judgment. """ return f"""## Vendor Information - Name: {ctx['vendor_name']} - Status: {ctx['vendor_status']} - History with this organization: {ctx['vendor_hist_text']} - Website verification: {ctx['vendor_verification_text']} ## Invoice - Number: {ctx['invoice_number']} - Amount: {ctx['currency']} {ctx['amount']:.2f} - Issue date: {ctx['issue_date']}, Due date: {ctx['due_date']} - Uploaded document SHA-256: {ctx['file_hash']} - On-chain evidence anchor: {ctx['evidence_hash']} - Line items: {ctx['line_items_text']} ## Deterministic Pre-check Flags {ctx['flags_text']} ## Procurement Policy {ctx['policy_text']} ## Deterministic Preliminary Risk Score {ctx['preliminary_score']}/100 (computed off-chain before this invoice reached GenLayer)""" def _analyze_invoice_risk(self, ctx: dict) -> tuple: """ Run one leader-only LLM call under a custom GenVM validator. Validators do not re-run the LLM. They inspect the leader's JSON deterministically and accept only bounded, well-formed, context-plausible analysis that is demonstrably grounded in the real, evidence-hash-anchored invoice facts (see _reasoning_grounded_in_evidence) - not just structurally well-formed and numerically plausible. This avoids unsupported nested equivalence helpers and gives the runtime a stable validation contract for the nondeterministic result. """ input_text = self._build_analysis_input_text(ctx) prompt = ( "You are an expert accounts payable fraud and compliance " "analyst. Using the invoice context provided, assess overall " "risk across vendor credibility, pricing plausibility, fraud " "indicators, and procurement policy compliance. Treat the " "uploaded document hash and evidence anchor as immutable " "references to the invoice evidence. Your reasoning must " "explicitly reference concrete details from the invoice " "(such as the vendor name or invoice number) to demonstrate " "it is grounded in the actual evidence provided, not generic " "boilerplate. Respond ONLY with valid JSON in this exact " "shape: " '{"final_risk_score": , "risk_factors": ' '[{"factor": "", "weight": <0-100>, "detail": ' '""}], "reasoning": ""}\n\n' f"{input_text}" ) def leader_fn(): return gl.nondet.exec_prompt(prompt, response_format="json") def validator_fn(leaders_res) -> bool: # In the real GenVM runtime, run_nondet_unsafe's validator_fn # receives a Result (Return | VMError | UserError) decoded # across the sub-VM boundary - a raised UserError/VMError from # the leader must be rejected outright. gltest's direct-mode # harness has no sub-VM boundary and passes the leader's raw # return value straight through without any wrapper, so that # case is accepted as-is rather than always rejected. if isinstance(leaders_res, gl.vm.Return): raw = leaders_res.calldata elif isinstance(leaders_res, (gl.vm.UserError, gl.vm.VMError)): return False else: raw = leaders_res return _accept_ai_analysis(ctx, raw) raw = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) parsed = _parse_ai_analysis(raw, ctx["preliminary_score"]) return ( parsed["final_risk_score"], parsed["risk_factors"], parsed["reasoning"], ) def _verify_vendor_website(self, vendor_name: str, vendor_id: str, website_url: str) -> tuple: """ One combined nondet block: the leader fetches the vendor's website and immediately asks the LLM to judge it, mirroring _analyze_invoice_risk's single leader/validator round rather than adding a second, separate consensus round just for the fetch itself. """ def leader_fn(): try: page_text = gl.nondet.web.render(website_url, mode="text") except Exception as exc: page_text = f"(failed to load page: {exc})" page_text = str(page_text)[:MAX_WEBSITE_TEXT_CHARS] prompt = ( "You are a fraud-prevention analyst checking whether a " "vendor's claimed website looks like a genuine, " "matching business - not proof of legal registration, " "just a real-world footprint check. Given the vendor's " "name and the fetched page content below, judge " "whether the site plausibly belongs to this business " "(consistent name/branding, real business content) " "versus a parked domain, unrelated site, or content " "with no connection to the vendor name. If the page " "failed to load, treat that as unverifiable rather " "than suspicious unless other signals suggest fraud. " "Respond ONLY with valid JSON in this exact shape: " '{"verdict": "legitimate"|"suspicious"|"unverifiable", ' '"confidence": , "reasoning": ""}\n\n' f"Vendor name: {vendor_name or vendor_id}\n" f"Declared website: {website_url}\n\n" f"Fetched page content:\n{page_text}" ) return gl.nondet.exec_prompt(prompt, response_format="json") def validator_fn(leaders_res) -> bool: if isinstance(leaders_res, gl.vm.Return): raw = leaders_res.calldata elif isinstance(leaders_res, (gl.vm.UserError, gl.vm.VMError)): return False else: raw = leaders_res return _accept_vendor_verification(raw) raw = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) parsed = _parse_vendor_verification(raw) return parsed["verdict"], parsed["confidence"], parsed["reasoning"] def _arbitrate(self, ctx: dict, evidence_text: str) -> tuple: """ One nondet block asking for a strictly binding approved/ rejected decision - unlike _analyze_invoice_risk, "escalated" is not a valid output here, since the entire point of arbitration is to resolve invoices already stuck in that state. Validators additionally confirm the ruling is grounded in the original invoice facts or the newly submitted evidence (see _accept_arbitration), not a generic restatement. """ input_text = self._build_analysis_input_text(ctx) prompt = ( "You are a senior accounts payable arbitrator making a " "FINAL, BINDING decision on an invoice that was already " "escalated for manual review. New evidence has been " "submitted by a finance reviewer. Weigh the original " "invoice context together with this new evidence and " "decide definitively. You MUST choose either \"approved\" " "or \"rejected\" - escalation is not a valid outcome of " "arbitration. Your reasoning must explicitly engage with " "the newly submitted evidence and reference concrete " "details from the invoice (such as the vendor name or " "invoice number), not generic boilerplate. Respond ONLY " "with valid JSON in this exact shape: " '{"decision": "approved"|"rejected", "reasoning": ""}\n\n' f"{input_text}\n\n" f"## Newly Submitted Evidence\n{evidence_text}" ) def leader_fn(): return gl.nondet.exec_prompt(prompt, response_format="json") def validator_fn(leaders_res) -> bool: if isinstance(leaders_res, gl.vm.Return): raw = leaders_res.calldata elif isinstance(leaders_res, (gl.vm.UserError, gl.vm.VMError)): return False else: raw = leaders_res return _accept_arbitration(ctx, evidence_text, raw) raw = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) parsed = _parse_arbitration(raw) return parsed["decision"], parsed["reasoning"] def _derive_decision(self, org_id: str, final_risk_score: int) -> str: """ Deterministically map a numeric risk score to a decision using this organization's configured thresholds (or the contract-wide defaults if none were set via set_organization_settings). Pure Python comparison - no consensus risk whatsoever. """ approve_below, reject_at_or_above = self._get_org_thresholds(org_id) if final_risk_score >= reject_at_or_above: return DECISION_REJECTED if final_risk_score < approve_below: return DECISION_APPROVED return DECISION_ESCALATED # ================================================================== # SECTION 6: Internal - finalization, history, and audit helpers # ================================================================== def _finalize_validation( self, org_id: str, invoice_id: str, decision: str, risk_score: int, risk_factors: list, reasoning: str, vendor_name: str, vendor_id: str, amount: float, flagged: bool, audit_event: str, actor_wallet: str, evidence_hash: str, ) -> None: """ Shared tail end of both the AI-driven and the deterministic fast-path validation flows: persist the ValidationResult, update vendor history and organization stats, and append the audit trail entry. An invoice can be finalized more than once - request_revalidation and arbitrate_invoice both call this again for an invoice that already has a stored ValidationResult. When that happens, this must move the invoice's counters from its previous decision to its new one rather than incrementing totals/decision counts a second time for the same invoice. """ if decision not in VALID_DECISIONS: decision = DECISION_ESCALATED key = self._invoice_key(org_id, invoice_id) now = self._now_iso() previous_decision = None previous_raw = self.validations.get(key) if previous_raw: try: previous_decision = json.loads(previous_raw).get("decision") except Exception: previous_decision = None self.validations[key] = json.dumps({ "org_id": org_id, "invoice_id": invoice_id, "decision": decision, "risk_score": risk_score, "risk_factors": risk_factors, "reasoning": reasoning, "evidence_hash": evidence_hash, "resolved_at": now, }) self._update_vendor_history( org_id=org_id, vendor_id=vendor_id, vendor_name=vendor_name, amount=amount, decision=decision, previous_decision=previous_decision, flagged=flagged, now=now, ) if previous_decision in VALID_DECISIONS: self._bump_org_stat(org_id, f"total_{previous_decision}", delta=-1) self._bump_org_stat(org_id, f"total_{decision}") self._append_audit_entry( entry_id=f"{key}:{audit_event}:{now}", org_id=org_id, invoice_id=invoice_id, event=audit_event, actor_wallet=actor_wallet, decision=decision, risk_score=risk_score, evidence_hash=evidence_hash, ) def _decision_counter_name(self, decision: str) -> str: return { DECISION_APPROVED: "approved_invoices", DECISION_REJECTED: "rejected_invoices", DECISION_ESCALATED: "escalated_invoices", }.get(decision, "escalated_invoices") def _update_vendor_history( self, org_id: str, vendor_id: str, vendor_name: str, amount: float, decision: str, previous_decision, flagged: bool, now: str, ) -> None: """ Update the running per-vendor summary used as context for future submissions. On a re-finalization (previous_decision is set), totals are not incremented again - only the decision bucket moves from the old decision to the new one. """ if not vendor_id: return vendor_key = self._vendor_key(org_id, vendor_id) hist = self._get_vendor_history_dict(org_id, vendor_id) or { "org_id": org_id, "vendor_id": vendor_id, "vendor_name": vendor_name, "total_invoices": 0, "approved_invoices": 0, "rejected_invoices": 0, "escalated_invoices": 0, "flagged_invoices": 0, "total_amount": 0.0, "avg_invoice_amount": 0.0, "last_invoice_at": now, } hist.setdefault("escalated_invoices", 0) if previous_decision is None: hist["total_invoices"] += 1 hist["total_amount"] = float(hist["total_amount"]) + amount hist["avg_invoice_amount"] = hist["total_amount"] / hist["total_invoices"] if flagged: hist["flagged_invoices"] += 1 hist["last_invoice_at"] = now if previous_decision in VALID_DECISIONS: counter = self._decision_counter_name(previous_decision) hist[counter] = max(0, int(hist.get(counter, 0)) - 1) counter = self._decision_counter_name(decision) hist[counter] = int(hist.get(counter, 0)) + 1 self.vendor_history[vendor_key] = json.dumps(hist) def _append_audit_entry( self, entry_id: str, org_id: str, invoice_id: str, event: str, actor_wallet: str, decision, risk_score, evidence_hash: str, ) -> None: """Append a single immutable audit log entry.""" self.audit_trail.append(json.dumps({ "entry_id": entry_id, "org_id": org_id, "invoice_id": invoice_id, "event": event, "actor_wallet": actor_wallet, "decision": decision, "risk_score": risk_score, "evidence_hash": evidence_hash, "timestamp": self._now_iso(), })) def _bump_org_stat(self, org_id: str, counter_name: str, delta: int = 1) -> None: """Adjust a single named counter in an organization's running stats.""" stats = self._get_org_stats_dict(org_id) stats[counter_name] = max(0, int(stats.get(counter_name, 0)) + delta) self.organization_stats[org_id] = json.dumps(stats) # ================================================================== # SECTION 7: Internal - parsing, formatting, and lookup helpers # ================================================================== def _current_sender(self) -> str: """Return the current transaction sender as a normalized address.""" return self._normalize_address(gl.message.sender_address) def _normalize_address(self, address) -> str: """Normalize an EVM address for storage-key comparisons.""" if isinstance(address, bytes): value = "0x" + address.hex() elif hasattr(address, "as_bytes"): value = "0x" + address.as_bytes.hex() else: value = str(address or "").strip().lower() if value.startswith("address(") and value.endswith(")"): value = value[8:-1].strip().lower() if value.startswith("b'") or value.startswith('b"'): raise gl.vm.UserError("address bytes must be hex encoded") if not value: return "" if not value.startswith("0x"): value = f"0x{value}" if len(value) != 42: raise gl.vm.UserError("address must be a 20-byte hex string") for ch in value[2:]: if ch not in "0123456789abcdef": raise gl.vm.UserError("address must be hex encoded") return value def _normalize_role(self, role: str) -> str: """Normalize caller roles accepted by the contract.""" return str(role or "").strip().lower() def _caller_key(self, org_id: str, caller: str) -> str: return f"{org_id}:{self._normalize_address(caller)}" def _get_org_caller_record(self, org_id: str, caller: str) -> object: raw = self.organization_callers.get(self._caller_key(org_id, caller)) if not raw: return None try: return json.loads(raw) except Exception: return None def _is_contract_owner(self, caller: str) -> bool: return self._normalize_address(caller) == self._normalize_address(self.contract_owner) def _require_org_role(self, org_id: str, allowed_roles: tuple) -> str: """Require the current sender to be bound to this org namespace.""" caller = self._current_sender() record = self._get_org_caller_record(org_id, caller) if not record: raise gl.vm.UserError("caller is not authorized for this organization") role = self._normalize_role(record.get("role", "")) if role not in allowed_roles: raise gl.vm.UserError("caller role is not authorized for this action") return caller def _verify_submitter_wallet(self, data: dict, caller: str) -> None: """Reject payloads that try to spoof the submitter wallet.""" supplied = data.get("submitter_wallet") if supplied and self._normalize_address(str(supplied)) != caller: raise gl.vm.UserError("submitter_wallet must match transaction sender") def _verify_evidence_anchor(self, org_id: str, invoice_id: str, data: dict) -> str: """Ensure supplied evidence_hash matches canonical invoice facts.""" expected = self._normalize_hash(data.get("evidence_hash", ""), "evidence_hash") file_hash = self._normalize_hash(data.get("file_hash", ""), "file_hash") data["file_hash"] = file_hash computed = self._compute_evidence_hash(org_id, invoice_id, data) if expected != computed: raise gl.vm.UserError("evidence_hash does not match invoice evidence") return f"0x{computed}" def _compute_evidence_hash(self, org_id: str, invoice_id: str, data: dict) -> str: canonical = json.dumps( self._canonical_evidence(org_id, invoice_id, data), sort_keys=True, separators=(",", ":"), ) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() def _canonical_evidence(self, org_id: str, invoice_id: str, data: dict) -> dict: return { "org_id": org_id, "invoice_id": invoice_id, "invoice_number": str(data.get("invoice_number", "")).strip(), "amount": self._format_amount(data.get("amount", 0)), "currency": str(data.get("currency", "USD")).strip().upper(), "issue_date": str(data.get("issue_date", "") or ""), "due_date": str(data.get("due_date", "") or ""), "vendor_id": str(data.get("vendor_id", "")).strip(), "vendor_name": str(data.get("vendor_name", "")).strip(), "file_hash": self._normalize_hash(data.get("file_hash", ""), "file_hash"), "line_items": self._normalize_line_items(data.get("line_items", [])), } def _normalize_line_items(self, line_items: object) -> list: if not isinstance(line_items, list): return [] normalized = [] for item in line_items: if not isinstance(item, dict): continue normalized.append({ "description": str(item.get("description", "")).strip(), "quantity": self._format_quantity(item.get("quantity", 0)), "unit_price": self._format_amount(item.get("unit_price", 0)), "total": self._format_amount(item.get("total", 0)), }) return normalized def _format_amount(self, value) -> str: try: return f"{float(value):.2f}" except Exception: return "0.00" def _format_quantity(self, value) -> str: try: formatted = f"{float(value):.4f}" except Exception: formatted = "0.0000" return formatted.rstrip("0").rstrip(".") or "0" def _normalize_hash(self, value, field_name: str) -> str: cleaned = str(value or "").strip().lower() if cleaned.startswith("sha256:"): cleaned = cleaned[7:] if cleaned.startswith("0x"): cleaned = cleaned[2:] if len(cleaned) != 64: raise gl.vm.UserError(f"{field_name} must be a sha256 hash") for ch in cleaned: if ch not in "0123456789abcdef": raise gl.vm.UserError(f"{field_name} must be hex encoded") return cleaned def _parse_submission_payload(self, payload: str) -> dict: """Parse and minimally sanity-check the incoming submission payload.""" try: data = json.loads(payload) except Exception as exc: raise gl.vm.UserError(f"payload is not valid JSON: {exc}") if not isinstance(data, dict): raise gl.vm.UserError("payload must decode to a JSON object") return data def _validate_required_fields(self, data: dict) -> None: """Raise if any required submission field is missing or falsy.""" for field in REQUIRED_SUBMISSION_FIELDS: if not data.get(field): raise gl.vm.UserError(f"Missing required field: {field}") def _index_invoice_number(self, org_id: str, data: dict) -> None: """Maintain the secondary index used for O(1) duplicate detection.""" vendor_id = data.get("vendor_id", "") invoice_number = data.get("invoice_number", "") if not vendor_id or not invoice_number: return index_key = f"{org_id}:{vendor_id}:{invoice_number}" if index_key not in self.invoice_number_index: self.invoice_number_index[index_key] = data.get("invoice_id", "") def _find_duplicate_invoice_number( self, org_id: str, vendor_id: str, invoice_number: str ) -> str: """Return the invoice_id already indexed under this vendor/number, or ''.""" if not vendor_id or not invoice_number: return "" index_key = f"{org_id}:{vendor_id}:{invoice_number}" return self.invoice_number_index.get(index_key) or "" def _blocked_vendor_reason(self, org_id: str, vendor_id: str) -> object: """Return the block reason string if this vendor is blocked, else None.""" if not vendor_id: return None vendor_key = self._vendor_key(org_id, vendor_id) raw = self.vendor_blocklist.get(vendor_key) if not raw: return None try: record = json.loads(raw) return record.get("reason", "blocked") except Exception: return "blocked" def _get_vendor_history_dict(self, org_id: str, vendor_id: str) -> object: """Return the parsed VendorSummary dict for a vendor, or None if absent.""" if not vendor_id: return None raw = self.vendor_history.get(self._vendor_key(org_id, vendor_id)) if not raw: return None try: return json.loads(raw) except Exception: return None def _get_vendor_verification_dict(self, org_id: str, vendor_id: str) -> object: """Return the parsed VendorVerification dict for a vendor, or None if absent.""" if not vendor_id: return None raw = self.vendor_verifications.get(self._vendor_key(org_id, vendor_id)) if not raw: return None try: return json.loads(raw) except Exception: return None def _get_org_stats_dict(self, org_id: str) -> dict: """Return the parsed OrganizationStats dict for an org, defaulting to empty.""" raw = self.organization_stats.get(org_id) if not raw: return {} try: return json.loads(raw) except Exception: return {} def _get_org_thresholds(self, org_id: str) -> tuple: """Return (approve_below, reject_at_or_above) for an org, falling back to defaults.""" raw = self.organization_settings.get(org_id) if not raw: return DEFAULT_APPROVE_BELOW, DEFAULT_REJECT_AT_OR_ABOVE try: settings = json.loads(raw) return ( int(settings.get("approve_below", DEFAULT_APPROVE_BELOW)), int(settings.get("reject_at_or_above", DEFAULT_REJECT_AT_OR_ABOVE)), ) except Exception: return DEFAULT_APPROVE_BELOW, DEFAULT_REJECT_AT_OR_ABOVE def _format_vendor_history(self, vendor_hist: object) -> str: """Render a VendorSummary dict as prompt-friendly text.""" if not vendor_hist: return "No prior history with this organization." return ( f"Total invoices: {vendor_hist.get('total_invoices', 0)}, " f"Approved: {vendor_hist.get('approved_invoices', 0)}, " f"Rejected: {vendor_hist.get('rejected_invoices', 0)}, " f"Escalated: {vendor_hist.get('escalated_invoices', 0)}, " f"Flagged: {vendor_hist.get('flagged_invoices', 0)}, " f"Avg invoice amount: ${float(vendor_hist.get('avg_invoice_amount', 0)):.2f}" ) def _format_vendor_verification(self, verification: object) -> str: """Render a VendorVerification dict as prompt-friendly text.""" if not verification: return "Not yet verified." return ( f"Verdict: {verification.get('verdict', 'unverifiable')} " f"(confidence {verification.get('confidence', 0)}/100). " f"{verification.get('reasoning', '')}" ) def _format_anomaly_flags(self, flags: object) -> str: """Render the deterministic pre-check anomaly flags as prompt-friendly text.""" if not flags: return "None" lines = [] for flag in flags: severity = str(flag.get("severity", "")).upper() flag_type = flag.get("type", "") detail = flag.get("detail", "") lines.append(f"- [{severity}] {flag_type}: {detail}") return "\n".join(lines) if lines else "None" def _format_line_items(self, line_items: object) -> str: """Render invoice line items as prompt-friendly text.""" if not line_items: return "No line items provided" lines = [] for item in line_items: description = item.get("description", "N/A") quantity = item.get("quantity", 1) unit_price = float(item.get("unit_price", 0)) total = float(item.get("total", 0)) lines.append( f"- {description}: qty {quantity} x ${unit_price:.2f} = ${total:.2f}" ) return "\n".join(lines) if lines else "No line items provided" def _invoice_key(self, org_id: str, invoice_id: str) -> str: return f"{org_id}:{invoice_id}" def _vendor_key(self, org_id: str, vendor_id: str) -> str: return f"{org_id}:{vendor_id}" def _now_iso(self) -> str: return datetime.datetime.now(datetime.UTC).replace(tzinfo=None).isoformat() + "Z" # ================================================================== # SECTION 8: Public read methods # ================================================================== @gl.public.view def get_validation_result(self, org_id: str, invoice_id: str) -> str: """ Returns JSON string of the ValidationResult, or empty string if not yet resolved. Called by the sync-genlayer-result Edge Function to detect when consensus has finalized. """ return self.validations.get(self._invoice_key(org_id, invoice_id)) or "" @gl.public.view def get_submission(self, org_id: str, invoice_id: str) -> str: """Returns JSON string of the InvoiceSubmission, or empty string.""" return self.submissions.get(self._invoice_key(org_id, invoice_id)) or "" @gl.public.view def get_vendor_history(self, org_id: str, vendor_id: str) -> str: """Returns JSON string of the VendorSummary, or empty string.""" return self.vendor_history.get(self._vendor_key(org_id, vendor_id)) or "" @gl.public.view def is_vendor_blocked(self, org_id: str, vendor_id: str) -> str: """Returns the JSON-encoded BlockRecord if blocked, or empty string.""" return self.vendor_blocklist.get(self._vendor_key(org_id, vendor_id)) or "" @gl.public.view def get_vendor_verification(self, org_id: str, vendor_id: str) -> str: """Returns JSON string of the cached VendorVerification, or empty string.""" return self.vendor_verifications.get(self._vendor_key(org_id, vendor_id)) or "" @gl.public.view def get_arbitration(self, org_id: str, invoice_id: str) -> str: """Returns JSON string of the ArbitrationRecord, or empty string.""" return self.arbitrations.get(self._invoice_key(org_id, invoice_id)) or "" @gl.public.view def get_authorized_org_caller(self, org_id: str, caller_address: str) -> str: """Returns a caller role binding for an organization, or empty string.""" caller = self._normalize_address(caller_address) return self.organization_callers.get(self._caller_key(org_id, caller)) or "" @gl.public.view def get_organization_settings(self, org_id: str) -> str: """ Returns JSON-encoded {approve_below, reject_at_or_above} for an organization, or the contract-wide defaults if never configured. """ approve_below, reject_at_or_above = self._get_org_thresholds(org_id) return json.dumps({ "org_id": org_id, "approve_below": approve_below, "reject_at_or_above": reject_at_or_above, }) @gl.public.view def get_organization_stats(self, org_id: str) -> str: """Returns JSON-encoded running counters for an organization.""" return json.dumps(self._get_org_stats_dict(org_id)) @gl.public.view def get_audit_trail(self, org_id: str, invoice_id: str) -> str: """ Returns a JSON array string of every audit entry for an organization, optionally filtered to a single invoice. Pass an empty string for invoice_id to get all entries for the whole organization. Equivalent to calling get_audit_trail_page(org_id, invoice_id, 0, ). """ return json.dumps( self._collect_audit_entries(org_id, invoice_id, offset=0, limit=-1) ) @gl.public.view def get_audit_trail_page( self, org_id: str, invoice_id: str, offset: int, limit: int ) -> str: """ Returns a JSON array string of up to `limit` matching audit entries, skipping the first `offset` matches. Intended for UIs that need to paginate a long-running organization's history instead of fetching every entry at once. Args: org_id: UUID of the organization to filter by. invoice_id: UUID to filter by, or "" for all invoices. offset: Number of matching entries to skip. limit: Maximum number of matching entries to return. """ return json.dumps( self._collect_audit_entries(org_id, invoice_id, offset, limit) ) @gl.public.view def get_contract_info(self) -> str: """Returns JSON-encoded contract metadata, useful for client-side compatibility checks.""" return json.dumps({ "contract": "InvoiceValidator", "version": CONTRACT_VERSION, "valid_decisions": list(VALID_DECISIONS), "contract_owner": self.contract_owner, "default_approve_below": DEFAULT_APPROVE_BELOW, "default_reject_at_or_above": DEFAULT_REJECT_AT_OR_ABOVE, }) def _collect_audit_entries( self, org_id: str, invoice_id: str, offset: int, limit: int ) -> list: """Shared filtering/pagination logic for the audit trail view methods.""" entries = [] matched = 0 for i in range(len(self.audit_trail)): try: entry = json.loads(self.audit_trail[i]) except Exception: continue if entry.get("org_id") != org_id: continue if invoice_id and entry.get("invoice_id") != invoice_id: continue if matched < offset: matched += 1 continue matched += 1 entries.append(entry) if limit >= 0 and len(entries) >= limit: break return entries