From c520a6e7a391f96d0065d9798c80fb2e5887e42f Mon Sep 17 00:00:00 2001
From: Muse <muse@axiom-code.com>
Date: Thu, 15 Oct 2026 17:21:51 -0600
Subject: [PATCH 1/2] Production API W1+W2: Ed25519 CA + durable store with
 hash-chained audit log

---
 services/api/NOTES.md             |  94 ++++++++++++
 services/api/billing.py           |  30 +++-
 services/api/ca.py                | 198 ++++++++++++++++++++++++
 services/api/mcp_server.py        | 105 ++++++++-----
 services/api/store.py             | 242 ++++++++++++++++++++++++++++++
 services/api/tests/conftest.py    |  16 ++
 services/api/tests/test_ca.py     |  78 ++++++++++
 services/api/tests/test_store.py  | 106 +++++++++++++
 services/api/tests/test_wiring.py | 108 +++++++++++++
 services/api/verify_api.py        | 119 +++++++++------
 10 files changed, 1003 insertions(+), 93 deletions(-)
 create mode 100644 services/api/NOTES.md
 create mode 100644 services/api/ca.py
 create mode 100644 services/api/store.py
 create mode 100644 services/api/tests/conftest.py
 create mode 100644 services/api/tests/test_ca.py
 create mode 100644 services/api/tests/test_store.py
 create mode 100644 services/api/tests/test_wiring.py

diff --git a/services/api/NOTES.md b/services/api/NOTES.md
new file mode 100644
index 0000000..98e1b9f
--- /dev/null
+++ b/services/api/NOTES.md
@@ -0,0 +1,94 @@
+# AxiomCode production build — W1 + W2
+
+Local working copy built from `api-prototype/` (state/secret files —
+keys.txt, meter.jsonl, revoked.txt — deliberately excluded). Nothing
+pushed anywhere; this tree is for review/checkpoint.
+
+## W1 — Ed25519 certification authority (`ca.py`, new)
+
+- `CA` class: Ed25519 keypair generation; `CA.load_or_create(path)`
+  persists the private key as PEM (PKCS8, no passphrase) with mode
+  0600 and reloads it afterwards. Key file path from
+  `AXIOMCODE_CA_KEY_FILE` (default: `ca_key.pem` beside the module).
+- Certificate format v2: the prototype fields — serial, issuer
+  "AxiomCode CA", valid_from / valid_until (90-day window),
+  code_sha256 (the artifact hash), verified (the verdict), toolchain,
+  attempts, key_id (the caller) — plus `version: 2`, `alg: "Ed25519"`,
+  `kid` (CA key id = first 16 hex of SHA-256 over the raw public key),
+  and `signature` (base64). Signature covers the canonical JSON of the
+  cert body — `json.dumps(..., sort_keys=True)`, the same
+  canonicalization certs.py used.
+- `verify_certificate(cert, public_key=...)` checks the signature with
+  a PUBLIC key only (accepts a key object, base64 raw key, or PEM —
+  i.e. exactly what the published payload carries) plus the validity
+  window; returns the prototype's `{valid, reason, payload}` shape.
+  Revocation is layered on top by the servers via the store, the same
+  split as X.509 vs CRL.
+- `public_keys_payload()` returns the document for
+  /.well-known/axiomcode-keys.json: issuer, keys[] with kid / alg /
+  use / public_key (base64) / public_key_pem, published_at.
+  verify_api serves it at `GET /.well-known/axiomcode-keys.json`.
+- verify_api.py and mcp_server.py issue + check via Ed25519 by
+  default. certs.py (HMAC) is unchanged and importable for legacy
+  fixtures; v2 verification rejects HMAC certs (no Ed25519 alg).
+
+## W2 — Durable store + hash-chained audit log (`store.py`, new)
+
+- sqlite3 (stdlib), one DB (`AXIOMCODE_DB`, default `axiomcode.db`
+  beside the module), WAL mode, lock-guarded single connection.
+- Tables: certificates (serial PK, cert_json, status, issued_at),
+  revocations (serial PK, reason, ts), meter_events (full billing
+  entry kept as JSON + indexed columns), audit_log (seq, ts,
+  event_type, payload_json, prev_hash, hash) with
+  hash = SHA-256(prev_hash + canonical payload_json); genesis
+  prev_hash is 64 zeros.
+- record_cert / revoke append their issuance / revocation audit
+  events by construction; the servers append "verification" events on
+  every /check and check_certificate call.
+- `verify_chain()` recomputes linkage + hashes and reports the first
+  broken seq.
+- billing.py: pricing math untouched (2.25× loaded-cost doctrine);
+  meter() / usage_for() now route through the store. A plain path
+  argument still uses the legacy JSONL code path for old fixtures.
+- revoked.txt and meter.jsonl are gone from both servers.
+- Temp Lean projects: created under `AXIOMCODE_WORK_ROOT` (default
+  system temp) and removed in a `finally` block in both servers.
+
+## Tests — 14 passed, 0 failed (`~/ax-venv/bin/python -m pytest tests/ -q`)
+
+- test_ca.py (6): key persistence + 0600 perms; issue→verify
+  round-trip (v2 fields, 90-day window); tampered verdict and mangled
+  signature fail; wrong key fails (bad_signature); published payload
+  verifies a real cert via both base64 and PEM; legacy HMAC cert
+  rejected.
+- test_store.py (5): record/get round-trip; revocation persists
+  across store close/reopen; audit chain verifies (issuance +
+  verification + revocation); one flipped byte in a stored payload is
+  detected with broken_seq = 2; meter events recorded and aggregated
+  (billing routes to the same numbers).
+- test_wiring.py (3): REST end-to-end with a stubbed prover —
+  /verify issues an Ed25519 cert, persists it, serves keys payload
+  that verifies it, /check valid → revoked after operator revocation,
+  /revoked lists it, temp dir cleaned, audit chain intact; MCP tools
+  (verify / check_certificate / usage_report) run on the same CA +
+  store; legacy HMAC module still works standalone.
+- The Lean prover is stubbed in wiring tests on purpose: they prove
+  CA/store wiring, not Lean. Real-Lean behaviour was proven by the
+  prototype demo (2026-09-26) and the engine suite.
+
+## Left stubbed / out of scope (later workstreams)
+
+- Revocation is operator-side only (store.revoke) — no public revoke
+  endpoint, same as the prototype; an authenticated admin path is
+  still to be designed.
+- API-key enforcement on REST (prototype had none; MCP bearer auth
+  unchanged, including its known mutable-identity issue — W3 scope).
+- Trial limits (10 cases), certificate withholding until payment,
+  Stripe entitlement/webhooks, quotas, rate limits, idempotency —
+  not in W1/W2 scope.
+- restart_rest.sh / restart_mcp.sh were not copied from the prototype
+  (they point at prototype paths/ports).
+- Ruff: clean except deliberate fail-closed broad `except` catches
+  (BLE001 in ca/certs/billing legacy path, S112 in the legacy JSONL
+  reader) and import-order notes where sys.path setup must precede
+  imports — all inherited patterns, left as-is intentionally.
diff --git a/services/api/billing.py b/services/api/billing.py
index 29c7ec8..4f7900a 100644
--- a/services/api/billing.py
+++ b/services/api/billing.py
@@ -8,6 +8,11 @@
 
 1 credit = $0.01. Tune via AXIOMCODE_PRICING_JSON or the defaults below.
 Currently: 2.25x multiple over fully-loaded cost (~55.6% gross margin).
