feat(cnf): as-of history fold + daemon :as-of op (thread 019f100f-eefe, part B)

live-as-of S = {claims born <= S} minus {claims with a cnf-supersedes marker
committed <= S}. Because retraction is append-only (a marker, never a delete),
this is EXACT: a claim later superseded/withdrawn is naturally RE-SEEN at an
earlier S — its tombstone hadn't been written yet. live-as-of-lp is the as-of
twin of live-cids-lp for one (te,pid) group. Folds the in-store tail, so it is
bounded by the snapshot floor, never O(total history).

Daemon :as-of {:seq S :query Q} runs the SAME q/run oracle over an as-of EDB
(the engine is untouched — it is just fed a historical claims vec); :as-of
{:seq S :te T :p P} resolves one group as-of, elected like :resolved.

cnf_causality (b): now-view shows only the latest; as-of before an overwrite
re-sees the superseded value; as-of current == live; as-of 0 is empty. 17/17.
Daemon loads clean.

Author: tompassarelli

cnf_coord.clj

@@ -105,6 +105,33 @@
 (defn causal-key [co cid]
   [(or (observed-of co cid) (seq-of co cid)) cid (str (agent-of co cid))])
 
+;; --- as-of: the history fold (thread H, Part B) ------------------------------
+;; "What was live AS OF seq S?" A claim is live-as-of-S iff it was BORN at a seq <= S
+;; AND no cnf-supersedes marker for it was committed at a seq <= S. Because retraction is
+;; append-only (a marker, never a delete), this is EXACT: a claim later superseded/withdrawn
+;; is naturally RE-SEEN at an earlier S — its tombstone hadn't been written yet (acceptance
+;; b). Folds the in-store tail, so it is bounded by thread D's snapshot floor (history
+;; compacted below a snapshot is gone: as-of before the floor is unavailable, not wrong) —
+;; never O(total history) (acceptance f).
+(defn- tx-seq-of [m cid] (get-in m [:txs (get (:tx-of m) cid) :seq] 0))
+(defn superseded-as-of [co s]
+  (let [m @(store co) sup (:supersedes-pred m)]
+    (set (for [[cid cl] (:claims m)
+               :when (and (= (:p cl) sup) (<= (tx-seq-of m cid) s))]
+           (:r cl)))))
+(defn live-as-of [co s]
+  (let [m @(store co) sup (:supersedes-pred m) gone (superseded-as-of co s)]
+    (set (for [[cid cl] (:claims m)
+               :when (and (<= (tx-seq-of m cid) s)
+                          (not= (:p cl) sup)              ; the markers are bookkeeping, not data
+                          (not (contains? gone cid)))]
+           cid))))
+;; the live cids of ONE (te,pid) group as of S — the as-of twin of live-cids-lp.
+(defn live-as-of-lp [co s te pid]
+  (let [m @(store co) all (live-as-of co s)]
+    (filterv (fn [cid] (let [cl (get (:claims m) cid)] (and (= (:l cl) te) (= (:p cl) pid))))
+             all)))
+
 ;; --- views-as-claims (thread E): per-branch isolation over the same log ------
 ;; A VIEW is a first-class subject; (view selects @cid) claims are its OVERLAY —
 ;; the cids it treats as facts. The object IS a claim id: cids live in the same

cnf_coord_daemon.clj

@@ -1283,6 +1283,33 @@
                   {:value (when (seq live) (render (elect @co live)))
                    :members (count vals) :ambiguous? (> (count vals) 1)
                    :values vals :version (current-seq @co)})
+      ;; :as-of — time-travel READ (thread H, Part B). Reconstruct the view as of seq S
+      ;; (cnf_coord/live-as-of: born <= S, no supersede/withdraw marker <= S) and either
+      ;; run a Datalog :query over that historical EDB (SAME q/run oracle, just fed an
+      ;; as-of claims vec — the engine is untouched) or resolve one (:te,:p) group as-of.
+      ;; Retraction-as-append makes this exact: a later-withdrawn claim is RE-SEEN at an
+      ;; earlier S. Bounded by the snapshot floor (live-as-of folds the in-store tail).
+      :as-of (let [s (:seq req) st (:store @co)]
+               (cond
+                 (nil? s) {:error ":as-of needs :seq"}
+                 (:query req)
+                 (let [cids   (live-as-of @co s)
+                       claims (vec (keep (fn [cid] (when-let [t (claim->triple st cid)]
+                                                     (ck/->Claim (nth t 0) (nth t 1) (nth t 2))))
+                                         cids))
+                       res    (q/run claims (:query req))]
+                   (assoc res :as-of s :version (current-seq @co)))
+                 (:te req)
+                 (let [e   (s/resolve-name st (:te req))
+                       pid (c/value-id st (:p req))
+                       live (if (and e pid) (live-as-of-lp @co s e pid) [])
+                       render (fn [cid] (let [r (:r (c/claim-of st cid))]
+                                          (if (c/value-object? st r) (c/literal st r) (s/name-of st r))))
+                       vals (mapv render live)]
+                   {:value (when (seq live) (render (elect @co (vec live))))
+                    :members (count vals) :ambiguous? (> (count vals) 1)
+                    :values vals :as-of s :version (current-seq @co)})
+                 :else {:error ":as-of needs :query or :te/:p"}))
       {:error "unknown op"}))))
 
 ;; ---- socket server (verbatim shape from the proven coord.clj) ---------------

tests/cnf_causality_test.clj

@@ -72,6 +72,31 @@
   (chk "(a) the elected member's value is the earliest decider's (B)"
        (= "B" (c/literal (store co) (:r (c/claim-of (store co) causal-win))))))
 
+;; ============================================================================
+;; (b) as-of — reconstruct a historical view; a later-superseded claim is RE-SEEN.
+;; ============================================================================
+(let [log "/tmp/cnf-causality-b.log"
+      co  (new-coord log)
+      _   (register-pred! co "status" "single" "literal")   ; single -> overwrite supersedes
+      pid (c/value-id (store co) "status")
+      tid (fn [] (s/resolve-name (store co) "TB"))
+      r1  (commit! co "w" "TB" "status" :assert "open" 0)   ; born at seq S1
+      s1  (:ok r1)
+      mid (current-seq co)                                  ; a seq AFTER open, BEFORE closed
+      r2  (commit! co "w" "TB" "status" :assert "closed" s1); supersedes "open" at seq S2 > S1
+      s2  (:ok r2)
+      val-of (fn [cids] (mapv #(c/literal (store co) (:r (c/claim-of (store co) %))) cids))]
+  (chk "(b) NOW: only the latest value is live (open superseded)"
+       (= ["closed"] (val-of (live-cids-lp co (tid) pid))))
+  (chk "(b) as-of a seq BEFORE the overwrite RE-SEES the superseded 'open'"
+       (= ["open"] (val-of (live-as-of-lp co mid (tid) pid))))
+  (chk "(b) as-of the CURRENT seq == the live view ('closed')"
+       (= ["closed"] (val-of (live-as-of-lp co s2 (tid) pid))))
+  (chk "(b) as-of seq 0 (pre-history) sees nothing on this group"
+       (empty? (live-as-of-lp co 0 (tid) pid)))
+  (chk "(b) as-of is bounded: live-as-of folds the in-store tail, returns a set"
+       (set? (live-as-of co s2))))
+
 ;; ---- summary ---------------------------------------------------------------
 (let [cs @checks fails (remove second cs)]
   (doseq [[nm ok] cs] (println (if ok "  [PASS] " "  [FAIL] ") nm))