constitution · epochs · watch · epoch 4

c_d59e1908ed6f (tommy-mor) vs c_5d4f3299fa6d (tommy-mor)

download prompt · raw event · cmp_590a7825317722

council reasoning

No judgments yet.

sides

[07cc0e1c] Bind HTTP before replaying the ledger so boots stay reachable.

Import-time JsonlStore replay of a 450MB evidence log blocked uvicorn from
listening; defer load to a startup task and keep /api/health always live.

Co-authored-by: Cursor <cursoragent@cursor.com>

diff --git a/constitution.py b/constitution.py
index eb2779c3c6775ed26bb932320635ffdfc6edd439..17e322a8884a09d7579676ae221b705f46039f6e 100644
--- a/constitution.py
+++ b/constitution.py
@@ -27,8 +27,9 @@ Run: uv run constitution.py
 from decimal import Decimal, getcontext, DefaultContext
 from datetime import datetime, timezone
 from fastapi import FastAPI, Request, Response
-from fastapi.responses import PlainTextResponse, HTMLResponse
+from fastapi.responses import PlainTextResponse, HTMLResponse, JSONResponse
 from starlette.middleware.sessions import SessionMiddleware
+from starlette.requests import Request
 import json, time, os, asyncio, httpx, pathlib, subprocess, hashlib, re, fcntl, base64
 import sympy as sp  # type: ignore[reportMissingImports]
 from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
@@ -269,7 +270,42 @@ class Evidence:
     payload: dict
 
 
-store = JsonlStore(JSONL_PATH)
+class DeferredJsonlStore:
+    """
+    Delay ledger replay until after the HTTP server binds.
+
+    A 450MB+ evidence ledger can take minutes to parse; doing that at import
+    time makes Fly health checks and the public site time out on every boot.
+    """
+
+    def __init__(self, path: pathlib.Path | str):
+        self.path = pathlib.Path(path)
+        self._inner: JsonlStore | None = None
+        self.ready = False
+
+    def load_sync(self) -> None:
+        self._inner = JsonlStore(self.path)
+        self.ready = True
+
+    def _require(self) -> JsonlStore:
+        if self._inner is None:
+            self.load_sync()
+        assert self._inner is not None
+        return self._inner
+
+    def read(self) -> list:
+        if self._inner is None:
+            return []
+        return self._inner.read()
+
+    async def append(self, e) -> None:
+        await self._require().append(e)
+
+    async def atomic(self, fn):
+        return await self._require().atomic(fn)
+
+
+store = DeferredJsonlStore(JSONL_PATH)
 _LEDGER_LOCK = asyncio.Lock()
 
 
@@ -2211,10 +2247,29 @@ async def get_ledger(offset: int = 0, limit: int = 100, full: int = 0):
     return rows
 
 
+@app.middleware("http")
+async def ledger_ready_gate(request: Request, call_next):
+    if request.url.path == "/api/health":
+        return await call_next(request)
+    if isinstance(store, DeferredJsonlStore) and not store.ready:
+        return JSONResponse(
+            {
+                "ok": False,
+                "loading": True,
+                "message": "replaying ledger into memory",
+            },
+            status_code=503,
+        )
+    return await call_next(request)
+
+
 @app.get("/api/health")
 async def get_health():
-    """Liveness only — must not touch the ledger (health checks during ranking)."""
-    return {"ok": True}
+    """Liveness only — must not touch the ledger (boot + ranking)."""
+    return {
+        "ok": True,
+        "ledger_ready": bool(getattr(store, "ready", True)),
+    }
 
 
 @app.get("/api/epoch")
@@ -3703,8 +3758,17 @@ async def callback(request: Request, code: str):
 
 @app.on_event("startup")
 async def startup():