+
+Production (W2): meter records go through the durable store
+(store.record_meter — sqlite + hash-chained audit trail), not the
+prototype's meter.jsonl. meter()/usage_for() accept a Store; a plain
+file path still works via the legacy JSONL path for old fixtures.
 """
 
 import json
@@ -44,7 +49,7 @@ def load_pricing() -> dict:
 def price_per_compute_s(pricing: dict | None = None) -> float:
     p = pricing or load_pricing()
     loaded = sum(v for k, v in p.items() if k != "multiple")
-    return float(loaded * p["multiple"])
+    return loaded * p["multiple"]
 
 
 def charge_for(compute_s: float, pricing: dict | None = None) -> dict:
@@ -72,10 +77,13 @@ def load_keys(keys_file: str) -> dict:
     return keys
 
 
-def meter(
-    meter_log: str, key_id: str, code_sha: str, verified: bool, compute_s: float, pricing: dict | None = None
-) -> dict:
-    """Append one metered call to the billing/audit log. Returns the entry."""
+def _is_store(target) -> bool:
+    return hasattr(target, "record_meter") and hasattr(target, "usage_for")
+
+
+def meter(target, key_id: str, code_sha: str, verified: bool, compute_s: float, pricing: dict | None = None) -> dict:
+    """Record one metered call. `target` is the durable Store
+    (production) or a legacy JSONL path. Returns the entry."""
     bill = charge_for(compute_s, pricing)
     entry = {
         "ts": int(time.time()),
@@ -84,14 +92,20 @@ def meter(
         "verified": verified,
         **bill,
     }
-    with open(meter_log, "a") as f:
+    if _is_store(target):
+        return target.record_meter(entry)
+    with open(target, "a") as f:
         f.write(json.dumps(entry) + "\n")
     return entry
 
 
-def usage_for(meter_log: str, key_id: str) -> dict:
+def usage_for(target, key_id: str) -> dict:
+    """Total metered spend for a key — from the Store (production)
+    or a legacy JSONL path."""
+    if _is_store(target):
+        return target.usage_for(key_id)
     total_c, total_usd, calls = 0.0, 0.0, 0
-    p = Path(meter_log)
+    p = Path(target)
     if p.exists():
         for line in p.read_text().splitlines():
             try:
diff --git a/services/api/ca.py b/services/api/ca.py
new file mode 100644
index 0000000..ac39c57
--- /dev/null
+++ b/services/api/ca.py
@@ -0,0 +1,198 @@
+"""AxiomCode CA — Ed25519 certification authority (production, W1).
+
+Replaces the prototype's HMAC-SHA256 signing (certs.py, kept for legacy
+fixtures) with asymmetric Ed25519 signatures: certificates can be
+verified by ANYONE holding only the public key, which is published at
+/.well-known/axiomcode-keys.json. The private key never leaves the CA
+key file (PEM, mode 0600).
+
+Certificate format v2: the prototype fields (serial, issuer,
+valid_from / valid_until — a 90-day window, code_sha256 [artifact hash],
+verified [verdict], toolchain, attempts, key_id [caller]) plus:
+  "version": 2, "alg": "Ed25519", "kid": <CA key id>,
+  "signature": base64 Ed25519 signature over the canonical JSON of the
+  certificate body (every field except "signature", json.dumps with
+  sort_keys=True — the same canonicalization as certs.py).
+"""
+
+import base64
+import hashlib
+import json
+import os
+import time
+import uuid
+from pathlib import Path
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+
+_HERE = Path(__file__).parent
+ISSUER = "AxiomCode CA"
+ALG = "Ed25519"
+VALIDITY_DAYS = int(os.environ.get("AXIOMCODE_CERT_VALIDITY_DAYS", "90"))
+DEFAULT_KEY_FILE = os.environ.get("AXIOMCODE_CA_KEY_FILE", str(_HERE / "ca_key.pem"))
+
+
+def canonical(body: dict) -> bytes:
+    """Canonical bytes of a certificate body — same approach as certs.py."""
+    return json.dumps(body, sort_keys=True).encode()
+
+
+def kid_for(public_key: Ed25519PublicKey) -> str:
+    """Stable key id: first 16 hex of SHA-256 over the raw public key."""
+    raw = public_key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    return hashlib.sha256(raw).hexdigest()[:16]
+
+
+def _pem_private(key: Ed25519PrivateKey) -> bytes:
+    return key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+
+
+def _pem_public(key: Ed25519PublicKey) -> str:
+    return key.public_bytes(
+        serialization.Encoding.PEM,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    ).decode()
+
+
+def coerce_public_key(key) -> Ed25519PublicKey:
+    """Accept an Ed25519PublicKey, a base64 raw-key string, raw bytes,
+    or a PEM string — so a key taken straight out of the published
+    /.well-known/axiomcode-keys.json payload can verify certificates."""
+    if isinstance(key, Ed25519PublicKey):
+        return key
+    if isinstance(key, str) and "BEGIN PUBLIC KEY" in key:
+        return serialization.load_pem_public_key(key.encode())
+    raw = base64.b64decode(key) if isinstance(key, str) else bytes(key)
+    return Ed25519PublicKey.from_public_bytes(raw)
+
+
+class CA:
+    """An Ed25519 certification authority bound to one keypair."""
+
+    def __init__(self, private_key: Ed25519PrivateKey):
+        self._private = private_key
+        self.public_key = private_key.public_key()
+        self.kid = kid_for(self.public_key)
+
+    # ─── key management ──────────────────────────────────────────────
+    @classmethod
+    def generate(cls) -> "CA":
+        return cls(Ed25519PrivateKey.generate())
+
+    @classmethod
+    def load_or_create(cls, key_file: str | Path | None = None) -> "CA":
+        """Load the CA key from `key_file`, generating and persisting a
+        new one (PEM, mode 0600) if it does not exist yet."""
+        path = Path(key_file or DEFAULT_KEY_FILE)
+        if path.exists():
+            key = serialization.load_pem_private_key(path.read_bytes(), password=None)
+            if not isinstance(key, Ed25519PrivateKey):
+                raise ValueError(f"{path} does not hold an Ed25519 key")
+            return cls(key)
+        ca = cls.generate()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        try:
+            os.write(fd, _pem_private(ca._private))
+        finally:
+            os.close(fd)
+        os.chmod(path, 0o600)
+        return ca
+
+    # ─── issuance ────────────────────────────────────────────────────
+    def sign_body(self, body: dict) -> str:
+        sig = self._private.sign(canonical(body))
+        return base64.b64encode(sig).decode()
+
+    def issue(self, subject: dict) -> dict:
+        """Issue a v2 certificate. `subject` holds the verdict fields
+        (code_sha256, verified, attempts, toolchain, key_id, ...)."""
+        now = int(time.time())
+        cert = {
+            "version": 2,
+            "alg": ALG,
+            "kid": self.kid,
+            "serial": uuid.uuid4().hex[:16],
+            "issuer": ISSUER,
+            "valid_from": now,
+            "valid_until": now + VALIDITY_DAYS * 86400,
+            **subject,
+        }
+        cert["signature"] = self.sign_body(cert)
+        return cert
+
+    # ─── publication ─────────────────────────────────────────────────
+    def public_keys_payload(self) -> dict:
+        """The document served at /.well-known/axiomcode-keys.json."""
+        raw = self.public_key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+        return {
+            "issuer": ISSUER,
+            "keys": [
+                {
+                    "kid": self.kid,
+                    "alg": ALG,
+                    "use": "cert-signing",
+                    "public_key": base64.b64encode(raw).decode(),
+                    "public_key_pem": _pem_public(self.public_key),
+                }
+            ],
+            "published_at": int(time.time()),
+        }
+
+
+def verify_certificate(cert: dict, public_key=None, ca: CA | None = None) -> dict:
+    """Validate a v2 certificate's signature and validity window using
+    ONLY a public key (the CA's own by default, or one supplied — e.g.
+    taken from the published keys payload). Revocation is layered on
+    top by the caller via store.is_revoked — a public key alone cannot
+    know the revocation list, same split as X.509 vs CRL.
+
+    Returns the certs.check shape: {valid, reason?, payload?}."""
+    if not isinstance(cert, dict) or cert.get("alg") != ALG:
+        return {"valid": False, "reason": "bad_signature"}
+    sig_b64 = cert.get("signature")
+    if not sig_b64:
+        return {"valid": False, "reason": "bad_signature"}
+    pub = coerce_public_key(public_key if public_key is not None else (ca or default_ca()).public_key)
+    body = {k: v for k, v in cert.items() if k != "signature"}
+    try:
+        pub.verify(base64.b64decode(sig_b64), canonical(body))
+    except Exception:  # InvalidSignature, malformed base64, wrong key type
+        return {"valid": False, "reason": "bad_signature"}
+    now = int(time.time())
+    if now < cert.get("valid_from", 0):
+        return {"valid": False, "reason": "not_yet_valid"}
+    if now > cert.get("valid_until", 0):
+        return {"valid": False, "reason": "expired", "payload": cert, "renew": True}
+    return {"valid": True, "payload": cert}
+
+
+_DEFAULT_CA: CA | None = None
+
+
+def default_ca() -> CA:
+    """Process-wide CA, loaded/created from AXIOMCODE_CA_KEY_FILE."""
+    global _DEFAULT_CA
+    if _DEFAULT_CA is None:
+        _DEFAULT_CA = CA.load_or_create()
+    return _DEFAULT_CA
+
+
+def encode(cert: dict) -> str:
+    return json.dumps(cert, sort_keys=True)
+
+
+def decode(raw: str) -> dict | None:
+    try:
+        c = json.loads(raw)
+        return c if isinstance(c, dict) else None
+    except Exception:
+        return None
diff --git a/services/api/mcp_server.py b/services/api/mcp_server.py
index 96b4220..cfc77fe 100644
--- a/services/api/mcp_server.py
+++ b/services/api/mcp_server.py
@@ -1,9 +1,15 @@
 """AxiomCode verification MCP server — the agent-native interface.
 
