""" AFC Protocol — Antonymified File Claim Protocol ================================================ Converts consumable information into accountable market objects. Protocol law: No full disclosure before payment. No payment without settlement. No settlement without an oracle. No oracle without a bond. Stack: File/answer → LLM antonymifier → hash/Merkle commitment → BlurHash64 fidelity label → lambda transferability score → oracle → bond → exclusivity window → settlement receipt MVP: Hidden Test Claim Market 1. Buyer posts task + hidden tests 2. Seller submits antonymified preview (not full answer) 3. Seller posts bond 4. Buyer escrows payment 5. Answer revealed only after commitment 6. Hidden test suite resolves pass/fail 7. Bond returned or slashed 8. Settlement receipt generated """ import os import json import time import hashlib import sqlite3 import base64 import uuid from pathlib import Path from fastapi import FastAPI, HTTPException, Header, Body, UploadFile, File from fastapi.responses import HTMLResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn app = FastAPI(title="AFC Protocol — Antonymified File Claim Protocol", version="1.0") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) DATA_DIR = Path(os.environ.get("AFC_DATA_DIR", "/data" if Path("/data").exists() else "/tmp/afc")) DATA_DIR.mkdir(parents=True, exist_ok=True) DB_PATH = DATA_DIR / "afc.db" def init_db(): conn = sqlite3.connect(str(DB_PATH)) conn.executescript(""" CREATE TABLE IF NOT EXISTS claims ( claim_id TEXT PRIMARY KEY, seller_id TEXT, buyer_id TEXT, task_description TEXT, antonymified_preview TEXT, full_answer_hash TEXT, full_answer_encrypted TEXT, bond_amount REAL DEFAULT 0, bond_posted INTEGER DEFAULT 0, payment_escrowed REAL DEFAULT 0, payment_escrowed_by TEXT, oracle_type TEXT, oracle_config TEXT, lambda_score REAL DEFAULT 0, fidelity_label TEXT, exclusivity_window_s INTEGER DEFAULT 0, exclusivity_expires REAL, status TEXT DEFAULT 'open', created_at REAL, committed_at REAL, revealed_at REAL, settled_at REAL, settlement_result TEXT, settlement_receipt TEXT ); CREATE TABLE IF NOT EXISTS hidden_tests ( test_id TEXT PRIMARY KEY, claim_id TEXT, test_hash TEXT, test_encrypted TEXT, created_at REAL ); CREATE TABLE IF NOT EXISTS oracle_results ( result_id TEXT PRIMARY KEY, claim_id TEXT, oracle_type TEXT, passed INTEGER, details TEXT, resolved_at REAL ); CREATE TABLE IF NOT EXISTS receipts ( receipt_id TEXT PRIMARY KEY, claim_id TEXT, type TEXT, payload TEXT, created_at REAL ); """) conn.commit() conn.close() init_db() def sha256(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def merkle_commitment(data: bytes, chunk_size: int = 4096) -> dict: chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)] if not chunks: chunks = [b""] leaf_hashes = [sha256(c) for c in chunks] tree = list(leaf_hashes) while len(tree) > 1: tree = [sha256((tree[i] + tree[i+1]).encode()) for i in range(0, len(tree)-1, 2)] if len(tree) % 2 == 0 else [sha256((tree[i] + tree[i+1]).encode()) for i in range(0, len(tree)-1, 2)] + [tree[-1]] return { "root": tree[0] if tree else sha256(b""), "leaf_count": len(leaf_hashes), "leaf_hashes": leaf_hashes[:8], } def blur_hash64(data: bytes) -> str: """BlurHash64-style fidelity label — non-reversible representation of content.""" h = hashlib.sha512(data).digest() return base64.b64encode(h[:48]).decode()[:64] def antonymify(content: str, filename: str = "") -> dict: """Generate non-consumable surrogate from file content. Exposes verifiable boundaries without revealing consumable content.""" data = content.encode() if isinstance(content, str) else content merkle = merkle_commitment(data) blur = blur_hash64(data) file_class = "unknown" if filename.endswith(".py") or "def " in content[:500]: file_class = "python_source" elif filename.endswith(".sql") or "SELECT" in content[:500].upper(): file_class = "sql_query" elif filename.endswith(".json"): file_class = "json_data" elif filename.endswith(".md"): file_class = "markdown_document" elif filename.endswith(".csv"): file_class = "csv_dataset" else: file_class = "text" lines = content.split("\n") if isinstance(content, str) else data.split(b"\n") line_count = len(lines) proof_hooks = [] if isinstance(content, str): for kw in ["def ", "class ", "SELECT", "CREATE TABLE", "import ", "function ", "return "]: if kw in content: proof_hooks.append(kw.strip()) excluded = ["full_source", "exact_algorithm", "alpha_signal", "raw_data_rows"] return { "file_class": file_class, "filename": filename, "size_bytes": len(data), "line_count": line_count, "merkle_root": merkle["root"], "merkle_leaves": merkle["leaf_count"], "blur_hash64": blur, "proof_hooks": proof_hooks, "excluded_content": excluded, "fidelity_label": "controlled_blur", "lambda_score": round(min(1.0, len(data) / 100000), 4), "preview": content[:200] + "..." if len(content) > 200 else content, } def encrypt_answer(answer: str, claim_id: str) -> str: """Simple XOR cipher with claim_id as key. Real impl would use proper crypto.""" key = (claim_id * 10).encode()[:32] data = answer.encode() return base64.b64encode(bytes(b ^ key[i % len(key)] for i, b in enumerate(data))).decode() def decrypt_answer(encrypted: str, claim_id: str) -> str: key = (claim_id * 10).encode()[:32] data = base64.b64decode(encrypted) return bytes(b ^ key[i % len(key)] for i, b in enumerate(data)).decode() # --- Endpoints --- @app.get("/health") async def health(): conn = sqlite3.connect(str(DB_PATH)) claim_count = conn.execute("SELECT COUNT(*) FROM claims").fetchone()[0] open_count = conn.execute("SELECT COUNT(*) FROM claims WHERE status='open'").fetchone()[0] settled_count = conn.execute("SELECT COUNT(*) FROM claims WHERE status='settled'").fetchone()[0] conn.close() return { "status": "ok", "protocol": "AFC/1.0", "claims_total": claim_count, "claims_open": open_count, "claims_settled": settled_count, "storage": str(DATA_DIR), "persistent": str(DATA_DIR).startswith("/data"), } @app.post("/claim/create") async def create_claim(body: dict = Body(...)): """Seller creates a claim with antonymified preview + encrypted full answer + bond.""" seller_id = body.get("seller_id", "") task_description = body.get("task_description", "") full_answer = body.get("full_answer", "") filename = body.get("filename", "") bond_amount = body.get("bond_amount", 0) oracle_type = body.get("oracle_type", "hidden_test") oracle_config = body.get("oracle_config", {}) exclusivity_window_s = body.get("exclusivity_window_s", 3600) if not full_answer: raise HTTPException(400, {"status": "error", "message": "full_answer required"}) if not seller_id: raise HTTPException(400, {"status": "error", "message": "seller_id required"}) claim_id = uuid.uuid4().hex[:12] surrogate = antonymify(full_answer, filename) answer_hash = sha256(full_answer.encode()) encrypted = encrypt_answer(full_answer, claim_id) conn = sqlite3.connect(str(DB_PATH)) conn.execute( """INSERT INTO claims (claim_id, seller_id, task_description, antonymified_preview, full_answer_hash, full_answer_encrypted, bond_amount, bond_posted, oracle_type, oracle_config, lambda_score, fidelity_label, exclusivity_window_s, exclusivity_expires, status, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (claim_id, seller_id, task_description, json.dumps(surrogate), answer_hash, encrypted, bond_amount, 1 if bond_amount > 0 else 0, oracle_type, json.dumps(oracle_config), surrogate["lambda_score"], surrogate["fidelity_label"], exclusivity_window_s, time.time() + exclusivity_window_s, "open", time.time()) ) conn.commit() conn.close() return { "claim_id": claim_id, "status": "open", "surrogate": surrogate, "answer_hash": answer_hash, "bond_posted": bond_amount > 0, "protocol": "AFC/1.0", "law": "No full disclosure before payment. No payment without settlement.", } @app.get("/claim/{claim_id}") async def get_claim(claim_id: str): """View a claim's antonymified surrogate — no full answer revealed.""" conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM claims WHERE claim_id = ?", [claim_id]).fetchone() conn.close() if not row: raise HTTPException(404, { "status": "claim_not_found", "claim_id": claim_id, "message": "This AFC claim is expired, settled, or does not exist.", "next": "Create a new claim." }) surrogate = json.loads(row["antonymified_preview"]) if row["antonymified_preview"] else {} return { "claim_id": claim_id, "seller_id": row["seller_id"], "task_description": row["task_description"], "surrogate": surrogate, "answer_hash": row["full_answer_hash"], "bond_amount": row["bond_amount"], "bond_posted": bool(row["bond_posted"]), "payment_escrowed": row["payment_escrowed"], "oracle_type": row["oracle_type"], "lambda_score": row["lambda_score"], "fidelity_label": row["fidelity_label"], "exclusivity_expires": row["exclusivity_expires"], "status": row["status"], "created_at": row["created_at"], "committed_at": row["committed_at"], "settled_at": row["settled_at"], "settlement_result": row["settlement_result"], "note": "Full answer is encrypted and hidden. This surrogate is non-consumable.", } @app.get("/claims") async def list_claims(status: str = "open"): conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row if status == "all": rows = conn.execute("SELECT claim_id, seller_id, task_description, status, lambda_score, bond_amount, created_at FROM claims ORDER BY created_at DESC").fetchall() else: rows = conn.execute("SELECT claim_id, seller_id, task_description, status, lambda_score, bond_amount, created_at FROM claims WHERE status = ? ORDER BY created_at DESC", [status]).fetchall() conn.close() return {"claims": [dict(r) for r in rows], "count": len(rows)} @app.post("/claim/{claim_id}/escrow") async def escrow_payment(claim_id: str, body: dict = Body(...)): """Buyer escrows payment — commits to purchase without seeing full answer.""" buyer_id = body.get("buyer_id", "") amount = body.get("amount", 0) if not buyer_id: raise HTTPException(400, {"status": "error", "message": "buyer_id required"}) conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM claims WHERE claim_id = ?", [claim_id]).fetchone() if not row: conn.close() raise HTTPException(404, {"status": "claim_not_found", "message": "Claim not found."}) if row["status"] != "open": conn.close() raise HTTPException(400, {"status": "error", "message": f"Claim is {row['status']}, not open."}) conn.execute( "UPDATE claims SET payment_escrowed = ?, payment_escrowed_by = ?, status = 'committed', committed_at = ? WHERE claim_id = ?", [amount, buyer_id, time.time(), claim_id] ) conn.commit() conn.close() return { "claim_id": claim_id, "status": "committed", "escrow_amount": amount, "buyer_id": buyer_id, "law": "No payment without settlement. No settlement without an oracle.", } @app.post("/claim/{claim_id}/reveal") async def reveal_answer(claim_id: str): """Reveal the full answer after payment escrow. Answer is decrypted.""" conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM claims WHERE claim_id = ?", [claim_id]).fetchone() if not row: conn.close() raise HTTPException(404, {"status": "claim_not_found", "message": "Claim not found."}) if row["status"] != "committed": conn.close() raise HTTPException(400, {"status": "error", "message": f"Claim must be committed (escrowed) first. Current: {row['status']}"}) answer = decrypt_answer(row["full_answer_encrypted"], claim_id) conn.execute("UPDATE claims SET status = 'revealed', revealed_at = ? WHERE claim_id = ?", [time.time(), claim_id]) conn.commit() conn.close() return { "claim_id": claim_id, "status": "revealed", "full_answer": answer, "answer_hash": row["full_answer_hash"], "verify": sha256(answer.encode()) == row["full_answer_hash"], "law": "Answer revealed. Oracle settlement pending.", } @app.post("/claim/{claim_id}/tests") async def submit_hidden_tests(claim_id: str, body: dict = Body(...)): """Buyer submits hidden tests for the claim (encrypted, hashed).""" tests = body.get("tests", []) if not tests: raise HTTPException(400, {"status": "error", "message": "tests array required"}) conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM claims WHERE claim_id = ?", [claim_id]).fetchone() if not row: conn.close() raise HTTPException(404, {"status": "claim_not_found", "message": "Claim not found."}) for i, test in enumerate(tests): test_id = uuid.uuid4().hex[:12] test_str = json.dumps(test) if not isinstance(test, str) else test test_hash = sha256(test_str.encode()) test_encrypted = encrypt_answer(test_str, claim_id) conn.execute( "INSERT INTO hidden_tests (test_id, claim_id, test_hash, test_encrypted, created_at) VALUES (?,?,?,?,?)", [test_id, claim_id, test_hash, test_encrypted, time.time()] ) conn.commit() conn.close() return { "claim_id": claim_id, "tests_submitted": len(tests), "status": "tests_ready", "law": "Oracle will resolve pass/fail on settlement.", } @app.post("/claim/{claim_id}/settle") async def settle_claim(claim_id: str, body: dict = Body(...)): """Oracle settles the claim — runs hidden tests against the revealed answer. Bond returned if pass, slashed if fail. Settlement receipt generated.""" conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT * FROM claims WHERE claim_id = ?", [claim_id]).fetchone() if not row: conn.close() raise HTTPException(404, {"status": "claim_not_found", "message": "Claim not found."}) if row["status"] not in ("revealed", "committed"): conn.close() raise HTTPException(400, {"status": "error", "message": f"Claim must be revealed first. Current: {row['status']}"}) answer = decrypt_answer(row["full_answer_encrypted"], claim_id) tests = conn.execute("SELECT * FROM hidden_tests WHERE claim_id = ?", [claim_id]).fetchall() oracle_type = row["oracle_type"] or "hidden_test" oracle_config = json.loads(row["oracle_config"]) if row["oracle_config"] else {} passed = 0 failed = 0 details = [] if oracle_type == "hidden_test" and tests: for t in tests: test_str = decrypt_answer(t["test_encrypted"], claim_id) test_obj = json.loads(test_str) test_type = test_obj.get("type", "contains") expected = test_obj.get("expected", "") if test_type == "contains": ok = expected in answer elif test_type == "equals": ok = answer.strip() == expected.strip() elif test_type == "regex": import re ok = bool(re.search(expected, answer)) elif test_type == "starts_with": ok = answer.strip().startswith(expected) elif test_type == "json_field": try: ans_json = json.loads(answer) ok = str(ans_json.get(test_obj.get("field", ""), "")) == str(expected) except: ok = False elif test_type == "python_exec": try: ns = {} exec(answer, ns) exec(test_obj.get("code", "result = True"), ns) ok = ns.get("result", False) except: ok = False else: ok = expected in answer if ok: passed += 1 details.append({"test_id": t["test_id"], "result": "pass"}) else: failed += 1 details.append({"test_id": t["test_id"], "result": "fail"}) elif oracle_type == "manual": passed = body.get("oracle_verdict", False) failed = 0 if passed else 1 else: passed = 1 failed = 0 details.append({"result": "no_tests", "note": "No hidden tests submitted. Default pass."}) all_passed = failed == 0 and passed > 0 settlement_result = "pass" if all_passed else "fail" bond_returned = row["bond_amount"] if all_passed else 0 bond_slashed = row["bond_amount"] if not all_passed else 0 payment_released = row["payment_escrowed"] if all_passed else 0 receipt_id = uuid.uuid4().hex[:12] receipt = { "receipt_id": receipt_id, "claim_id": claim_id, "seller_id": row["seller_id"], "buyer_id": row["payment_escrowed_by"], "result": settlement_result, "tests_passed": passed, "tests_failed": failed, "bond_amount": row["bond_amount"], "bond_returned": bond_returned, "bond_slashed": bond_slashed, "payment_escrowed": row["payment_escrowed"], "payment_released": payment_released, "answer_hash": row["full_answer_hash"], "oracle_type": oracle_type, "settled_at": time.time(), "protocol": "AFC/1.0", "law_verified": "No full disclosure before payment. No payment without settlement. No settlement without an oracle. No oracle without a bond.", } conn.execute( "UPDATE claims SET status = 'settled', settled_at = ?, settlement_result = ?, settlement_receipt = ? WHERE claim_id = ?", [time.time(), settlement_result, json.dumps(receipt), claim_id] ) conn.execute( "INSERT INTO oracle_results (result_id, claim_id, oracle_type, passed, details, resolved_at) VALUES (?,?,?,?,?,?)", [uuid.uuid4().hex[:12], claim_id, oracle_type, 1 if all_passed else 0, json.dumps(details), time.time()] ) conn.execute( "INSERT INTO receipts (receipt_id, claim_id, type, payload, created_at) VALUES (?,?,?,?,?)", [receipt_id, claim_id, "settlement", json.dumps(receipt), time.time()] ) conn.commit() conn.close() return receipt @app.get("/claim/{claim_id}/receipt") async def get_receipt(claim_id: str): """Get the settlement receipt for a claim.""" conn = sqlite3.connect(str(DB_PATH)) conn.row_factory = sqlite3.Row row = conn.execute("SELECT settlement_receipt FROM claims WHERE claim_id = ?", [claim_id]).fetchone() conn.close() if not row or not row["settlement_receipt"]: raise HTTPException(404, {"status": "no_receipt", "message": "No settlement receipt. Claim may not be settled yet."}) return json.loads(row["settlement_receipt"]) @app.get("/protocol") async def protocol_manifest(): """AFC Protocol manifest — describes the full stack.""" return { "protocol": "AFC/1.0", "name": "Antonymified File Claim Protocol", "thesis": "We do not sell answers. We sell bonded answer-claims whose value can be priced through controlled blur and settled through an oracle.", "law": [ "No full disclosure before payment.", "No payment without settlement.", "No settlement without an oracle.", "No oracle without a bond.", ], "stack": [ {"layer": 1, "name": "source_object", "description": "File or answer — the thing being claimed"}, {"layer": 2, "name": "antonymifier", "description": "LLM-generated non-consumable surrogate"}, {"layer": 3, "name": "merkle_commitment", "description": "Cryptographic identity and integrity"}, {"layer": 4, "name": "blur_hash64", "description": "Fidelity label — disclosure level"}, {"layer": 5, "name": "lambda_score", "description": "Transferability / usability friction"}, {"layer": 6, "name": "oracle", "description": "Truth-resolution mechanism"}, {"layer": 7, "name": "bond", "description": "Economic accountability"}, {"layer": 8, "name": "exclusivity_window", "description": "Urgency / first-mover value"}, {"layer": 9, "name": "settlement_receipt", "description": "Final proof of outcome"}, ], "market_unit": "bonded, partially disclosed, oracle-settled claim about an answer", "endpoints": { "create_claim": "POST /claim/create", "view_claim": "GET /claim/{id}", "list_claims": "GET /claims?status=open", "escrow_payment": "POST /claim/{id}/escrow", "reveal_answer": "POST /claim/{id}/reveal", "submit_tests": "POST /claim/{id}/tests", "settle": "POST /claim/{id}/settle", "receipt": "GET /claim/{id}/receipt", "protocol": "GET /protocol", "health": "GET /health", }, "oracle_types": ["hidden_test", "manual", "on_chain_event", "expert_arbitration"], "test_types": ["contains", "equals", "regex", "starts_with", "json_field", "python_exec"], } LANDING_HTML = """
Antonymified File Claim Protocol
We do not sell answers. We sell bonded answer-claims whose value can be priced through controlled blur and settled through an oracle.
Loading...