sx-pub Phase 4: anchoring — Merkle trees, OpenTimestamps, verification

New endpoints:
- POST /pub/anchor — batch unanchored Publish activities into Merkle tree,
  pin tree to IPFS, submit root to OpenTimestamps, store OTS proof on IPFS
- GET /pub/verify/<cid> — verify a CID's Merkle proof against anchored tree

Uses existing shared/utils/anchoring.py infrastructure:
- build_merkle_tree (SHA256, deterministic sort)
- get_merkle_proof / verify_merkle_proof (inclusion proofs)
- submit_to_opentimestamps (3 calendar servers with fallback)

Tested: anchored 1 activity, Merkle tree + OTS proof pinned to IPFS,
verification returns :verified true with full proof chain.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

sx/sx/handlers/pub-api.sx

@@ -337,3 +337,59 @@
                 " :actor-url \"" (get f "actor-url") "\")"))
               data)))
         (str "(SxFollowing" (join "" items) ")")))))
+
+
+;; ==========================================================================
+;; Phase 4: Anchoring — Merkle trees, OTS, verification
+;; ==========================================================================
+
+
+;; --------------------------------------------------------------------------
+;; Anchor pending activities
+;; --------------------------------------------------------------------------
+
+(defhandler pub-anchor
+  :path "/pub/anchor"
+  :method :post
+  :csrf false
+  :returns "element"
+  (&key)
+  (let ((result (helper "pub-anchor-pending")))
+    (do
+      (set-response-header "Content-Type" "text/sx; charset=utf-8")
+      (if (= (get result "status") "nothing-to-anchor")
+        "(Anchor :status \"nothing-to-anchor\" :count 0)"
+        (str
+          "(Anchor"
+          "\n  :status \"" (get result "status") "\""
+          "\n  :count " (get result "count")
+          "\n  :merkle-root \"" (get result "merkle-root") "\""
+          "\n  :tree-cid \"" (get result "tree-cid") "\""
+          "\n  :ots-proof-cid \"" (get result "ots-proof-cid") "\")")))))
+
+
+;; --------------------------------------------------------------------------
+;; Verify a CID's anchor
+;; --------------------------------------------------------------------------
+
+(defhandler pub-verify
+  :path "/pub/verify/<cid>"
+  :method :get
+  :returns "element"
+  (&key cid)
+  (let ((data (helper "pub-verify-anchor" cid)))
+    (do
+      (set-response-header "Content-Type" "text/sx; charset=utf-8")
+      (if (get data "error")
+        (do
+          (set-response-status 404)
+          (str "(Error :message \"" (get data "error") "\")"))
+        (str
+          "(AnchorVerification"
+          "\n  :cid \"" (get data "cid") "\""
+          "\n  :status \"" (get data "status") "\""
+          "\n  :verified " (get data "verified")
+          "\n  :merkle-root \"" (get data "merkle-root") "\""
+          "\n  :tree-cid \"" (get data "tree-cid") "\""
+          "\n  :ots-proof-cid \"" (get data "ots-proof-cid") "\""
+          "\n  :published \"" (get data "published") "\")")))))

sx/sxc/pages/helpers.py

@@ -52,6 +52,8 @@ def _register_sx_helpers() -> None:
         "pub-process-inbox": _pub_process_inbox,
         "pub-deliver-to-followers": _pub_deliver_to_followers,
         "pub-request-body": _pub_request_body,
+        "pub-anchor-pending": _pub_anchor_pending,
+        "pub-verify-anchor": _pub_verify_anchor,
     })
 
 
@@ -1806,3 +1808,13 @@ async def _pub_deliver_to_followers(activity_sx):
 async def _pub_request_body():
     from .pub_helpers import get_request_body
     return await get_request_body()
+
+
+async def _pub_anchor_pending():
+    from .pub_helpers import anchor_pending
+    return await anchor_pending()
+
+
+async def _pub_verify_anchor(cid):
+    from .pub_helpers import verify_cid_anchor
+    return await verify_cid_anchor(cid)

sx/sxc/pages/pub_helpers.py

@@ -777,3 +777,165 @@ async def get_request_path() -> str:
     """Get the request path."""
     from quart import request
     return request.path