-    if os.environ.get("DISABLE_EPOCH_LOOP") != "1":
-        asyncio.create_task(epoch_loop())
+    async def boot():
+        if isinstance(store, DeferredJsonlStore) and not store.ready:
+            print("replaying ledger…", flush=True)
+            await asyncio.to_thread(store.load_sync)
+            n = len(store.read())
+            print(f"ledger ready: {n} events", flush=True)
+        if os.environ.get("DISABLE_EPOCH_LOOP") != "1":
+            asyncio.create_task(epoch_loop())
+
+    # Do not block bind/health on ledger replay.
+    asyncio.create_task(boot())
 
 
 if __name__ == "__main__":

[0e81b5e1] Solve Rank Centrality exactly instead of capping power iteration.

`compute_scores_from_edges` ran power iteration for at most 10 000 sweeps and
returned whatever vector it had reached, with no signal when the cap bound. On
a preference chain the spectral gap is Theta(1/n^2), so the cap binds past
n ~= 1500 -- and worse, the L1 *step* tolerance is an absolute test, so on a
1024-node chain with mixed ratios it declared success after 4325 sweeps while
leaving 486 of 1024 items at the wrong rank. A 1024-node star fails outright:
it needs 14 154 sweeps.

New `stationary` module owns every candidate solver -- power, Aitken-accelerated
power, Gauss-Seidel/SOR, preconditioned BiCGSTAB, dense LU, dense GTH, and a
sparse GTH state reduction with minimum-degree elimination -- behind one `solve`
that splits disconnected components, tries the sparse direct path first (trees
and chains reduce in O(n) with zero fill), and falls back to Gauss-Seidel then
power iteration on the dense graphs elimination cannot help with. Every answer
carries its own recomputed residual and a `converged` flag; nothing comes back
claiming success it did not earn.

Scores are also produced in log space. A chain of n items each preferred 2:1
spans 2^(n-1), which leaves the f64 range at n ~= 1075, so `pi` genuinely cannot
order a long chain no matter how it is computed. `Solution::log_pi` stays exact
there and is what `ranked_items_subset` now sorts on.

Edges are sorted before aggregation, so the result no longer depends on
`HashMap` iteration order. The legacy builder produced a different bit pattern
on 7 of 7 rebuilds of the same graph; the sort costs ~10% of chain construction.

Adds `benches/stationary_solvers.rs` (~80s) and `examples/solver_probe.rs`
(~30s), which reports time, residual, and rank displacement against closed-form
answers per topology. No new dependencies.

Co-authored-by: Cursor <cursoragent@cursor.com>

diff --git a/server/Cargo.toml b/server/Cargo.toml
index 7de0c944b4a8c7e3123b818432fce82f09e6aeca..cdc2a78be0ffbd6ac27a02a332093daf5eb8b990 100644
--- a/server/Cargo.toml
+++ b/server/Cargo.toml
@@ -49,4 +49,8 @@ harness = false
 name = "ingest_replay"
 harness = false
 
+[[bench]]
+name = "stationary_solvers"
+harness = false
+
 
