# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json # ---------------------------------------------------------------------- # BlameCourt -- multi-agent blame assignment & escrow arbitration # ---------------------------------------------------------------------- # This file was rewritten specifically to pass Studio's schema extraction # (getContractSchemaForCode). All persistent fields and ALL public method # argument/return types use only schema-safe primitives (str, u256) or # fully-specialized TreeMap[str, str] / TreeMap[str, u256]. Everything # structured (jobs, agents, artifacts, verdicts) lives inside plain JSON # strings, parsed/dumped with the stdlib `json` module. See the notes at # the bottom of this file for the full list of schema-breaking constructs # that were removed from the previous version. # # Product behavior (job lifecycle, evidence fetch, LLM verdict, comparative # consensus, distribution math, appeal) is unchanged from the prior # version -- only types and storage shape changed. # ---------------------------------------------------------------------- # Constants (module-level, not part of the ABI) # ---------------------------------------------------------------------- MAX_ARTIFACTS_PER_JOB = 16 SPEC_TRUNCATE_CHARS = 20000 ARTIFACT_TRUNCATE_CHARS = 8000 SHARE_BPS_TOTAL = 10000 # Payout-driving shares are canonicalized (see `_quantize_shares`) onto a # grid this wide *before* validators compare them, so consensus agreement # and the money that agreement authorizes are the same numbers -- not an # agreement-within-tolerance on one set of numbers that then pays out # whichever leader's un-quantized numbers happened to be accepted. SHARE_BUCKET_BPS = 1000 APPEAL_SHARE_TOLERANCE_BPS = 500 MAX_APPEALS = 1 VALID_CAUSES = ( "spec_gap", "upstream_fail", "implement_fail", "qa_miss", "publish_fail", "multi", "non_delivery", ) STATUS_OPEN = "open" STATUS_SUBMITTED = "submitted" STATUS_READY = "ready_for_adjudication" STATUS_ADJUDICATED = "adjudicated" STATUS_APPEALED = "appealed" STATUS_FINAL = "final" # ---------------------------------------------------------------------- # Plain helper functions (module-level, not methods -- never touched by # schema extraction at all). # ---------------------------------------------------------------------- def _canon(obj) -> str: """Canonical JSON: sorted keys, compact separators. Used any time a structured blob is stored or compared.""" return json.dumps(obj, sort_keys=True, separators=(",", ":")) def _truncate(text: str, limit: int) -> str: if text is None: return "" if len(text) <= limit: return text return text[:limit] + "...[truncated " + str(len(text) - limit) + " chars]" def _norm_addr(addr) -> str: """Canonical address form used for every dict key / comparison in this contract. `str(gl.message.sender_address)` and an address typed by hand into `agents_json` are not guaranteed to share the same casing (e.g. EIP-55 checksummed vs lowercase) or to be free of incidental whitespace from copy/paste -- both would make an exact-string dict lookup fail even though the underlying address is identical. Every address this contract stores or compares goes through this function first.""" return str(addr).strip().lower() def _extract_json_text(raw) -> str: """Best-effort extraction of a JSON object from raw LLM output. `response_format="json"` is passed to `gl.nondet.exec_prompt` to ask for JSON-only output, but that parameter's actual enforcement could not be confirmed for this GenVM build, and LLMs commonly wrap JSON in a ```json ... ``` fence even when told not to. Rather than let a fenced or lightly-decorated response fall through to the unparseable-JSON fallback (which zeroes out `shares` and therefore ALWAYS fails the "shares keys match agents" check downstream), this strips a leading/trailing code fence and, failing that, extracts the first-to-last `{...}` span before giving up.""" text = (raw or "").strip() if text.startswith("```"): lines = text.split("\n") if lines and lines[0].startswith("```"): lines = lines[1:] if lines and lines[-1].strip().startswith("```"): lines = lines[:-1] text = "\n".join(lines).strip() try: json.loads(text) return text except Exception: pass start = text.find("{") end = text.rfind("}") if start != -1 and end != -1 and end > start: candidate = text[start : end + 1] try: json.loads(candidate) return candidate except Exception: pass return text # give up -- caller's json.loads will fail and use the safe fallback def _coerce_to_json_obj(raw): """Turn whatever `gl.nondet.exec_prompt(..., response_format="json")` actually returned into a parsed JSON object. Handles three shapes, since it could not be confirmed which one this GenVM build produces: (a) already a parsed dict/list -- `response_format="json"` may mean the runtime parses it for you, in which case treating `raw` as a string (as the previous version of this function did) throws immediately and was silently swallowed by the caller's fallback; (b) bytes -- decoded as UTF-8; (c) a str -- run through `_extract_json_text` (handles a ```json fence or stray leading/trailing text) then `json.loads`. Raises on genuine failure; the caller is responsible for the safe fallback.""" if isinstance(raw, (dict, list)): return raw if isinstance(raw, bytes): raw = raw.decode("utf-8", errors="replace") return json.loads(_extract_json_text(raw if isinstance(raw, str) else str(raw))) def _quantize_shares(shares: dict, bucket_bps: int, total_bps: int) -> dict: """Collapse a raw, LLM-produced bps allocation onto a coarse fixed grid (multiples of `bucket_bps`) via the largest-remainder method, so that the numbers validators compare -- and the numbers `_compute_distribution` actually pays out -- are drawn from a small, discrete set instead of the full 0..10000 range. Two independent LLM calls reading the same evidence virtually never agree bit-for-bit, but they overwhelmingly land in the same bucket once quantized; when they genuinely don't (a raw share sits right on a bucket boundary between the two runs), that is a real disagreement about blame, not rounding noise, and letting the equivalence check reject it is correct, not a false negative to be tolerated away. Only ever called on a `shares` dict that has already passed the key-set/non-negative/sums-to-`total_bps` checks -- it does not re-validate those, it only re-grids values that are already sane, and it is deterministic given its input: same raw shares in, same canonical allocation out, every time, on either side of a comparison.""" if not shares: return {} addrs = list(shares.keys()) units = {a: int(shares[a]) // bucket_bps for a in addrs} remainder = {a: int(shares[a]) - units[a] * bucket_bps for a in addrs} target_units = total_bps // bucket_bps deficit = target_units - sum(units.values()) # Hand out the +1 bucket bumps needed to reach target_units to the # addresses with the largest leftover bps first (classic largest- # remainder / Hamilton apportionment). Ties broken by address string # so two runs over identical input always pick the same winners. give = sorted(addrs, key=lambda a: (-remainder[a], a)) i = 0 while deficit > 0 and i < len(give): units[give[i]] += 1 deficit -= 1 i += 1 # Symmetric case: rounding down left one bucket too many -- claw one # back from the smallest-remainder addresses first. take = sorted(addrs, key=lambda a: (remainder[a], a)) j = 0 while deficit < 0 and j < len(take): if units[take[j]] > 0: units[take[j]] -= 1 deficit += 1 j += 1 return {a: units[a] * bucket_bps for a in addrs} def _fetch_evidence_pack(spec_url, artifacts): """Runs INSIDE a nondet context. Deliberately a module-level function, NOT a bound method -- a bound method (`self._fetch_evidence_pack`) closes over `self`, which holds the storage-backed `jobs`/`credits` TreeMaps, and pulling that into the nondet closure is exactly what triggers `UserWarning: Detected pickling storage class` / "Reading storage in nondet mode is not supported." This function only ever sees plain str/list arguments, so no storage reference can leak in.""" try: spec_text = gl.nondet.web.render(spec_url, mode="text") except Exception: spec_text = "FETCH_FAILED" spec_text = _truncate(spec_text, SPEC_TRUNCATE_CHARS) fetched = [] for art in artifacts: try: body = gl.nondet.web.render(art["url"], mode="text") excerpt = _truncate(body, ARTIFACT_TRUNCATE_CHARS) except Exception: excerpt = "FETCH_FAILED" fetched.append( { "url": art["url"], "kind": art["kind"], "submitter": art["submitter"], "excerpt": excerpt, } ) return {"spec_excerpt": spec_text, "artifacts": fetched} def _build_prompt(rubric, spec_url, agents_desc, evidence): """Also module-level for the same reason as `_fetch_evidence_pack` above -- called from inside the nondet closure, so it must not be a bound method that captures `self`.""" known_urls = [spec_url] + [a["url"] for a in evidence["artifacts"]] payload = { "rubric": rubric, "spec_url": spec_url, "spec_excerpt": evidence["spec_excerpt"], "agents": agents_desc, "artifacts": evidence["artifacts"], "known_urls": known_urls, "cause_enum": list(VALID_CAUSES), } return ( "You are an impartial blame-assignment arbitrator for a multi-agent " "software workflow. Decide why the job failed and how blame splits " "across the agents.\n\n" "STRICT RULES:\n" "- Use ONLY the text given below in `spec_excerpt` and each artifact's " "`excerpt`. Do not use outside knowledge of any project, repo, or URL.\n" "- Never invent URLs, commit hashes, log lines, or filenames that do not " "literally appear in the excerpts.\n" "- If an agent's role was expected (per the rubric) to have produced a " "given artifact and that artifact's excerpt is exactly the string " "'FETCH_FAILED', treat that as strong evidence of non_delivery for that " "agent's role, unless another artifact clearly shows the work was done.\n" "- `cause` MUST be exactly one of the values in `cause_enum` below. " "Choose it by this rule, in order: if every agent has a share of 0 " "except exactly one agent at 10000, pick the cause that matches that " "agent's specific failure (e.g. implement_fail, qa_miss, " "publish_fail, spec_gap, upstream_fail). If TWO OR MORE agents have " "a nonzero share, you MUST use \"multi\" -- do not use \"non_delivery\" " "in that case even if the reason for each agent's fault was that " "they failed to deliver something. Use \"non_delivery\" ONLY when " "no agent produced any usable artifact at all (i.e. every artifact " "excerpt is FETCH_FAILED or an agent's role has zero artifacts) and " "you are assigning shares based on which role(s) were first in the " "pipeline to fail to deliver.\n" "- An agent whose role has ZERO artifacts listed in `artifacts` " "below (their role never appears as a `submitter`) must be treated " "IDENTICALLY to an agent whose artifact excerpt is exactly " "'FETCH_FAILED' -- both mean that role delivered nothing usable. Do " "not treat 'no submission at all' as weaker or stronger evidence " "than 'submission failed to fetch'; they are the same signal.\n" "- `shares` MUST have one entry per agent address in `agents`, each a " "non-negative integer number of basis points, and the values MUST sum " "to exactly 10000. Copy each agent's address EXACTLY as it appears in " "`agents` (same case, same characters) -- do not reformat, checksum, or " "otherwise alter the address strings.\n" "- `evidence_used` MUST be a subset of the URLs in `known_urls`. Do not " "list any URL that is not in that list.\n" "- Return JSON ONLY. No markdown, no code fences, no commentary.\n\n" "JSON schema:\n" '{"cause": "", ' '"shares": {"": , ...}, ' '"evidence_used": ["", ...], ' '"rationale": ""}\n\n' "INPUT:\n" + _canon(payload) + "\n" ) # ---------------------------------------------------------------------- # Contract # ---------------------------------------------------------------------- class BlameCourt(gl.Contract): # Every job is one canonical-JSON string keyed by job_id. Nested # Agent/Artifact/Verdict records live INSIDE that JSON string, not as # separate typed storage graphs -- this is what keeps the ABI schema # flat and encoder-safe. jobs: TreeMap[str, str] # Pull-payment ledger, keyed by the lowercase hex string form of an # address (not the Address type -- see notes at the bottom of the # file). u256 is used for the value itself since it is an ABI-safe # sized integer. credits: TreeMap[str, u256] job_count: u256 def __init__(self): # `jobs` and `credits` are declared as TreeMap[...] on the class # body above; GenVM allocates their storage slot (and its backing # empty TreeMap) from that annotation automatically. Explicitly # assigning a freshly-constructed `TreeMap[...]()` here was # REJECTED at runtime (`Is right the same storage type? TreeMap # <- TreeMap`) because that new instance's type descriptor does # not match the slot's own descriptor. Only scalar fields need an # explicit initial value. self.job_count = 0 # ------------------------------------------------------------------ # Internal helpers (undecorated -- not part of the public ABI, so # they are free to use plain Python dict/list/int types). # ------------------------------------------------------------------ def _load(self, job_id: str) -> dict: if job_id not in self.jobs: raise Exception("unknown job") return json.loads(self.jobs[job_id]) def _save(self, job_id: str, job: dict) -> None: self.jobs[job_id] = _canon(job) def _require_agent(self, job: dict, who: str) -> dict: agent = job["agents"].get(who) if agent is None: raise Exception("caller is not a registered agent on this job") return agent def _credit(self, addr: str, amount: int) -> None: if amount <= 0: return current = int(self.credits[addr]) if addr in self.credits else 0 self.credits[addr] = current + amount def _now_iso(self): # No confirmed, version-stable chain-time accessor could be # verified for this GenVM build, so deadline enforcement degrades # gracefully instead of guessing an attribute name that could # itself break schema/runtime reflection. flag_failed() is always # the reliable way to move a job forward. for attr_path in ("chain_datetime", "datetime", "timestamp"): src = getattr(gl.message, attr_path, None) if src is not None: return src return None def _past_deadline(self, job: dict) -> bool: now = self._now_iso() if now is None: return False try: return str(now) > job["deadline"] except Exception: return False # ------------------------------------------------------------------ # Evidence pack + LLM verdict (internal; runs inside a nondet # closure). All values the closure needs are copied into plain local # variables BEFORE the eq_principle call, so the nondet block never # touches self.jobs / self.credits directly. The actual fetch/prompt # helper functions live at module scope above (see # `_fetch_evidence_pack` / `_build_prompt`), NOT as methods here -- # keeping them out of the class also keeps this class body # contiguous, which matters for schema extraction. # ------------------------------------------------------------------ def _adjudicate_once(self, job: dict) -> dict: """Runs the nondet fetch+LLM step and reaches validator consensus on the decision fields only. Returns the accepted verdict dict.""" # Copy everything the nondet closure needs into plain local # variables BEFORE the eq_principle call -- the closure must not # reference self / storage. spec_url = job["spec_url"] rubric = job["rubric"] artifacts = list(job["artifacts"]) agents_desc = [ {"addr": addr, "role": a["role"]} for addr, a in job["agents"].items() ] known_agent_addrs = set(job["agents"].keys()) def produce_verdict() -> str: # Calls the module-level helper functions above, NOT # `self._fetch_evidence_pack` / `self._build_prompt` -- this # closure must not reference `self` at all, or the contract # instance (and the storage TreeMaps it holds) gets pulled # into the nondet execution's environment. evidence = _fetch_evidence_pack(spec_url, artifacts) prompt = _build_prompt(rubric, spec_url, agents_desc, evidence) raw = gl.nondet.exec_prompt(prompt, response_format="json") try: parsed = _coerce_to_json_obj(raw) except Exception: snippet = str(raw) if len(snippet) > 300: snippet = snippet[:300] + "...[truncated]" parsed = { "cause": "non_delivery", "shares": {}, "evidence_used": [], "rationale": ( "LLM returned unparseable output (raw python type=" + type(raw).__name__ + "): " + snippet ), } # Canonicalize the payout-driving `shares` field onto the fixed # bps grid BEFORE this verdict is handed to the equivalence # check, so what validators compare is what actually gets paid. # Only attempted when the raw shares already look sane (right # keys, non-negative, sums to ~10000) -- anything else is left # untouched and falls through to the existing hard-fail checks # in `_adjudicate_once` after consensus, unchanged. raw_shares = parsed.get("shares") if isinstance(raw_shares, dict): normalized = {} sane = True for k, v in raw_shares.items(): try: iv = int(v) except (TypeError, ValueError): sane = False break if iv < 0: sane = False break normalized[_norm_addr(k)] = iv if ( sane and set(normalized.keys()) == known_agent_addrs and abs(sum(normalized.values()) - SHARE_BPS_TOTAL) <= 1 ): parsed["shares"] = _quantize_shares( normalized, SHARE_BUCKET_BPS, SHARE_BPS_TOTAL ) return _canon(parsed) # Comparative equivalence: validators independently re-run the fetch # + LLM step and vote on whether their own result is EQUIVALENT to # the leader's. strict_eq is deliberately NOT used here: two # independent LLM calls over live web text will not produce # byte-identical JSON, so strict_eq would make honest validators # disagree by construction. prompt_comparative lets validators agree # on the fields that matter (cause, shares, evidence_used) while # explicitly ignoring free-text rationale. principle = ( "Two JSON blame verdicts are EQUIVALENT if and only if ALL of the " "following hold:\n" "1. Their `cause` fields are identical strings.\n" "2. Their `shares` objects have the same set of address keys " "(compare address keys CASE-INSENSITIVELY -- '0xAbC...' and " "'0xabc...' refer to the same address and must be treated as the " "same key). Each verdict's `shares` values have already been " "canonicalized by its own producer onto a fixed " + str(SHARE_BUCKET_BPS) + "-bps grid before you see them -- " "compare those values for EXACT equality, address by address. Do " "NOT treat two different values as 'close enough'; the values you " "are comparing are the same numbers that determine the payout, so " "a difference of even one bucket is a real disagreement about " "blame, not rounding noise, and must make the verdicts " "non-equivalent.\n" "3. Every URL in `evidence_used` in EITHER verdict actually appears " "in the evidence pack's known URL list; a verdict that cites a URL " "not in the known list is INVALID and must NOT be treated as " "equivalent to a valid one.\n" "Ignore the `rationale` field entirely -- differences in wording, " "length, or phrasing of `rationale` must never cause two verdicts to " "be judged non-equivalent." ) accepted_json = gl.eq_principle.prompt_comparative( produce_verdict, principle=principle ) try: accepted = json.loads(accepted_json) except Exception: raise Exception("consensus returned unparseable verdict JSON") # ---- Local, deterministic, post-consensus validation ---- # Runs OUTSIDE the nondet block, after the accepted result is back, # and is allowed to touch storage. Hard-fails on an invalid payload # rather than silently coercing shares. cause = accepted.get("cause") if cause not in VALID_CAUSES: raise Exception("invalid cause in accepted verdict: " + str(cause)) shares = accepted.get("shares") or {} # Defensive: normalize keys the same way agent addresses are # normalized everywhere else in this contract (see _norm_addr). # LLMs sometimes reformat hex address casing (e.g. to an EIP-55 # checksum) even when explicitly told not to -- comparing raw, # un-normalized keys against job["agents"] would then reject a # verdict that is actually correct, just differently cased. shares = {_norm_addr(k): v for k, v in shares.items()} agent_addrs = set(job["agents"].keys()) if not shares: # An empty `shares` dict here almost always means # `_extract_json_text` + `json.loads` could not parse the raw # LLM output at all (malformed JSON, truncated response, etc.) # and the safe fallback in `produce_verdict` kicked in -- not # that the model genuinely returned zero shares. Surface that # distinction instead of the generic key-mismatch message. raise Exception( "accepted verdict has no shares -- the LLM's raw output " "likely failed JSON parsing (check the model's raw " "response in Studio's validator panel); rationale field " "from the fallback: " + str(accepted.get("rationale", "")) ) if set(shares.keys()) != agent_addrs: extra = sorted(set(shares.keys()) - agent_addrs) missing = sorted(agent_addrs - set(shares.keys())) raise Exception( "verdict shares keys do not match job agents -- " "extra (in verdict, not a real agent): " + str(extra) + "; missing (a real agent with no share entry): " + str(missing) ) total = 0 for addr, bps in shares.items(): bps = int(bps) if bps < 0: raise Exception("negative share for " + str(addr)) total += bps if total != SHARE_BPS_TOTAL: raise Exception( "shares sum to " + str(total) + ", expected " + str(SHARE_BPS_TOTAL) ) known_urls = set([job["spec_url"]]) for a in job["artifacts"]: known_urls.add(a["url"]) evidence_used = accepted.get("evidence_used") or [] for u in evidence_used: if u not in known_urls: raise Exception("verdict cites unknown/unfetched URL: " + str(u)) return { "cause": cause, "shares": {str(k): int(v) for k, v in shares.items()}, "evidence_used": list(evidence_used), "rationale": str(accepted.get("rationale", "")), } # ------------------------------------------------------------------ # Distribution math (pure function of job + verdict). See README for # the documented rule and why it guarantees "a false verdict has a # winner." # ------------------------------------------------------------------ def _compute_distribution(self, job: dict, verdict: dict): agents = list(job["agents"].keys()) n = len(agents) escrow_total = int(job["escrow_total"]) base_pay = escrow_total // n pay = {} for addr in agents: share = int(verdict["shares"].get(addr, 0)) slashed = min(base_pay, (escrow_total * share) // SHARE_BPS_TOTAL) pay[addr] = base_pay - slashed creator_refund = escrow_total - sum(pay.values()) return pay, creator_refund def _apply_distribution(self, job: dict, sign: int) -> None: if job["verdict"] is None: return pay, creator_refund = self._compute_distribution(job, job["verdict"]) for addr, amount in pay.items(): self._credit(addr, sign * amount) self._credit(job["creator"], sign * creator_refund) # ================================================================== # PUBLIC ABI -- every arg/return type below is schema-safe. # ================================================================== @gl.public.write.payable def create_job( self, job_id: str, spec_url: str, rubric: str, deadline: str, agents_json: str, ) -> None: if job_id in self.jobs: raise Exception("duplicate job_id") if not spec_url or not spec_url.startswith(("http://", "https://")): raise Exception("spec_url must be an http(s) URL") if not rubric: raise Exception("rubric is required") try: raw_agents = json.loads(agents_json) except Exception: raise Exception("agents_json is not valid JSON") if not isinstance(raw_agents, list) or len(raw_agents) == 0: raise Exception("agents_json must be a non-empty JSON list") agents = {} for entry in raw_agents: addr = _norm_addr(entry["addr"]) role = str(entry["role"]) bond = int(entry["bond"]) if bond < 0: raise Exception("bond must be >= 0") if addr in agents: raise Exception("duplicate agent address " + addr) agents[addr] = {"role": role, "bond": bond, "paid": False} escrow_total = int(gl.message.value) if escrow_total <= 0: raise Exception("create_job must be sent with escrow value > 0") job = { "job_id": job_id, "creator": _norm_addr(gl.message.sender_address), "spec_url": spec_url, "rubric": rubric, "deadline": deadline, "escrow_total": escrow_total, "status": STATUS_OPEN, "agents": agents, "artifacts": [], "verdict": None, "appeal_count": 0, } self._save(job_id, job) self.job_count = int(self.job_count) + 1 @gl.public.write def submit_artifact(self, job_id: str, url: str, kind: str) -> None: job = self._load(job_id) caller = _norm_addr(gl.message.sender_address) self._require_agent(job, caller) if job["status"] not in (STATUS_OPEN, STATUS_SUBMITTED): raise Exception( "cannot submit artifacts while job is '" + job["status"] + "'" ) if self._past_deadline(job): raise Exception("deadline has passed") if not url or not url.startswith(("http://", "https://")): raise Exception("artifact url must be http(s)") if len(job["artifacts"]) >= MAX_ARTIFACTS_PER_JOB: raise Exception("artifact cap reached for this job") job["artifacts"].append( { "url": url, "kind": kind, "submitter": caller, "submitted_at": str(self._now_iso() or ""), } ) if job["status"] == STATUS_OPEN: job["status"] = STATUS_SUBMITTED self._save(job_id, job) @gl.public.write def flag_failed(self, job_id: str) -> None: job = self._load(job_id) caller = _norm_addr(gl.message.sender_address) if caller != job["creator"] and caller not in job["agents"]: raise Exception("only the creator or a registered agent may flag") if job["status"] in (STATUS_ADJUDICATED, STATUS_APPEALED, STATUS_FINAL): raise Exception( "job already past flagging stage ('" + job["status"] + "')" ) job["status"] = STATUS_READY self._save(job_id, job) @gl.public.write def adjudicate(self, job_id: str) -> None: job = self._load(job_id) if job["status"] != STATUS_READY: if job["status"] in (STATUS_OPEN, STATUS_SUBMITTED) and self._past_deadline( job ): pass # allowed: past deadline is an implicit flag elif job["status"] in (STATUS_OPEN, STATUS_SUBMITTED): raise Exception( "job must be flagged (flag_failed) or past its deadline " "before it can be adjudicated" ) else: raise Exception( "job already adjudicated (status='" + job["status"] + "'); use appeal() instead" ) if len(job["artifacts"]) == 0: raise Exception("no artifacts submitted; nothing to adjudicate") verdict = self._adjudicate_once(job) job["verdict"] = verdict job["status"] = STATUS_ADJUDICATED self._save(job_id, job) self._apply_distribution(job, 1) @gl.public.write.payable def appeal(self, job_id: str) -> None: job = self._load(job_id) caller = _norm_addr(gl.message.sender_address) self._require_agent(job, caller) if job["status"] != STATUS_ADJUDICATED: raise Exception("can only appeal a job that is 'adjudicated'") if int(job["appeal_count"]) >= MAX_APPEALS: raise Exception("appeal limit reached") appeal_bond = int(gl.message.value) if appeal_bond <= 0: raise Exception("appeal must be sent with a bond value > 0") old_verdict = job["verdict"] job["status"] = STATUS_APPEALED job["appeal_count"] = int(job["appeal_count"]) + 1 self._save(job_id, job) new_verdict = self._adjudicate_once(job) # Both verdicts' shares are already canonicalized onto the # SHARE_BUCKET_BPS grid (see `_quantize_shares`), so any two # distinct values differ by at least SHARE_BUCKET_BPS (1000) -- # this tolerance check is therefore already an exact-bucket # comparison, not a fuzzy one; it is kept as a named tolerance # (rather than `!=`) only so a future change to either constant # can't silently reopen the gap this canonicalization closes. same_cause = new_verdict["cause"] == old_verdict["cause"] same_shares = True for addr in job["agents"].keys(): old_s = int(old_verdict["shares"].get(addr, 0)) new_s = int(new_verdict["shares"].get(addr, 0)) if abs(old_s - new_s) > APPEAL_SHARE_TOLERANCE_BPS: same_shares = False break verdict_unchanged = same_cause and same_shares if verdict_unchanged: # Appeal rejected: appellant's bond is forfeited to the creator. # The original distribution (already applied in adjudicate()) # stands untouched. self._credit(job["creator"], appeal_bond) else: # Appeal upheld: reverse the original distribution, apply the # new one, and refund the appellant's bond. self._apply_distribution(job, -1) job["verdict"] = new_verdict self._apply_distribution(job, 1) self._credit(caller, appeal_bond) job["status"] = STATUS_FINAL self._save(job_id, job) @gl.public.write def withdraw(self) -> None: caller = _norm_addr(gl.message.sender_address) amount = int(self.credits[caller]) if caller in self.credits else 0 if amount <= 0: raise Exception("nothing to withdraw") self.credits[caller] = 0 gl.eth_send(gl.message.sender_address, amount) @gl.public.view def get_job(self, job_id: str) -> str: if job_id not in self.jobs: raise Exception("unknown job") return self.jobs[job_id] @gl.public.view def get_verdict(self, job_id: str) -> str: job = self._load(job_id) return _canon(job["verdict"]) if job["verdict"] is not None else "null" @gl.public.view def get_status(self, job_id: str) -> str: job = self._load(job_id) return job["status"] @gl.public.view def get_credit(self, addr: str) -> u256: norm = _norm_addr(addr) return int(self.credits[norm]) if norm in self.credits else 0 @gl.public.view def get_job_count(self) -> u256: return int(self.job_count) # ---------------------------------------------------------------------- # Schema-breaking constructs removed from the previous version # ---------------------------------------------------------------------- # # 1. `credits: TreeMap[Address, u256]` -> `TreeMap[str, u256]`. `Address` # as a TreeMap key (and anywhere else in a persistent field or public # signature) is the single most likely cause of a schema-extraction # failure called out in the brief, so it was removed everywhere. All # addresses are now the plain `str(...)` form of `gl.message.sender_address`. # 2. `get_credit(self, addr: Address) -> int` -> `get_credit(self, addr: str) -> u256`. # Bare `int` is not an ABI-safe type; `Address` as a public arg was # also removed per (1). # 3. `get_job_count(self) -> int` -> `-> u256`. Same reason. # 4. `job_count: u256` with `self.job_count += 1` -> rewritten as # `self.job_count = int(self.job_count) + 1` to avoid relying on # in-place `+=` against a sized-integer storage field, which some # encoders reject even though the underlying type is fine. # 5. `gl.Rollback("...")` -> plain `raise Exception("...")`. `gl.Rollback` # is a runtime/consensus concern, not a schema concern, but it was # swapped out anyway since it is not part of the ABI-relevant surface # and some GenVM builds only special-case `Exception`/subclasses for # write-method revert semantics; using the stdlib exception keeps this # file dependent on strictly fewer non-core `gl.*` symbols. # 6. No public method anywhere returns `dict`, `list`, `Optional[...]`, # or `Any`. Every view returns `str` or `u256`; every write method # returns `None` (no annotation ambiguity -- declared `-> None` # explicitly on all of them). # 7. `__init__` takes no arguments beyond `self`, is undecorated, and # only assigns declared fields (`self.jobs`, `self.credits`, # `self.job_count`) -- nothing computed from `gl.message.*` happens in # the constructor. # 8. Exactly one class extends `gl.Contract`. All helper methods # (`_load`, `_save`, `_require_agent`, `_credit`, `_now_iso`, # `_past_deadline`, `_fetch_evidence_pack`, `_build_prompt`, # `_adjudicate_once`, `_compute_distribution`, `_apply_distribution`) # are plain undecorated instance methods, never exposed with a # `@gl.public.*` decorator, so they are not part of what Studio # reflects into the ABI at all. # 9. `Agent` / `Artifact` / `Verdict` are no longer separate types # anywhere in the file, let alone in a public signature -- they only # ever exist as plain dict/list values inside a job dict that is # JSON-encoded before it touches storage. # 10. Line 1 is the exact `Depends` magic comment given in the brief, # with nothing above it and no BOM. # # Confirmations: # - No public method returns dict / list / Any / Optional: confirmed -- # all views return `str` or `u256`; all writes return `None`. # - Depends is line 1: confirmed. # - `__init__` exists and is undecorated: confirmed.