""" AFC Protocol Server — Unified API ================================== Integrates: 1. BlurHash64 — adjustable-fidelity glyph encodings 2. GlyphForge — recursive glyph production engine 3. OverLanguage 2.0 — meta-language parser/compiler 4. Layer4Meter — latent compute substrate capture 5. AFC Protocol — bonded claim market with oracle settlement Run: python3 afc_server.py """ import os import json import time import hashlib import base64 import uuid import sqlite3 from pathlib import Path from fastapi import FastAPI, HTTPException, Header, Body, UploadFile, File from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse from fastapi.middleware.cors import CORSMiddleware import uvicorn from blurhash64 import BlurHash64Encoder, Glyph from glyphforge import GlyphForge, MASTER_GLYPH from overlanguage import OverLanguageCompiler, Layer4Meter from afc_protocol import ( app as afc_app, create_claim, get_claim, list_claims, escrow_payment, reveal_answer, submit_hidden_tests, settle_claim, get_receipt, protocol_manifest, antonymify, sha256, merkle_commitment, blur_hash64 ) app = FastAPI(title="AFC Protocol — Unified Server", version="1.0") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # Mount AFC endpoints for route in afc_app.routes: if hasattr(route, 'path') and route.path not in ('/', '/health'): app.router.routes.append(route) encoder = BlurHash64Encoder() forge = GlyphForge(max_generations=5, min_score=15.0) compiler = OverLanguageCompiler() l4meter = Layer4Meter() @app.get("/health") async def health(): return { "status": "ok", "systems": ["blurhash64", "glyphforge", "overlanguage", "layer4meter", "afc_protocol"], "version": "1.0", } # --- BlurHash64 endpoints --- @app.post("/bh64/encode") async def bh64_encode(body: dict = Body(...)): """Encode a file at a specified fidelity level (0-9).""" content = body.get("content", "") filename = body.get("filename", "") fidelity = body.get("fidelity", 6) if not content: raise HTTPException(400, {"status": "error", "message": "content required"}) glyph = encoder.encode(content, filename, fidelity) return glyph.to_dict() @app.post("/bh64/ladder") async def bh64_ladder(body: dict = Body(...)): """Generate all 10 fidelity levels for a file.""" content = body.get("content", "") filename = body.get("filename", "") if not content: raise HTTPException(400, {"status": "error", "message": "content required"}) return {"ladder": encoder.ladder(content, filename), "filename": filename} @app.get("/bh64/levels") async def bh64_levels(): """Describe the 10 fidelity levels.""" return { "levels": [ {"level": 0, "name": "null", "discloses": "nothing", "recoverable": False, "executable": False}, {"level": 1, "name": "presence", "discloses": "file exists", "recoverable": False, "executable": False}, {"level": 2, "name": "type", "discloses": "file class", "recoverable": False, "executable": False}, {"level": 3, "name": "metadata", "discloses": "size, ext, timestamps", "recoverable": False, "executable": False}, {"level": 4, "name": "feature", "discloses": "imports, functions, deps", "recoverable": False, "executable": False}, {"level": 5, "name": "sketch", "discloses": "lossy summary, preview", "recoverable": False, "executable": False}, {"level": 6, "name": "receipt", "discloses": "hash, merkle, provenance, claims", "recoverable": False, "executable": False}, {"level": 7, "name": "partial_body", "discloses": "selected chunks", "recoverable": "partial", "executable": False}, {"level": 8, "name": "encrypted_body", "discloses": "full body (key-gated)", "recoverable": "full", "executable": False}, {"level": 9, "name": "full_transport", "discloses": "complete Base64 body", "recoverable": "full", "executable": True}, ] } # --- GlyphForge endpoints --- @app.post("/forge/run") async def forge_run(body: dict = Body(...)): """Run the glyph forge for N generations.""" generations = body.get("generations", 5) seed = body.get("seed", MASTER_GLYPH) glyphs = forge.forge(generations=generations) return { "master_glyph": MASTER_GLYPH, "total_glyphs": len(glyphs), "generations": max(g.generation for g in glyphs) if glyphs else 0, "top_10": forge.top_glyphs(10), "stream": forge.stream(20), } @app.get("/forge/top") async def forge_top(n: int = 10): return {"top_glyphs": forge.top_glyphs(n)} @app.get("/forge/stream") async def forge_stream(n: int = 20): return {"ticker": forge.stream(n)} # --- OverLanguage endpoints --- @app.post("/over/compile") async def over_compile(body: dict = Body(...)): """Compile an .over program into production artifacts.""" source = body.get("source", "") if not source: raise HTTPException(400, {"status": "error", "message": "source required"}) result = compiler.compile(source) return result.to_dict() @app.get("/over/grammar") async def over_grammar(): """Return the OverLanguage 2.0 grammar and glyph alphabet.""" from overlanguage import OverLanguageParser from glyphforge import ALPHABET, GRAMMAR return { "root_glyph": MASTER_GLYPH, "root_meaning": "stationary artifact at location → hash-bound → receipt-bound → transferable → verified → financeable", "alphabet": ALPHABET, "grammar": GRAMMAR, "layers": [ {"layer": 0, "name": "glyph", "description": "compressed symbolic substrate"}, {"layer": 1, "name": "intent", "description": "human-level objective"}, {"layer": 2, "name": "contract", "description": "enforceable requirements"}, {"layer": 3, "name": "agent", "description": "production operators"}, {"layer": 4, "name": "substrate", "description": "latent compute capture"}, {"layer": 5, "name": "receipt", "description": "proof binding"}, {"layer": 6, "name": "transfer", "description": "lambda friction / transferability"}, {"layer": 7, "name": "economic", "description": "buyer / value / price"}, ], "compiler_passes": ["parse", "expand", "contract", "assign", "execute", "capture", "hash", "receipt", "score", "package"], "primitives": ["overprogram", "glyph", "receipt", "lambda", "monetize"], } # --- Layer4Meter endpoints --- @app.post("/l4/sample") async def l4_sample(): """Capture a substrate sample.""" s = l4meter.sample() lci = l4meter.compute_lci(s) return {"sample": asdict_safe(s), "lci": lci} @app.post("/l4/baseline") async def l4_baseline(body: dict = Body(...)): """Set a baseline LCI (idle or human). Either provide lci directly or capture samples.""" mode = body.get("mode", "idle") lci = body.get("lci", None) if lci is not None: l4meter.set_baseline(mode, lci) return {"mode": mode, "baseline_lci": lci, "method": "manual"} else: samples = body.get("samples", 5) return l4meter.capture_baseline(mode, samples) @app.post("/l4/lift") async def l4_lift(body: dict = Body(...)): """Compute hidden compute lift = Agent LCI - Human Baseline - Idle Baseline.""" workload_lci = body.get("workload_lci", None) return l4meter.hidden_compute_lift(workload_lci) @app.post("/l4/business") async def l4_business(body: dict = Body(...)): """Compute business metrics from substrate data.""" return l4meter.business_metrics( artifact_value=body.get("artifact_value", 0), lci=body.get("lci", 0), useful_outputs=body.get("useful_outputs", 1), retries=body.get("retries", 0), total_events=body.get("total_events", 100), ) @app.post("/l4/rank") async def l4_rank(body: dict = Body(...)): """Rank workflows by value per LCI.""" workflows = body.get("workflows", []) if not workflows: workflows = [ {"name": "Workflow A", "lci": 220, "artifact_value": 2000}, {"name": "Workflow B", "lci": 80, "artifact_value": 10000}, ] return {"ranked": l4meter.rank_workflows(workflows)} @app.post("/l4/session") async def l4_session(body: dict = Body(...)): """Run a full L4 session: capture idle baseline, human baseline, agent workload, then compute lift + receipt.""" project = body.get("project", "unnamed") idle_samples = body.get("idle_samples", 3) human_samples = body.get("human_samples", 3) agent_samples = body.get("agent_samples", 5) idle_result = l4meter.capture_baseline("idle", idle_samples) human_result = l4meter.capture_baseline("human", human_samples) for _ in range(agent_samples): l4meter.sample(mode="agent") lift = l4meter.hidden_compute_lift() receipt = l4meter.receipt(project, time.time() - 3600) return { "project": project, "idle_baseline": idle_result, "human_baseline": human_result, "agent_samples": agent_samples, "hidden_compute_lift": lift, "business_metrics": l4meter.business_metrics(), "receipt": receipt, } @app.post("/l4/receipt") async def l4_receipt(body: dict = Body(...)): """Generate an L4 substrate receipt with 5-plane breakdown.""" project = body.get("project", "unnamed") session_start = body.get("session_start", time.time()) return l4meter.receipt(project, session_start) @app.get("/l4/planes") async def l4_planes(): """Describe the 5 capture planes.""" return { "planes": [ {"plane": 1, "name": "visual", "description": "Screen state changes, active app, windows visible", "production_api": "ScreenCaptureKit"}, {"plane": 2, "name": "file", "description": "File events, creations, modifications, deletions, git deltas", "production_api": "FSEvents"}, {"plane": 3, "name": "process", "description": "Process spawns, child processes, security events", "production_api": "Endpoint Security"}, {"plane": 4, "name": "power", "description": "CPU seconds, GPU activity, disk writes, network bytes, memory pressure", "production_api": "MetricKit + powermetrics"}, {"plane": 5, "name": "time_snapshot", "description": "Snapshot delta MB, temporal anchors", "production_api": "Time Machine local snapshots"}, ], "modes": ["idle", "human", "agent"], "formula": "LCI = α·CPU + β·GPU + γ·disk + δ·files + ε·procs + ζ·net + η·mem + θ·snap + ι·screen + κ·idle", "lift_formula": "Hidden Compute Lift = Agent LCI - Human Baseline - Idle Baseline", "business_metrics": ["cost_per_artifact", "proof_density", "agent_efficiency", "waste_ratio", "revenue_readiness", "value_per_lci"], "receipt_format": ".l4receipt/{manifest.json, events.sqlite, shards/*, hashes/merkle_root.txt, proofs/*}", "stages": [ {"stage": "V1", "name": "Capture", "description": "Screen checkpoints, file deltas, git diffs, command logs, disk growth, power samples"}, {"stage": "V2", "name": "Quantify", "description": "LCI score, baseline comparison, waste ratio, artifact yield, cost per artifact"}, {"stage": "V3", "name": "Prove", "description": "Merkle roots, signed receipts, selective disclosure, verifier CLI"}, {"stage": "V4", "name": "Quantum-sharded zkReceipt", "description": "Post-quantum signatures, sharded proofs, zero-knowledge claims, external notarization"}, ], } # --- Paper endpoint --- @app.get("/paper", response_class=PlainTextResponse) async def paper(): """Serve the research paper abstract.""" return """Antonymified File Receipts: LLM-Mediated Non-Consumable Disclosure for Hash-Bound, Oracle-Settled Digital Artifacts Abstract: Information goods are economically difficult because inspection can consume the good. A buyer can inspect a car without owning it, but inspecting an answer, trading signal, source file, dataset, or proprietary analysis may reveal the thing being sold. This creates a market failure: the seller cannot fully reveal the information before payment, while the buyer cannot confidently value it without some form of inspection. This paper proposes Antonymified File Receipts, a controlled-disclosure framework for representing digital files without directly revealing, copying, summarizing, or reconstructing their consumable content. The framework introduces antonymification as a semantic transformation in which a large language model produces a controlled opposite-representation of a file: enough to classify, route, price, verify, or settle claims about the file, but not enough to consume, execute, or reconstruct it. The LLM is not treated as the security layer. It acts as a semantic blur engine. Security and accountability are supplied by cryptographic hashes, receipts, leakage tests, oracle settlement, bonds, and controlled access windows. The resulting object, an Antonymified File Receipt, combines: - A non-consumable surrogate - A hash commitment to the sealed source - A leakage-risk score - A transferability coefficient (lambda friction) - A declared correctness oracle The system addresses Arrow's information paradox by replacing direct inspection with verifiable non-seeing. The buyer does not consume the answer before purchase. Instead, the buyer inspects a bundle of controlled signals: surrogate, hash commitment, proof hooks, seller bond, oracle, disclosure level, and settlement terms. The central claim: answers do not become sellable by being encoded. They become sellable when controlled disclosure is paired with settlement accountability. BlurHash64 solves the pre-sale visibility problem with a 10-level fidelity ladder. Antonymification turns file content into non-consumable market evidence. Bonds and oracles solve the truth problem. Together, they form a practical architecture for markets in answers, files, software artifacts, datasets, and AI-generated work. Protocol law: 1. No full disclosure before payment. 2. No payment without settlement. 3. No settlement without an oracle. 4. No oracle without a bond. Master glyph: ⧉◇@L → H@L Æ R Æ λ⁻¹ = ◎ → $ Live system: https://josephrw-afc-protocol.hf.space """ def asdict_safe(obj): from dataclasses import asdict return asdict(obj) # --- Unified landing page --- LANDING = """ AFC Protocol — Antonymified File Claim Protocol

AFC Protocol

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.

⧉◇@L → H@L Æ R Æ λ⁻¹ = ◎ → $
Protocol Law
1.No full disclosure before payment.
2.No payment without settlement.
3.No settlement without an oracle.
4.No oracle without a bond.
Five Systems
BlurHash64
Adjustable-fidelity glyph encodings
/bh64/encode /bh64/ladder
GlyphForge
Recursive glyph production engine
/forge/run /forge/top
OverLanguage 2.0
Meta-language for production reality
/over/compile /over/grammar
Layer4Meter
Latent compute substrate capture
/l4/sample /l4/receipt
AFC Protocol
Bonded claim market + oracle
/claim/create /claim/settle
BlurHash64 — Fidelity Ladder
GlyphForge — Recursive Production
OverLanguage 2.0 — Compile .over Program
Layer4Meter — Latent Compute Substrate
AFC Protocol — Bonded Claim Market
""" @app.get("/", response_class=HTMLResponse) async def landing(): return LANDING if __name__ == "__main__": port = int(os.environ.get("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)