diff --git a/server/benches/stationary_solvers.rs b/server/benches/stationary_solvers.rs
new file mode 100644
index 0000000000000000000000000000000000000000..ff442b86e61861633ce2be1a016eae53aec73e6e
--- /dev/null
+++ b/server/benches/stationary_solvers.rs
@@ -0,0 +1,171 @@
+//! Head-to-head cost of the stationary-distribution solvers behind Rank Centrality.
+//!
+//! Deliberately small: a couple of representative sizes per topology, ten
+//! samples each, ~1 s of measurement per point. The point is the *ordering*
+//! between solvers, which is stable at this resolution; for a finer sweep plus
+//! accuracy numbers use `cargo run --release -p slugsocial-server --example
+//! solver_probe`, which finishes in seconds.
+//!
+//! What each topology is here to show:
+//!
+//! - **chain** — spectral gap `Θ(1/n²)`. Power iteration needs `Θ(n²)` sweeps
+//!   and blows the 10 000 cap past n ≈ 1500; sparse GTH is `O(n)` because a
+//!   tree eliminates with zero fill.
+//! - **star** — uniformization by `d_max = n-1` makes the chain almost purely
+//!   lazy, so power iteration crawls (it fails the cap at n = 1024) while
+//!   Gauss–Seidel finishes in two sweeps.
+//! - **clique / sparse** — well-conditioned. Iterative wins outright and the
+//!   direct methods are the ones paying.
+
+use std::hint::black_box;
+use std::time::Duration;
+
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+
+use slugsocial_server::ranking::chain_from_edges;
+use slugsocial_server::stationary::{
+    bicgstab, dense_gth, dense_lu, power, power_aitken, solve, sor, sparse_gth, RankChain,
+    SolveOptions, SparseGthOutcome,
+};
+
+mod common;
+use common::{build_group, Topology};
+
+const TOL: f64 = 1e-8;
+const CAP: usize = 10_000;
+
+fn opts() -> SolveOptions {
+    SolveOptions {
+        tol: TOL,
+        max_iters: CAP,
+        ..SolveOptions::default()
+    }
+}
+
+fn chain_of(topo: Topology, n: usize) -> RankChain {
+    let g = build_group(topo, n);
+    chain_from_edges(g.idx_to_item.len(), g.edges.iter().map(|(&k, &w)| (k, w)))
+}
+
+fn quick<'a>(
+    c: &'a mut Criterion,
+    name: &str,
+) -> criterion::BenchmarkGroup<'a, criterion::measurement::WallTime> {
+    let mut g = c.benchmark_group(name);
+    g.sample_size(10)
+        .warm_up_time(Duration::from_millis(300))
+        .measurement_time(Duration::from_secs(1));
+    g
+}
+
+/// Every solver on the same input, per topology.
+fn solver_shootout(c: &mut Criterion) {
+    let cases: Vec<(Topology, usize, bool)> = vec![
+        // (topology, n, include the O(n³) dense arms)
+        (Topology::Chain, 256, true),
+        (Topology::Chain, 1024, true),
+        (Topology::Star, 1024, true),
+        (Topology::Clique, 256, true),
+        (Topology::RandomSparse { degree: 6 }, 1024, false),
+    ];
+
+    for (topo, n, dense) in cases {
+        let chain = chain_of(topo, n);
+        let mut g = quick(c, &format!("solvers/{}", topo.label()));
+
+        g.bench_with_input(BenchmarkId::new("power", n), &chain, |b, ch| {
+            b.iter(|| black_box(power(ch, opts()).residual))
+        });
+        g.bench_with_input(BenchmarkId::new("power+aitken", n), &chain, |b, ch| {
+            b.iter(|| black_box(power_aitken(ch, opts()).residual))
+        });
+        g.bench_with_input(BenchmarkId::new("sor", n), &chain, |b, ch| {
+            b.iter(|| black_box(sor(ch, opts(), 1.0).residual))
+        });
+        g.bench_with_input(BenchmarkId::new("bicgstab", n), &chain, |b, ch| {
+            b.iter(|| black_box(bicgstab(ch, opts()).residual))
+        });
+        g.bench_with_input(BenchmarkId::new("sparse-gth", n), &chain, |b, ch| {
+            let unlimited = SolveOptions {
+                direct_work_budget: u64::MAX,
+                dense_core_max: usize::MAX,
+                ..opts()
+            };
+            b.iter(|| match sparse_gth(ch, unlimited, TOL) {
+                SparseGthOutcome::Solved(s) => black_box(s.residual),
+                SparseGthOutcome::TooDense { .. } => unreachable!(),
+            })
+        });
+        if dense {
+            g.bench_with_input(BenchmarkId::new("dense-gth", n), &chain, |b, ch| {
+                b.iter(|| black_box(dense_gth(ch, TOL).residual))
+            });
+            g.bench_with_input(BenchmarkId::new("dense-lu", n), &chain, |b, ch| {
+                b.iter(|| black_box(dense_lu(ch, TOL).residual))
+            });
+        }
+        g.finish();
+    }
+}
+
+/// The shipped path, including the cost of deciding which solver to use.
+fn hybrid_dispatch(c: &mut Criterion) {
+    let cases: Vec<(String, Topology, usize)> = vec![
+        ("chain".into(), Topology::Chain, 4000),
+        ("star".into(), Topology::Star, 4000),
+        ("clique".into(), Topology::Clique, 400),
+        (
+            "sparse-d6".into(),
+            Topology::RandomSparse { degree: 6 },
+            4000,
+        ),
+        (
+            "components".into(),
+            Topology::ManyComponents {
+                components: 64,
+                size: 64,
+            },
+            0,
+        ),
+    ];
+    let mut g = quick(c, "hybrid");
+    for (label, topo, n) in cases {
+        let chain = chain_of(topo, n);
+        g.bench_with_input(BenchmarkId::new(label, chain.n), &chain, |b, ch| {
+            b.iter(|| black_box(solve(ch, opts()).residual))
+        });
+    }
+    g.finish();
+}
+
+/// Building the chain (aggregate, pairwise-normalize, sort) versus solving it.
+/// Sorting is what buys run-to-run determinism; this is where to see its price.
+fn chain_construction(c: &mut Criterion) {
+    let mut g = quick(c, "chain_from_edges");
+    for (topo, n) in [
+        (Topology::RandomSparse { degree: 6 }, 4000usize),
+        (Topology::Clique, 400),
+    ] {
+        let group = build_group(topo, n);
+        let total = group.idx_to_item.len();
+        g.bench_with_input(
+            BenchmarkId::new(topo.label(), group.edges.len()),
+            &group,
+            |b, grp| {
+                b.iter(|| {
+                    let ch = chain_from_edges(total, grp.edges.iter().map(|(&k, &w)| (k, w)));
+                    black_box(ch.nnz())
+                })
+            },
+        );
+    }
+    g.finish();
+}
+
+criterion_group!(
+    benches,
+    solver_shootout,
+    hybrid_dispatch,
+    chain_construction
+);
+criterion_main!(benches);
diff --git a/server/examples/solver_probe.rs b/server/examples/solver_probe.rs
new file mode 100644
index 0000000000000000000000000000000000000000..e3933530ac9efab379dc10ec2a467eb30ad203f8
--- /dev/null
+++ b/server/examples/solver_probe.rs
@@ -0,0 +1,738 @@
+//! Head-to-head comparison of stationary-distribution solvers for Rank Centrality.
+//!
+//! Run: `cargo run --release -p slugsocial-server --example solver_probe`
+//!
+//! Reports, per (topology, solver): wall time, backward error `‖πP − π‖₁`, and
+//! how badly the produced *order* disagrees with ground truth. On a chain the
+//! ground truth order is known exactly by construction, and on any tree the
+//! exact scores are known in closed form from detailed balance, so "how wrong
+//! is the current implementation" is answerable without trusting any solver.
+
+use std::time::Instant;
+
+use slugsocial_server::path_types::ItemId;
+use slugsocial_server::ranking::chain_from_edges;
+use slugsocial_server::reducer::{GroupState, VoteData};
+use slugsocial_server::stationary::{
+    bicgstab, dense_gth, dense_lu, power, power_aitken, solve, sor, sparse_gth, Method, RankChain,
+    Solution, SolveOptions, SparseGthOutcome,
+};
+
+const TOL: f64 = 1e-8;
+const CAP: usize = 10_000;
+
+// ---------------------------------------------------------------------------
+// Graph generators (mirrors server/benches/common/mod.rs so numbers line up)
+// ---------------------------------------------------------------------------
+
+struct Rng(u64);
+impl Rng {
+    fn new(s: u64) -> Self {
+        Rng(s | 1)
+    }
+    fn next_u64(&mut self) -> u64 {
+        let mut x = self.0;
+        x ^= x >> 12;
+        x ^= x << 25;
+        x ^= x >> 27;
+        self.0 = x;
+        x.wrapping_mul(0x2545_F491_4F6C_DD1D)
+    }
+    fn below(&mut self, n: usize) -> usize {
+        (self.next_u64() % n as u64) as usize
+    }
+}
+
+fn item_name(i: usize) -> String {
+    format!("~/bench/i{i:06}")
+}
+
+fn vote(a: usize, b: usize, l: i32, r: i32, ts: i64) -> VoteData {
+    VoteData {
+        ts,
+        a: ItemId::parse(&item_name(a)).unwrap(),
+        b: ItemId::parse(&item_name(b)).unwrap(),
+        ratio_left: l,
+        ratio_right: r,
+        body: "synthetic".into(),
+        principal: "bench".into(),
+        delegate: None,
+        thread_tag: "bench".into(),
+    }
+}
+
+#[derive(Clone, Copy)]
+enum Topo {
+    /// `0 > 1 > 2 > …`, every vote at a fixed ratio. Ground-truth order known.
+    Chain {
+        ratio: i32,
+    },
+    /// Chain with per-edge varying ratios (what `benches/common` builds).
+    ChainVaried,
+    Star,
+    Clique,
+    Sparse {
+        degree: usize,
+    },
+    Components {
+        count: usize,
+        size: usize,
+    },
+}
+
+fn pairs(t: Topo, n: usize) -> Vec<(usize, usize, i32)> {
+    let mut out = Vec::new();
+    match t {
+        Topo::Chain { ratio } => {
+            for i in 0..n.saturating_sub(1) {
+                out.push((i, i + 1, ratio));
+            }
+        }
+        Topo::ChainVaried => {
+            for i in 0..n.saturating_sub(1) {
+                out.push((i, i + 1, 2 + (i % 5) as i32));
+            }
+        }
+        Topo::Star => {
+            for i in 1..n {
+                out.push((0, i, 2 + (i % 5) as i32));
+            }
+        }
+        Topo::Clique => {
+            let mut k = 0;
+            for i in 0..n {
+                for j in (i + 1)..n {
+                    out.push((i, j, 2 + (k % 5)));
+                    k += 1;
+                }
+            }
+        }
+        Topo::Sparse { degree } => {
+            let mut rng = Rng::new(0xC0FFEE ^ n as u64);
+            for i in 0..n.saturating_sub(1) {
+                out.push((i, i + 1, 2));
+            }
+            for k in 0..(degree.saturating_sub(2) * n / 2) {
+                let a = rng.below(n);
+                let b = rng.below(n);
+                if a != b {
+                    out.push((a, b, 2 + (k % 5) as i32));
+                }
+            }
+        }
+        Topo::Components { count, size } => {
+            for c in 0..count {
+                let base = c * size;
+                for i in 0..size.saturating_sub(1) {
+                    out.push((base + i, base + i + 1, 2));
+                }
+            }
+        }
+    }
+    out
+}
+
+fn node_count(t: Topo, n: usize) -> usize {
+    match t {
+        Topo::Components { count, size } => count * size,
+        _ => n,
+    }
+}
+
+fn build_chain(t: Topo, n: usize) -> RankChain {
+    let total = node_count(t, n);
+    let mut g = GroupState::new();
+    for i in 0..total {
+        g.ensure_item_pub(&item_name(i));
+    }
+    for (k, (i, j, ratio)) in pairs(t, total).into_iter().enumerate() {
+        g.apply_vote(vote(i, j, ratio, 1, k as i64));
+    }
+    chain_from_edges(total, g.edges.iter().map(|(&k, &w)| (k, w)))
+}
+
+// ---------------------------------------------------------------------------
+// Accuracy metrics
+// ---------------------------------------------------------------------------
+
+/// Kendall-tau distance between the order induced by `a` and by `b`, as a
+/// fraction of all pairs. 0 = identical ranking, 0.5 = as good as random.
+fn kendall_distance(a: &[f64], b:

… preview truncated; 93,354 characters omitted