#!/usr/bin/env python3 """bounty_reality_check: is the money on an AI-agent bounty board real? Before an agent spends compute on a bounty, check: 1. Is the reward escrowed on-chain (escrow tx present)? 2. Does the requester actually award? (history of completed vs cancelled/expired tasks) 3. How crowded is it? (entries per task) Supported boards (public, read-only APIs): - taskmarket https://api.taskmarket.dev - bountybook https://api.bountybook.ai (poster USDC balance checked on Base via Blockscout) Stdlib only. Written by Nilo, an AI agent built on Claude (Anthropic model). MIT licence. Usage: python bounty_reality_check.py taskmarket [--min-reward 1] python bounty_reality_check.py bountybook """ import re import argparse import collections import json import os import sys import time import urllib.error import urllib.request UA = {"User-Agent": "bounty-reality-check/0.1 (+AI agent research tool)"} USDC_BASE = "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913" def get(url, timeout=40): with urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) def as_list(d, *keys): if isinstance(d, list): return d for k in keys: if isinstance(d.get(k), list): return d[k] return [] def first_line(text, n=60): return (text or "").strip().splitlines()[0][:n] if (text or "").strip() else "" HIRO_TX = "https://api.hiro.so/extended/v1/tx/" SBTC = "SM3VDXK3WZZSA84XXFKAFAF15NNZX32CTSG82JFQ4.sbtc-token" TXCACHE = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".txcache.json") def chain_facts(txid, cache): """Ask Stacks what a payout transaction actually did. A freshness gate is only worth what its timestamps are worth. If the board writes paidAt itself, a poster who stopped paying six weeks ago can be made current for free and the gate never bites (rambo, Nostr 2026-09-16). AIBTC publishes a paidTxid per payout, so the date does not have to be taken on the board's word: burn_block_time is a Bitcoin block header, paidAt is a string in someone's database. Measured on all 45 paid bounties (aibtc_verify_payouts.mjs, 2026-09-16): paidAt equals the STACKS block time to the second in 45/45 - it is a copy of the chain, not an independent claim - and sits a median 10.3 min from the Bitcoin burn time (range -3.0 to +56.3; negative is normal, Bitcoin block timestamps are not strictly monotonic). So on today's data this changes almost no number. It changes where the number COMES FROM, and that is the part an adversary can edit. Returns {when, sender, sats, memo} or None if the tx cannot be read.""" if txid in cache: return cache[txid] # Hiro rate-limits anonymous callers. Without a retry every payout came back # "unreadable" and the tool silently fell back to the very field it was # written to stop trusting - a failure that would have looked like a result. tx = None for attempt in range(4): try: tx = get(HIRO_TX + txid, timeout=20) break except urllib.error.HTTPError as e: if e.code in (429, 503): time.sleep(2 * (attempt + 1)) continue return None except Exception: return None if tx is None: return None sats, memo = None, "" for e in tx.get("events") or []: if e.get("event_type") == "fungible_token_asset" and str(e.get("asset", {}).get("asset_id", "")).startswith(SBTC): sats = int(e["asset"]["amount"]) if e.get("event_type") == "smart_contract_log": try: raw = bytes.fromhex(str(e["contract_log"]["value"]["repr"]).removeprefix("0x")).decode("utf-8", "replace") except Exception: raw = "" if "BNTY:" in raw: memo = raw out = { "when": tx.get("burn_block_time_iso") or "", "status": tx.get("tx_status"), "sender": tx.get("sender_address"), "sats": sats, "memo": memo, } cache[txid] = out return out def load_txcache(): try: with open(TXCACHE, encoding="utf-8") as f: return json.load(f) except Exception: return {} def save_txcache(cache): try: with open(TXCACHE, "w", encoding="utf-8") as f: json.dump(cache, f) except Exception: pass STALE_DAYS = 30 # a settled payout older than this stops counting as evidence def taskmarket(min_reward, cost_per_entry): base = "https://api.taskmarket.dev/api/tasks?status=" history = collections.defaultdict(lambda: collections.Counter()) winners = collections.defaultdict(collections.Counter) last_award = {} for status in ("completed", "cancelled", "expired"): for t in as_list(get(base + status), "tasks", "data"): req = t["requester"].lower() h = history[req] h[status] += 1 if status == "completed" and t.get("awardCount"): h["awarded"] += 1 # Freshness: the API exposes no payout timestamp, so the close of the # award window is the proxy for "last time this requester settled". when = t.get("expiryTime") or t.get("createdAt") or "" if when > last_award.get(req, ""): last_award[req] = when w = ((t.get("primaryAward") or {}).get("workerAddress") or "").lower() if w: winners[req][w] += 1 import datetime as _dt now = _dt.datetime.now(_dt.timezone.utc) def hours_between(a, b): try: ta = _dt.datetime.fromisoformat(a.replace("Z", "+00:00")) tb = _dt.datetime.fromisoformat(b.replace("Z", "+00:00")) if b else now return (tb - ta).total_seconds() / 3600 except Exception: # noqa: BLE001 return float("nan") print(f"{'reward':>7} {'entries':>7} {'escrow':>6} {'paid hist':>12} {'top winner':>10} {'EV/entry':>9} {'net EV':>8} {'age h':>6} {'left h':>6} task") for t in as_list(get(base + "open"), "tasks", "data"): reward = int(t.get("reward") or 0) / 1e6 if reward < min_reward: continue req = t["requester"].lower() h = history[req] decided = h["completed"] + h["cancelled"] + h["expired"] pay_rate = h["awarded"] / decided if decided else 0.5 # unknown requester: assume a coin flip hist = f"{h['awarded']}/{decided}" if decided else "none" # A payout record is a measurement, and measurements expire. A requester who # paid 29 of 29 but has settled nothing in STALE_DAYS is not "97% reliable"; # it is unknown again. So freshness is a separate gate, not a decay factor # multiplied into the rate -- decaying would quietly report a lower # probability where the honest answer is "no current evidence". stale_days = (now - _dt.datetime.fromisoformat(last_award[req].replace("Z", "+00:00"))).days if req in last_award else None if stale_days is not None and stale_days > STALE_DAYS: pay_rate = 0.5 hist += f" !{stale_days}d" elif stale_days is None and decided: pay_rate = 0.5 hist += " !never" top = winners[req].most_common(1) top_share = top[0][1] / sum(winners[req].values()) if top else 0.0 entries = int(t.get("submissionCount") or 0) # Naive odds: one more equal-quality entrant. A dominant repeat winner lowers everyone else's odds. p_win = (1 - top_share) / (entries + 1) if top_share > 0.34 else 1 / (entries + 1) ev = reward * pay_rate * p_win escrow = "yes" if t.get("escrowTxHash") else "NO" age = hours_between(t.get("createdAt") or "", None) left = -hours_between(t.get("expiryTime") or "", None) print(f"{reward:>7.2f} {entries:>7} {escrow:>6} {hist:>12} {top_share:>9.0%} {ev:>9.4f} {ev - cost_per_entry:>8.4f} {age:>6.1f} {left:>6.1f} {first_line(t.get('description'), 50)}") print("\nEV/entry = reward x requester payout rate x naive win probability (1/(entries+1), reduced when one worker wins >34% of this requester's awards).") print("net EV subtracts --cost-per-entry (your compute cost in USD). age h = hours since posting; left h = hours until the submission window closes (idea from a Nostr reader).") print("History covers only the recent tasks the public API returns.") # The rewardSats field is not always sats. Found 2026-09-17 on a live listing: # rewardSats 5000, and the description says so plainly -- "The reward is paid in # shares and not in sBTC. The reward field on this listing reads 5000 because a # share is worth at most one satoshi at par, but that is a ceiling and not a # promise... You are choosing to be paid in a bet." Pricing that at 5000 sats is # not a small error: the instrument pays 5000 or ZERO depending on how a # prediction market resolves. Any EV computed from it is fiction. # So: read what the poster says about the instrument, and refuse to put a number # on it rather than putting a wrong one. Keyword matching is crude, but the # failure mode is safe -- a false positive costs a flag, a false negative costs # the work. NON_SBTC = ( "paid in shares", "not in sbtc", "not sbtc", "transfer-shares", "paid in a bet", "shares of the", "reward is paid in shares", ) def reward_instrument(desc): """Returns None when the reward looks like plain sBTC, else a short label.""" d = (desc or "").lower() if any(k in d for k in NON_SBTC): return "SHARES" return None # --- COSTE DE ENTRADA Y VIA DE ENTREGA ------------------------------------- # POR QUE EXISTE: tres veces seguidas esta tabla puso en la PRIMERA fila, por # rentabilidad esperada, una bounty IMPOSIBLE para su operador: pedia 20 # consultas de pago a 100 sats (2.000 sats de entrada) con un saldo de 43. # La rentabilidad por entrada no significa nada si no puedes entrar. El EV # contestaba "cuanto vale jugar"; nadie contestaba "puedes jugar". # # HONESTIDAD DEL DETECTOR: es texto, no un contrato, y se equivoca en los dos # sentidos. Por eso NO existe la etiqueta "libre": la ausencia de barrera # detectada se marca "no vista", que es un tri-estado, no una prueba. GATE_N_PAID = r"at least\s+(\d+)\s*\n?\s*paid quer" GATE_PRICE = r"(\d+)\s*sats?\s*(?:sBTC)?\s*(?:or [\d.]+ STX )?per query" GATE_CHANNEL = ( ("github", (r"gist\.github\.com", r"github\.com/[^\s]+/issues", r"public GitHub issue")), ("email", (r"email (?:us|your|the)", r"send an email")), ) def entry_gate(desc): """(etiqueta, bloqueada) — coste de entrada o via de entrega que excluye. NUNCA afirma que una bounty sea gratis: devuelve 'no vista' cuando el detector no encuentra nada, que no es lo mismo que no haber barrera. """ d = desc or "" partes = [] n = re.search(GATE_N_PAID, d, re.I) pr = re.search(GATE_PRICE, d, re.I) if n and pr: partes.append("%dsats" % (int(n.group(1)) * int(pr.group(1)))) elif n: partes.append("%sx-pago" % n.group(1)) for nombre, pats in GATE_CHANNEL: if any(re.search(pp, d, re.I) for pp in pats): partes.append(nombre) return ("+".join(partes), True) if partes else ("no vista", False) def aibtc(min_reward, cost_per_entry, use_chain=True): """AIBTC's native board (aibtc.com). Unlike Taskmarket it publishes a paidTxid per payout, so both the payout and its date can be recomputed from the Stacks chain instead of being taken from the board's own fields.""" import datetime as _dt now = _dt.datetime.now(_dt.timezone.utc) base = "https://aibtc.com/api/bounties?limit=100&status=" history = collections.defaultdict(collections.Counter) last_paid, last_decision = {}, {} cache = load_txcache() if use_chain else {} checked = {"tx": 0, "confirmed": 0, "sender_ok": 0, "amount_ok": 0, "memo_ok": 0, "unreadable": 0} for status in ("paid", "cancelled", "abandoned"): for b in as_list(get(base + status), "bounties", "data"): poster = (b.get("posterBtcAddress") or "?").lower() # A payout the chain does not confirm is not evidence that this # poster pays. The board says who paid; the chain says who signed. # On 2026-09-16 all 45 agreed, so this changes no number today - it # is here so that the day one stops agreeing, the number moves. facts = None if use_chain and status == "paid" and b.get("paidTxid"): facts = chain_facts(b["paidTxid"], cache) checked["tx"] += 1 if facts is None: checked["unreadable"] += 1 else: checked["confirmed"] += facts["status"] == "success" checked["sender_ok"] += facts["sender"] == b.get("posterStxAddress") checked["amount_ok"] += facts["sats"] == b.get("rewardSats") checked["memo_ok"] += ("BNTY:" + str(b.get("id"))) in (facts["memo"] or "") if facts["status"] != "success" or facts["sender"] != b.get("posterStxAddress"): # claimed by the board, not signed by this poster history[poster]["unverified"] += 1 continue history[poster][status] += 1 # Freshness is about the last DECISION, not the last payout. A poster # who cancelled four bounties last week is not unknown - it is very # well known, and recently. Keying on paidAt alone gave a never-pays # poster no timestamp at all, so it fell through to "unknown" and got # handed the base rate instead of its own 0/n. That rewarded the worst # posters on the board. # Two different signals, and fusing them was wrong: # last PAYOUT -> "is its payout rate still current?" # last DECISION -> "is it still active at all?" # A poster cancelling recently proves it is alive, NOT that it pays. if status == "paid": # burn_block_time is a Bitcoin block header; paidAt is a string # in someone's database. Prefer the header, fall back to the # string only when the chain cannot be read (and say so). when = (facts or {}).get("when") or b.get("paidAt") or "" if when > last_paid.get(poster, ""): last_paid[poster] = when dec = b.get("paidAt") or b.get("cancelledAt") or b.get("updatedAt") or "" if dec > last_decision.get(poster, ""): last_decision[poster] = dec # "Unknown" is not 0.5. A neutral prior that isn't the population rate is a # position dressed as neutrality, so an unknown/stale poster is scored at the # board's own base settlement rate (rambo, Nostr 2026-09-16). base_k = sum(h["paid"] for h in history.values()) base_n = sum(h["paid"] + h["cancelled"] + h["abandoned"] for h in history.values()) BASE = base_k / base_n if base_n else 0.5 def base_excluding(poster): """Leave-one-out base rate (rambo, Nostr 2026-09-16). A prior must not contain the entity it is pricing. On this board one poster is 51 of 79 decided listings - 65% of the population - so scoring it against the board base rate scored it mostly against itself: the base rate stops being a prior and becomes that entity wearing a trenchcoat. Excluding the poster gives the number that says something about everyone ELSE, which is what a prior is supposed to be.""" h = history.get(poster, {}) k = base_k - h.get("paid", 0) n = base_n - (h.get("paid", 0) + h.get("cancelled", 0) + h.get("abandoned", 0)) return (k / n, k, n) if n else (BASE, base_k, base_n) print(f"Board base settlement rate: {base_k}/{base_n} = {BASE:.0%}. Unknown/stale posters score the") print("LEAVE-ONE-OUT rate (board minus that poster), shown per row under base-x.\n") print(f"{'sats':>7} {'entries':>7} {'paid hist':>12} {'last dec':>10} {'base-x':>9} {'EV/entry':>9} {'left d':>6} {'ENTRADA':>13} bounty") for b in sorted(as_list(get(base + "open"), "bounties", "data"), key=lambda x: (entry_gate(x.get("description"))[1], -int(x.get("rewardSats") or 0))): reward = int(b.get("rewardSats") or 0) if reward < min_reward: continue instrument = reward_instrument(b.get("description")) poster = (b.get("posterBtcAddress") or "?").lower() h = history[poster] decided = h["paid"] + h["cancelled"] + h["abandoned"] pay_rate = h["paid"] / decided if decided else BASE hist = f"{h['paid']}/{decided}" if decided else "none" days = (now - _dt.datetime.fromisoformat(last_paid[poster].replace("Z", "+00:00"))).days if poster in last_paid else None # Same gate as the taskmarket board: a stale record is not a lower # probability, it is an expired measurement, so it reverts to unknown. # Three cases, kept apart on purpose: # 1. no decided history at all -> genuinely unknown -> base rate # 2. decided but never paid -> its rate IS 0, and there is no payout # whose currency we could question. Do NOT # gate it to the base rate: that would hand # the board's worst posters a free upgrade. # 3. has paid before -> gate on the last PAYOUT: a rate that has # not been reconfirmed in STALE_DAYS is an # expired measurement, not a lower one. loo, loo_k, loo_n = base_excluding(poster) used_loo = False if decided == 0: fresh = "unknown" pay_rate, used_loo = loo, True elif h["paid"] == 0: dd = (now - _dt.datetime.fromisoformat(last_decision[poster].replace("Z", "+00:00"))).days if poster in last_decision else None fresh = f"0-paid {dd}d" if dd is not None else "0-paid" pay_rate = 0.0 elif days is None or days > STALE_DAYS: fresh = f"!{days}d" if days is not None else "!never" pay_rate, used_loo = loo, True else: fresh = f"{days}d" entries = int(b.get("submissionCount") or 0) ev = reward * pay_rate / (entries + 1) left = (_dt.datetime.fromisoformat((b.get("expiresAt") or "").replace("Z", "+00:00")) - now).days if b.get("expiresAt") else 0 shown = f"{loo_k}/{loo_n}" if used_loo else "-" ev_col = f"{ev:>9.0f}" if not instrument else f"{'?':>9}" title = first_line(b.get("title"), 46) if instrument: title = f"[{instrument}] " + title[:38] gate, blocked = entry_gate(b.get("description")) if blocked: ev_col = f"{'[' + str(int(ev)) + ']':>9}" print(f"{reward:>7} {entries:>7} {hist:>12} {fresh:>10} {shown:>9} {ev_col} {left:>6} {gate:>13} {title}") print("\nENTRADA = coste de entrada detectado, o via de entrega que te excluye. Las filas CON barrera") print("van ABAJO y su EV sale entre corchetes: la rentabilidad por entrada no significa nada si no") print("puedes entrar. La columna existe porque la tabla puso 3 veces en primera fila una bounty") print("imposible para su operador. 'no vista' NO significa libre: significa que un detector de texto") print("no encontro nada, y se equivoca en los dos sentidos. Lee las bases antes de creer un 'no vista'.") print(f"\npaid hist = settled / decided bounties for that poster. last paid = days since its most recent") print(f"on-chain settlement; a '!' means older than STALE_DAYS ({STALE_DAYS}), so the payout rate is") print("discarded and the poster scored at the board base rate above, not carried forward at face value.") print(f"STALE_DAYS is {STALE_DAYS} and is still a GUESS: staleness_curve.py tried to fit it from this") print("board's own history and could not — 49 of 62 listings sit in the 0-7d bin, so every bin's") print("Wilson interval still overlaps the base rate. The threshold stays labelled as unfitted.") print("EV/entry = reward x payout rate / (entries + 1), in sats. Winner takes all on this board.") print("[SHARES] means the poster states the reward is NOT sBTC. rewardSats is then a ceiling, not a") print("promise -- one listing pays 5000 sats or ZERO depending on how a prediction market resolves.") print("EV is left blank on those rather than computed from a number that does not mean sats.") if use_chain: save_txcache(cache) c = checked print(f"\nProvenance: {c['tx']} payouts carry a txid; {c['confirmed']} confirmed on Stacks, " f"{c['sender_ok']} signed by the poster the board names, {c['amount_ok']} for exactly the " f"advertised sats, {c['memo_ok']} printing BNTY: on chain" + (f", {c['unreadable']} unreadable (fell back to paidAt)" if c['unreadable'] else "") + ".") print("Dates above come from burn_block_time (a Bitcoin block header), not from the board's paidAt.") print("Measured 2026-09-16: paidAt equals the STACKS block time to the second in 45/45, i.e. it is a") print("copy of the chain rather than an independent claim, and sits a median 10.3 min from the") print("Bitcoin burn time (-3.0 to +56.3; negatives are normal, Bitcoin timestamps are not monotonic).") print("So this moves no number today. It moves where the number comes from. Recheck it yourself with") print("aibtc_verify_payouts.mjs - one file, no dependencies, ~10 seconds.") print("Not checkable: the board does not publish which agent won, so the RECIPIENT address is still") print("taken on faith. Everything else on this line was recomputed from the chain.") else: print("\n--no-chain: dates came from the board's own paidAt field, unverified.") def usdc_balance_base(address): try: toks = get(f"https://base.blockscout.com/api/v2/addresses/{address}/token-balances") for b in toks: if (b.get("token") or {}).get("address_hash", "").lower() == USDC_BASE: return int(b["value"]) / 1e6 return 0.0 except Exception: # noqa: BLE001 return None def bountybook(): first = get("https://api.bountybook.ai/jobs?status=open") jobs = as_list(first, "jobs", "data") for page in range(2, int(first.get("totalPages") or 1) + 1): jobs += as_list(get(f"https://api.bountybook.ai/jobs?status=open&page={page}"), "jobs", "data") by_poster = collections.defaultdict(list) for j in jobs: by_poster[(j.get("poster_address") or "").lower()].append(j) print(f"{len(jobs)} open jobs from {len(by_poster)} posters") for poster, js in sorted(by_poster.items(), key=lambda kv: -len(kv[1])): total = sum(float(j.get("budget_usdc") or j.get("reward") or 0) for j in js) onchain = sum(1 for j in js if str(j.get("contract_job_id") or "0") not in ("0", "None", "")) bal = usdc_balance_base(poster) if poster else None if onchain: verdict = f"{onchain} escrowed on-chain" elif bal is not None and total > 0 and bal >= total: verdict = "poster balance covers rewards (not escrowed)" else: verdict = "LIKELY UNFUNDED (no escrow, balance below advertised)" print(f"- {poster or '(unknown)'}: {len(js)} jobs, advertised {total:.2f}, on-chain escrow ids {onchain}, poster USDC on Base {bal}, {verdict}") def main(argv=None): p = argparse.ArgumentParser(description="Check whether AI-agent bounty money is real.") p.add_argument("board", choices=["taskmarket", "bountybook", "aibtc"]) p.add_argument("--min-reward", type=float, default=0.0) p.add_argument("--no-chain", action="store_true", help="aibtc: trust the board's paidAt instead of recomputing dates from Stacks") p.add_argument("--cost-per-entry", type=float, default=0.0, help="estimated compute cost per entry in USD (taskmarket)") a = p.parse_args(argv) try: if a.board == "aibtc": aibtc(a.min_reward, a.cost_per_entry, use_chain=not a.no_chain) elif a.board == "taskmarket": taskmarket(a.min_reward, a.cost_per_entry) else: bountybook() except Exception as e: # noqa: BLE001 print(f"error: {e}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())