# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json ALLOWED_REVIEW_TYPES = [ "SUPPLIER", "ORDER", "TRANSPORT", "LOGISTICS_PLAN", "INVOICE", "INSURANCE", "STOCK", ] ALLOWED_DECISIONS = ["APPROVE", "NEEDS_REVIEW", "REJECT"] ALLOWED_RISK_LEVELS = ["LOW", "MEDIUM", "HIGH", "CRITICAL"] ALLOWED_SOURCE_TYPES = [ "CUSTOMS", "GPS", "WAREHOUSE", "ERP", "INVOICE", "CONTRACT", "WEATHER", "CARRIER", "HISTORICAL_RECORDS", "INSURANCE_FILE", "OTHER", ] def _normalize_choice(value, allowed: list[str], fallback: str) -> str: normalized = str(value or "").strip().upper().replace(" ", "_").replace("-", "_") if normalized in allowed: return normalized return fallback def _to_int_0_100(value) -> int: try: number = int(round(float(str(value).strip()))) except Exception: raise gl.vm.UserError("[LLM_ERROR] numeric field is invalid") if number < 0: return 0 if number > 100: return 100 return number def _to_bool(value) -> bool: if isinstance(value, bool): return value if isinstance(value, str): return value.strip().lower() in ["true", "yes", "1", "required", "needed"] return bool(value) def _short_text(value, limit: int) -> str: text = str(value or "").strip() if len(text) > limit: return text[:limit] return text def _parse_review(raw: str) -> dict: first = raw.find("{") last = raw.rfind("}") if first < 0 or last < first: raise gl.vm.UserError("[LLM_ERROR] missing JSON object") try: data = json.loads(raw[first:last + 1]) except Exception: raise gl.vm.UserError("[LLM_ERROR] invalid JSON") decision = _normalize_choice(data.get("decision"), ALLOWED_DECISIONS, "NEEDS_REVIEW") risk_level = _normalize_choice(data.get("risk_level"), ALLOWED_RISK_LEVELS, "MEDIUM") return { "decision": decision, "risk_level": risk_level, "risk_score": _to_int_0_100(data.get("risk_score")), "confidence": _to_int_0_100(data.get("confidence")), "requires_human_review": _to_bool(data.get("requires_human_review")), "main_findings": _short_text(data.get("main_findings"), 320), "recommended_action": _short_text(data.get("recommended_action"), 260), "evidence_gap": _short_text(data.get("evidence_gap"), 220), } def _decision_family(decision: str) -> str: if decision == "APPROVE": return "pass" if decision == "NEEDS_REVIEW": return "review" if decision == "REJECT": return "block" return "review" def _risk_family(risk_level: str) -> str: if risk_level == "LOW": return "pass" if risk_level == "MEDIUM": return "review" if risk_level in ["HIGH", "CRITICAL"]: return "block" return "review" class ChainGuardRiskReview(gl.Contract): reviews: DynArray[str] review_count: u32 owner: Address def __init__(self): self.owner = gl.message.sender_address self.review_count = 0 @gl.public.view def get_review_count(self) -> int: return self.review_count @gl.public.view def get_reviews(self) -> list: return list(self.reviews) @gl.public.view def get_review(self, review_id: int) -> dict: if review_id < 1 or review_id > self.review_count: raise gl.vm.UserError("[EXPECTED] review not found") return json.loads(self.reviews[review_id - 1]) @gl.public.write def review_supply_chain_decision( self, review_type: str, entity_name: str, decision_request: str, evidence_summary: str, source_types: str, evidence_hash: str, ) -> None: normalized_type = _normalize_choice(review_type, ALLOWED_REVIEW_TYPES, "") entity = entity_name.strip() request = decision_request.strip() evidence = evidence_summary.strip() sources = source_types.strip() ev_hash = evidence_hash.strip() if normalized_type == "": raise gl.vm.UserError("[EXPECTED] invalid review type") if len(entity) < 3: raise gl.vm.UserError("[EXPECTED] entity name is required") if len(request) < 30: raise gl.vm.UserError("[EXPECTED] decision request must be detailed") if len(evidence) < 80: raise gl.vm.UserError("[EXPECTED] trusted evidence summary is required") if len(sources) < 3: raise gl.vm.UserError("[EXPECTED] source types are required") if len(request) > 1400 or len(evidence) > 1800 or len(sources) > 500: raise gl.vm.UserError("[EXPECTED] review input is too long") def leader_fn() -> dict: prompt = f"""You are ChainGuard, a supply-chain risk decision support reviewer. Review a logistics or supply-chain decision using only the provided trusted evidence summary. Do not invent facts. If evidence is weak, incomplete, old, or not from trusted sources, require human review. Your role is to support professional human judgment, not replace it. REVIEW TYPE: {normalized_type} ENTITY NAME: {entity} DECISION REQUEST: {request} TRUSTED EVIDENCE SUMMARY: {evidence} SOURCE TYPES: {sources} Return ONLY JSON: {{ "decision": "APPROVE" | "NEEDS_REVIEW" | "REJECT", "risk_level": "LOW" | "MEDIUM" | "HIGH" | "CRITICAL", "risk_score": integer 0-100, "confidence": integer 0-100, "requires_human_review": true/false, "main_findings": "key evidence-based findings, max 320 chars", "recommended_action": "practical next step for the logistics team, max 260 chars", "evidence_gap": "missing or weak evidence, or 'none', max 220 chars" }} Decision rules: - APPROVE only if evidence strongly supports the plan and risk is low. - NEEDS_REVIEW if evidence is incomplete, risk is medium, or human judgment is needed. - REJECT if there is high fraud, safety, legal, quantity, invoice, customs, theft, or delivery risk. - High or critical risk must require human review. - Never claim external facts that are not in the evidence summary. Pure JSON only.""" result = ( gl.nondet.exec_prompt(prompt) .replace("```json", "") .replace("```", "") ) return _parse_review(result) def validator_fn(leaders_res: gl.vm.Result) -> bool: if not isinstance(leaders_res, gl.vm.Return): return False leader = leaders_res.calldata validator = leader_fn() if len(leader["main_findings"]) < 20: return False if len(leader["recommended_action"]) < 12: return False same_decision_family = ( _decision_family(leader["decision"]) == _decision_family(validator["decision"]) ) same_risk_family = ( _risk_family(leader["risk_level"]) == _risk_family(validator["risk_level"]) ) close_risk_score = abs(leader["risk_score"] - validator["risk_score"]) <= 30 close_confidence = abs(leader["confidence"] - validator["confidence"]) <= 35 if leader["risk_level"] in ["HIGH", "CRITICAL"] and not leader["requires_human_review"]: return False return same_decision_family or (same_risk_family and close_risk_score and close_confidence) review = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) record = { "id": str(self.review_count + 1), "submitter": str(gl.message.sender_address), "review_type": normalized_type, "entity_name": entity, "decision_request": request, "evidence_summary": evidence, "source_types": sources, "evidence_hash": ev_hash, "decision": review["decision"], "risk_level": review["risk_level"], "risk_score": review["risk_score"], "confidence": review["confidence"], "requires_human_review": review["requires_human_review"], "main_findings": review["main_findings"], "recommended_action": review["recommended_action"], "evidence_gap": review["evidence_gap"], } self.reviews.append(json.dumps(record)) self.review_count += 1