# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json # --------------------------------------------------------------------------- # AGENT NEGOTIATOR # # An Intelligent Contract where autonomous buyer/seller agents negotiate # real-world tasks. GenLayer validators reach consensus on whether a proposal # satisfies both parties' natural-language constraints, whether the final # agreement is internally consistent, and how a challenge should be resolved. # # Every non-deterministic block returns a STRUCTURED JSON result, never # free-form prose, so the contract state stays machine-readable. # --------------------------------------------------------------------------- class AgentNegotiator(gl.Contract): owner: Address agent_ids: DynArray[str] agents: TreeMap[str, str] deal_ids: DynArray[str] deals: TreeMap[str, str] proposals: TreeMap[str, str] evaluations: TreeMap[str, str] verifications: TreeMap[str, str] challenges: TreeMap[str, str] deal_count: u256 def __init__(self) -> None: """Deploys an empty negotiation marketplace.""" self.owner = gl.message.sender_address self.deal_count = u256(0) # -- helpers ------------------------------------------------------------ def _consensus_json(self, task: str, principle: str) -> str: """Runs a prompt under validator consensus and returns a JSON object string.""" def evaluate() -> str: raw = gl.nondet.exec_prompt(task) cleaned = raw.strip() if cleaned.startswith("```"): cleaned = cleaned.split("```")[1] if cleaned.startswith("json"): cleaned = cleaned[4:] start = cleaned.find("{") end = cleaned.rfind("}") if start == -1 or end == -1: raise gl.vm.UserError("model did not return a JSON object") return cleaned[start : end + 1] result = gl.eq_principle.prompt_comparative(evaluate, principle) return result def _require_deal(self, deal_id: str) -> dict: raw = self.deals.get(deal_id) if raw is None: raise gl.vm.UserError("unknown deal") return json.loads(raw) # -- agents ------------------------------------------------------------- @gl.public.write def register_agent(self, agent_id: str, agent_json: str) -> None: """Registers a buyer or seller agent. agent_json carries: name, role, identity, wallet, objective, budget, preferences, constraints, strategy. """ if self.agents.get(agent_id) is not None: raise gl.vm.UserError("agent already registered") json.loads(agent_json) self.agents[agent_id] = agent_json self.agent_ids.append(agent_id) @gl.public.view def get_agent(self, agent_id: str) -> str: raw = self.agents.get(agent_id) if raw is None: raise gl.vm.UserError("unknown agent") return raw @gl.public.view def list_agents(self) -> str: return json.dumps(list(self.agent_ids)) # -- marketplace -------------------------------------------------------- @gl.public.write def post_deal(self, deal_id: str, deal_json: str) -> None: """Posts a natural-language request, e.g. 'landing page, $100, 3 days'.""" if self.deals.get(deal_id) is not None: raise gl.vm.UserError("deal already exists") deal = json.loads(deal_json) deal["status"] = "open" deal["rounds"] = 0 self.deals[deal_id] = json.dumps(deal) self.deal_ids.append(deal_id) self.deal_count = u256(len(self.deal_ids)) @gl.public.view def get_deal(self, deal_id: str) -> str: raw = self.deals.get(deal_id) if raw is None: raise gl.vm.UserError("unknown deal") return raw @gl.public.view def list_deals(self) -> str: return json.dumps(list(self.deal_ids)) # -- negotiation -------------------------------------------------------- @gl.public.write def submit_proposal(self, deal_id: str, proposal_id: str, proposal_json: str) -> str: """Submits a proposal / counteroffer and evaluates it under consensus. Returns the structured evaluation: proposal_status, accepted, price, deadline, requirements_satisfied, requirements_failed, hidden_conditions, fairness_score, risk_score, negotiation_score, confidence, reason, final_terms. """ deal = self._require_deal(deal_id) if deal.get("status") in ("agreed", "verified", "rejected"): raise gl.vm.UserError("deal is already closed") proposal = json.loads(proposal_json) buyer_raw = self.agents.get(str(deal.get("buyer_agent_id", ""))) seller_raw = self.agents.get(str(proposal.get("agent_id", ""))) buyer = json.loads(buyer_raw) if buyer_raw is not None else {} seller = json.loads(seller_raw) if seller_raw is not None else {} task = f""" You are a neutral negotiation adjudicator for a decentralized marketplace. Judge whether the PROPOSAL satisfies the BUYER REQUEST and both agents' constraints. BUYER REQUEST: {json.dumps(deal, indent=2)} BUYER AGENT (objective, budget, preferences, constraints): {json.dumps(buyer, indent=2)} PROPOSING AGENT: {json.dumps(seller, indent=2)} PROPOSAL (price, deadline, deliverables, quality, penalties, payment terms): {json.dumps(proposal, indent=2)} Rules: - A requirement counts as satisfied only if the proposal covers it in substance. - Flag hidden or unreasonable conditions (non-refundable fees, unlimited scope, exclusive rights transfer, auto-renewal, unilateral discretion clauses). - Flag terms that conflict with each other. - accepted = true only if there are no failed requirements, no hidden conditions, fairness_score >= 55 and risk_score <= 45. - Scores are integers from 0 to 100. Respond with ONLY this JSON object and no other text: {{ "proposal_status": "valid" | "conflicting" | "invalid", "accepted": true | false, "price": , "deadline": , "requirements_satisfied": [], "requirements_failed": [], "hidden_conditions": [], "conflicting_terms": [], "fairness_score": , "risk_score": , "negotiation_score": , "confidence": , "reason": "", "final_terms": "" }} """ principle = ( "The values of proposal_status, accepted, price and deadline must be identical, " "and the sets of requirements_satisfied and requirements_failed must agree in " "substance. Numeric scores may differ by at most 15 points." ) evaluation = self._consensus_json(task, principle) self.proposals[proposal_id] = proposal_json self.evaluations[proposal_id] = evaluation parsed = json.loads(evaluation) rounds = int(deal.get("rounds", 0)) + 1 deal["rounds"] = rounds deal["last_proposal_id"] = proposal_id if bool(parsed.get("accepted", False)): deal["status"] = "agreed" deal["final_terms"] = parsed.get("final_terms", "") deal["negotiation_score"] = parsed.get("negotiation_score", 0) else: deal["status"] = "negotiating" self.deals[deal_id] = json.dumps(deal) return evaluation @gl.public.write def generate_counteroffer(self, deal_id: str, proposal_id: str, agent_id: str) -> str: """Smart counteroffer: builds terms that respect the agent's constraints.""" deal = self._require_deal(deal_id) proposal_raw = self.proposals.get(proposal_id) evaluation_raw = self.evaluations.get(proposal_id) agent_raw = self.agents.get(agent_id) if proposal_raw is None or evaluation_raw is None or agent_raw is None: raise gl.vm.UserError("missing proposal, evaluation or agent") task = f""" You are the negotiation engine of an autonomous agent. The last proposal violated at least one of your constraints. Produce the next counteroffer using your negotiation strategy. YOUR AGENT (strategy, budget, constraints, preferences): {json.dumps(json.loads(agent_raw), indent=2)} DEAL: {json.dumps(deal, indent=2)} LAST PROPOSAL: {proposal_raw} EVALUATION OF THAT PROPOSAL: {evaluation_raw} Strategy behaviour: - aggressive: large price move in your favour, concede schedule last. - balanced: move price and schedule toward the midpoint. - cooperative: small concession now to close quickly. - best_value: accept a small premium for faster, better-specified delivery. Respond with ONLY this JSON object: {{ "price": , "deadline": , "deliverables": [], "quality": "", "penalties": "", "payment_terms": "", "resolves": [], "reason": "" }} """ principle = ( "price and deadline must match, and the deliverables must cover the same scope. " "Wording may differ." ) return self._consensus_json(task, principle) # -- verification ------------------------------------------------------- @gl.public.write def verify_agreement(self, deal_id: str) -> str: """Reviews the final terms before finalisation: VERIFIED / NEEDS REVIEW / REJECTED.""" deal = self._require_deal(deal_id) if deal.get("status") != "agreed": raise gl.vm.UserError("no agreement to verify") proposal_id = str(deal.get("last_proposal_id", "")) proposal_raw = self.proposals.get(proposal_id) evaluation_raw = self.evaluations.get(proposal_id) if proposal_raw is None or evaluation_raw is None: raise gl.vm.UserError("missing final proposal") task = f""" Review the FINAL AGREEMENT of a negotiation for internal consistency before it is finalised on chain. ORIGINAL REQUEST: {json.dumps(deal, indent=2)} FINAL PROPOSAL: {proposal_raw} EVALUATION: {evaluation_raw} Check: price within budget, deadline within the requested window, deliverables matching the requested scope, penalties and payment terms not contradicting each other, and no clause that silently reverses an earlier agreed term. Respond with ONLY this JSON object: {{ "verification_status": "VERIFIED" | "NEEDS REVIEW" | "REJECTED", "contradictions": [], "risk_level": "low" | "medium" | "high", "confidence": , "reason": "" }} """ principle = ( "verification_status and risk_level must be identical; confidence may differ by at " "most 15 points." ) verification = self._consensus_json(task, principle) self.verifications[deal_id] = verification parsed = json.loads(verification) status = str(parsed.get("verification_status", "NEEDS REVIEW")) deal["status"] = "verified" if status == "VERIFIED" else "needs_review" if status == "REJECTED": deal["status"] = "rejected" deal["verification"] = status self.deals[deal_id] = json.dumps(deal) return verification @gl.public.view def get_verification(self, deal_id: str) -> str: raw = self.verifications.get(deal_id) if raw is None: raise gl.vm.UserError("no verification for this deal") return raw # -- challenges --------------------------------------------------------- @gl.public.write def challenge_agreement(self, deal_id: str, claim: str, history_json: str) -> str: """Re-reads the whole negotiation history and returns UPHELD or OVERTURNED.""" deal = self._require_deal(deal_id) verification_raw = self.verifications.get(deal_id) original = "VERIFIED" if verification_raw is None else str( json.loads(verification_raw).get("verification_status", "VERIFIED") ) task = f""" A participant challenges the interpretation of a finalised agreement. Re-read the ENTIRE negotiation history and decide whether the original decision stands. DEAL: {json.dumps(deal, indent=2)} ORIGINAL DECISION: {original} FULL NEGOTIATION HISTORY (every offer, counteroffer, rejection, acceptance): {history_json} CHALLENGE CLAIM: {claim} Overturn only if the history shows the accepted terms do not reflect what both parties actually agreed, or a term was materially altered without acknowledgement. Respond with ONLY this JSON object: {{ "challenge_status": "reviewed", "original_decision": "{original}", "final_decision": "UPHELD" | "OVERTURNED", "reason": "", "confidence": }} """ principle = "final_decision must be identical; confidence may differ by at most 15 points." challenge = self._consensus_json(task, principle) self.challenges[deal_id] = challenge parsed = json.loads(challenge) if str(parsed.get("final_decision", "UPHELD")) == "OVERTURNED": deal["status"] = "needs_review" self.deals[deal_id] = json.dumps(deal) return challenge @gl.public.view def get_challenge(self, deal_id: str) -> str: raw = self.challenges.get(deal_id) if raw is None: raise gl.vm.UserError("no challenge for this deal") return raw # -- AI deal finder ----------------------------------------------------- @gl.public.write def find_deals(self, description: str, catalog_json: str) -> str: """Matches a natural-language need against open deals by requirement fit.""" task = f""" Match a natural-language need against a catalog of open marketplace deals. Judge compatibility of requirements, budget and timeline, not keyword overlap. NEED: {description} CATALOG: {catalog_json} Respond with ONLY this JSON object: {{ "matches": [ {{ "deal_id": "", "compatibility": , "matched_requirements": [], "gaps": [], "reason": "" }} ] }} """ principle = ( "The set of deal_id values and their ordering must match; compatibility may differ " "by at most 15 points." ) return self._consensus_json(task, principle) # -- external evidence -------------------------------------------------- @gl.public.write def attach_evidence(self, deal_id: str, url: str) -> str: """Fetches external evidence (e.g. a delivered URL) and judges the claim.""" deal = self._require_deal(deal_id) def review() -> str: web_data = gl.nondet.web.render(url, mode="text") task = f""" Judge whether the published page satisfies the agreed deliverables. AGREED TERMS: {json.dumps(deal, indent=2)} PAGE CONTENT: {web_data} Respond with ONLY this JSON object: {{ "evidence_status": "SATISFIED" | "PARTIAL" | "NOT SATISFIED", "satisfied": [], "missing": [], "confidence": , "reason": "" }} """ raw = gl.nondet.exec_prompt(task).strip() start = raw.find("{") end = raw.rfind("}") if start == -1 or end == -1: raise gl.vm.UserError("model did not return a JSON object") return raw[start : end + 1] principle = "evidence_status must be identical; confidence may differ by at most 15 points." result = gl.eq_principle.prompt_comparative(review, principle) return result