+
+
+# ---------------------------------------------------------------------------
+# Phase 4: Anchoring — Merkle trees, OTS, verification
+# ---------------------------------------------------------------------------
+
+async def anchor_pending() -> dict[str, Any]:
+    """Anchor all unanchored Publish activities into a Merkle tree.
+
+    1. Collect unanchored activities (by CID)
+    2. Build Merkle tree
+    3. Pin tree to IPFS
+    4. Submit root to OpenTimestamps
+    5. Store proof on IPFS
+    6. Record anchor in DB
+    """
+    from quart import g
+    from sqlalchemy import select
+    from shared.models.sx_pub import SxPubActivity
+    from shared.utils.anchoring import (
+        build_merkle_tree, submit_to_opentimestamps,
+    )
+    from shared.utils.ipfs_client import add_json, add_bytes, is_available
+
+    # Find unanchored Publish activities with CIDs
+    result = await g.s.execute(
+        select(SxPubActivity).where(
+            SxPubActivity.activity_type == "Publish",
+            SxPubActivity.ipfs_cid.isnot(None),
+            SxPubActivity.object_data["anchor_tree_cid"].is_(None),
+        ).order_by(SxPubActivity.created_at.asc())
+        .limit(100)
+    )
+    activities = result.scalars().all()
+
+    if not activities:
+        return {"status": "nothing-to-anchor", "count": 0}
+
+    # Build Merkle tree from CIDs
+    cids = [a.ipfs_cid for a in activities if a.ipfs_cid]
+    if not cids:
+        return {"status": "no-cids", "count": 0}
+
+    tree = build_merkle_tree(cids)
+    merkle_root = tree["root"]
+
+    # Pin tree to IPFS
+    tree_cid = None
+    ots_proof_cid = None
+
+    if await is_available():
+        try:
+            tree_data = {
+                "root": merkle_root,
+                "leaves": tree["leaves"],
+                "cids": cids,
+                "created_at": datetime.now(timezone.utc).isoformat(),
+            }
+            tree_cid = await add_json(tree_data)
+            logger.info("Merkle tree pinned: %s (%d leaves)", tree_cid, len(cids))
+        except Exception as e:
+            logger.error("IPFS tree storage failed: %s", e)
+
+    # Submit to OpenTimestamps
+    ots_proof = await submit_to_opentimestamps(merkle_root)
+    if ots_proof and await is_available():
+        try:
+            ots_proof_cid = await add_bytes(ots_proof)
+            logger.info("OTS proof pinned: %s", ots_proof_cid)
+        except Exception as e:
+            logger.error("IPFS OTS proof storage failed: %s", e)
+
+    # Record anchor in activities (store in object_data)
+    anchor_info = {
+        "merkle_root": merkle_root,
+        "tree_cid": tree_cid,
+        "ots_proof_cid": ots_proof_cid,
+        "activity_count": len(activities),
+        "anchored_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+    for a in activities:
+        data = dict(a.object_data or {})
+        data["anchor_tree_cid"] = tree_cid
+        data["anchor_merkle_root"] = merkle_root
+        data["anchor_ots_cid"] = ots_proof_cid
+        a.object_data = data
+
+    # Also record anchor as its own activity
+    from shared.models.sx_pub import SxPubActivity as SPA
+    g.s.add(SPA(
+        activity_type="Anchor",
+        object_type="MerkleTree",
+        object_data=anchor_info,
+        ipfs_cid=tree_cid,
+    ))
+    await g.s.flush()
+
+    logger.info("Anchored %d activities, root=%s, tree=%s, ots=%s",
+                len(activities), merkle_root, tree_cid, ots_proof_cid)
+
+    return {
+        "status": "anchored",
+        "count": len(activities),
+        "merkle-root": merkle_root,
+        "tree-cid": tree_cid or "",
+        "ots-proof-cid": ots_proof_cid or "",
+    }
+
+
+async def verify_cid_anchor(cid: str) -> dict[str, Any]:
+    """Verify the anchor proof for a specific CID."""
+    from quart import g
+    from sqlalchemy import select
+    from shared.models.sx_pub import SxPubActivity
+    from shared.utils.anchoring import build_merkle_tree, verify_merkle_proof
+    from shared.utils.ipfs_client import get_bytes
+
+    # Find the Publish activity for this CID
+    result = await g.s.execute(
+        select(SxPubActivity).where(
+            SxPubActivity.activity_type == "Publish",
+            SxPubActivity.ipfs_cid == cid,
+        )
+    )
+    activity = result.scalar_one_or_none()
+    if not activity:
+        return {"error": "not-found", "cid": cid}
+
+    data = activity.object_data or {}
+    tree_cid = data.get("anchor_tree_cid")
+    merkle_root = data.get("anchor_merkle_root")
+    ots_cid = data.get("anchor_ots_cid")
+
+    if not tree_cid:
+        return {"status": "not-anchored", "cid": cid}
+
+    # Fetch the Merkle tree from IPFS to verify
+    verified = False
+    if tree_cid:
+        tree_bytes = await get_bytes(tree_cid)
+        if tree_bytes:
+            import json
+            try:
+                tree_data = json.loads(tree_bytes)
+                tree = build_merkle_tree(tree_data["cids"])
+                from shared.utils.anchoring import get_merkle_proof
+                proof = get_merkle_proof(tree, cid)
+                if proof is not None:
+                    verified = verify_merkle_proof(cid, proof, tree["root"])
+            except Exception as e:
+                logger.warning("Merkle verification failed: %s", e)
+
+    return {
+        "status": "anchored" if verified else "unverified",
+        "cid": cid,
+        "merkle-root": merkle_root or "",
+        "tree-cid": tree_cid or "",
+        "ots-proof-cid": ots_cid or "",
+        "verified": "true" if verified else "false",
+        "published": activity.published.isoformat() if activity.published else "",
+    }