# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } import hashlib import json import re from datetime import datetime, timedelta, timezone from dataclasses import dataclass from genlayer import * ERROR_EXPECTED = "[EXPECTED]" ERROR_LLM = "[LLM_ERROR]" MAX_SOURCE_EXCERPT_LENGTH = 2000 MAX_AGREEMENT_ID_LENGTH = 128 MAX_EVIDENCE_ID_LENGTH = 128 MAX_OBLIGATION_ID_LENGTH = 128 MAX_DESCRIPTION_LENGTH = 2000 MAX_SOURCE_REFERENCE_LENGTH = 2048 MAX_PARTY_CONTENT_LENGTH = 4000 MAX_SOURCE_CONTENT_LENGTH = 16000 MAX_PARTIES = 32 MAX_OBLIGATIONS = 64 MAX_EVIDENCE_ITEMS = 64 APPEAL_WINDOW_SECONDS = 24 * 60 * 60 MAX_RECOVERY_ATTEMPTS = 1 @allow_storage @dataclass class Obligation: obligation_id: str description: str responsible_party: str deadline: str rule_type: str # "NUMERIC", "DEADLINE", "SEMANTIC" target_value: str comparison_op: str # ">=", "<=", "==", "<", ">", "EVAL" is_mandatory: bool @allow_storage @dataclass class Evidence: evidence_id: str submitted_by: str description: str source_reference: str content: str source_acquisition_status: str # "FETCHED", "UNAVAILABLE", "INVALID", "NOT_PROVIDED" source_verification_status: str # "VERIFIED" or "UNVERIFIED" source_hash: str evidence_commitment: str source_excerpt: str canonical_source_content: str @allow_storage @dataclass class ObligationResult: obligation_id: str status: str # "FULFILLED", "NOT_FULFILLED", "INSUFFICIENT_EVIDENCE" evidence_sufficient: bool rule_type: str observed_fact: str target_value: str comparison_op: str explanation: str @allow_storage @dataclass class Agreement: agreement_id: str creator: Address debtor: Address creditor: Address escrow_amount: u256 escrow_status: str # "NONE", "UNFUNDED", "FUNDED", "LOCKED", "RELEASED", "REFUNDED" appeal_count: u256 appeal_deadline: str deadline: str status: str # "ACTIVE", "VERIFIED", "REVERIFIED", "FINALIZED" verification_status: str # "NONE", "FULFILLED", "NOT_FULFILLED", "INSUFFICIENT_EVIDENCE" verification_explanation: str parties_json: str obligations_json: str evidence_json: str results_json: str recovery_baseline_evidence_json: str recovery_count: u256 def _handle_leader_error(leaders_res: gl.vm.Result, leader_fn) -> bool: leader_msg = leaders_res.message if hasattr(leaders_res, "message") else "" try: leader_fn() return False # Leader errored, validator succeeded — disagree except gl.vm.UserError as e: validator_msg = e.message if hasattr(e, "message") else str(e) if validator_msg.startswith(ERROR_EXPECTED): return validator_msg == leader_msg return False except Exception: return False def _parse_llm_json(raw: dict | str) -> dict: if isinstance(raw, dict): return raw if isinstance(raw, str): cleaned = raw.strip() first_brace = cleaned.find("{") last_brace = cleaned.rfind("}") if first_brace != -1 and last_brace != -1: cleaned = cleaned[first_brace : last_brace + 1] try: return json.loads(cleaned) except Exception as e: raise gl.vm.UserError(f"{ERROR_LLM} Failed to parse LLM JSON: {e}") raise gl.vm.UserError(f"{ERROR_LLM} Unexpected LLM output format: {type(raw)}") def _normalize_address(addr) -> Address: if isinstance(addr, Address): return addr if isinstance(addr, (bytes, bytearray)): return Address(addr) if isinstance(addr, str): return Address(addr) return Address(addr) def _normalize_address_str(addr) -> str: if isinstance(addr, Address): return str(addr) if isinstance(addr, (bytes, bytearray)): return str(Address(addr)) if isinstance(addr, str): return str(Address(addr)) if addr.startswith("0x") else addr return str(Address(addr)) def _is_valid_iso_date(val: str) -> bool: if not isinstance(val, str) or len(val) != 10: return False parts = val.split("-") if len(parts) != 3: return False try: year = int(parts[0]) month = int(parts[1]) day = int(parts[2]) if not (1900 <= year <= 2100 and 1 <= month <= 12): return False leap_year = year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) days_by_month = [31, 29 if leap_year else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31] return 1 <= day <= days_by_month[month - 1] except (ValueError, TypeError): return False def _parse_integer_safely(val) -> int | None: if val is None: return None if isinstance(val, int) and not isinstance(val, bool): return val try: s = str(val).strip().replace(",", "") if "." in s: return None return int(s) except (ValueError, TypeError): return None def _evaluate_numeric_obligation(ob: dict, extracted: dict) -> dict: target_raw = ob.get("target_value", "0") target_int = _parse_integer_safely(target_raw) if target_int is None: return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "NUMERIC", "observed_fact": "", "target_value": target_raw, "comparison_op": ob.get("comparison_op", ">="), "explanation": f"Invalid target_value integer in obligation definition: '{target_raw}'", } if not extracted.get("evidence_sufficient", False): return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "NUMERIC", "observed_fact": "", "target_value": str(target_int), "comparison_op": ob.get("comparison_op", ">="), "explanation": str(extracted.get("explanation", "Insufficient evidence for numeric value.")), } raw_observed = extracted.get("observed_numeric") observed_int = _parse_integer_safely(raw_observed) if observed_int is None: return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "NUMERIC", "observed_fact": str(raw_observed or ""), "target_value": str(target_int), "comparison_op": ob.get("comparison_op", ">="), "explanation": f"Observed numeric value could not be reliably parsed as integer: '{raw_observed}'", } op = ob.get("comparison_op", ">=").strip() if op == ">=": is_ok = observed_int >= target_int elif op == "<=": is_ok = observed_int <= target_int elif op == "==": is_ok = observed_int == target_int elif op == "<": is_ok = observed_int < target_int elif op == ">": is_ok = observed_int > target_int else: is_ok = observed_int >= target_int return { "id": ob["obligation_id"], "status": "FULFILLED" if is_ok else "NOT_FULFILLED", "evidence_sufficient": True, "rule_type": "NUMERIC", "observed_fact": str(observed_int), "target_value": str(target_int), "comparison_op": op, "explanation": str(extracted.get("explanation", "")), } def _evaluate_deadline_obligation(ob: dict, extracted: dict) -> dict: target_deadline = ob.get("deadline", "") or ob.get("target_value", "") if not _is_valid_iso_date(target_deadline): return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "DEADLINE", "observed_fact": "", "target_value": target_deadline, "comparison_op": "<=", "explanation": f"Invalid target deadline date in obligation definition: '{target_deadline}'", } if not extracted.get("evidence_sufficient", False): return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "DEADLINE", "observed_fact": "", "target_value": target_deadline, "comparison_op": "<=", "explanation": str(extracted.get("explanation", "Insufficient evidence for delivery date.")), } observed_date = str(extracted.get("observed_date", "")).strip() if not _is_valid_iso_date(observed_date): return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "DEADLINE", "observed_fact": observed_date, "target_value": target_deadline, "comparison_op": "<=", "explanation": f"Observed date could not be parsed as valid YYYY-MM-DD: '{observed_date}'", } is_on_or_before = observed_date <= target_deadline return { "id": ob["obligation_id"], "status": "FULFILLED" if is_on_or_before else "NOT_FULFILLED", "evidence_sufficient": True, "rule_type": "DEADLINE", "observed_fact": observed_date, "target_value": target_deadline, "comparison_op": "<=", "explanation": str(extracted.get("explanation", "")), } def _evaluate_semantic_obligation(ob: dict, extracted: dict) -> dict: if not extracted.get("evidence_sufficient", False): return { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": "SEMANTIC", "observed_fact": "INSUFFICIENT_EVIDENCE", "target_value": ob.get("target_value", ""), "comparison_op": "EVAL", "explanation": str(extracted.get("explanation", "Insufficient evidence for semantic compliance.")), } raw_status = str(extracted.get("semantic_status", "INSUFFICIENT_EVIDENCE")).upper().strip() if raw_status not in ("FULFILLED", "NOT_FULFILLED", "INSUFFICIENT_EVIDENCE"): raw_status = "INSUFFICIENT_EVIDENCE" is_suff = raw_status != "INSUFFICIENT_EVIDENCE" return { "id": ob["obligation_id"], "status": raw_status, "evidence_sufficient": is_suff, "rule_type": "SEMANTIC", "observed_fact": raw_status, "target_value": ob.get("target_value", ""), "comparison_op": "EVAL", "explanation": str(extracted.get("explanation", "")), } def _derive_agreement_status(obligation_results: list[dict], obs_defs: list[dict]) -> str: if not obligation_results: return "INSUFFICIENT_EVIDENCE" def_map = {d["obligation_id"]: d for d in obs_defs} mandatory_statuses = [] for r in obligation_results: d = def_map.get(r["id"], {}) if d.get("is_mandatory", True): mandatory_statuses.append(r["status"]) if not mandatory_statuses: return "INSUFFICIENT_EVIDENCE" if any(s == "NOT_FULFILLED" for s in mandatory_statuses): return "NOT_FULFILLED" if any(s == "INSUFFICIENT_EVIDENCE" for s in mandatory_statuses): return "INSUFFICIENT_EVIDENCE" if all(s == "FULFILLED" for s in mandatory_statuses): return "FULFILLED" return "INSUFFICIENT_EVIDENCE" def _normalize_source_content(content: str) -> str: if not isinstance(content, str): return "" content = content.replace("\r\n", "\n").replace("\r", "\n") return content.strip() def _compute_sha256_hash(text: str) -> str: if not text: return "" return hashlib.sha256(text.encode("utf-8")).hexdigest() def _current_time() -> datetime: # GenVM exposes the transaction clock through its deterministic datetime # implementation; direct mode's warp() patches this same clock. return datetime.now(timezone.utc) def _appeal_deadline() -> str: return (_current_time() + timedelta(seconds=APPEAL_WINDOW_SECONDS)).isoformat().replace("+00:00", "Z") def _appeal_window_active(deadline: str) -> bool: if not deadline: return True return _current_time() < datetime.fromisoformat(deadline.replace("Z", "+00:00")) def _unverified_evidence_result(obs_data: list[dict], evidence_sources: list[dict]) -> dict: results = [] for ob in obs_data: results.append({ "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": ob.get("rule_type", "SEMANTIC"), "observed_fact": "", "target_value": ob.get("target_value", ""), "comparison_op": ob.get("comparison_op", ">="), "explanation": "Referenced evidence was not acquired and verified by the contract.", }) return { "agreement_status": "INSUFFICIENT_EVIDENCE", "obligations": results, "evidence_sources": evidence_sources, "explanation": "Settlement is locked because every submitted evidence source must be acquired and verified.", } def _has_complete_unique_id_set(items: list[dict], id_key: str, expected_ids: set[str]) -> bool: """Require a one-to-one, canonical ID set for a consequential result.""" if not isinstance(items, list): return False max_len = MAX_OBLIGATION_ID_LENGTH if id_key == "id" else MAX_EVIDENCE_ID_LENGTH actual_ids = [] for item in items: if not isinstance(item, dict): return False item_id = item.get(id_key) if not isinstance(item_id, str) or not item_id.strip(): return False if len(item_id) > max_len: return False actual_ids.append(item_id) return len(actual_ids) == len(set(actual_ids)) and set(actual_ids) == expected_ids def _consensus_results_equivalent( leader_data: dict, validator_data: dict, expected_obligation_ids: set[str], expected_evidence_ids: set[str], ) -> bool: """Compare every leader/validator field that can affect settlement.""" if not isinstance(leader_data, dict) or not isinstance(validator_data, dict): return False if leader_data.get("agreement_status") != validator_data.get("agreement_status"): return False if leader_data.get("agreement_status") not in ( "FULFILLED", "NOT_FULFILLED", "INSUFFICIENT_EVIDENCE", ): return False leader_obs = leader_data.get("obligations", []) validator_obs = validator_data.get("obligations", []) if not _has_complete_unique_id_set(leader_obs, "id", expected_obligation_ids): return False if not _has_complete_unique_id_set(validator_obs, "id", expected_obligation_ids): return False validator_by_id = {item["id"]: item for item in validator_obs} for leader_obligation in leader_obs: validator_obligation = validator_by_id[leader_obligation["id"]] if any( leader_obligation.get(field) != validator_obligation.get(field) for field in ( "status", "evidence_sufficient", "observed_fact", "rule_type", "target_value", "comparison_op", ) ): return False leader_evs = leader_data.get("evidence_sources", []) validator_evs = validator_data.get("evidence_sources", []) if not _has_complete_unique_id_set(leader_evs, "evidence_id", expected_evidence_ids): return False if not _has_complete_unique_id_set(validator_evs, "evidence_id", expected_evidence_ids): return False validator_evidence_by_id = {item["evidence_id"]: item for item in validator_evs} for leader_evidence in leader_evs: validator_evidence = validator_evidence_by_id[leader_evidence["evidence_id"]] if any( leader_evidence.get(field) != validator_evidence.get(field) for field in ( "source_reference", "source_acquisition_status", # finalize() consumes this exact persisted field. "source_verification_status", "source_hash", "evidence_commitment", "canonical_source_content", ) ): return False return True def _evidence_commitments_match(previous_evidence: list[dict], fresh_evidence: list[dict]) -> bool: """Keep original evidence commitments stable across recovery retries.""" expected_ids = { item.get("evidence_id") for item in previous_evidence if isinstance(item, dict) and isinstance(item.get("evidence_id"), str) } if not _has_complete_unique_id_set(previous_evidence, "evidence_id", expected_ids): return False fresh_ids = { item.get("evidence_id") for item in fresh_evidence if isinstance(item, dict) and isinstance(item.get("evidence_id"), str) } # Recovery may append new evidence, but every fresh ID must still be # non-empty, unique, valid, and represented by the acquired result. if not _has_complete_unique_id_set(fresh_evidence, "evidence_id", fresh_ids): return False fresh_by_id = {item["evidence_id"]: item for item in fresh_evidence} for previous in previous_evidence: if previous["evidence_id"] not in fresh_by_id: return False fresh = fresh_by_id[previous["evidence_id"]] if any( previous.get(field) != fresh.get(field) for field in ( "source_reference", "source_acquisition_status", "source_hash", "evidence_commitment", "canonical_source_content", "source_verification_status", ) ): return False return True def _acquire_evidence_source(ev: dict) -> dict: ev_id = str(ev.get("evidence_id", "")) submitted_by = str(ev.get("submitted_by", "")) desc = str(ev.get("description", "")) source_ref = str(ev.get("source_reference", "")).strip() if not source_ref: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": "", "content": "", "source_acquisition_status": "NOT_PROVIDED", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } is_http_url = source_ref.startswith("http://") or source_ref.startswith("https://") if not is_http_url: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, "content": "", "source_acquisition_status": "INVALID", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } try: resp = gl.nondet.web.get(source_ref) if resp is None: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, "content": "", "source_acquisition_status": "UNAVAILABLE", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } status_code = getattr(resp, "status", None) body = getattr(resp, "body", None) if status_code != 200 or body is None: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, "content": "", "source_acquisition_status": "UNAVAILABLE", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } if isinstance(body, bytes): raw_text = body.decode("utf-8", errors="replace") else: raw_text = str(body) normalized_text = _normalize_source_content(raw_text) if not normalized_text: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, "content": "", "source_acquisition_status": "UNAVAILABLE", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } if len(normalized_text) > MAX_SOURCE_CONTENT_LENGTH: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, "content": "", "source_acquisition_status": "INVALID", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } source_hash = _compute_sha256_hash(normalized_text) evidence_commitment = _compute_sha256_hash(source_ref + "\n" + source_hash) bounded_excerpt = normalized_text[:MAX_SOURCE_EXCERPT_LENGTH] return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, # Party content is retained only in the local call frame. It is # deliberately not persisted as the evaluated evidence. "content": "", "source_acquisition_status": "FETCHED", "source_verification_status": "VERIFIED", "source_hash": source_hash, "evidence_commitment": evidence_commitment, "source_excerpt": bounded_excerpt, "canonical_source_content": normalized_text, } except Exception: return { "evidence_id": ev_id, "submitted_by": submitted_by, "description": desc, "source_reference": source_ref, "content": "", "source_acquisition_status": "UNAVAILABLE", "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } def _run_hybrid_consensus(obs_data: list[dict], ev_data: list[dict]) -> dict: def leader_fn() -> dict: if len(ev_data) == 0: empty_results = [] for ob in obs_data: empty_results.append( { "id": ob["obligation_id"], "status": "INSUFFICIENT_EVIDENCE", "evidence_sufficient": False, "rule_type": ob["rule_type"], "observed_fact": "", "target_value": ob.get("target_value", ""), "comparison_op": ob.get("comparison_op", ">="), "explanation": "No evidence was submitted for this agreement.", } ) return { "agreement_status": "INSUFFICIENT_EVIDENCE", "obligations": empty_results, "evidence_sources": [], "explanation": "No evidence submitted to evaluate obligations.", } acquired_evidence = [_acquire_evidence_source(ev) for ev in ev_data] if any( ev.get("source_acquisition_status") != "FETCHED" or ev.get("source_verification_status") != "VERIFIED" or not ev.get("source_hash") or not ev.get("evidence_commitment") for ev in acquired_evidence ): return _unverified_evidence_result(obs_data, acquired_evidence) prompt = f"""You are the ObliGen Intelligent Contract fact extraction engine. Carefully examine the acquired evidence sources and submitted data to extract observed facts for each obligation. AGREEMENT OBLIGATIONS: {json.dumps(obs_data, indent=2)} CONTRACT-ACQUIRED AND VERIFIED EVIDENCE: {json.dumps(acquired_evidence, indent=2)} CRITICAL EVALUATION AND EVIDENCE AUTHENTICATION RULES: 1. PRIMARY SOURCE OF TRUTH: For any evidence item with source_acquisition_status == "FETCHED", the fetched "source_excerpt" and authenticated source is the primary authoritative source of truth. 2. EVALUATION INPUT: Evaluate only canonical_source_content returned by the contract-side acquisition. The party-submitted content is never an evaluation input and cannot override the acquired source. 3. SOURCE COMMITMENT: source_hash commits to the normalized acquired representation; evidence_commitment binds that hash to source_reference. These commitments establish representation integrity, not truth of the external source. 4. UNAVAILABLE / INVALID SOURCES: These are rejected before this prompt is reached and cannot produce a consequential status. 5. EXTRACTION REQUIREMENTS: - "id": The obligation ID. - "evidence_sufficient": true ONLY if authentic, verified evidence conclusively establishes the required fact. If evidence is missing, unverified, contradictory, or insufficient, set false. - "observed_numeric": If rule_type is "NUMERIC", extract the exact integer quantity delivered/completed (e.g. 1000). If not mentioned or insufficient, return null. Do NOT perform arithmetic or inequality comparison. - "observed_date": If rule_type is "DEADLINE", extract the exact fulfillment date formatted strictly as YYYY-MM-DD (e.g. "2026-08-28"). If not mentioned or insufficient, return null. Do NOT perform date ordering comparison. - "semantic_status": If rule_type is "SEMANTIC", evaluate whether authenticated evidence proves fulfillment ("FULFILLED"), non-fulfillment ("NOT_FULFILLED"), or is inconclusive ("INSUFFICIENT_EVIDENCE"). - "explanation": Concise explanation citing the verified source and observed facts. Return JSON ONLY in this exact structure: {{ "extractions": [ {{ "id": "obligation_id", "evidence_sufficient": true | false, "observed_numeric": 1000 | null, "observed_date": "YYYY-MM-DD" | null, "semantic_status": "FULFILLED" | "NOT_FULFILLED" | "INSUFFICIENT_EVIDENCE" | null, "explanation": "rationale citing verified source" }} ], "explanation": "overall extraction summary" }}""" raw_response = gl.nondet.exec_prompt(prompt, response_format="json") parsed = _parse_llm_json(raw_response) raw_extractions = parsed.get("extractions", []) if not isinstance(raw_extractions, list): raise gl.vm.UserError( f"{ERROR_LLM} Extractions must be a list in LLM response" ) extracted_map = {} for item in raw_extractions: if isinstance(item, dict) and "id" in item: extracted_map[item["id"]] = item evaluated_obligations = [] for ob in obs_data: oid = ob["obligation_id"] ext = extracted_map.get( oid, {"evidence_sufficient": False, "explanation": "Obligation omitted in extractor response."}, ) rule_type = ob.get("rule_type", "SEMANTIC") if rule_type == "NUMERIC": res = _evaluate_numeric_obligation(ob, ext) elif rule_type == "DEADLINE": res = _evaluate_deadline_obligation(ob, ext) else: res = _evaluate_semantic_obligation(ob, ext) evaluated_obligations.append(res) overall_status = _derive_agreement_status(evaluated_obligations, obs_data) overall_explanation = str(parsed.get("explanation", "")) return { "agreement_status": overall_status, "obligations": evaluated_obligations, "evidence_sources": acquired_evidence, "explanation": overall_explanation, } def validator_fn(leaders_res: gl.vm.Result) -> bool: if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, leader_fn) leader_data = leaders_res.calldata validator_data = leader_fn() expected_obligation_ids = { ob.get("obligation_id") for ob in obs_data if isinstance(ob, dict) and isinstance(ob.get("obligation_id"), str) } expected_evidence_ids = { ev.get("evidence_id") for ev in ev_data if isinstance(ev, dict) and isinstance(ev.get("evidence_id"), str) } return _consensus_results_equivalent( leader_data, validator_data, expected_obligation_ids, expected_evidence_ids, ) return gl.vm.run_nondet_unsafe(leader_fn, validator_fn) class ObliGen(gl.Contract): owner: Address agreements: TreeMap[str, Agreement] agreement_ids: DynArray[str] def __init__(self): self.owner = gl.message.sender_address @gl.public.write def create_agreement( self, agreement_id: str, parties: list[Address], obligations: list[dict], debtor: Address, creditor: Address, escrow_amount: u256 = u256(0), deadline: str = "", ) -> None: if not agreement_id or agreement_id.strip() == "": raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement ID cannot be empty") if len(agreement_id) > MAX_AGREEMENT_ID_LENGTH: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement ID exceeds {MAX_AGREEMENT_ID_LENGTH} characters") if agreement_id in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' already exists") if not parties or len(parties) == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement must have at least one party") if len(parties) > MAX_PARTIES: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement cannot have more than {MAX_PARTIES} parties") if not obligations or len(obligations) == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement must have at least one obligation") if len(obligations) > MAX_OBLIGATIONS: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement cannot have more than {MAX_OBLIGATIONS} obligations") debtor_addr = _normalize_address(debtor) creditor_addr = _normalize_address(creditor) parties_list = [] for p in parties: parties_list.append(_normalize_address_str(p)) debtor_str = _normalize_address_str(debtor_addr) creditor_str = _normalize_address_str(creditor_addr) if debtor_str not in parties_list: parties_list.append(debtor_str) if creditor_str not in parties_list: parties_list.append(creditor_str) seen_obs = set() obs_list = [] for ob_dict in obligations: ob_id = ob_dict.get("obligation_id", "") if not isinstance(ob_id, str) or not ob_id.strip() or ob_id in seen_obs: raise gl.vm.UserError( f"{ERROR_EXPECTED} Invalid or duplicate obligation ID: '{ob_id}'" ) if len(ob_id) > MAX_OBLIGATION_ID_LENGTH: raise gl.vm.UserError(f"{ERROR_EXPECTED} Obligation ID exceeds {MAX_OBLIGATION_ID_LENGTH} characters") seen_obs.add(ob_id) rule_type = str(ob_dict.get("rule_type", "SEMANTIC")).upper() if rule_type not in ("NUMERIC", "DEADLINE", "SEMANTIC"): raise gl.vm.UserError( f"{ERROR_EXPECTED} Invalid rule_type '{rule_type}'. Must be NUMERIC, DEADLINE, or SEMANTIC" ) resp_party_raw = ob_dict.get("responsible_party") if resp_party_raw is not None: resp_party = _normalize_address_str(resp_party_raw) else: resp_party = str(gl.message.sender_address) obligation_description = str(ob_dict.get("description", "")) if len(obligation_description) > MAX_DESCRIPTION_LENGTH: raise gl.vm.UserError( f"{ERROR_EXPECTED} Obligation description exceeds {MAX_DESCRIPTION_LENGTH} characters" ) ob_entry = { "obligation_id": ob_id, "description": obligation_description, "responsible_party": resp_party, "deadline": str(ob_dict.get("deadline", "")), "rule_type": rule_type, "target_value": str(ob_dict.get("target_value", "")), "comparison_op": str(ob_dict.get("comparison_op", ">=")), "is_mandatory": bool(ob_dict.get("is_mandatory", True)), } obs_list.append(ob_entry) escrow_amt = u256(escrow_amount) escrow_st = "UNFUNDED" if escrow_amt > 0 else "NONE" agreement = Agreement( agreement_id=agreement_id, creator=gl.message.sender_address, debtor=debtor_addr, creditor=creditor_addr, escrow_amount=escrow_amt, escrow_status=escrow_st, appeal_count=u256(0), appeal_deadline="", deadline=deadline, status="ACTIVE", verification_status="NONE", verification_explanation="", parties_json=json.dumps(parties_list), obligations_json=json.dumps(obs_list), evidence_json=json.dumps([]), results_json=json.dumps([]), recovery_baseline_evidence_json="", recovery_count=u256(0), ) self.agreements[agreement_id] = agreement self.agreement_ids.append(agreement_id) @gl.public.write.payable def fund_escrow(self, agreement_id: str) -> None: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] if agreement.status != "ACTIVE": raise gl.vm.UserError( f"{ERROR_EXPECTED} Cannot fund escrow for non-active agreement (status: {agreement.status})" ) sender = gl.message.sender_address if sender != agreement.debtor: raise gl.vm.UserError( f"{ERROR_EXPECTED} Only the designated debtor ({agreement.debtor}) can fund escrow" ) if agreement.escrow_amount <= 0: raise gl.vm.UserError( f"{ERROR_EXPECTED} Agreement '{agreement_id}' has zero escrow and cannot be funded" ) if agreement.escrow_status != "UNFUNDED": raise gl.vm.UserError( f"{ERROR_EXPECTED} Escrow is already funded (status: {agreement.escrow_status})" ) msg_val = gl.message.value if msg_val != agreement.escrow_amount: raise gl.vm.UserError( f"{ERROR_EXPECTED} Incorrect funding amount: expected {agreement.escrow_amount}, received {msg_val}" ) agreement.escrow_status = "FUNDED" self.agreements[agreement_id] = agreement @gl.public.write def submit_evidence( self, agreement_id: str, evidence_id: str, description: str, source_reference: str, content: str, ) -> None: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] if agreement.status != "ACTIVE": raise gl.vm.UserError( f"{ERROR_EXPECTED} Cannot submit evidence for non-active agreement (status: {agreement.status})" ) parties = json.loads(agreement.parties_json) sender_str = str(gl.message.sender_address) if sender_str not in parties: raise gl.vm.UserError( f"{ERROR_EXPECTED} Sender {sender_str} is not a recognized party of agreement '{agreement_id}'" ) if not isinstance(evidence_id, str) or not evidence_id.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} Evidence ID cannot be empty") if len(evidence_id) > MAX_EVIDENCE_ID_LENGTH: raise gl.vm.UserError(f"{ERROR_EXPECTED} Evidence ID exceeds {MAX_EVIDENCE_ID_LENGTH} characters") if len(description) > MAX_DESCRIPTION_LENGTH: raise gl.vm.UserError(f"{ERROR_EXPECTED} Evidence description exceeds {MAX_DESCRIPTION_LENGTH} characters") if len(source_reference) > MAX_SOURCE_REFERENCE_LENGTH: raise gl.vm.UserError(f"{ERROR_EXPECTED} Source reference exceeds {MAX_SOURCE_REFERENCE_LENGTH} characters") if len(content) > MAX_PARTY_CONTENT_LENGTH: raise gl.vm.UserError(f"{ERROR_EXPECTED} Party-supplied content exceeds {MAX_PARTY_CONTENT_LENGTH} characters") evidence_list = json.loads(agreement.evidence_json) if len(evidence_list) >= MAX_EVIDENCE_ITEMS: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement cannot contain more than {MAX_EVIDENCE_ITEMS} evidence items") for ev in evidence_list: if ev.get("evidence_id") == evidence_id: raise gl.vm.UserError( f"{ERROR_EXPECTED} Duplicate evidence ID '{evidence_id}' in agreement '{agreement_id}'" ) source_ref_clean = (source_reference or "").strip() status = "NOT_PROVIDED" if not source_ref_clean else "PENDING" new_ev = { "evidence_id": evidence_id, "submitted_by": sender_str, "description": description, "source_reference": source_reference, "content": content, "source_acquisition_status": status, "source_verification_status": "UNVERIFIED", "source_hash": "", "evidence_commitment": "", "source_excerpt": "", "canonical_source_content": "", } evidence_list.append(new_ev) agreement.evidence_json = json.dumps(evidence_list) self.agreements[agreement_id] = agreement @gl.public.write def recover_insufficient_evidence(self, agreement_id: str) -> None: """Re-open only an insufficient-evidence result for more evidence. Escrow remains LOCKED and the agreement cannot settle while ACTIVE. A recognized party may add evidence, after which verify() performs a fresh validator-backed result and restores the normal appeal flow. """ if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] if agreement.status != "VERIFIED" or agreement.verification_status != "INSUFFICIENT_EVIDENCE": raise gl.vm.UserError( f"{ERROR_EXPECTED} Can only recover VERIFIED insufficient-evidence agreements" ) if agreement.escrow_amount > 0 and agreement.escrow_status != "LOCKED": raise gl.vm.UserError( f"{ERROR_EXPECTED} Insufficient-evidence escrow must remain LOCKED" ) parties = json.loads(agreement.parties_json) if str(gl.message.sender_address) not in parties: raise gl.vm.UserError( f"{ERROR_EXPECTED} Sender {gl.message.sender_address} is not a recognized party of agreement '{agreement_id}'" ) if agreement.recovery_count >= MAX_RECOVERY_ATTEMPTS: raise gl.vm.UserError( f"{ERROR_EXPECTED} Agreement '{agreement_id}' has already reached the recovery limit (limit: {MAX_RECOVERY_ATTEMPTS})" ) agreement.recovery_count = u256(int(agreement.recovery_count) + 1) # This is the sole recovery transition. It does not release/refund # escrow, reset the appeal count, or permit settlement. # Capture the pre-recovery evidence before additional evidence can be # appended; verify() checks this baseline before replacing evidence. agreement.recovery_baseline_evidence_json = agreement.evidence_json agreement.status = "ACTIVE" agreement.appeal_deadline = "" self.agreements[agreement_id] = agreement @gl.public.write def verify(self, agreement_id: str) -> dict: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] if agreement.status != "ACTIVE": raise gl.vm.UserError( f"{ERROR_EXPECTED} Agreement '{agreement_id}' is not in ACTIVE state (status: {agreement.status})" ) retrying_locked_insufficient = ( agreement.escrow_amount > 0 and agreement.escrow_status == "LOCKED" and agreement.verification_status == "INSUFFICIENT_EVIDENCE" ) retrying_recovery = bool(agreement.recovery_baseline_evidence_json) or retrying_locked_insufficient if agreement.escrow_amount > 0 and agreement.escrow_status != "FUNDED" and not retrying_locked_insufficient: raise gl.vm.UserError( f"{ERROR_EXPECTED} Cannot verify agreement with unfunded escrow (status: {agreement.escrow_status})" ) obs_data = json.loads(agreement.obligations_json) ev_data = json.loads(agreement.evidence_json) previous_evidence = ( json.loads(agreement.recovery_baseline_evidence_json) if agreement.recovery_baseline_evidence_json else [] ) consensus_result = _run_hybrid_consensus(obs_data, ev_data) if retrying_recovery and previous_evidence and not _evidence_commitments_match( previous_evidence, consensus_result.get("evidence_sources", []) ): raise gl.vm.UserError( f"{ERROR_EXPECTED} Recovery evidence commitment changed" ) agreement.status = "VERIFIED" if agreement.escrow_amount > 0: agreement.escrow_status = "LOCKED" agreement.verification_status = consensus_result["agreement_status"] agreement.appeal_deadline = _appeal_deadline() agreement.verification_explanation = consensus_result["explanation"] agreement.evidence_json = json.dumps(consensus_result.get("evidence_sources", ev_data)) agreement.results_json = json.dumps(consensus_result["obligations"]) agreement.recovery_baseline_evidence_json = "" self.agreements[agreement_id] = agreement return consensus_result @gl.public.write def appeal(self, agreement_id: str) -> dict: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] parties = json.loads(agreement.parties_json) sender_str = str(gl.message.sender_address) if sender_str not in parties: raise gl.vm.UserError( f"{ERROR_EXPECTED} Sender {sender_str} is not a recognized party of agreement '{agreement_id}'" ) if agreement.status != "VERIFIED": raise gl.vm.UserError( f"{ERROR_EXPECTED} Can only appeal agreements in VERIFIED status (current: {agreement.status})" ) if agreement.appeal_count > 0: raise gl.vm.UserError( f"{ERROR_EXPECTED} Agreement '{agreement_id}' has already been appealed (limit: 1)" ) if not _appeal_window_active(agreement.appeal_deadline): raise gl.vm.UserError(f"{ERROR_EXPECTED} Appeal window has expired") obs_data = json.loads(agreement.obligations_json) ev_data = json.loads(agreement.evidence_json) # Fresh execution of hybrid consensus with fresh source acquisition fresh_result = _run_hybrid_consensus(obs_data, ev_data) previous_evidence = json.loads(agreement.evidence_json) fresh_evidence = fresh_result.get("evidence_sources", []) if len(previous_evidence) != len(fresh_evidence): raise gl.vm.UserError(f"{ERROR_EXPECTED} Appeal source set changed during re-verification") fresh_by_id = { ev.get("evidence_id"): ev for ev in fresh_evidence if isinstance(ev, dict) and ev.get("evidence_id") } for previous in previous_evidence: current = fresh_by_id.get(previous.get("evidence_id")) if current is None: raise gl.vm.UserError(f"{ERROR_EXPECTED} Appeal source set changed during re-verification") if ( current.get("source_reference") != previous.get("source_reference") or current.get("source_hash") != previous.get("source_hash") or current.get("source_verification_status") != previous.get("source_verification_status") or current.get("evidence_commitment") != previous.get("evidence_commitment") or current.get("canonical_source_content") != previous.get("canonical_source_content") ): raise gl.vm.UserError(f"{ERROR_EXPECTED} Appeal source commitment changed during re-verification") agreement.appeal_count = u256(1) agreement.status = "REVERIFIED" agreement.verification_status = fresh_result["agreement_status"] agreement.appeal_deadline = _appeal_deadline() agreement.verification_explanation = fresh_result["explanation"] agreement.evidence_json = json.dumps(fresh_result.get("evidence_sources", ev_data)) agreement.results_json = json.dumps(fresh_result["obligations"]) self.agreements[agreement_id] = agreement return fresh_result @gl.public.write def finalize(self, agreement_id: str) -> None: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] if agreement.status not in ("VERIFIED", "REVERIFIED"): raise gl.vm.UserError( f"{ERROR_EXPECTED} Can only finalize agreements in VERIFIED or REVERIFIED status (current: {agreement.status})" ) if _appeal_window_active(agreement.appeal_deadline): raise gl.vm.UserError(f"{ERROR_EXPECTED} Cannot finalize while appeal window is active") if agreement.verification_status not in ("FULFILLED", "NOT_FULFILLED"): raise gl.vm.UserError(f"{ERROR_EXPECTED} Cannot finalize without a consequential validator consensus") persisted_evidence = json.loads(agreement.evidence_json) if not persisted_evidence or any( ev.get("source_acquisition_status") != "FETCHED" or ev.get("source_verification_status") != "VERIFIED" or not ev.get("source_hash") or not ev.get("evidence_commitment") for ev in persisted_evidence ): raise gl.vm.UserError(f"{ERROR_EXPECTED} Cannot finalize without acquired and verified evidence") agreement.status = "FINALIZED" self.agreements[agreement_id] = agreement @gl.public.write def settle(self, agreement_id: str) -> None: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] if agreement.status != "FINALIZED": raise gl.vm.UserError( f"{ERROR_EXPECTED} Cannot settle agreement before FINALIZED state (current: {agreement.status})" ) if agreement.escrow_amount <= 0: raise gl.vm.UserError( f"{ERROR_EXPECTED} Agreement '{agreement_id}' has zero escrow to settle" ) if agreement.escrow_status != "LOCKED": raise gl.vm.UserError( f"{ERROR_EXPECTED} Cannot settle escrow with status '{agreement.escrow_status}' (must be LOCKED)" ) if agreement.verification_status == "FULFILLED": recipient = agreement.creditor agreement.escrow_status = "RELEASED" self.agreements[agreement_id] = agreement gl.get_contract_at(recipient).emit_transfer( value=agreement.escrow_amount, on="finalized" ) elif agreement.verification_status == "NOT_FULFILLED": recipient = agreement.debtor agreement.escrow_status = "REFUNDED" self.agreements[agreement_id] = agreement gl.get_contract_at(recipient).emit_transfer( value=agreement.escrow_amount, on="finalized" ) elif agreement.verification_status == "INSUFFICIENT_EVIDENCE": raise gl.vm.UserError( f"{ERROR_EXPECTED} Cannot settle escrow with INSUFFICIENT_EVIDENCE; funds remain LOCKED" ) else: raise gl.vm.UserError( f"{ERROR_EXPECTED} Unknown verification status '{agreement.verification_status}'" ) @gl.public.view def get_agreement(self, agreement_id: str) -> dict: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] return { "agreement_id": agreement.agreement_id, "creator": str(agreement.creator), "debtor": str(agreement.debtor), "creditor": str(agreement.creditor), "escrow_amount": int(agreement.escrow_amount), "escrow_status": agreement.escrow_status, "appeal_count": agreement.appeal_count, "appeal_deadline": agreement.appeal_deadline, "deadline": agreement.deadline, "status": agreement.status, "verification_status": agreement.verification_status, "verification_explanation": agreement.verification_explanation, "parties": json.loads(agreement.parties_json), "obligations": json.loads(agreement.obligations_json), "evidence_list": json.loads(agreement.evidence_json), "results": json.loads(agreement.results_json), "recovery_count": agreement.recovery_count, } @gl.public.view def get_verification(self, agreement_id: str) -> dict: if agreement_id not in self.agreements: raise gl.vm.UserError(f"{ERROR_EXPECTED} Agreement '{agreement_id}' not found") agreement = self.agreements[agreement_id] return { "agreement_id": agreement.agreement_id, "status": agreement.status, "verification_status": agreement.verification_status, "verification_explanation": agreement.verification_explanation, "appeal_count": agreement.appeal_count, "appeal_deadline": agreement.appeal_deadline, "escrow_status": agreement.escrow_status, "obligations": json.loads(agreement.results_json), "recovery_count": agreement.recovery_count, }