-Tool: verify(code, max_attempts) -> {verified, attempts, certificate, usage}
-Auth: Authorization: Bearer <api_key> (keys in API_KEYS_FILE, one per line: <key_id>:<key>)
-Metering: every call appended to METER_LOG (JSONL) — the audit log doubles as
-the billing record. Priced in compute-seconds; see CREDITS_PER_COMPUTE_S.
+Tools: verify(code, max_attempts), check_certificate(certificate),
+pricing(), usage_report().
+Auth: Authorization: Bearer <api_key> (keys in API_KEYS_FILE, one per
+line: <key_id>:<key>).
+
+Production build: certificates are Ed25519-signed (ca.py, format v2);
+metering, revocation and the hash-chained audit log live in the
+durable store (store.py) — no more meter.jsonl / revoked.txt.
+Temp Lean projects are created under AXIOMCODE_WORK_ROOT and always
+removed after the verification finishes.
 
 Run: uvicorn mcp_server:mcp_app --port 8092
 (Uses Streamable HTTP transport so remote agents can connect.)
@@ -11,6 +17,7 @@
 
 import hashlib
 import os
+import shutil
 import sys
 import tempfile
 import time
@@ -19,22 +26,23 @@
 sys.path.insert(0, "/home/hatch/workspace/axiomcode/repo")
 sys.path.insert(0, os.path.dirname(__file__))
 
+import ca
 from billing import load_keys, meter, price_per_compute_s, usage_for
-from certs import issue as issue_cert
-
-# mcp v1 API; typeshed's mcp.server.fastmcp stub does not declare FastMCP
-from mcp.server.fastmcp import FastMCP  # type: ignore[attr-defined]
+from core.prover import iterative_proof_search
+from mcp.server.fastmcp import FastMCP
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.responses import JSONResponse
-
-from core.prover import iterative_proof_search
+from store import Store
 
 # ─── config ──────────────────────────────────────────────────────────────
-API_KEYS_FILE = os.environ.get("AXIOMCODE_API_KEYS", "/home/hatch/workspace/axiomcode/api-prototype/keys.txt")
-METER_LOG = os.environ.get("AXIOMCODE_METER_LOG", "/home/hatch/workspace/axiomcode/api-prototype/meter.jsonl")
-CERT_SECRET = os.environ.get("AXIOMCODE_CERT_SECRET", "dev-secret-change-me").encode()
+_HERE = Path(__file__).parent
+API_KEYS_FILE = os.environ.get("AXIOMCODE_API_KEYS", str(_HERE / "keys.txt"))
+WORK_ROOT = os.environ.get("AXIOMCODE_WORK_ROOT", tempfile.gettempdir())
 TOOLCHAIN = "leanprover/lean4:v4.35.0-rc3"
 
+CA = ca.CA.load_or_create()
+STORE = Store()
+
 mcp = FastMCP("axiomcode-verify")
 
 
