# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json import hashlib JOB_CATEGORIES = ( "SOFTWARE", "RESEARCH", "WRITING", "DESIGN", "DATA", "BUSINESS", "PROFESSIONAL_SERVICES", "EDUCATION", "MARKETING", "CUSTOMER_SUPPORT", "DIGITAL_ACTIONS", "PHYSICAL_SERVICES", "AI_AGENT", "OTHER", ) LEGACY_JOB_CATEGORIES = ("GENERAL", "CODE", "CONTENT", "WEB") ACCEPTED_JOB_CATEGORIES = JOB_CATEGORIES + LEGACY_JOB_CATEGORIES LEGACY_CATEGORY_MAP = { "GENERAL": "OTHER", "CODE": "SOFTWARE", "CONTENT": "WRITING", "WEB": "SOFTWARE", "DATA": "DATA", "RESEARCH": "RESEARCH", } CATEGORY_CONTEXTS = { "SOFTWARE": "This is a SOFTWARE task. Judge functional completeness and correctness.", "RESEARCH": "This is a RESEARCH task. Judge research rigor, factual consistency, and source quality.", "WRITING": "This is a WRITING task. Judge structure, accuracy, tone, and completeness.", "DESIGN": "This is a DESIGN task. Judge compliance with requirements and quality signals.", "DATA": "This is a DATA task. Judge schema conformance, completeness, and format consistency.", "BUSINESS": "This is a BUSINESS task. Judge whether the stated business outcome was achieved.", "PROFESSIONAL_SERVICES": "This is a PROFESSIONAL SERVICES task. Judge domain correctness and documentation.", "EDUCATION": "This is an EDUCATION task. Judge coverage of learning objectives and clarity.", "MARKETING": "This is a MARKETING task. Judge compliance with brief and campaign asset completeness.", "CUSTOMER_SUPPORT": "This is a CUSTOMER SUPPORT task. Judge whether the issue was resolved correctly.", "DIGITAL_ACTIONS": "This is a DIGITAL ACTIONS task. Judge verifiable completion of on-chain or off-chain actions.", "PHYSICAL_SERVICES": "This is a PHYSICAL SERVICES task. Judge completion evidence like photos and confirmations.", "AI_AGENT": "This is an AI AGENT task. Judge workflow completion and correct tool behavior.", "OTHER": "This is a general task. Judge whether the agreed outcome was achieved.", } ERROR_EXPECTED = "[EXPECTED]" ERROR_EXTERNAL = "[EXTERNAL]" ERROR_TRANSIENT = "[TRANSIENT]" ERROR_LLM = "[LLM_ERROR]" def _resolve_category(category: str) -> str: if not category: return "OTHER" return LEGACY_CATEGORY_MAP.get(category, category) DEFAULT_MIN_SCORE = 75 def _parse_verification_plan(policy: str) -> dict: plan: dict = {"min_score": DEFAULT_MIN_SCORE, "gates": []} if not policy: return plan for segment in policy.split("|"): segment = segment.strip() if ":" not in segment: continue key, _, value = segment.partition(":") key = key.strip().upper() value = value.strip() if key == "MIN_SCORE": try: plan["min_score"] = max(30, min(100, int(value))) except ValueError: pass elif key == "GATES": for gate in value.replace("+", ",").split(","): gate = gate.strip().lower() if gate and gate not in plan["gates"]: plan["gates"].append(gate) return plan def _new_job(job_id, title, spec, fmt, budget, poster, deadline, task_type, policy, category, sources=None): return { "id": job_id, "title": title, "spec": spec, "deliverable_format": fmt, "budget": int(budget), "poster": poster, "agent": "", "status": "OPEN", "deliverable": "", "verdict": "", "score": 0, "judgment_reason": "", "created_at": 0, "deadline": int(deadline), "completed_at": 0, "dispute_round": 0, "task_type": task_type, "policy": policy, "sources": sources if sources else [], "category": category, } def _default_agent_stats() -> dict: return { "completed": 0, "approved": 0, "rejected": 0, "partial": 0, "total_earned": 0, "score_sum": 0, "disputes": 0, "delivery_time_sum": 0, } def _parse_llm_json(text: str) -> dict: cleaned = text.strip() if cleaned.startswith("```json"): cleaned = cleaned[7:] if cleaned.startswith("```"): cleaned = cleaned[3:] if cleaned.endswith("```"): cleaned = cleaned[:-3] cleaned = cleaned.strip() first = cleaned.find("{") last = cleaned.rfind("}") if first >= 0 and last > first: cleaned = cleaned[first:last + 1] return json.loads(cleaned) class AgentPact(gl.Contract): jobs: TreeMap[str, str] job_sources: TreeMap[str, str] verifications: TreeMap[str, str] agent_stats: TreeMap[str, str] total_escrowed: u256 total_released: u256 open_job_ids: DynArray[str] all_job_ids: DynArray[str] platform_fee_bps: u256 def __init__(self): self.platform_fee_bps = 250 def _get_job(self, job_id: str) -> dict: raw = self.jobs.get(job_id, "") if not raw: raise gl.vm.UserError("Job not found") return json.loads(raw) def _save_job(self, job_id: str, job: dict): self.jobs[job_id] = json.dumps(job) def _get_stats(self, addr: str) -> dict: raw = self.agent_stats.get(addr, "") if not raw: return _default_agent_stats() return json.loads(raw) def _save_stats(self, addr: str, stats: dict): self.agent_stats[addr] = json.dumps(stats) @gl.public.write.payable def post_job( self, title: str, spec: str, deliverable_format: str, deadline_blocks: int, policy: str = "", category: str = "GENERAL", ) -> str: if not title or len(title) > 80: raise gl.vm.UserError(f"{ERROR_EXPECTED} Title must be non-empty and 80 characters or fewer") if not spec or len(spec) > 2000: raise gl.vm.UserError(f"{ERROR_EXPECTED} Spec must be non-empty and 2000 characters or fewer") if deliverable_format not in ["url", "text", "github_url"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} deliverable_format must be one of: url, text, github_url") if deadline_blocks <= 0 or deadline_blocks > 302400: raise gl.vm.UserError(f"{ERROR_EXPECTED} deadline_blocks must be between 1 and 302400") if len(policy) > 500: raise gl.vm.UserError(f"{ERROR_EXPECTED} Policy must be 500 characters or fewer") if category not in ACCEPTED_JOB_CATEGORIES: raise gl.vm.UserError(f"{ERROR_EXPECTED} category must be one of: " + ", ".join(ACCEPTED_JOB_CATEGORIES)) if gl.message.value <= 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Budget must be greater than 0") job_id = hashlib.sha256( (gl.message.sender_address + str(gl.block.number)).encode() ).hexdigest()[:16] deadline = gl.block.timestamp + (deadline_blocks * 8) job = _new_job(job_id, title, spec, deliverable_format, gl.message.value, gl.message.sender_address, deadline, "TASK", policy, category) job["created_at"] = gl.block.timestamp self._save_job(job_id, job) self.open_job_ids.append(job_id) self.all_job_ids.append(job_id) self.total_escrowed += gl.message.value return job_id @gl.public.write.payable def post_claim( self, claim: str, sources: DynArray[str], deadline_blocks: int, policy: str = "", ) -> str: if not claim or len(claim) > 2000: raise gl.vm.UserError(f"{ERROR_EXPECTED} Claim must be non-empty and 2000 characters or fewer") if not isinstance(sources, list) or len(sources) == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} At least one source URL is required") if len(sources) > 10: raise gl.vm.UserError(f"{ERROR_EXPECTED} Maximum 10 sources allowed") for src in sources: if not isinstance(src, str) or not src.startswith("https://"): raise gl.vm.UserError(f"{ERROR_EXPECTED} All source URLs must start with https://") if len(policy) > 500: raise gl.vm.UserError(f"{ERROR_EXPECTED} Policy must be 500 characters or fewer") if deadline_blocks <= 0 or deadline_blocks > 302400: raise gl.vm.UserError(f"{ERROR_EXPECTED} deadline_blocks must be between 1 and 302400") if gl.message.value <= 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Budget must be greater than 0") job_id = hashlib.sha256( (gl.message.sender_address + str(gl.block.number)).encode() ).hexdigest()[:16] deadline = gl.block.timestamp + (deadline_blocks * 8) title = ("VERIFY CLAIM: " + claim)[:80] job = _new_job(job_id, title, claim, "text", gl.message.value, gl.message.sender_address, deadline, "CLAIM", policy, "GENERAL", sources=sources) job["created_at"] = gl.block.timestamp self._save_job(job_id, job) self.job_sources[job_id] = ",".join(sources) self.open_job_ids.append(job_id) self.all_job_ids.append(job_id) self.total_escrowed += gl.message.value return job_id @gl.public.write def accept_job(self, job_id: str) -> None: job = self._get_job(job_id) if job["status"] != "OPEN": raise gl.vm.UserError(f"{ERROR_EXPECTED} Job is not available for acceptance") if gl.message.sender_address == job["poster"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} Poster cannot accept their own job") if gl.block.timestamp > job["deadline"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} Job deadline has passed") job["agent"] = gl.message.sender_address job["status"] = "ACTIVE" self._save_job(job_id, job) self.open_job_ids.remove(job_id) @gl.public.write def submit_deliverable(self, job_id: str, deliverable: str) -> None: job = self._get_job(job_id) if job["status"] != "ACTIVE": raise gl.vm.UserError(f"{ERROR_EXPECTED} Job is not in ACTIVE status") if gl.message.sender_address != job["agent"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} Only the assigned agent can submit deliverables") if not deliverable: raise gl.vm.UserError(f"{ERROR_EXPECTED} Deliverable cannot be empty") if job["deliverable_format"] in ["url", "github_url"]: if not deliverable.startswith("https://"): raise gl.vm.UserError(f"{ERROR_EXPECTED} URL deliverables must start with https://") if gl.block.timestamp > job["deadline"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} Job deadline has passed") job["deliverable"] = deliverable job["status"] = "SUBMITTED" self._save_job(job_id, job) self._judge_deliverable(job_id) @gl.public.write def dispute_verdict(self, job_id: str) -> None: job = self._get_job(job_id) if job["status"] != "COMPLETE": raise gl.vm.UserError(f"{ERROR_EXPECTED} Job must be COMPLETE to dispute") if gl.message.sender_address != job["poster"] and gl.message.sender_address != job["agent"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} Only poster or agent can dispute") if job["dispute_round"] >= 3: raise gl.vm.UserError(f"{ERROR_EXPECTED} Maximum dispute rounds reached") blocks_since = gl.block.timestamp - job["completed_at"] if blocks_since > 860: raise gl.vm.UserError(f"{ERROR_EXPECTED} Dispute window has passed") job["dispute_round"] += 1 job["status"] = "DISPUTED" stats = self._get_stats(job["agent"]) stats["disputes"] += 1 self._save_stats(job["agent"], stats) self._save_job(job_id, job) self._judge_deliverable(job_id) @gl.public.write def cancel_job(self, job_id: str) -> None: job = self._get_job(job_id) if job["status"] != "OPEN": raise gl.vm.UserError(f"{ERROR_EXPECTED} Only OPEN jobs can be cancelled") if gl.message.sender_address != job["poster"]: raise gl.vm.UserError(f"{ERROR_EXPECTED} Only the poster can cancel") job["status"] = "CANCELLED" self._save_job(job_id, job) self.open_job_ids.remove(job_id) gl.transfer(job["poster"], job["budget"]) self.total_escrowed -= job["budget"] @gl.public.view def get_job(self, job_id: str) -> dict: return self._get_job(job_id) @gl.public.view def get_open_jobs(self, limit: int = 20) -> list: if limit > 50: limit = 50 result = [] for i in range(len(self.open_job_ids) - 1, -1, -1): if len(result) >= limit: break raw = self.jobs.get(self.open_job_ids[i], "") if raw: result.append(json.loads(raw)) return result @gl.public.view def get_all_jobs(self, limit: int = 50) -> list: if limit > 100: limit = 100 result = [] for i in range(len(self.all_job_ids) - 1, -1, -1): if len(result) >= limit: break raw = self.jobs.get(self.all_job_ids[i], "") if raw: result.append(json.loads(raw)) return result @gl.public.view def get_verification(self, job_id: str) -> dict: raw = self.verifications.get(job_id, "") if not raw: raise gl.vm.UserError("No verification found for this job") records = json.loads(raw) if not records: raise gl.vm.UserError("No verification found for this job") return records[-1] @gl.public.view def get_verifications(self, job_id: str) -> list: raw = self.verifications.get(job_id, "") if not raw: return [] return json.loads(raw) @gl.public.view def get_agent_stats(self, agent_address: str) -> dict: return self._get_stats(agent_address) @gl.public.view def get_platform_stats(self) -> dict: return { "total_escrowed": self.total_escrowed, "total_released": self.total_released, "total_jobs": len(self.all_job_ids), "open_jobs_count": len(self.open_job_ids), } def _collect_evidence(self, job: dict, deliverable_content: str) -> list: evidence = [] fetch_failed = "[Could not fetch" in deliverable_content if job["deliverable_format"] in ["url", "github_url"]: evidence.append({"label": "Deliverable URL reachable", "passed": not fetch_failed}) if job["deliverable_format"] == "github_url": evidence.append({ "label": "Repository hosted on github.com", "passed": "github.com" in job["deliverable"].lower(), }) content_lower = deliverable_content.lower() spec_tokens = [] seen_tokens = [] for token in job["spec"].lower().replace("\n", " ").split(" "): cleaned = "".join(c for c in token if c.isalnum()) if len(cleaned) >= 5 and cleaned not in ( "about", "should", "would", "there", "their", "which", "these", "those", "where", "using", "based", "must", "with", "that", "this", "from", "have", "will", ) and cleaned not in seen_tokens: spec_tokens.append(cleaned) seen_tokens.append(cleaned) if len(spec_tokens) >= 5: break for token in spec_tokens[:5]: evidence.append({ "label": f"Spec keyword '{token}' present in deliverable", "passed": token in content_lower, }) if job["deliverable_format"] == "text": evidence.append({ "label": "Deliverable meets minimum length (50 chars)", "passed": len(deliverable_content.strip()) >= 50, }) return evidence def _fetch_claim_sources(self, job: dict): source_texts = [] source_evidence = [] sources_str = self.job_sources.get(job["id"], "") sources = sources_str.split(",") if sources_str else [] for idx, src in enumerate(sources[:5], start=1): try: text = get_webpage(src, mode="text")[:1500] fetched = "[Could not fetch" not in text and len(text.strip()) > 0 source_texts.append(f"SOURCE {idx} ({src}):\n{text}") except Exception: text = "" fetched = False source_texts.append(f"SOURCE {idx} ({src}): [Could not fetch source]") source_evidence.append({ "label": f"Source #{idx} accessible ({src})", "passed": fetched, "source": src, "source_index": idx, }) return "\n\n".join(source_texts), source_evidence def _judge_deliverable(self, job_id: str) -> None: job = self._get_job(job_id) deliverable_content = "" evidence = [] if job["task_type"] == "CLAIM": deliverable_content, source_evidence = self._fetch_claim_sources(job) evidence.extend(source_evidence) elif job["deliverable_format"] in ["url", "github_url"]: try: deliverable_content = get_webpage(job["deliverable"], mode="text")[:3000] except Exception: deliverable_content = "[Could not fetch URL - treating as submission failure]" else: deliverable_content = job["deliverable"][:3000] evidence.extend(self._collect_evidence(job, deliverable_content)) plan = _parse_verification_plan(job["policy"]) min_score = plan["min_score"] gate_failure = None if "url" in plan["gates"] and job["deliverable_format"] in ["url", "github_url"]: url_ok = any( e.get("label") == "Deliverable URL reachable" and e.get("passed") for e in evidence if isinstance(e, dict) ) if not url_ok: gate_failure = "Hard gate failed: the deliverable URL is unreachable." if "sources" in plan["gates"] and job["task_type"] == "CLAIM": dead_source = any( "accessible" in e.get("label", "") and not e.get("passed") for e in evidence if isinstance(e, dict) ) if dead_source: gate_failure = "Hard gate failed: one or more claim sources are unreachable." dispute_context = "" if job["dispute_round"] > 0: dispute_context = ( f"NOTE: This is dispute round {job['dispute_round']}. " f"The previous verdict was: {job['verdict']} (score: {job['score']}/100). " f"Previous reasoning: {job['judgment_reason']}. Apply higher scrutiny." ) policy_context = "" if job["policy"]: policy_context = f"ADDITIONAL VERIFICATION POLICY: {job['policy']}" claim_context = "" if job["task_type"] == "CLAIM": claim_context = "This is a CLAIM VERIFICATION job." category_context = "" if job["task_type"] == "TASK": resolved = _resolve_category(job["category"]) if resolved in CATEGORY_CONTEXTS: category_context = CATEGORY_CONTEXTS[resolved] if gate_failure is not None: parsed = {"verdict": "REJECTED", "score": 0, "judgment_reason": gate_failure, "release_percentage": 0} else: try: prompt = f"""You are a neutral judge evaluating delivered work. JOB TITLE: {job['title']} JOB SPEC: {job['spec']} DELIVERABLE FORMAT: {job['deliverable_format']} DELIVERABLE SUBMITTED: {job['deliverable']} REFERENCE CONTENT: {deliverable_content} EVIDENCE CHECKS: {json.dumps(evidence)} {policy_context} {claim_context} {category_context} {dispute_context} Rules: - APPROVED (score {min_score}-100): Full payment released. - PARTIAL (score 30-{min_score - 1}): Partial payment based on release_percentage. - REJECTED (score 0-29): No payment released. Return ONLY valid JSON: {{"verdict": "APPROVED"|"REJECTED"|"PARTIAL", "score": <0-100>, "judgment_reason": "<2-4 sentences>", "release_percentage": <0-100>, "validators": [{{"id":1,"verdict":"...","score":N}},{{"id":2,"verdict":"...","score":N}},{{"id":3,"verdict":"...","score":N}}]}}""" def leader_fn(): return gl.nondet.exec_prompt(prompt, response_format="json") def validator_fn(leaders_res: gl.vm.Result) -> bool: if not isinstance(leaders_res, gl.vm.Return): return False leader_data = leaders_res.calldata if not isinstance(leader_data, dict): return False v = leader_data.get("verdict") if v not in ("APPROVED", "REJECTED", "PARTIAL"): return False s = leader_data.get("score", 0) if not isinstance(s, (int, float)) or s < 0 or s > 100: return False rp = leader_data.get("release_percentage", 0) if not isinstance(rp, (int, float)) or rp < 0 or rp > 100: return False validators = leader_data.get("validators", []) if not isinstance(validators, list) or len(validators) != 3: return False for val in validators: if not isinstance(val, dict): return False if val.get("verdict") not in ("APPROVED", "REJECTED", "PARTIAL"): return False return True result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) parsed = _parse_llm_json(str(result)) if isinstance(result, str) else result except Exception: parsed = {"verdict": "REJECTED", "score": 0, "judgment_reason": "Judge failed to evaluate", "release_percentage": 0} job["verdict"] = parsed.get("verdict", "REJECTED") job["score"] = int(parsed.get("score", 0)) job["judgment_reason"] = parsed.get("judgment_reason", "Evaluation failed") release_pct = int(parsed.get("release_percentage", 0)) if job["verdict"] == "APPROVED" and job["score"] < min_score: job["verdict"] = "PARTIAL" job["judgment_reason"] = f"Score {job['score']} is below minimum {min_score}; downgraded to PARTIAL. {job['judgment_reason']}" release_pct = max(10, min(90, job["score"])) if job["verdict"] == "APPROVED": release_pct = 100 elif job["verdict"] == "REJECTED": release_pct = 0 elif job["verdict"] == "PARTIAL": release_pct = max(10, min(90, release_pct)) validators = parsed.get("validators") if ( not isinstance(validators, list) or len(validators) == 0 or not all(isinstance(v, dict) and "verdict" in v and "score" in v for v in validators) ): validators = [{"id": i + 1, "verdict": job["verdict"], "score": job["score"]} for i in range(3)] consensus_score = sum(int(v["score"]) for v in validators) / len(validators) record = { "round": job["dispute_round"], "verdict": job["verdict"], "score": job["score"], "judgment_reason": job["judgment_reason"], "release_percentage": release_pct, "consensus_score": consensus_score, "validators": validators, "evidence": evidence, "timestamp": gl.block.timestamp, "deliverable": job["deliverable"], "spec": job["spec"], } existing = [] raw = self.verifications.get(job_id, "") if raw: existing = json.loads(raw) existing.append(record) self.verifications[job_id] = json.dumps(existing) platform_fee = (job["budget"] * self.platform_fee_bps) // 10000 net_budget = job["budget"] - platform_fee release_amount = (net_budget * release_pct) // 100 refund_amount = net_budget - release_amount if release_amount > 0: gl.transfer(job["agent"], release_amount) self.total_released += release_amount stats = self._get_stats(job["agent"]) stats["completed"] += 1 v = job["verdict"].lower() if v == "approved": stats["approved"] += 1 elif v == "rejected": stats["rejected"] += 1 elif v == "partial": stats["partial"] += 1 stats["total_earned"] += release_amount stats["score_sum"] += job["score"] stats["delivery_time_sum"] += max(0, gl.block.timestamp - job["created_at"]) self._save_stats(job["agent"], stats) if refund_amount > 0: gl.transfer(job["poster"], refund_amount) if platform_fee > 0: gl.transfer(gl.contract_address, platform_fee) job["status"] = "COMPLETE" job["completed_at"] = gl.block.timestamp self._save_job(job_id, job)