feed: conversation threading — :reply-to transitive closure (thread/replies/thread-size) + 12 tests

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

lib/feed/conformance.sh

@@ -13,7 +13,7 @@ if [ ! -x "$SX_SERVER" ]; then
   exit 1
 fi
 
-SUITES=(basic fanout rank integration content notify home dedupe trending mute page)
+SUITES=(basic fanout rank integration content notify home dedupe trending mute page thread)
 
 OUT_JSON="lib/feed/scoreboard.json"
 OUT_MD="lib/feed/scoreboard.md"
@@ -43,6 +43,7 @@ run_suite() {
 (load "lib/feed/trending.sx")
 (load "lib/feed/mute.sx")
 (load "lib/feed/page.sx")
+(load "lib/feed/thread.sx")
 (epoch 2)
 (eval "(define feed-test-pass 0)")
 (eval "(define feed-test-fail 0)")

lib/feed/scoreboard.json

@@ -10,9 +10,10 @@
     "dedupe": {"pass": 9, "fail": 0},
     "trending": {"pass": 11, "fail": 0},
     "mute": {"pass": 9, "fail": 0},
-    "page": {"pass": 14, "fail": 0}
+    "page": {"pass": 14, "fail": 0},
+    "thread": {"pass": 12, "fail": 0}
   },
-  "total_pass": 177,
+  "total_pass": 189,
   "total_fail": 0,
-  "total": 177
+  "total": 189
 }

lib/feed/scoreboard.md

@@ -15,4 +15,5 @@ _Generated by `lib/feed/conformance.sh`_
 | trending | 11 | 0 | 11 |
 | mute | 9 | 0 | 9 |
 | page | 14 | 0 | 14 |
-| **Total** | **177** | **0** | **177** |
+| thread | 12 | 0 | 12 |
+| **Total** | **189** | **0** | **189** |

lib/feed/tests/thread.sx

@@ -0,0 +1,49 @@
+; Follow-up — conversation threading via :reply-to closure. (feed-test name got expected)
+
+(define
+  S
+  (feed/stream
+    (list
+      (feed/normalize {:actor "a" :object "root" :at 1})
+      (feed/normalize {:actor "b" :object "r1" :at 2 :verb "reply" :reply-to "root"})
+      (feed/normalize {:actor "c" :object "r2" :at 3 :verb "reply" :reply-to "root"})
+      (feed/normalize {:actor "d" :object "r3" :at 4 :verb "reply" :reply-to "r1"})
+      (feed/normalize {:actor "e" :object "x" :at 5}))))
+
+; ---------- direct replies ----------
+
+(feed-test "direct replies to root" (feed/reply-count S "root") 2)
+(feed-test "direct replies to r1" (feed/reply-count S "r1") 1)
+(feed-test "no replies to r3" (feed/reply-count S "r3") 0)
+(feed-test
+  "replies objects to root"
+  (map (fn (a) (get a :object)) (feed/items (feed/replies S "root")))
+  (list "r1" "r2"))
+
+; ---------- thread closure ----------
+
+(feed-test
+  "thread objects root (transitive)"
+  (feed/thread-objects S "root")
+  (list "root" "r1" "r2" "r3"))
+(feed-test
+  "thread root chronological"
+  (map (fn (a) (get a :object)) (feed/items (feed/thread S "root")))
+  (list "root" "r1" "r2" "r3"))
+(feed-test "thread size root" (feed/thread-size S "root") 4)
+(feed-test
+  "thread excludes unrelated x"
+  (feed/-elem?
+    "x"
+    (map (fn (a) (get a :object)) (feed/items (feed/thread S "root"))))
+  false)
+
+; ---------- sub-thread ----------
+
+(feed-test
+  "thread from r1 (sub-tree)"
+  (map (fn (a) (get a :object)) (feed/items (feed/thread S "r1")))
+  (list "r1" "r3"))
+(feed-test "thread size r1" (feed/thread-size S "r1") 2)
+(feed-test "leaf thread is itself" (feed/thread-size S "r3") 1)
+(feed-test "unrelated thread is itself" (feed/thread-size S "x") 1)

lib/feed/thread.sx

@@ -0,0 +1,59 @@
+; feed/thread — conversation threading. A reply carries :reply-to <parent-object>
+; (normalize preserves it). A thread is the transitive closure over :reply-to from
+; a root object: root + replies + replies-to-replies, gathered chronologically.
+;
+; Requires: lib/feed/normalize.sx, lib/feed/stream.sx, lib/feed/fanout.sx
+; (feed/-elem?, feed/-distinct).
+
+; direct replies to an object
+(define
+  feed/replies
+  (fn
+    (stream object)
+    (feed/filter stream (fn (a) (equal? (get a :reply-to) object)))))
+
+(define
+  feed/reply-count
+  (fn (stream object) (feed/count (feed/replies stream object))))
+
+; iterate f from x until the result stops growing (set-closure fixpoint)
+(define
+  feed/-fixpoint
+  (fn
+    (f x)
+    (let
+      ((nx (f x)))
+      (if (= (len nx) (len x)) x (feed/-fixpoint f nx)))))
+
+; the set of object-ids in the thread rooted at `root`
+(define
+  feed/thread-objects
+  (fn
+    (stream root)
+    (let
+      ((all (feed/items stream)))
+      (feed/-fixpoint
+        (fn
+          (acc)
+          (feed/-distinct
+            (append
+              acc
+              (map
+                (fn (a) (get a :object))
+                (filter (fn (a) (feed/-elem? (get a :reply-to) acc)) all)))))
+        (list root)))))
+
+; the full thread as a chronological stream (root + all descendants)
+(define
+  feed/thread
+  (fn
+    (stream root)
+    (let
+      ((objs (feed/thread-objects stream root)))
+      (feed/sort-by-at
+        (feed/filter stream (fn (a) (feed/-elem? (get a :object) objs)))))))
+
+; how many activities are in the thread (root counts as 1)
+(define
+  feed/thread-size
+  (fn (stream root) (feed/count (feed/thread stream root))))

plans/feed-on-sx.md

@@ -14,7 +14,7 @@ APL, ACL visibility filtering via `lib/acl/`, federation via fed-sx.
 
 ## Status (rolling)
 
-`bash lib/feed/conformance.sh` → **177/177** (Phases 1–4 + TF-IDF, notifications, home, smart-dedupe, trending, mute, pagination)
+`bash lib/feed/conformance.sh` → **189/189** (Phases 1–4 + TF-IDF, notifications, home, smart-dedupe, trending, mute, pagination, threading)
 
 ## Ground rules
 
@@ -160,6 +160,9 @@ are function parameters. Real acl-sx / fed-sx wire in at the call site unchanged
 - [x] Pagination — `feed/page`/`feed/page-count` (offset) + `feed/before`/
   `feed/after`/`feed/page-before`/`feed/next-cursor` (cursor by :at, stable under
   inserts) (`page.sx`); 14 tests. (177/177 total.)
+- [x] Threading — `feed/replies`/`feed/reply-count`/`feed/thread`/
+  `feed/thread-objects`/`feed/thread-size`: conversation closure over `:reply-to`
+  (transitive fixpoint), chronological (`thread.sx`); 12 tests. (189/189 total.)
 
 (none)