@@ -63,32 +71,35 @@ def verify(code: str, max_attempts: int = 1) -> dict:
     Anyone can re-check the certificate without trusting us:
     verification-as-evidence, not verification-as-opinion."""
     key_id = _KEY.id
-    proj = Path(tempfile.mkdtemp(prefix="ax-mcp-"))
-    (proj / "lean-toolchain").write_text(TOOLCHAIN)
-    (proj / "lakefile.lean").write_text(
-        "import Lake\nopen Lake DSL\npackage v where\nlean_lib V where\n  roots := #[`V]\n"
-    )
-    src = proj / "V.lean"
-
-    def writer(c: str) -> Path:
-        src.write_text(c)
-        return src
-
-    t0 = time.time()
-    verified, _, attempts, _ = iterative_proof_search(
-        writer,
-        proj,
-        code,
-        None,
-        "none",
-        max_attempts=max_attempts,
-        use_pantograph=False,
-    )
-    compute_s = time.time() - t0
+    proj = Path(tempfile.mkdtemp(prefix="ax-mcp-", dir=WORK_ROOT))
+    try:
+        (proj / "lean-toolchain").write_text(TOOLCHAIN)
+        (proj / "lakefile.lean").write_text(
+            "import Lake\nopen Lake DSL\npackage v where\nlean_lib V where\n  roots := #[`V]\n"
+        )
+        src = proj / "V.lean"
+
+        def writer(c: str) -> Path:
+            src.write_text(c)
+            return src
+
+        t0 = time.time()
+        verified, _, attempts, _ = iterative_proof_search(
+            writer,
+            proj,
+            code,
+            None,
+            "none",
+            max_attempts=max_attempts,
+            use_pantograph=False,
+        )
+        compute_s = time.time() - t0
+    finally:
+        shutil.rmtree(proj, ignore_errors=True)
 
     code_sha = hashlib.sha256(code.encode()).hexdigest()
-    usage = meter(METER_LOG, key_id, code_sha, verified, compute_s)
-    cert = issue_cert(
+    usage = meter(STORE, key_id, code_sha, verified, compute_s)
+    cert = CA.issue(
         {
             "code_sha256": code_sha,
             "verified": verified,
@@ -97,6 +108,7 @@ def writer(c: str) -> Path:
             "key_id": key_id,
         }
     )
+    STORE.record_cert(cert)
     return {
         "verified": verified,
         "attempts": attempts,
@@ -108,16 +120,27 @@ def writer(c: str) -> Path:
 @mcp.tool()
 def usage_report() -> dict:
     """Total metered spend for the calling API key, in credits and USD."""
-    return usage_for(METER_LOG, _KEY.id)
+    return usage_for(STORE, _KEY.id)
 
 
 @mcp.tool()
 def check_certificate(certificate: dict) -> dict:
     """Re-verify a verification certificate: signature, validity window,
     revocation status. Anyone can check — never take our word for it."""
-    from certs import check as check_cert
-
-    return check_cert(certificate)
+    result = ca.verify_certificate(certificate, ca=CA)
+    serial = certificate.get("serial") if isinstance(certificate, dict) else None
+    if result.get("valid") and serial and STORE.is_revoked(serial):
+        result = {"valid": False, "reason": "revoked", "payload": certificate}
+    STORE.append_audit(
+        "verification",
+        {
+            "serial": serial,
+            "valid": bool(result.get("valid")),
+            "reason": result.get("reason"),
+            "via": "mcp",
+        },
+    )
+    return result
 
 
 @mcp.tool()
diff --git a/services/api/store.py b/services/api/store.py
new file mode 100644
index 0000000..714ba16
--- /dev/null
+++ b/services/api/store.py
@@ -0,0 +1,242 @@
+"""AxiomCode durable store (production, W2) — sqlite3, stdlib only.
+
+Replaces the prototype's revoked.txt + meter.jsonl flat files with one
+durable database:
+
+  certificates  (serial PK, cert_json, status, issued_at)
+  revocations   (serial PK, reason, ts)
+  meter_events  (id, ts, key_id, code_sha256, verified, credits, usd,
+                 entry_json) — the billing record
+  audit_log     (seq, ts, event_type, payload_json, prev_hash, hash)
+
+The audit log is hash-chained: every row's hash is
+SHA-256(prev_hash + canonical payload_json), where the payload is
+stored as its canonical JSON string (json.dumps, sort_keys=True), so
+any later edit to a stored payload — even one byte — breaks the chain
+at that seq, and verify_chain() reports the first broken seq.
+
+Issuance and revocation append their audit events by construction
+(record_cert / revoke do it themselves); verification events are
+appended by the API layer via append_audit().
+"""
+
+import hashlib
+import json
+import os
+import sqlite3
+import threading
+import time
+from pathlib import Path
+
+_HERE = Path(__file__).parent
+DEFAULT_DB = os.environ.get("AXIOMCODE_DB", str(_HERE / "axiomcode.db"))
+GENESIS_HASH = "0" * 64
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS certificates (
+    serial TEXT PRIMARY KEY,
+    cert_json TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT 'issued',
+    issued_at INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS revocations (
+    serial TEXT PRIMARY KEY,
+    reason TEXT NOT NULL DEFAULT '',
+    ts INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS meter_events (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    ts INTEGER NOT NULL,
+    key_id TEXT NOT NULL,
+    code_sha256 TEXT NOT NULL,
+    verified INTEGER NOT NULL,
+    credits REAL NOT NULL DEFAULT 0,
+    usd REAL NOT NULL DEFAULT 0,
+    entry_json TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS audit_log (
+    seq INTEGER PRIMARY KEY AUTOINCREMENT,
+    ts INTEGER NOT NULL,
+    event_type TEXT NOT NULL,
+    payload_json TEXT NOT NULL,
+    prev_hash TEXT NOT NULL,
+    hash TEXT NOT NULL
+);
+"""
+
+
+def _canonical(payload: dict) -> str:
+    return json.dumps(payload, sort_keys=True)
+
+
+def _chain_hash(prev_hash: str, payload_json: str) -> str:
+    return hashlib.sha256((prev_hash + payload_json).encode()).hexdigest()
+
+
+class Store:
+    """One durable AxiomCode store. Safe for the threadpool-style
+    concurrency of the API servers (single connection + lock, WAL)."""
+
+    def __init__(self, db_path: str | Path | None = None):
+        self.path = Path(db_path or DEFAULT_DB)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+        self._db = sqlite3.connect(self.path, check_same_thread=False)
+        self._db.row_factory = sqlite3.Row
+        self._db.execute("PRAGMA journal_mode=WAL")
+        self._db.executescript(_SCHEMA)
+        self._db.commit()
+
+    def close(self) -> None:
+        with self._lock:
+            self._db.close()
+
+    # ─── audit log ───────────────────────────────────────────────────
+    def append_audit(self, event_type: str, payload: dict) -> dict:
+        """Append one hash-chained audit event. Returns the stored row."""
+        payload_json = _canonical(payload)
+        ts = int(time.time())
+        with self._lock:
+            row = self._db.execute("SELECT hash FROM audit_log ORDER BY seq DESC LIMIT 1").fetchone()
+            prev_hash = row["hash"] if row else GENESIS_HASH
+            h = _chain_hash(prev_hash, payload_json)
+            cur = self._db.execute(
+                "INSERT INTO audit_log (ts, event_type, payload_json, prev_hash, hash) VALUES (?, ?, ?, ?, ?)",
+                (ts, event_type, payload_json, prev_hash, h),
+            )
+            self._db.commit()
+            return {
+                "seq": cur.lastrowid,
+                "ts": ts,
+                "event_type": event_type,
+                "payload": payload,
+                "prev_hash": prev_hash,
+                "hash": h,
+            }
+
+    def verify_chain(self) -> dict:
+        """Recompute the whole chain. {"ok": True, "events": n} or
+        {"ok": False, "broken_seq": <first seq that fails>}."""
+        with self._lock:
+            rows = self._db.execute("SELECT seq, payload_json, prev_hash, hash FROM audit_log ORDER BY seq").fetchall()
+        prev = GENESIS_HASH
+        for r in rows:
+            if r["prev_hash"] != prev:
+                return {"ok": False, "broken_seq": r["seq"]}
+            if _chain_hash(r["prev_hash"], r["payload_json"]) != r["hash"]:
+                return {"ok": False, "broken_seq": r["seq"]}
+            prev = r["hash"]
+        return {"ok": True, "events": len(rows)}
+
+    def audit_events(self, event_type: str | None = None) -> list[dict]:
+        with self._lock:
+            if event_type:
+                rows = self._db.execute(
+                    "SELECT * FROM audit_log WHERE event_type = ? ORDER BY seq", (event_type,)
+                ).fetchall()
+            else:
+                rows = self._db.execute("SELECT * FROM audit_log ORDER BY seq").fetchall()
+        return [{**dict(r), "payload": json.loads(r["payload_json"])} for r in rows]
+
+    # ─── certificates & revocation ───────────────────────────────────
+    def record_cert(self, cert: dict) -> dict:
+        """Persist an issued certificate + its issuance audit event."""
+        with self._lock:
+            self._db.execute(
+                "INSERT OR REPLACE INTO certificates (serial, cert_json, status, issued_at) VALUES (?, ?, ?, ?)",
+                (cert["serial"], _canonical(cert), "issued", int(cert.get("valid_from", time.time()))),
+            )
+            self._db.commit()
+        self.append_audit(
+            "issuance",
+            {
+                "serial": cert["serial"],
+                "kid": cert.get("kid"),
+                "code_sha256": cert.get("code_sha256"),
+                "verified": cert.get("verified"),
+            },
+        )
+        return cert
+
+    def get_cert(self, serial: str) -> dict | None:
+        with self._lock:
+            row = self._db.execute("SELECT cert_json FROM certificates WHERE serial = ?", (serial,)).fetchone()
+        return json.loads(row["cert_json"]) if row else None
+
+    def revoke(self, serial: str, reason: str = "") -> None:
+        ts = int(time.time())
+        with self._lock:
+            self._db.execute(
+                "INSERT OR REPLACE INTO revocations (serial, reason, ts) VALUES (?, ?, ?)", (serial, reason, ts)
+            )
+            self._db.execute("UPDATE certificates SET status = 'revoked' WHERE serial = ?", (serial,))
+            self._db.commit()
+        self.append_audit("revocation", {"serial": serial, "reason": reason})
+
+    def is_revoked(self, serial: str) -> bool:
+        with self._lock:
+            row = self._db.execute("SELECT 1 FROM revocations WHERE serial = ?", (serial,)).fetchone()
+        return row is not None
+
+    def revoked_serials(self) -> set:
+        with self._lock:
+            rows = self._db.execute("SELECT serial FROM revocations").fetchall()
+        return {r["serial"] for r in rows}
+
+    # ─── metering ────────────────────────────────────────────────────
+    def record_meter(self, entry: dict) -> dict:
+        """Persist one metered call (the full billing entry is kept)."""
+        with self._lock:
+            self._db.execute(
+                "INSERT INTO meter_events (ts, key_id, code_sha256,"
+                " verified, credits, usd, entry_json)"
+                " VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (
+                    int(entry.get("ts", time.time())),
+                    entry.get("key_id", ""),
+                    entry.get("code_sha256", ""),
+                    1 if entry.get("verified") else 0,
+                    float(entry.get("credits", 0)),
+                    float(entry.get("usd", 0)),
+                    _canonical(entry),
+                ),
+            )
+            self._db.commit()
+        return entry
+
+    def usage_for(self, key_id: str) -> dict:
+        """Aggregate metered spend — same shape as billing.usage_for."""
+        with self._lock:
+            row = self._db.execute(
+                "SELECT COUNT(*) AS calls, COALESCE(SUM(credits), 0) AS c,"
+                " COALESCE(SUM(usd), 0) AS u FROM meter_events"
+                " WHERE key_id = ?",
+                (key_id,),
+            ).fetchone()
+        return {
+            "key_id": key_id,
+            "calls": row["calls"],
+            "credits_used": round(row["c"], 4),
+            "usd_used": round(row["u"], 4),
+        }
+
+    def meter_events(self, key_id: str | None = None) -> list[dict]:
+        with self._lock:
+            if key_id:
+                rows = self._db.execute(
+                    "SELECT entry_json FROM meter_events WHERE key_id = ? ORDER BY id", (key_id,)
+                ).fetchall()
+            else:
+                rows = self._db.execute("SELECT entry_json FROM meter_events ORDER BY id").fetchall()
+        return [json.loads(r["entry_json"]) for r in rows]
+
+
+_DEFAULT_STORE: Store | None = None
+
+
+def default_store() -> Store:
+    """Process-wide store, opened at AXIOMCODE_DB."""
+    global _DEFAULT_STORE
+    if _DEFAULT_STORE is None:
+        _DEFAULT_STORE = Store()
+    return _DEFAULT_STORE
diff --git a/services/api/tests/conftest.py b/services/api/tests/conftest.py
new file mode 100644
index 0000000..be15e2f
--- /dev/null
+++ b/services/api/tests/conftest.py
@@ -0,0 +1,16 @@
+"""Test bootstrap: prod-build modules importable, and all default
+state (CA key, store DB, work root) pointed at a throwaway tmp dir so
+tests never create real state files inside the deliverable tree."""
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+_TMP = tempfile.mkdtemp(prefix="ax-prod-test-")
+os.environ.setdefault("AXIOMCODE_CA_KEY_FILE", os.path.join(_TMP, "ca_key.pem"))
+os.environ.setdefault("AXIOMCODE_DB", os.path.join(_TMP, "axiomcode.db"))
+os.environ.setdefault("AXIOMCODE_WORK_ROOT", os.path.join(_TMP, "work"))
+Path(os.environ["AXIOMCODE_WORK_ROOT"]).mkdir(parents=True, exist_ok=True)
diff --git a/services/api/tests/test_ca.py b/services/api/tests/test_ca.py
new file mode 100644
index 0000000..89f11d5
--- /dev/null
+++ b/services/api/tests/test_ca.py
@@ -0,0 +1,78 @@
+"""W1 — Ed25519 certification authority."""
+
+import os
+import stat
+
+import ca
+import certs as legacy_certs
+import pytest
+
+SUBJECT = {
+    "code_sha256": "ab" * 32,
+    "verified": True,
+    "attempts": 1,
+    "toolchain": "leanprover/lean4:v4.35.0-rc3",
+    "key_id": "test-caller",
+}
+
+
+@pytest.fixture()
+def authority(tmp_path):
+    return ca.CA.load_or_create(tmp_path / "ca_key.pem")
+
+
+def test_key_persistence_and_permissions(tmp_path):
+    key_file = tmp_path / "ca_key.pem"
+    first = ca.CA.load_or_create(key_file)
+    assert key_file.exists()
+    assert stat.S_IMODE(os.stat(key_file).st_mode) == 0o600
+    second = ca.CA.load_or_create(key_file)
+    assert second.kid == first.kid  # same key reloaded, not regenerated
+
+
+def test_issue_verify_round_trip(authority):
+    cert = authority.issue(dict(SUBJECT))
+    assert cert["version"] == 2
+    assert cert["alg"] == "Ed25519"
+    assert cert["kid"] == authority.kid
+    assert cert["issuer"] == "AxiomCode CA"
+    assert cert["valid_until"] - cert["valid_from"] == 90 * 86400
+    result = ca.verify_certificate(cert, ca=authority)
+    assert result["valid"] is True
+    assert result["payload"]["serial"] == cert["serial"]
+
+
+def test_tampered_cert_fails(authority):
+    cert = authority.issue(dict(SUBJECT))
+    forged = dict(cert, verified=False)  # flip the verdict after signing
+    assert ca.verify_certificate(forged, ca=authority)["valid"] is False
+    mangled = dict(cert, signature="A" + cert["signature"][1:])
+    result = ca.verify_certificate(mangled, ca=authority)
+    assert result == {"valid": False, "reason": "bad_signature"}
+
+
+def test_wrong_key_fails(authority, tmp_path):
+    other = ca.CA.load_or_create(tmp_path / "other_key.pem")
+    cert = authority.issue(dict(SUBJECT))
+    result = ca.verify_certificate(cert, public_key=other.public_key)
+    assert result["valid"] is False
+    assert result["reason"] == "bad_signature"
+
+
+def test_public_payload_verifies_issued_cert(authority):
+    cert = authority.issue(dict(SUBJECT))
+    payload = authority.public_keys_payload()
+    assert payload["issuer"] == "AxiomCode CA"
+    (entry,) = payload["keys"]
+    assert entry["kid"] == cert["kid"]
+    assert entry["alg"] == "Ed25519"
+    # A third party with ONLY the published payload can verify:
+    by_b64 = ca.verify_certificate(cert, public_key=entry["public_key"])
+    assert by_b64["valid"] is True
+    by_pem = ca.verify_certificate(cert, public_key=entry["public_key_pem"])
+    assert by_pem["valid"] is True
+
+
+def test_legacy_hmac_cert_rejected(authority):
+    old = legacy_certs.issue(dict(SUBJECT))
+    assert ca.verify_certificate(old, ca=authority)["valid"] is False
diff --git a/services/api/tests/test_store.py b/services/api/tests/test_store.py
new file mode 100644
index 0000000..8fa21ea
--- /dev/null
+++ b/services/api/tests/test_store.py
@@ -0,0 +1,106 @@
+"""W2 — durable store with hash-chained audit log."""
+
+import sqlite3
+
+import billing
+import ca
+import pytest
+from store import Store
+
+
+@pytest.fixture()
+def store(tmp_path):
+    s = Store(tmp_path / "test.db")
+    yield s
+    s.close()
+
+
+@pytest.fixture()
+def cert(authority_keyfile):
+    return authority_keyfile.issue(
+        {
+            "code_sha256": "cd" * 32,
+            "verified": True,
+            "attempts": 1,
+            "toolchain": "leanprover/lean4:v4.35.0-rc3",
+            "key_id": "caller-1",
+        }
+    )
+
+
+@pytest.fixture()
+def authority_keyfile(tmp_path):
+    return ca.CA.load_or_create(tmp_path / "k.pem")
+
+
+def test_record_and_get_cert(store, cert):
+    store.record_cert(cert)
+    assert store.get_cert(cert["serial"]) == cert
+    assert store.get_cert("no-such-serial") is None
+
+
+def test_revocation_persists_across_reopen(tmp_path, cert):
+    db = tmp_path / "durable.db"
+    first = Store(db)
+    first.record_cert(cert)
+    first.revoke(cert["serial"], reason="toolchain defect")
+    assert first.is_revoked(cert["serial"])
+    first.close()
+
+    second = Store(db)  # reopen — revocation must survive
+    assert second.is_revoked(cert["serial"])
+    assert cert["serial"] in second.revoked_serials()
+    assert second.get_cert(cert["serial"])["serial"] == cert["serial"]
+    second.close()
+
+
+def test_audit_chain_verifies(store, cert):
+    store.record_cert(cert)  # issuance event
+    store.append_audit("verification", {"serial": cert["serial"], "valid": True, "via": "test"})
+    store.revoke(cert["serial"], reason="test")  # revocation event
+    result = store.verify_chain()
+    assert result == {"ok": True, "events": 3}
+    types = [e["event_type"] for e in store.audit_events()]
+    assert types == ["issuance", "verification", "revocation"]
+
+
+def test_audit_chain_detects_flipped_byte(tmp_path, cert):
+    db = tmp_path / "tamper.db"
+    s = Store(db)
+    s.record_cert(cert)
+    s.append_audit("verification", {"serial": cert["serial"], "valid": True})
+    s.revoke(cert["serial"], reason="test")
+    assert s.verify_chain()["ok"] is True
+    s.close()
+
+    raw = sqlite3.connect(db)
+    seq, payload = raw.execute("SELECT seq, payload_json FROM audit_log WHERE seq = 2").fetchone()
+    flipped = payload[:6] + ("X" if payload[6] != "X" else "Y") + payload[7:]
+    raw.execute("UPDATE audit_log SET payload_json = ? WHERE seq = ?", (flipped, seq))
+    raw.commit()
+    raw.close()
+
+    reopened = Store(db)
+    result = reopened.verify_chain()
+    assert result["ok"] is False
+    assert result["broken_seq"] == 2
+    reopened.close()
+
+
+def test_meter_events_recorded(store):
+    e1 = billing.meter(store, "key-1", "aa" * 32, True, 2.0)
+    billing.meter(store, "key-1", "bb" * 32, False, 2.0)
+    billing.meter(store, "key-2", "cc" * 32, True, 1.0)
+
+    events = store.meter_events("key-1")
+    assert len(events) == 2
+    assert events[0]["code_sha256"] == "aa" * 32
+    assert e1["credits"] == pytest.approx(1.8)  # 2.0s x 0.9 credits/s
+
+    usage = store.usage_for("key-1")
+    assert usage["calls"] == 2
+    assert usage["credits_used"] == pytest.approx(3.6)
+    assert usage["usd_used"] == pytest.approx(0.036)
+    # billing.usage_for routes to the same store numbers
+    assert billing.usage_for(store, "key-1") == usage
+    assert store.usage_for("key-2")["calls"] == 1
diff --git a/services/api/tests/test_wiring.py b/services/api/tests/test_wiring.py
new file mode 100644
index 0000000..419c76f
--- /dev/null
+++ b/services/api/tests/test_wiring.py
@@ -0,0 +1,108 @@
+"""Wiring — verify_api (REST) and mcp_server issue/verify via Ed25519
+and persist through the durable store. The Lean prover is stubbed:
+these tests prove the CA/store wiring, not Lean itself."""
+
+import os
+from pathlib import Path
+
+import ca
+import mcp_server
+import pytest
+import verify_api
+from fastapi.testclient import TestClient
+
+CODE = "theorem t : True := trivial"
+
+
+def _fake_search(writer, proj, code, *args, **kwargs):
+    writer(code)
+    return True, code, 1, "stubbed prover log"
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    monkeypatch.setattr(verify_api, "iterative_proof_search", _fake_search)
+    monkeypatch.setattr(verify_api, "lean_version", lambda: "lean-test")
+    return TestClient(verify_api.app)
+
+
+def _work_root_leftovers(prefix):
+    root = Path(os.environ["AXIOMCODE_WORK_ROOT"])
+    return [p.name for p in root.iterdir() if p.name.startswith(prefix)]
+
+
+def test_rest_end_to_end(client):
+    r = client.post("/verify", json={"code": CODE})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["verified"] is True
+    cert = body["certificate"]
+    assert cert["alg"] == "Ed25519" and cert["version"] == 2
+
+    # temp Lean project cleaned up by the finally block
+    assert _work_root_leftovers("ax-api-") == []
+
+    # cert persisted in the store
+    assert verify_api.STORE.get_cert(cert["serial"]) == cert
+
+    # published keys endpoint serves the payload for this CA
+    keys = client.get("/.well-known/axiomcode-keys.json")
+    assert keys.status_code == 200
+    payload = keys.json()
+    (entry,) = payload["keys"]
+    assert entry["kid"] == cert["kid"]
+    # a stranger with only the payload can verify the cert
+    assert ca.verify_certificate(cert, public_key=entry["public_key"])["valid"] is True
+
+    # check endpoint: valid, then revoked after operator revocation
+    ok = client.post("/check", json={"certificate": cert})
+    assert ok.json()["valid"] is True
+    verify_api.STORE.revoke(cert["serial"], reason="test revocation")
+    revoked = client.post("/check", json={"certificate": cert})
+    assert revoked.json() == {"valid": False, "reason": "revoked", "payload": cert}
+    listing = client.get("/revoked")
+    assert cert["serial"] in listing.json()["revoked"]
+
+    # metering + audit landed in the store; chain intact
+    assert verify_api.STORE.usage_for("rest-anonymous")["calls"] >= 1
+    chain = verify_api.STORE.verify_chain()
+    assert chain["ok"] is True
+    types = [e["event_type"] for e in verify_api.STORE.audit_events()]
+    assert "issuance" in types and "verification" in types
+    assert "revocation" in types
+
+
+def _unwrap(tool):
+    return getattr(tool, "fn", tool)
+
+
+def test_mcp_tools_use_ed25519_and_store(monkeypatch):
+    monkeypatch.setattr(mcp_server, "iterative_proof_search", _fake_search)
+    mcp_server._KEY.id = "mcp-test-key"
+
+    verify_fn = _unwrap(mcp_server.verify)
+    out = verify_fn(code=CODE)
+    cert = out["certificate"]
+    assert out["verified"] is True
+    assert cert["alg"] == "Ed25519"
+    assert cert["key_id"] == "mcp-test-key"
+    assert _work_root_leftovers("ax-mcp-") == []
+    assert mcp_server.STORE.get_cert(cert["serial"]) == cert
+
+    check_fn = _unwrap(mcp_server.check_certificate)
+    assert check_fn(certificate=cert)["valid"] is True
+
+    usage_fn = _unwrap(mcp_server.usage_report)
+    report = usage_fn()
+    assert report["key_id"] == "mcp-test-key"
+    assert report["calls"] >= 1
+    assert mcp_server.STORE.verify_chain()["ok"] is True
+
+
+def test_legacy_hmac_module_still_importable():
+    import certs  # legacy fixtures must keep working
+
+    old = certs.issue({"code_sha256": "ee" * 32, "verified": True, "attempts": 1, "toolchain": "x", "key_id": "k"})
+    assert certs.check(old)["valid"] is True  # HMAC path untouched
+    # ...but the production CA refuses it (no Ed25519 alg)
+    assert ca.verify_certificate(old)["valid"] is False
diff --git a/services/api/verify_api.py b/services/api/verify_api.py
index 5f37574..3898766 100644
--- a/services/api/verify_api.py
+++ b/services/api/verify_api.py
@@ -1,16 +1,26 @@
-"""AxiomCode verification API prototype.
-
-POST /verify {code} -> {verified, attempts, certificate}
-POST /check  {certificate} -> {valid, payload}
-
-The certificate is an HMAC-signed attestation binding:
-  code hash + verified verdict + toolchain version + timestamp.
-Anyone holding the cert can confirm *what* was checked and *what* the
-machine said — without trusting the server's word for it.
+"""AxiomCode verification API (production build).
+
+POST /verify {code} -> {verified, attempts, certificate, usage}
+POST /check  {certificate} -> {valid, ...}
+GET  /revoked -> {issuer, revoked: [serial...]}
+GET  /.well-known/axiomcode-keys.json -> published CA public keys
+
+Certificates are Ed25519-signed attestations (ca.py, format v2)
+binding: code hash + verified verdict + toolchain version + validity
+window. Anyone holding the cert and the published public key can
+confirm *what* was checked and *what* the machine said — without
+trusting the server's word for it. All state (certificates,
+revocations, meter, audit) lives in the durable store (store.py);
+issuance / revocation / verification events land in its hash-chained
+audit log.
+
+Temp Lean projects are created under AXIOMCODE_WORK_ROOT (default:
+system temp dir) and always removed after the verification finishes.
 """
 
 import hashlib
 import os
+import shutil
 import sys
 import tempfile
 import time
@@ -19,16 +29,18 @@
 sys.path.insert(0, "/home/hatch/workspace/axiomcode/repo")
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
-import certs
+import ca
 from billing import meter
+from core.prover import iterative_proof_search, lean_version
 from fastapi import FastAPI
 from pydantic import BaseModel
+from store import Store
 
-from core.prover import iterative_proof_search, lean_version
-
-SECRET = os.environ.get("AXIOMCODE_CERT_SECRET", "dev-secret-change-me").encode()
 TOOLCHAIN = "leanprover/lean4:v4.35.0-rc3"
-METER_LOG = os.environ.get("AXIOMCODE_METER_LOG", "/home/hatch/workspace/axiomcode/api-prototype/meter.jsonl")
+WORK_ROOT = os.environ.get("AXIOMCODE_WORK_ROOT", tempfile.gettempdir())
+
+CA = ca.CA.load_or_create()
+STORE = Store()
 
 app = FastAPI(
     title="AxiomCode Verify",
@@ -46,38 +58,38 @@ class CheckRequest(BaseModel):
     certificate: dict
 
 
-class UsageRequest(BaseModel):
-    key_id: str
-
-
 @app.post("/verify")
 def verify(req: VerifyRequest):
-    proj = Path(tempfile.mkdtemp(prefix="ax-api-"))
-    (proj / "lean-toolchain").write_text(TOOLCHAIN)
-    (proj / "lakefile.lean").write_text(
-        "import Lake\nopen Lake DSL\npackage v where\nlean_lib V where\n  roots := #[`V]\n"
-    )
-    src = proj / "V.lean"
-
-    def writer(code: str) -> Path:
-        src.write_text(code)
-        return src
-
-    t0 = time.time()
-    verified, final_code, attempts, log = iterative_proof_search(
-        writer,
-        proj,
-        req.code,
-        None,
-        "none",
-        max_attempts=req.max_attempts,
-        use_pantograph=False,
-    )
-    compute_s = time.time() - t0
+    proj = Path(tempfile.mkdtemp(prefix="ax-api-", dir=WORK_ROOT))
+    try:
+        (proj / "lean-toolchain").write_text(TOOLCHAIN)
+        (proj / "lakefile.lean").write_text(
+            "import Lake\nopen Lake DSL\npackage v where\nlean_lib V where\n  roots := #[`V]\n"
+        )
+        src = proj / "V.lean"
+
+        def writer(code: str) -> Path:
+            src.write_text(code)
+            return src
+
+        t0 = time.time()
+        verified, _final_code, attempts, _log = iterative_proof_search(
+            writer,
+            proj,
+            req.code,
+            None,
+            "none",
+            max_attempts=req.max_attempts,
+            use_pantograph=False,
+        )
+        compute_s = time.time() - t0
+    finally:
+        shutil.rmtree(proj, ignore_errors=True)
+
     code_sha = hashlib.sha256(req.code.encode()).hexdigest()
     # REST callers are metered under a per-request key; key auth lives on MCP.
-    usage = meter(METER_LOG, "rest-anonymous", code_sha, verified, compute_s)
-    cert = certs.issue(
+    usage = meter(STORE, "rest-anonymous", code_sha, verified, compute_s)
+    cert = CA.issue(
         {
             "code_sha256": code_sha,
             "verified": verified,
@@ -87,6 +99,7 @@ def writer(code: str) -> Path:
             "key_id": "rest-anonymous",
         }
     )
+    STORE.record_cert(cert)
     return {
         "verified": verified,
         "attempts": attempts,
@@ -97,9 +110,27 @@ def writer(code: str) -> Path:
 
 @app.post("/check")
 def check(req: CheckRequest):
-    return certs.check(req.certificate)
+    result = ca.verify_certificate(req.certificate, ca=CA)
+    serial = req.certificate.get("serial") if isinstance(req.certificate, dict) else None
+    if result.get("valid") and serial and STORE.is_revoked(serial):
+        result = {"valid": False, "reason": "revoked", "payload": req.certificate}
+    STORE.append_audit(
+        "verification",
+        {
+            "serial": serial,
+            "valid": bool(result.get("valid")),
+            "reason": result.get("reason"),
+            "via": "rest",
+        },
+    )
+    return result
 
 
 @app.get("/revoked")
 def revoked():
-    return {"issuer": certs.ISSUER, "revoked": sorted(certs.revoked_serials())}
+    return {"issuer": ca.ISSUER, "revoked": sorted(STORE.revoked_serials())}
+
+
+@app.get("/.well-known/axiomcode-keys.json")
+def ca_public_keys():
+    return CA.public_keys_payload()

From 1be30f82d4db6d86a37990f4e46155ebc23e0e27 Mon Sep 17 00:00:00 2001
From: Muse <muse@axiom-code.com>
Date: Thu, 15 Oct 2026 17:29:10 -0600
Subject: [PATCH 2/2] Fix CI in full-repo context: first-party import order
 (core), mypy Any returns + PEM key narrowing

---
 services/api/billing.py    | 8 +++++---
 services/api/ca.py         | 5 ++++-
 services/api/mcp_server.py | 3 ++-
 services/api/verify_api.py | 3 ++-
 4 files changed, 13 insertions(+), 6 deletions(-)

diff --git a/services/api/billing.py b/services/api/billing.py
index 4f7900a..e305a7c 100644
--- a/services/api/billing.py
+++ b/services/api/billing.py
@@ -49,7 +49,7 @@ def load_pricing() -> dict:
 def price_per_compute_s(pricing: dict | None = None) -> float:
     p = pricing or load_pricing()
     loaded = sum(v for k, v in p.items() if k != "multiple")
-    return loaded * p["multiple"]
+    return float(loaded * p["multiple"])
 
 
 def charge_for(compute_s: float, pricing: dict | None = None) -> dict:
@@ -93,7 +93,8 @@ def meter(target, key_id: str, code_sha: str, verified: bool, compute_s: float,
         **bill,
     }
     if _is_store(target):
-        return target.record_meter(entry)
+        entry_out: dict = target.record_meter(entry)
+        return entry_out
     with open(target, "a") as f:
         f.write(json.dumps(entry) + "\n")
     return entry
@@ -103,7 +104,8 @@ def usage_for(target, key_id: str) -> dict:
     """Total metered spend for a key — from the Store (production)
     or a legacy JSONL path."""
     if _is_store(target):
-        return target.usage_for(key_id)
+        usage: dict = target.usage_for(key_id)
+        return usage
     total_c, total_usd, calls = 0.0, 0.0, 0
     p = Path(target)
     if p.exists():
diff --git a/services/api/ca.py b/services/api/ca.py
index ac39c57..b56d6fd 100644
--- a/services/api/ca.py
+++ b/services/api/ca.py
@@ -69,7 +69,10 @@ def coerce_public_key(key) -> Ed25519PublicKey:
     if isinstance(key, Ed25519PublicKey):
         return key
     if isinstance(key, str) and "BEGIN PUBLIC KEY" in key:
-        return serialization.load_pem_public_key(key.encode())
+        loaded = serialization.load_pem_public_key(key.encode())
+        if not isinstance(loaded, Ed25519PublicKey):
+            raise ValueError("PEM does not hold an Ed25519 public key")
+        return loaded
     raw = base64.b64decode(key) if isinstance(key, str) else bytes(key)
     return Ed25519PublicKey.from_public_bytes(raw)
 
diff --git a/services/api/mcp_server.py b/services/api/mcp_server.py
index cfc77fe..39a5b94 100644
--- a/services/api/mcp_server.py
+++ b/services/api/mcp_server.py
@@ -28,12 +28,13 @@
 
 import ca
 from billing import load_keys, meter, price_per_compute_s, usage_for
-from core.prover import iterative_proof_search
 from mcp.server.fastmcp import FastMCP
 from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.responses import JSONResponse
 from store import Store
 
+from core.prover import iterative_proof_search
+
 # ─── config ──────────────────────────────────────────────────────────────
 _HERE = Path(__file__).parent
 API_KEYS_FILE = os.environ.get("AXIOMCODE_API_KEYS", str(_HERE / "keys.txt"))
diff --git a/services/api/verify_api.py b/services/api/verify_api.py
index 3898766..40df951 100644
--- a/services/api/verify_api.py
+++ b/services/api/verify_api.py
@@ -31,11 +31,12 @@
 
 import ca
 from billing import meter
-from core.prover import iterative_proof_search, lean_version
 from fastapi import FastAPI
 from pydantic import BaseModel
 from store import Store
 
+from core.prover import iterative_proof_search, lean_version
+
 TOOLCHAIN = "leanprover/lean4:v4.35.0-rc3"
 WORK_ROOT = os.environ.get("AXIOMCODE_WORK_ROOT", tempfile.gettempdir())