You are a constitutional council ranking individual git commits for ownership allocation.

Compare these two commits. Decide which contributed more lasting value to the project.

Judge substance, not spectacle:
- Prefer correct, lasting design and real bugfixes over churn, formatting, renames, or generated noise.
- Prefer clarity and necessity over sheer line count. A small precise change can beat a large diffuse one.
- Do not favor a side merely because its patch is longer or noisier.
- Weight what the change does for the project, not the contributor's name.

Return ONLY a JSON object: {"winner": "A" or "B", "ratio": "N:M", "explanation": "..."}
The explanation must cite concrete differences in the patches (1-3 sentences).

Side A — contributor: tommy-mor
Side A — commit message:
[15e1037a] url stuff

Side A — unified diff (full patch):
diff --git a/server/src/url_rules/graph.rs b/server/src/url_rules/graph.rs
new file mode 100644
index 0000000000000000000000000000000000000000..f7ac0f9a551a1727cb2f9294778c283b9885b147
--- /dev/null
+++ b/server/src/url_rules/graph.rs
@@ -0,0 +1,831 @@
+//! Semantic URL graph: DFA traversal on host + path, query in context, generic fallback.
+
+use std::collections::HashMap;
+use std::sync::OnceLock;
+
+use url::Url;
+
+use super::graph_builder::GraphBuilder;
+use super::parse::{normalize_match_host, strip_tracking_query, UrlParts};
+
+#[derive(Debug, Clone, Default)]
+pub struct Context {
+    pub vars: HashMap<String, String>,
+    pub query: HashMap<String, String>,
+}
+
+pub type CanonicalFn = fn(&Context) -> Option<String>;
+
+#[derive(Clone, Copy)]
+pub enum EdgePattern {
+    Literal(&'static str),
+    Variable(&'static str),
+    /// Absorb any trailing segment without leaving this node (e.g. post title slug).
+    AbsorbAny,
+    /// Absorb segment when `cond(seg)` (e.g. subreddit listing suffix).
+    AbsorbIf(fn(&str) -> bool),
+}
+
+pub struct Edge {
+    pub pattern: EdgePattern,
+    pub target: &'static str,
+}
+
+pub struct Node {
+    pub edges: Vec<Edge>,
+    pub canonical: CanonicalFn,
+    pub parent: Option<&'static str>,
+}
+
+impl Node {
+    pub(crate) fn empty() -> Self {
+        Self {
+            edges: Vec::new(),
+            canonical: |_| None,
+            parent: None,
+        }
+    }
+}
+
+pub struct Graph {
+    pub nodes: HashMap<&'static str, Node>,
+}
+
+static GRAPH: OnceLock<Graph> = OnceLock::new();
+
+pub fn graph() -> &'static Graph {
+    GRAPH.get_or_init(build_graph)
+}
+
+impl Graph {
+    pub fn resolve_canonical(&self, parts: &UrlParts) -> Option<String> {
+        let mut query = parts.query.clone();
+        strip_tracking_query(&mut query);
+        let mut ctx = Context {
+            vars: HashMap::new(),
+            query,
+        };
+
+        if let Some(node_id) = self.traverse(parts, &mut ctx) {
+            if let Some(canon) = (self.nodes.get(node_id)?.canonical)(&ctx) {
+                return Some(canon);
+            }
+        }
+        Some(generic_canonical(parts))
+    }
+
+    pub fn breadcrumbs(&self, parts: &UrlParts) -> Vec<String> {
+        let mut query = parts.query.clone();
+        strip_tracking_query(&mut query);
+        let mut ctx = Context {
+            vars: HashMap::new(),
+            query,
+        };
+
+        if let Some(mut node_id) = self.traverse(parts, &mut ctx) {
+            let mut paths = Vec::new();
+            loop {
+                let node = match self.nodes.get(node_id) {
+                    Some(n) => n,
+                    None => break,
+                };
+                if let Some(url) = (node.canonical)(&ctx) {
+                    if paths.last() != Some(&url) {
+                        paths.push(url);
+                    }
+                }
+                match node.parent {
+                    Some(p) => node_id = p,
+                    None => break,
+                }
+            }
+            paths.reverse();
+            if !paths.is_empty() {
+                return paths;
+            }
+        }
+        generic_breadcrumbs(parts)
+    }
+
+    fn traverse(&self, parts: &UrlParts, ctx: &mut Context) -> Option<&'static str> {
+        let host = parts.match_host();
+        let mut node_id = match host.as_str() {
+            "reddit.com" => "reddit_root",
+            "youtube.com" => "youtube_root",
+            "youtu.be" => "youtu_be_entry",
+            _ => return None,
+        };
+
+        let segs: Vec<&str> = parts.path_segments.iter().map(String::as_str).collect();
+        let mut i = 0;
+        while i < segs.len() {
+            let seg = segs[i];
+            match self.follow_edge(node_id, seg, ctx) {
+                Ok(next) => {
+                    node_id = next;
+                    i += 1;
+                }
+                Err(()) => {
+                    if self.try_absorb(node_id, seg) {
+                        i += 1;
+                        continue;
+                    }
+                    return None;
+                }
+            }
+        }
+        Some(node_id)
+    }
+
+    fn follow_edge(
+        &self,
+        node_id: &'static str,
+        seg: &str,
+        ctx: &mut Context,
+    ) -> Result<&'static str, ()> {
+        let node = self.nodes.get(node_id).ok_or(())?;
+        for edge in &node.edges {
+            match edge.pattern {
+                EdgePattern::Literal(lit) if lit == seg => return Ok(edge.target),
+                EdgePattern::Variable(name) => {
+                    ctx.vars.insert(name.to_string(), seg.to_string());
+                    return Ok(edge.target);
+                }
+                EdgePattern::AbsorbAny
+                | EdgePattern::AbsorbIf(_)
+                | EdgePattern::Literal(_)
+                | EdgePattern::Variable(_) => {}
+            }
+        }
+        Err(())
+    }
+
+    fn try_absorb(&self, node_id: &'static str, seg: &str) -> bool {
+        let node = match self.nodes.get(node_id) {
+            Some(n) => n,
+            None => return false,
+        };
+        for edge in &node.edges {
+            match edge.pattern {
+                EdgePattern::AbsorbAny => return true,
+                EdgePattern::AbsorbIf(cond) if cond(seg) => return true,
+                EdgePattern::AbsorbIf(_) | EdgePattern::Literal(_) | EdgePattern::Variable(_) => {}
+            }
+        }
+        false
+    }
+
+    /// Test hook: terminal graph node and captured context after traversal.
+    #[cfg(test)]
+    pub fn traverse_terminal(&self, parts: &UrlParts) -> Option<(&'static str, Context)> {
+        let mut query = parts.query.clone();
+        strip_tracking_query(&mut query);
+        let mut ctx = Context {
+            vars: HashMap::new(),
+            query,
+        };
+        let node = self.traverse(parts, &mut ctx)?;
+        Some((node, ctx))
+    }
+}
+
+fn is_reddit_listing_suffix(seg: &str) -> bool {
+    matches!(seg, "hot" | "top" | "new" | "rising" | "controversial")
+}
+
+/// Percent-encode a path or query fragment so `&`, `?`, etc. cannot break URL structure.
+fn enc(s: &str) -> String {
+    urlencoding::encode(s).into_owned()
+}
+
+// --- Canonical formatters ---
+
+fn canon_reddit_root(_: &Context) -> Option<String> {
+    Some("https://reddit.com".to_string())
+}
+
+fn canon_reddit_r_hub(_: &Context) -> Option<String> {
+    Some("https://reddit.com/r".to_string())
+}
+
+fn canon_reddit_subreddit(ctx: &Context) -> Option<String> {
+    let sub = ctx.vars.get("subreddit")?;
+    Some(format!(
+        "https://reddit.com/r/{}",
+        enc(&sub.to_ascii_lowercase())
+    ))
+}
+
+fn canon_reddit_post(ctx: &Context) -> Option<String> {
+    let sub = ctx.vars.get("subreddit")?.to_ascii_lowercase();
+    let id = ctx.vars.get("post_id")?;
+    Some(format!(
+        "https://reddit.com/r/{}/comments/{}",
+        enc(&sub),
+        enc(id)
+    ))
+}
+
+fn canon_youtube_root(_: &Context) -> Option<String> {
+    Some("https://youtube.com".to_string())
+}
+
+fn canon_youtube_watch(ctx: &Context) -> Option<String> {
+    let v = ctx
+        .query
+        .get("v")
+        .or_else(|| ctx.vars.get("video_id"))?;
+    Some(format!("https://youtube.com/watch?v={}", enc(v)))
+}
+
+fn canon_youtu_be(ctx: &Context) -> Option<String> {
+    let v = ctx.vars.get("vid_id")?;
+    Some(format!("https://youtube.com/watch?v={}", enc(v)))
+}
+
+pub fn build_graph() -> Graph {
+    GraphBuilder::new()
+        .node("reddit_root")
+        .canonical(canon_reddit_root)
+        .edge(EdgePattern::Literal("r"), "reddit_r_hub")
+        .node("reddit_r_hub")
+        .parent("reddit_root")
+        .canonical(canon_reddit_r_hub)
+        .edge(EdgePattern::Variable("subreddit"), "reddit_subreddit")
+        .node("reddit_subreddit")
+        .parent("reddit_r_hub")
+        .canonical(canon_reddit_subreddit)
+        .edge(
+            EdgePattern::AbsorbIf(is_reddit_listing_suffix),
+            "reddit_subreddit",
+        )
+        .edge(EdgePattern::Literal("comments"), "reddit_comments_gate")
+        .node("reddit_comments_gate")
+        .parent("reddit_subreddit")
+        .canonical(canon_reddit_subreddit)
+        .edge(EdgePattern::Variable("post_id"), "reddit_post")
+        .node("reddit_post")
+        .parent("reddit_subreddit")
+        .canonical(canon_reddit_post)
+        .edge(EdgePattern::AbsorbAny, "reddit_post")
+        .node("youtube_root")
+        .canonical(canon_youtube_root)
+        .edge(EdgePattern::Literal("watch"), "youtube_watch")
+        .edge(EdgePattern::Literal("shorts"), "youtube_shorts_gate")
+        .node("youtube_watch")
+        .parent("youtube_root")
+        .canonical(canon_youtube_watch)
+        .node("youtube_shorts_gate")
+        .parent("youtube_root")
+        .canonical(canon_youtube_root)
+        .edge(EdgePattern::Variable("video_id"), "youtube_watch")
+        .node("youtu_be_entry")
+        .canonical(canon_youtube_root)
+        .edge(EdgePattern::Variable("vid_id"), "youtu_be_video")
+        .node("youtu_be_video")
+        .parent("youtube_root")
+        .canonical(canon_youtu_be)
+        .build()
+}
+
+// --- Generic internet fallback ---
+
+pub fn generic_canonical(parts: &UrlParts) -> String {
+    let host = normalize_match_host(&parts.host);
+    let path_segments: Vec<String> = parts.path_segments.clone();
+    let mut query = parts.query.clone();
+    strip_tracking_query(&mut query);
+
+    let mut url = if path_segments.is_empty() {
+        Url::parse(&format!("https://{host}"))
+            .unwrap_or_else(|_| Url::parse("https://invalid").unwrap())
+    } else {
+        let path = format!("/{}", path_segments.join("/"));
+        Url::parse(&format!("https://{host}{path}"))
+            .unwrap_or_else(|_| Url::parse("https://invalid").unwrap())
+    };
+
+    if !query.is_empty() {
+        let mut pairs: Vec<_> = query.iter().collect();
+        pairs.sort_by(|a, b| a.0.cmp(b.0));
+        url.query_pairs_mut().clear();
+        for (k, v) in pairs {
+            url.query_pairs_mut().append_pair(k, v);
+        }
+    }
+
+    let mut s = url.to_string();
+    if path_segments.is_empty() {
+        s = s.trim_end_matches('/').to_string();
+    }
+    s
+}
+
+pub fn generic_breadcrumbs(parts: &UrlParts) -> Vec<String> {
+    let host = normalize_match_host(&parts.host);
+    let n = parts.path_segments.len();
+    let mut out = Vec::new();
+
+    let base = generic_canonical(&UrlParts {
+        scheme: "https".to_string(),
+        host: host.clone(),
+        path_segments: vec![],
+        query: HashMap::new(),
+    });
+    out.push(base);
+
+    for i in 0..n {
+        let segs: Vec<String> = parts.path_segments[..=i].to_vec();
+        let url = generic_canonical(&UrlParts {
+            scheme: "https".to_string(),
+            host: host.clone(),
+            path_segments: segs,
+            query: HashMap::new(),
+        });
+        if out.last() != Some(&url) {
+            out.push(url);
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::url_rules::parse::test_parts;
+
+    fn g() -> &'static Graph {
+        graph()
+    }
+
+    fn canon(parts: &UrlParts) -> String {
+        g().resolve_canonical(parts).unwrap()
+    }
+
+    fn crumbs(parts: &UrlParts) -> Vec<String> {
+        g().breadcrumbs(parts)
+    }
+
+    fn terminal(parts: &UrlParts) -> Option<&'static str> {
+        g().traverse_terminal(parts).map(|(n, _)| n)
+    }
+
+    fn vars(parts: &UrlParts) -> HashMap<String, String> {
+        g().traverse_terminal(parts)
+            .map(|(_, c)| c.vars)
+            .unwrap_or_default()
+    }
+
+    #[test]
+    fn youtu_be_malicious_segment_encoded_not_injected() {
+        let p = test_parts("youtu.be", &["abc&t=1"], &[]);
+        assert_eq!(canon(&p), "https://youtube.com/watch?v=abc%26t%3D1");
+        assert!(!canon(&p).contains("abc&t=1"));
+    }
+
+    #[test]
+    fn youtube_query_v_encoded() {
+        let p = test_parts("youtube.com", &["watch"], &[("v", "a&b=c")]);
+        assert_eq!(canon(&p), "https://youtube.com/watch?v=a%26b%3Dc");
+    }
+
+    #[test]
+    fn absorb_patterns_live_on_edges_not_in_engine() {
+        let g = build_graph();
+        let sub = g.nodes.get("reddit_subreddit").unwrap();
+        assert!(sub
+            .edges
+            .iter()
+            .any(|e| matches!(e.pattern, EdgePattern::AbsorbIf(_))));
+        let post = g.nodes.get("reddit_post").unwrap();
+        assert!(post
+            .edges
+            .iter()
+            .any(|e| matches!(e.pattern, EdgePattern::AbsorbAny)));
+    }
+
+    #[test]
+    fn builder_rejects_missing_parent() {
+        let result = std::panic::catch_unwind(|| {
+            GraphBuilder::new()
+                .node("orphan")
+                .parent("nonexistent_parent")
+                .build();
+        });
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn generic_canonical_forces_https_and_strips_www() {
+        let p = test_parts("www.example.com", &["blog", "post"], &[]);
+        assert_eq!(canon(&p), "https://example.com/blog/post");
+    }
+
+    #[test]
+    fn generic_canonical_sorts_query_keys() {
+        let p = test_parts("example.com", &["search"], &[("q", "rust"), ("page", "2")]);
+        assert_eq!(canon(&p), "https://example.com/search?page=2&q=rust");
+    }
+
+    #[test]
+    fn generic_canonical_strips_tracking_from_query() {
+        let p = test_parts(
+            "news.ycombinator.com",
+            &["item"],
+            &[("id", "1"), ("utm_medium", "social")],
+        );
+        assert_eq!(canon(&p), "https://news.ycombinator.com/item?id=1");
+    }
+
+    #[test]
+    fn generic_breadcrumbs_cumulative_path() {
+        let p = test_parts("paulgraham.com", &["articles", "lisp.html"], &[]);
+        assert_eq!(
+            crumbs(&p),
+            vec![
+                "https://paulgraham.com",
+                "https://paulgraham.com/articles",
+                "https://paulgraham.com/articles/lisp.html"
+            ]
+        );
+    }
+
+    #[test]
+    fn generic_breadcrumbs_domain_only() {
+        let p = test_parts("example.com", &[], &[]);
+        assert_eq!(crumbs(&p), vec!["https://example.com"]);
+    }
+
+    #[test]
+    fn unknown_host_uses_generic_not_graph() {
+        let p = test_parts("hackernews.com", &["item", "123"], &[]);
+        assert_eq!(terminal(&p), None);
+        assert_eq!(canon(&p), "https://hackernews.com/item/123");
+    }
+
+    #[test]
+    fn traverse_captures_subreddit_variable() {
+        let p = test_parts("reddit.com", &["r", "Rust"], &[]);
+        assert_eq!(terminal(&p), Some("reddit_subreddit"));
+        assert_eq!(vars(&p).get("subreddit").map(String::as_str), Some("Rust"));
+    }
+
+    #[test]
+    fn traverse_captures_post_id() {
+        let p = test_parts("reddit.com", &["r", "aww", "comments", "abc123"], &[]);
+        assert_eq!(terminal(&p), Some("reddit_post"));
+        assert_eq!(vars(&p).get("post_id").map(String::as_str), Some("abc123"));
+    }
+
+    #[test]
+    fn traverse_absorbs_listing_suffix_stays_on_subreddit() {
+        let p = test_parts("reddit.com", &["r", "rust", "hot"], &[]);
+        assert_eq!(terminal(&p), Some("reddit_subreddit"));
+        assert_eq!(canon(&p), "https://reddit.com/r/rust");
+    }
+
+    #[test]
+    fn traverse_absorbs_all_listing_suffixes() {
+        for suffix in ["hot", "top", "new", "rising", "controversial"] {
+            let p = test_parts("reddit.com", &["r", "test", suffix], &[]);
+            assert_eq!(terminal(&p), Some("reddit_subreddit"), "suffix {suffix}");
+            assert_eq!(canon(&p), "https://reddit.com/r/test", "suffix {suffix}");
+        }
+    }
+
+    #[test]
+    fn traverse_absorbs_post_title_slug() {
+        let p = test_parts(
+            "reddit.com",
+            &["r", "rust", "comments", "aaa", "my_great_post_title"],
+            &[],
+        );
+        assert_eq!(terminal(&p), Some("reddit_post"));
+        assert_eq!(canon(&p), "https://reddit.com/r/rust/comments/aaa");
+    }
+
+    #[test]
+    fn traverse_unknown_segment_falls_back_to_generic() {
+        let p = test_parts("reddit.com", &["r", "rust", "wiki", "faq"], &[]);
+        assert_eq!(terminal(&p), None);
+        assert_eq!(canon(&p), "https://reddit.com/r/rust/wiki/faq");
+    }
+
+    #[test]
+    fn traverse_youtube_watch_requires_v_in_query() {
+        let p = test_parts("youtube.com", &["watch"], &[("v", "xyz")]);
+        assert_eq!(terminal(&p), Some("youtube_watch"));
+    }
+
+    #[test]
+    fn traverse_youtu_be_captures_vid_id() {
+        let p = test_parts("youtu.be", &["dQw4w9WgXcQ"], &[]);
+        assert_eq!(terminal(&p), Some("youtu_be_video"));
+        assert_eq!(vars(&p).get("vid_id").map(String::as_str), Some("dQw4w9WgXcQ"));
+    }
+
+    #[test]
+    fn traverse_shorts_sets_video_id_var() {
+        let p = test_parts("youtube.com", &["shorts", "abc99"], &[]);
+        assert_eq!(terminal(&p), Some("youtube_watch"));
+        assert_eq!(vars(&p).get("video_id").map(String::as_str), Some("abc99"));
+    }
+
+    #[test]
+    fn reddit_domain_canonical() {
+        let p = test_parts("reddit.com", &[], &[]);
+        assert_eq!(canon(&p), "https://reddit.com");
+    }
+
+    #[test]
+    fn reddit_r_hub_canonical() {
+        let p = test_parts("reddit.com", &["r"], &[]);
+        assert_eq!(terminal(&p), Some("reddit_r_hub"));
+        assert_eq!(canon(&p), "https://reddit.com/r");
+    }
+
+    #[test]
+    fn reddit_subreddit_lowercases_name() {
+        let p = test_parts("reddit.com", &["r", "AmITheAsshole"], &[]);
+        assert_eq!(canon(&p), "https://reddit.com/r/amitheasshole");
+    }
+
+    #[test]
+    fn reddit_host_aliases_old_new_www() {
+        for host in ["old.reddit.com", "new.reddit.com", "www.reddit.com"] {
+            let p = test_parts(host, &["r", "rust"], &[]);
+            assert_eq!(canon(&p), "https://reddit.com/r/rust", "host {host}");
+        }
+    }
+
+    #[test]
+    fn reddit_post_strips_slug_and_query() {
+        let p = test_parts(
+            "old.reddit.com",
+            &["r", "Rust", "comments", "1abc", "title_slug_here"],
+            &[("sort", "new")],
+        );
+        assert_eq!(canon(&p), "https://reddit.com/r/rust/comments/1abc");
+    }
+
+    #[test]
+    fn reddit_post_multiple_slugs_absorbed() {
+        let p = test_parts(
+            "reddit.com",
+            &["r", "x", "comments", "id1", "slug1", "extra"],
+            &[],
+        );
+        assert_eq!(canon(&p), "https://reddit.com/r/x/comments/id1");
+    }
+
+    #[test]
+    fn reddit_listing_with_query_only() {
+        let p = test_parts("www.reddit.com", &["r", "programming"], &[("sort", "top")]);
+        assert_eq!(canon(&p), "https://reddit.com/r/programming");
+    }
+
+    #[test]
+    fn reddit_subreddit_breadcrumbs_include_r_hub() {
+        let p = test_parts("reddit.com", &["r", "movies"], &[]);
+        assert_eq!(
+            crumbs(&p),
+            vec![
+                "https://reddit.com",
+                "https://reddit.com/r",
+                "https://reddit.com/r/movies"
+            ]
+        );
+    }
+
+    #[test]
+    fn reddit_post_breadcrumbs_skip_comments_node() {
+        let p = test_parts("reddit.com", &["r", "aww", "comments", "1trnvdl"], &[]);
+        let c = crumbs(&p);
+        assert!(!c.iter().any(|u| u.ends_with("/comments")));
+        assert_eq!(
+            c.last().map(String::as_str),
+            Some("https://reddit.com/r/aww/comments/1trnvdl")
+        );
+        assert!(c.contains(&"https://reddit.com/r/aww".to_string()));
+    }
+
+    #[test]
+    fn reddit_post_parent_is_subreddit_not_comments() {
+        let p = test_parts("reddit.com", &["r", "aww", "comments", "1trnvdl"], &[]);
+        let c = crumbs(&p);
+        let parent = c.get(c.len() - 2).unwrap();
+        assert_eq!(parent, "https://reddit.com/r/aww");
+    }
+
+    #[test]
+    fn reddit_domain_parent_is_none_in_breadcrumb_chain() {
+        let p = test_parts("reddit.com", &[], &[]);
+        assert_eq!(crumbs(&p), vec!["https://reddit.com"]);
+    }
+
+    #[test]
+    fn youtube_watch_canonical_uses_v_only() {
+        let p = test_parts("youtube.com", &["watch"], &[("v", "abc"), ("t", "99")]);
+        assert_eq!(canon(&p), "https://youtube.com/watch?v=abc");
+    }
+
+    #[test]
+    fn youtube_query_order_independent() {
+        let a = test_parts("youtube.com", &["watch"], &[("v", "abc"), ("t", "4")]);
+        let b = test_parts("youtube.com", &["watch"], &[("t", "4"), ("v", "abc")]);
+        assert_eq!(canon(&a), canon(&b));
+    }
+
+    #[test]
+    fn youtube_host_aliases() {
+        for host in ["www.youtube.com", "m.youtube.com"] {
+            let p = test_parts(host, &["watch"], &[("v", "x")]);
+            assert_eq!(canon(&p), "https://youtube.com/watch?v=x", "host {host}");
+        }
+    }
+
+    #[test]
+    fn youtube_shorts_canonical_matches_watch() {
+        let shorts = test_parts("youtube.com", &["shorts", "vid123"], &[]);
+        let watch = test_parts("youtube.com", &["watch"], &[("v", "vid123")]);
+        assert_eq!(canon(&shorts), canon(&watch));
+        assert_eq!(canon(&shorts), "https://youtube.com/watch?v=vid123");
+    }
+
+    #[test]
+    fn youtu_be_matches_youtube_watch() {
+        let be = test_parts("youtu.be", &["dQw4w9WgXcQ"], &[]);
+        let watch = test_parts("youtube.com", &["watch"], &[("v", "dQw4w9WgXcQ")]);
+        assert_eq!(canon(&be), canon(&watch));
+    }
+
+    #[test]
+    fn youtube_breadcrumbs_domain_then_watch() {
+        let p = test_parts("youtube.com", &["watch"], &[("v", "abc")]);
+        assert_eq!(
+            crumbs(&p),
+            vec!["https://youtube.com", "https://youtube.com/watch?v=abc"]
+        );
+    }
+
+    #[test]
+    fn youtu_be_breadcrumbs_include_youtube_domain() {
+        let p = test_parts("youtu.be", &["abc"], &[]);
+        let c = crumbs(&p);
+        assert_eq!(c.first().map(String::as_str), Some("https://youtube.com"));
+        assert_eq!(
+            c.last().map(String::as_str),
+            Some("https://youtube.com/watch?v=abc")
+        );
+    }
+
+    #[test]
+    fn parsed_urls_match_hand_built_parts() {
+        let raw = "https://www.reddit.com/r/rust/comments/aaa/title/?utm=x";
+        let parsed = UrlParts::parse(raw).unwrap();
+        let hand = test_parts(
+            "www.reddit.com",
+            &["r", "rust", "comments", "aaa", "title"],
+            &[("utm", "x")],
+        );
+        assert_eq!(canon(&parsed), canon(&hand));
+    }
+
+    #[test]
+    fn equivalence_cluster_youtube_formats() {
+        let urls = [
+            "https://youtu.be/abc123",
+            "https://www.youtube.com/watch?v=abc123",
+            "https://youtube.com/watch?v=abc123&t=1",
+            "https://m.youtube.com/watch?t=1&v=abc123",
+        ];
+        let canonical: Vec<_> = urls
+            .iter()
+            .map(|u| canon(&UrlParts::parse(u).unwrap()))
+            .collect();
+        assert!(canonical.iter().all(|c| *c == "https://youtube.com/watch?v=abc123"));
+    }
+
+    #[test]
+    fn equivalence_cluster_reddit_post_formats() {
+        let urls = [
+            "https://old.reddit.com/r/Rust/comments/aaa/slug/",
+            "reddit.com/r/rust/comments/aaa/other_slug",
+            "https://reddit.com/r/RUST/comments/aaa",
+        ];
+        let canonical: Vec<_> = urls
+            .iter()
+            .map(|u| canon(&UrlParts::parse(u).unwrap()))
+            .collect();
+        assert!(
+            canonical
+                .iter()
+                .all(|c| *c == "https://reddit.com/r/rust/comments/aaa")
+        );
+    }
+
+    #[test]
+    fn graph_nodes_all_have_valid_parent_links() {
+        let g = build_graph();
+        for (id, node) in &g.nodes {
+            if let Some(parent) = node.parent {
+                assert!(g.nodes.contains_key(parent), "node {id} parent {parent}");
+            }
+        }
+    }
+
+    #[test]
+    fn graph_terminal_canonical_always_succeeds_for_reddit_paths() {
+        let cases: &[(&[&str], &str)] = &[
+            (&["r", "rust"], "https://reddit.com/r/rust"),
+            (
+                &["r", "rust", "comments", "x"],
+                "https://reddit.com/r/rust/comments/x",
+            ),
+        ];
+        for (segs, want) in cases {
+            let p = test_parts("reddit.com", segs, &[]);
+            assert_eq!(canon(&p), *want);
+        }
+    }
+
+    #[test]
+    fn breadcrumb_parent_walk_matches_parent_url_semantics() {
+        let p = test_parts("reddit.com", &["r", "aww", "comments", "id1"], &[]);
+        let c = crumbs(&p);
+        assert_eq!(c.len(), 4);
+        assert_eq!(
+            c.get(c.len() - 2).map(String::as_str),
+            Some("https://reddit.com/r/aww")
+        );
+    }
+
+    #[test]
+    fn youtube_watch_without_v_falls_back_to_generic() {
+        let p = test_parts("youtube.com", &["watch"], &[]);
+        assert_eq!(terminal(&p), Some("youtube_watch"));
+        assert_eq!(canon(&p), "https://youtube.com/watch");
+    }
+
+    #[test]
+    fn reddit_only_comments_path_stops_at_gate() {
+        let p = test_parts("reddit.com", &["r", "rust", "comments"], &[]);
+        assert_eq!(terminal(&p), Some("reddit_comments_gate"));
+        assert_eq!(canon(&p), "https://reddit.com/r/rust");
+    }
+
+    #[test]
+    fn generic_deep_path_many_segments() {
+        let segs: Vec<&str> = (0..10)
+            .map(|i| match i {
+                0 => "a",
+                1 => "b",
+                2 => "c",
+                3 => "d",
+                4 => "e",
+                5 => "f",
+                6 => "g",
+                7 => "h",
+                8 => "i",
+                _ => "j",
+            })
+            .collect();
+        let p = test_parts("site.com", &segs, &[]);
+        assert_eq!(crumbs(&p).len(), 11);
+    }
+
+    #[test]
+    fn traverse_literal_r_required_for_subreddit() {
+        let p = test_parts("reddit.com", &["rust"], &[]);
+        assert_eq!(terminal(&p), None);
+    }
+
+    #[test]
+    fn http_scheme_upgraded_via_generic_fallback_host() {
+        let parsed = UrlParts::parse("http://example.com/page").unwrap();
+        assert_eq!(canon(&parsed), "https://example.com/page");
+    }
+
+    #[test]
+    fn each_graph_node_canonical_is_invokable() {
+        let g = build_graph();
+        let empty = Context::default();
+        for (id, node) in &g.nodes {
+            let _ = (node.canonical)(&empty);
+            let _ = id;
+        }
+    }
+
+    #[test]
+    fn reddit_double_listing_suffix_both_absorbed() {
+        let p = test_parts("reddit.com", &["r", "rust", "hot", "new"], &[]);
+        assert_eq!(terminal(&p), Some("reddit_subreddit"));
+        assert_eq!(canon(&p), "https://reddit.com/r/rust");
+    }
+
+    #[test]
+    fn youtu_be_empty_path_stays_at_entry() {
+        let p = test_parts("youtu.be", &[], &[]);
+        assert_eq!(terminal(&p), Some("youtu_be_entry"));
+    }
+}
diff --git a/server/src/url_rules/graph_builder.rs b/server/src/url_rules/graph_builder.rs
new file mode 100644
index 0000000000000000000000000000000000000000..243204ad5ca514fff459057956080cacb5bf38e2
--- /dev/null
+++ b/server/src/url_rules/graph_builder.rs
@@ -0,0 +1,73 @@
+//! Declarative construction of the URL graph with build-time link validation.
+
+use std::collections::HashMap;
+
+use super::graph::{CanonicalFn, Edge, EdgePattern, Graph, Node};
+
+pub struct GraphBuilder {
+    nodes: HashMap<&'static str, Node>,
+    current: Option<&'static str>,
+}
+
+impl GraphBuilder {
+    pub fn new() -> Self {
+        Self {
+            nodes: HashMap::new(),
+            current: None,
+        }
+    }
+
+    pub fn node(mut self, id: &'static str) -> Self {
+        self.nodes.entry(id).or_insert_with(Node::empty);
+        self.current = Some(id);
+        self
+    }
+
+    pub fn canonical(mut self, f: CanonicalFn) -> Self {
+        let id = self.current.expect("canonical() without node()");
+        self.nodes.get_mut(id).expect("node missing").canonical = f;
+        self
+    }
+
+    pub fn parent(mut self, parent_id: &'static str) -> Self {
+        let id = self.current.expect("parent() without node()");
+        self.nodes.get_mut(id).expect("node missing").parent = Some(parent_id);
+        self
+    }
+
+    pub fn edge(mut self, pattern: EdgePattern, target: &'static str) -> Self {
+        let id = self.current.expect("edge() without node()");
+        self.nodes
+            .get_mut(id)
+            .expect("node missing")
+            .edges
+            .push(Edge { pattern, target });
+        self
+    }
+
+    pub fn build(self) -> Graph {
+        for (id, node) in &self.nodes {
+            if let Some(parent) = node.parent {
+                assert!(
+                    self.nodes.contains_key(parent),
+                    "node {id}: parent {parent} does not exist"
+                );
+            }
+            for edge in &node.edges {
+                if !matches!(
+                    edge.pattern,
+                    EdgePattern::AbsorbAny | EdgePattern::AbsorbIf(_)
+                ) {
+                    assert!(
+                        self.nodes.contains_key(edge.target),
+                        "node {id}: edge target {} does not exist",
+                        edge.target
+                    );
+                }
+            }
+        }
+        Graph {
+            nodes: self.nodes,
+        }
+    }
+}
diff --git a/server/src/url_rules/mod.rs b/server/src/url_rules/mod.rs
index 9e1445346ce77a49dd6a7e7713bf9c57aef353cc..ba4ac662acc7bd4d50ee34613eb7eb6fccbfb17b 100644
--- a/server/src/url_rules/mod.rs
+++ b/server/src/url_rules/mod.rs
@@ -1,6 +1,7 @@
 //! URL canonicalization and hierarchy via a semantic graph (DFA + generic fallback).
 
 mod graph;
+mod graph_builder;
 mod parse;
 mod registry;
 
diff --git a/server/src/url_rules/parse.rs b/server/src/url_rules/parse.rs
new file mode 100644
index 0000000000000000000000000000000000000000..19d0c82feb718817168b8b445fcbfa39cf6aa6ba
--- /dev/null
+++ b/server/src/url_rules/parse.rs
@@ -0,0 +1,169 @@
+//! Parse raw strings into host, path segments, and query (order-independent).
+
+use std::collections::HashMap;
+
+use url::Url;
+
+#[derive(Debug, Clone)]
+pub struct UrlParts {
+    pub scheme: String,
+    pub host: String,
+    pub path_segments: Vec<String>,
+    pub query: HashMap<String, String>,
+}
+
+impl UrlParts {
+    pub fn parse(raw: &str) -> Option<Self> {
+        let trimmed = raw.trim();
+        if trimmed.is_empty() {
+            return None;
+        }
+
+        let with_scheme = if trimmed.contains("://") {
+            trimmed.to_string()
+        } else if trimmed.starts_with("r/") || trimmed.starts_with("/r/") {
+            let rest = trimmed.trim_start_matches('/').trim_start_matches("r/");
+            format!("https://reddit.com/r/{rest}")
+        } else if trimmed.contains('.') && !trimmed.starts_with('/') {
+            format!("https://{trimmed}")
+        } else {
+            trimmed.to_string()
+        };
+
+        let url = Url::parse(&with_scheme).ok()?;
+        let host = url.host_str()?.to_string();
+        let path_segments: Vec<String> = url
+            .path_segments()
+            .map(|segs| segs.filter(|s| !s.is_empty()).map(str::to_string).collect())
+            .unwrap_or_default();
+
+        let mut query = HashMap::new();
+        for (k, v) in url.query_pairs() {
+            query.insert(k.into_owned(), v.into_owned());
+        }
+
+        Some(Self {
+            scheme: url.scheme().to_string(),
+            path_segments,
+            query,
+            host,
+        })
+    }
+
+    /// Host normalized for graph entry matching (lowercase, aliases).
+    pub fn match_host(&self) -> String {
+        normalize_match_host(&self.host)
+    }
+}
+
+pub fn normalize_match_host(host: &str) -> String {
+    let h = host
+        .strip_prefix("www.")
+        .unwrap_or(host)
+        .to_ascii_lowercase();
+    match h.as_str() {
+        "old.reddit.com" | "new.reddit.com" => "reddit.com".to_string(),
+        "m.youtube.com" => "youtube.com".to_string(),
+        _ => h,
+    }
+}
+
+pub fn strip_tracking_query(query: &mut HashMap<String, String>) {
+    query.retain(|k, _| {
+        let lower = k.to_ascii_lowercase();
+        !(lower.starts_with("utm_")
+            || matches!(
+                lower.as_str(),
+                "fbclid" | "gclid" | "ref" | "ref_src" | "ref_source" | "mc_cid" | "mc_eid"
+            ))
+    });
+}
+
+#[cfg(test)]
+pub(crate) fn test_parts(host: &str, segs: &[&str], query: &[(&str, &str)]) -> UrlParts {
+    UrlParts {
+        scheme: "https".to_string(),
+        host: host.to_string(),
+        path_segments: segs.iter().map(|s| (*s).to_string()).collect(),
+        query: query
+            .iter()
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_full_url_splits_host_path_query() {
+        let p = UrlParts::parse("https://www.youtube.com/watch?v=abc&t=4").unwrap();
+        assert_eq!(p.host, "www.youtube.com");
+        assert_eq!(p.path_segments, vec!["watch"]);
+        assert_eq!(p.query.get("v").map(String::as_str), Some("abc"));
+        assert_eq!(p.query.get("t").map(String::as_str), Some("4"));
+    }
+
+    #[test]
+    fn parse_r_shortcut_expands_to_reddit() {
+        let p = UrlParts::parse("r/rust").unwrap();
+        assert_eq!(p.match_host(), "reddit.com");
+        assert_eq!(p.path_segments, vec!["r", "rust"]);
+    }
+
+    #[test]
+    fn parse_slash_r_shortcut() {
+        let p = UrlParts::parse("/r/aww").unwrap();
+        assert_eq!(p.path_segments, vec!["r", "aww"]);
+    }
+
+    #[test]
+    fn parse_schemeless_host_path() {
+        let p = UrlParts::parse("reddit.com/r/rust/comments/aaa/slug").unwrap();
+        assert_eq!(p.match_host(), "reddit.com");
+        assert_eq!(
+            p.path_segments,
+            vec!["r", "rust", "comments", "aaa", "slug"]
+        );
+    }
+
+    #[test]
+    fn parse_empty_returns_none() {
+        assert!(UrlParts::parse("").is_none());
+        assert!(UrlParts::parse("   ").is_none());
+    }
+
+    #[test]
+    fn normalize_match_host_reddit_aliases() {
+        assert_eq!(normalize_match_host("old.reddit.com"), "reddit.com");
+        assert_eq!(normalize_match_host("NEW.reddit.com"), "reddit.com");
+        assert_eq!(normalize_match_host("www.reddit.com"), "reddit.com");
+    }
+
+    #[test]
+    fn normalize_match_host_youtube_aliases() {
+        assert_eq!(normalize_match_host("m.youtube.com"), "youtube.com");
+        assert_eq!(normalize_match_host("www.youtube.com"), "youtube.com");
+    }
+
+    #[test]
+    fn strip_tracking_query_removes_known_params() {
+        let mut q = HashMap::from([
+            ("v".into(), "1".into()),
+            ("utm_source".into(), "x".into()),
+            ("fbclid".into(), "y".into()),
+            ("ref".into(), "z".into()),
+        ]);
+        strip_tracking_query(&mut q);
+        assert_eq!(q.len(), 1);
+        assert_eq!(q.get("v").map(String::as_str), Some("1"));
+    }
+
+    #[test]
+    fn strip_tracking_query_utm_prefix() {
+        let mut q = HashMap::from([("utm_campaign".into(), "email".into())]);
+        strip_tracking_query(&mut q);
+        assert!(q.is_empty());
+    }
+}
diff --git a/server/src/url_rules/registry_tests.rs b/server/src/url_rules/registry_tests.rs
new file mode 100644
index 0000000000000000000000000000000000000000..7b76b550f808f590e011469fdae02adf603b43d2
--- /dev/null
+++ b/server/src/url_rules/registry_tests.rs
@@ -0,0 +1,195 @@
+//! End-to-end tests for the public registry API (`canonicalize_raw`, breadcrumbs, parent).
+
+use super::registry::{
+    canonicalize_raw, looks_like_url, navigable_breadcrumbs, parent_url, resolve_id,
+};
+
+fn canon(raw: &str) -> String {
+    canonicalize_raw(raw).unwrap().canonical
+}
+
+#[test]
+fn looks_like_url_positive_cases() {
+    for raw in [
+        "https://reddit.com/r/rust",
+        "r/rust",
+        "/r/aww",
+        "reddit.com/r/x",
+        "www.example.com/path",
+        "youtu.be/abc",
+    ] {
+        assert!(looks_like_url(raw), "{raw}");
+    }
+}
+
+#[test]
+fn looks_like_url_negative_cases() {
+    for raw in ["alpha", "beta", "", "hello world", "no-dots"] {
+        assert!(!looks_like_url(raw), "{raw}");
+    }
+}
+
+#[test]
+fn resolve_id_matches_canonicalize_raw() {
+    let raw = "https://youtu.be/xyz";
+    assert_eq!(
+        resolve_id(raw).as_deref(),
+        Some(canon(raw).as_str())
+    );
+}
+
+#[test]
+fn canonicalize_empty_returns_none() {
+    assert!(canonicalize_raw("").is_none());
+}
+
+#[test]
+fn alias_when_slug_stripped() {
+    let r = canonicalize_raw(
+        "https://reddit.com/r/rust/comments/aaa/very_long_title_slug",
+    )
+    .unwrap();
+    assert_eq!(r.canonical, "https://reddit.com/r/rust/comments/aaa");
+    assert!(r.alias_of.is_some());
+}
+
+#[test]
+fn alias_none_when_already_canonical() {
+    let raw = "https://reddit.com/r/rust";
+    let r = canonicalize_raw(raw).unwrap();
+    assert_eq!(r.canonical, raw);
+    assert!(r.alias_of.is_none());
+}
+
+#[test]
+fn parent_url_subreddit_under_r_hub() {
+    assert_eq!(
+        parent_url("https://reddit.com/r/movies").as_deref(),
+        Some("https://reddit.com/r")
+    );
+}
+
+#[test]
+fn parent_url_domain_has_none() {
+    assert_eq!(parent_url("https://reddit.com").as_deref(), None);
+}
+
+#[test]
+fn parent_url_generic_site() {
+    assert_eq!(
+        parent_url("https://example.com/a/b").as_deref(),
+        Some("https://example.com/a")
+    );
+}
+
+#[test]
+fn breadcrumbs_from_canonical_string_roundtrip() {
+    let id = "https://reddit.com/r/golang/comments/abc123";
+    let crumbs = navigable_breadcrumbs(id);
+    assert_eq!(crumbs.last().map(String::as_str), Some(id));
+}
+
+// --- Table: Reddit raw URLs → canonical ---
+
+#[test]
+fn reddit_canonical_matrix() {
+    let cases: &[(&str, &str)] = &[
+        ("r/rust", "https://reddit.com/r/rust"),
+        ("/r/aww", "https://reddit.com/r/aww"),
+        ("https://reddit.com/r/rust", "https://reddit.com/r/rust"),
+        (
+            "https://www.reddit.com/r/programming/new",
+            "https://reddit.com/r/programming",
+        ),
+        (
+            "https://old.reddit.com/r/test/comments/xyz/slug/",
+            "https://reddit.com/r/test/comments/xyz",
+        ),
+        (
+            "reddit.com/r/Movies/comments/abc/Title_Case_Slug",
+            "https://reddit.com/r/movies/comments/abc",
+        ),
+    ];
+    for (raw, want) in cases {
+        assert_eq!(canon(raw), *want, "raw={raw}");
+    }
+}
+
+// --- Table: YouTube raw URLs → canonical ---
+
+#[test]
+fn youtube_canonical_matrix() {
+    let cases: &[(&str, &str)] = &[
+        (
+            "https://youtube.com/watch?v=abc",
+            "https://youtube.com/watch?v=abc",
+        ),
+        (
+            "https://www.youtube.com/watch?v=abc&t=1&feature=share",
+            "https://youtube.com/watch?v=abc",
+        ),
+        ("https://youtu.be/abc", "https://youtube.com/watch?v=abc"),
+        (
+            "https://youtube.com/shorts/abc",
+            "https://youtube.com/watch?v=abc",
+        ),
+    ];
+    for (raw, want) in cases {
+        assert_eq!(canon(raw), *want, "raw={raw}");
+    }
+}
+
+// --- Table: generic sites ---
+
+#[test]
+fn generic_canonical_matrix() {
+    let cases: &[(&str, &str)] = &[
+        (
+            "https://news.ycombinator.com/item?id=38472",
+            "https://news.ycombinator.com/item?id=38472",
+        ),
+        (
+            "https://www.github.com/rust-lang/rust/issues/1?utm_source=x",
+            "https://github.com/rust-lang/rust/issues/1",
+        ),
+        ("https://example.com", "https://example.com"),
+    ];
+    for (raw, want) in cases {
+        assert_eq!(canon(raw), *want, "raw={raw}");
+    }
+}
+
+// --- Phantom /comments/ regression (sorter2-specific) ---
+
+#[test]
+fn phantom_comments_not_in_breadcrumbs_for_post() {
+    let crumbs = navigable_breadcrumbs("https://reddit.com/r/rust/comments/aaa");
+    assert!(!crumbs.iter().any(|c| c.ends_with("/comments")));
+}
+
+#[test]
+fn phantom_comments_not_sibling_of_subreddit_in_breadcrumb_chain() {
+    let crumbs = navigable_breadcrumbs("https://reddit.com/r/rust/comments/aaa");
+    let subs: Vec<_> = crumbs
+        .iter()
+        .filter(|c| c.contains("/r/rust") && !c.contains("/comments/"))
+        .collect();
+    assert_eq!(subs, vec!["https://reddit.com/r/rust"]);
+}
+
+// --- Distinct items must stay distinct ---
+
+#[test]
+fn different_posts_different_canonical() {
+    let a = canon("https://reddit.com/r/rust/comments/aaa");
+    let b = canon("https://reddit.com/r/rust/comments/bbb");
+    assert_ne!(a, b);
+}
+
+#[test]
+fn different_subreddits_different_canonical() {
+    assert_ne!(
+        canon("https://reddit.com/r/rust"),
+        canon("https://reddit.com/r/golang")
+    );
+}


Side B — contributor: tommy-mor
Side B — commit message:
[1d14ff09] Ephemeral Reddit content; log structure only (#46)

* Keep Reddit content ephemeral; log structure only

Remove EntityImported and EntityStore. Reddit fetches write display
content directly to the projection with a fetched_at timestamp, while
the event log records NodeEnsured for discovered identities only.

A background task evicts cached display content after 48 hours. Votes,
tree structure, and ItemIds remain in the log and projection.

Co-authored-by: tommy <thmorriss@gmail.com>

* Fix reddit import test assertions and Clojure syntax

Co-authored-by: tommy <thmorriss@gmail.com>

---------

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

Side B — unified diff (full patch):
diff --git a/server/src/bin/storage_bench.rs b/server/src/bin/storage_bench.rs
index 4762d0d23bc8e2df439afe09a3235ba7f72bd486..3d988416ad36b27a7d3dc84280cfdbdcafa43e69 100644
--- a/server/src/bin/storage_bench.rs
+++ b/server/src/bin/storage_bench.rs
@@ -6,8 +6,8 @@ use std::{
 };
 
 use sorter2_server::{
-    entity_store::EntityStore, event_log::EventLog, events::Event, journal::JournalClient,
-    projection_apply, projection_store::ProjectionStore,
+    event_log::EventLog, events::Event, journal::JournalClient, projection_apply,
+    projection_store::ProjectionStore,
 };
 
 #[tokio::main]
@@ -18,12 +18,10 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     let data_dir = opts.data_dir.to_string_lossy().into_owned();
     let event_log = Arc::new(EventLog::new(format!("{data_dir}/events.jsonl")));
     let db = durable::Db::open(opts.data_dir.join("store"))?;
-    let entity_store = EntityStore::from_db(&db)?;
     let projection_store = ProjectionStore::from_db(&db)?;
 
     let journal = JournalClient::spawn(
         event_log.clone(),
-        entity_store.clone(),
         projection_store.clone(),
         event_log.last_sequence().await? + 1,
     );
@@ -46,12 +44,9 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     drop(journal);
 
     let rebuild_start = Instant::now();
-    entity_store.reset()?;
     projection_store.reset()?;
     let rebuild = event_log
-        .replay(|record| {
-            projection_apply::apply_records(&projection_store, &entity_store, &[record])
-        })
+        .replay(|record| projection_apply::apply_records(&projection_store, &[record]))
         .await?;
     let rebuild_elapsed = rebuild_start.elapsed();
 
diff --git a/server/src/entity_store.rs b/server/src/entity_store.rs
deleted file mode 100644
index d29f39aecca6f12cdcf263cf77c3654eb4ee6cfa..0000000000000000000000000000000000000000
--- a/server/src/entity_store.rs
+++ /dev/null
@@ -1,134 +0,0 @@
-//! Off-heap storage for full entity payloads (Reddit API JSON).
-//!
-//! Derived [`crate::reducer::EntityData`] is stored on the node; the raw JSON
-//! lives here, in the shared durable [`Store`] schema.
-
-use std::path::Path;
-
-use durable::{Batch, Db, Durability};
-use serde_json::Value;
-
-use crate::{
-    path_types::ItemId,
-    storage_dto::{decode_entity_payload, encode_entity_payload},
-    storage_schema::{Store, StoreFields},
-};
-
-const ENTITY_SCHEMA_KEY: &str = "schema_version";
-const ENTITY_SCHEMA_VERSION: u64 = 2;
-
-#[derive(Debug, thiserror::Error)]
-pub enum EntityStoreError {
-    #[error("durable error: {0}")]
-    Durable(#[from] durable::Error),
-    #[error("json error: {0}")]
-    Json(#[from] serde_json::Error),
-    #[error("storage decode error: {0}")]
-    Storage(String),
-    #[error("io error: {0}")]
-    Io(#[from] std::io::Error),
-}
-
-/// Disk-backed map of entity id → raw JSON payload.
-#[derive(Clone)]
-pub struct EntityStore {
-    db: Db,
-}
-
-impl EntityStore {
-    /// Open (or create) the entity database under `dir`.
-    pub fn open(dir: &Path) -> Result<Self, EntityStoreError> {
-        std::fs::create_dir_all(dir)?;
-        let db = Db::open(dir)?;
-        Self::from_db(&db)
-    }
-
-    /// Create an entity store backed by an already-open database.
-    pub fn from_db(db: &Db) -> Result<Self, EntityStoreError> {
-        let store = Self { db: db.clone() };
-        let version = Store::root()
-            .entity_meta()
-            .key(&ENTITY_SCHEMA_KEY.to_string())
-            .get(db)?;
-        if version != Some(ENTITY_SCHEMA_VERSION) {
-            store.reset()?;
-        }
-        Ok(store)
-    }
-
-    /// Clear rebuildable entity payloads and reset storage schema metadata.
-    pub fn reset(&self) -> Result<(), EntityStoreError> {
-        let root = Store::root();
-        self.db.apply(
-            &[root.entities().clear(), root.entity_meta().clear()],
-            Durability::SyncWal,
-        )?;
-        self.db.run(
-            root.entity_meta()
-                .key(&ENTITY_SCHEMA_KEY.to_string())
-                .set(&ENTITY_SCHEMA_VERSION),
-            Durability::SyncWal,
-        )?;
-        Ok(())
-    }
-
-    /// Persist a payload for `id` (overwrites any existing entry).
-    pub fn put(&self, id: &ItemId, payload: &Value) -> Result<(), EntityStoreError> {
-        self.db.run(
-            Store::root()
-                .entities()
-                .key(&id.as_str().to_string())
-                .set(&encode_entity_payload(payload)),
-            Durability::SyncWal,
-        )?;
-        Ok(())
-    }
-
-    /// Add a payload write to the caller's batch.
-    pub fn put_in_batch(
-        &self,
-        batch: &mut Batch,
-        id: &ItemId,
-        payload: &Value,
-    ) -> Result<(), EntityStoreError> {
-        batch.write(
-            Store::root()
-                .entities()
-                .key(&id.as_str().to_string())
-                .set(&encode_entity_payload(payload)),
-        );
-        Ok(())
-    }
-
-    /// Load a stored payload, if present.
-    pub fn get(&self, id: &ItemId) -> Result<Option<Value>, EntityStoreError> {
-        match Store::root()
-            .entities()
-            .key(&id.as_str().to_string())
-            .get(&self.db)?
-        {
-            Some(record) => decode_entity_payload(record)
-                .map(Some)
-                .map_err(EntityStoreError::Storage),
-            None => Ok(None),
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn round_trip_payload() {
-        let tmp = tempfile::tempdir().unwrap();
-        let store = EntityStore::open(tmp.path()).unwrap();
-        let id = ItemId::from_url("https://reddit.com/r/rust").unwrap();
-        let payload = json!({"kind": "t5", "data": {"display_name": "rust"}});
-
-        store.put(&id, &payload).unwrap();
-        let loaded = store.get(&id).unwrap().unwrap();
-        assert_eq!(loaded, payload);
-    }
-}
diff --git a/server/src/events.rs b/server/src/events.rs
index a3d88e285c645a96d33e6f0aed1b487b843aceb8..d76c3bb4277216b0d39c9422ba7a50db10a95e05 100644
--- a/server/src/events.rs
+++ b/server/src/events.rs
@@ -1,5 +1,4 @@
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 
 /// Schema version for JSONL log records. Bump when event semantics change.
 pub const CURRENT_LOG_SCHEMA: u32 = 1;
@@ -30,7 +29,7 @@ pub type ViewRecord = LogRecord<ViewEvent>;
 /// Wall-clock timestamp carried on the log envelope for domain events.
 pub fn event_timestamp(event: &Event) -> i64 {
     match event {
-        Event::VoteRecorded { ts, .. } | Event::EntityImported { ts, .. } => *ts,
+        Event::VoteRecorded { ts, .. } => *ts,
         Event::NodeEnsured { .. } => crate::fetch::now_ms(),
     }
 }
@@ -57,6 +56,4 @@ pub enum Event {
     },
     /// Register a node path in the fractal tree (no external fetch).
     NodeEnsured { id: String },
-    /// Full upstream API payload for a node (domain-specific view derived at replay/render time).
-    EntityImported { id: String, ts: i64, payload: Value },
 }
diff --git a/server/src/journal.rs b/server/src/journal.rs
index 50bc89f976edb82b7b0e49e954a8eccbbe82bf87..d50023aca7c3e74068de988449b11baee32d1ecf 100644
--- a/server/src/journal.rs
+++ b/server/src/journal.rs
@@ -5,7 +5,6 @@ use std::sync::Arc;
 use tokio::sync::{mpsc, oneshot};
 
 use crate::{
-    entity_store::EntityStore,
     event_log::EventLog,
     events::{event_timestamp, Event, EventRecord},
     projection_apply,
@@ -25,7 +24,6 @@ pub struct JournalClient {
 impl JournalClient {
     pub fn spawn(
         event_log: Arc<EventLog>,
-        entity_store: EntityStore,
         projection_store: ProjectionStore,
         next_seq: u64,
     ) -> Self {
@@ -33,7 +31,6 @@ impl JournalClient {
         tokio::spawn(journal_worker(
             rx,
             event_log,
-            entity_store,
             projection_store,
             next_seq,
         ));
@@ -62,7 +59,6 @@ impl JournalClient {
 async fn journal_worker(
     mut rx: mpsc::Receiver<JournalCommand>,
     event_log: Arc<EventLog>,
-    entity_store: EntityStore,
     projection_store: ProjectionStore,
     mut next_seq: u64,
 ) {
@@ -75,7 +71,6 @@ async fn journal_worker(
         let result = append_and_project_batch(
             &event_log,
             &projection_store,
-            &entity_store,
             &mut next_seq,
             &batch,
         )
@@ -99,7 +94,6 @@ async fn journal_worker(
 async fn append_and_project_batch(
     event_log: &EventLog,
     projection_store: &ProjectionStore,
-    entity_store: &EntityStore,
     next_seq: &mut u64,
     commands: &[JournalCommand],
 ) -> Result<(), String> {
@@ -117,7 +111,7 @@ async fn append_and_project_batch(
         .await
         .map_err(|e| e.to_string())?;
     *next_seq = seq;
-    projection_apply::apply_records(projection_store, entity_store, &records)
+    projection_apply::apply_records(projection_store, &records)
         .map_err(|e| format!("projection apply failed after durable append: {e}"))
 }
 
@@ -132,10 +126,9 @@ mod tests {
         let log_path = tmp.path().join("events.jsonl");
         let event_log = Arc::new(EventLog::new(log_path));
         let db = durable::Db::open(tmp.path().join("store")).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let projection_store = ProjectionStore::from_db(&db).unwrap();
 
-        let journal = JournalClient::spawn(event_log, entity_store, projection_store.clone(), 1);
+        let journal = JournalClient::spawn(event_log, projection_store.clone(), 1);
 
         let j1 = journal.clone();
         let j2 = journal.clone();
@@ -177,11 +170,9 @@ mod tests {
             .unwrap();
 
         let db = durable::Db::open(tmp.path().join("store")).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let projection_store = ProjectionStore::from_db(&db).unwrap();
         projection_apply::apply_records(
             &projection_store,
-            &entity_store,
             &[EventRecord::new(
                 1,
                 1,
@@ -196,7 +187,6 @@ mod tests {
 
         let journal = JournalClient::spawn(
             event_log.clone(),
-            entity_store,
             projection_store.clone(),
             next_seq,
         );
@@ -219,10 +209,9 @@ mod tests {
         let log_path = tmp.path().join("events.jsonl");
         let event_log = Arc::new(EventLog::new(log_path));
         let db = durable::Db::open(tmp.path().join("store")).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let projection_store = ProjectionStore::from_db(&db).unwrap();
         let journal =
-            JournalClient::spawn(event_log.clone(), entity_store, projection_store.clone(), 1);
+            JournalClient::spawn(event_log.clone(), projection_store.clone(), 1);
 
         journal
             .append_many(vec![
diff --git a/server/src/lib.rs b/server/src/lib.rs
index 5c02c8e704e4664453bad75d819df8a067668176..3dfc7c8acb8ed61bb73ade63e72768e402042cc5 100644
--- a/server/src/lib.rs
+++ b/server/src/lib.rs
@@ -1,5 +1,4 @@
 pub mod api;
-pub mod entity_store;
 pub mod event_log;
 pub mod events;
 pub mod fetch;
diff --git a/server/src/projection_apply.rs b/server/src/projection_apply.rs
index 5644557a41b3e9497c7421b444155ae629fa79f1..9c8990a8af927f35d3344c8d0872a516aba56b86 100644
--- a/server/src/projection_apply.rs
+++ b/server/src/projection_apply.rs
@@ -1,22 +1,20 @@
 //! Apply event-log records to the durable projection as precise point updates.
 //!
 //! Each batch of records lowers to reified durable writes (edge merges, child
-//! links, voted-pair flags, recent-vote pushes, entity payloads) plus a cursor
-//! advance, all committed in one atomic `DisableWal` batch. The cursor moving in
-//! the same batch as the (non-idempotent) edge merges guarantees exactly-once
-//! application across replay.
+//! links, voted-pair flags, recent-vote pushes) plus a cursor advance, all
+//! committed in one atomic `DisableWal` batch. The cursor moving in the same
+//! batch as the (non-idempotent) edge merges guarantees exactly-once application
+//! across replay.
 
 use std::collections::BTreeSet;
 
 use crate::{
-    entity_store::EntityStore,
     event_log::EventLogError,
     events::{Event, EventRecord},
     path_types::ItemId,
     projection_store::ProjectionStore,
-    reddit::entity_view_from_payload,
     reducer::VoteData,
-    storage_schema::{ensure_path_writes, entity_view_writes, vote_writes},
+    storage_schema::{ensure_path_writes, vote_writes},
 };
 
 fn parse_event_id(id: &str) -> Result<ItemId, EventLogError> {
@@ -38,7 +36,6 @@ fn parent_from_event_scope(scope: &str) -> ItemId {
 
 pub fn apply_records(
     projection_store: &ProjectionStore,
-    entity_store: &EntityStore,
     records: &[EventRecord],
 ) -> Result<(), EventLogError> {
     if records.is_empty() {
@@ -79,14 +76,6 @@ pub fn apply_records(
                 let parsed = parse_event_id(id)?;
                 ensure_path_writes(&mut batch, &parsed);
             }
-            Event::EntityImported { id, payload, .. } => {
-                let parsed = parse_event_id(id)?;
-                let view = entity_view_from_payload(&parsed, payload);
-                entity_view_writes(&mut batch, &parsed, view.as_ref());
-                entity_store
-                    .put_in_batch(&mut batch, &parsed, payload)
-                    .map_err(|e| EventLogError::Apply(e.to_string()))?;
-            }
         }
         last_seq = record.seq;
     }
diff --git a/server/src/projection_store.rs b/server/src/projection_store.rs
index 30ee478f953e80d7322bdbfa521ae559ff08fa51..8576d671f351004426207894ac35594ddb0f70cf 100644
--- a/server/src/projection_store.rs
+++ b/server/src/projection_store.rs
@@ -9,13 +9,16 @@ use durable::{Db, Durability, Write};
 
 use crate::{
     path_types::ItemId,
-    reducer::{GlobalTree, NodeState},
-    storage_schema::{load_node_state, node, NodeSchemaFields, Store, StoreFields},
+    reducer::{EntityData, GlobalTree, NodeState},
+    storage_schema::{
+        entity_content_clear_writes, entity_content_writes, load_node_state, node, NodeSchemaFields,
+        Store, StoreFields,
+    },
 };
 
 const PROJECTION_CURSOR_KEY: &str = "cursor";
 const PROJECTION_SCHEMA_KEY: &str = "schema_version";
-const PROJECTION_SCHEMA_VERSION: u64 = 2;
+const PROJECTION_SCHEMA_VERSION: u64 = 3;
 
 #[derive(Debug, thiserror::Error)]
 pub enum ProjectionStoreError {
@@ -148,6 +151,45 @@ impl ProjectionStore {
         )?;
         Ok(())
     }
+
+    /// Cache Reddit display content outside the event log (must be evicted per policy).
+    pub fn put_ephemeral_content(
+        &self,
+        id: &ItemId,
+        view: &EntityData,
+        fetched_at: i64,
+    ) -> Result<(), ProjectionStoreError> {
+        let mut batch = self.db.batch();
+        entity_content_writes(&mut batch, id, view, fetched_at);
+        batch
+            .commit_with(Durability::DisableWal)
+            .map_err(ProjectionStoreError::from)?;
+        Ok(())
+    }
+
+    /// Drop cached display content older than `cutoff_ms` (votes and tree structure remain).
+    pub fn evict_content_older_than(&self, cutoff_ms: i64) -> Result<usize, ProjectionStoreError> {
+        let keys = Store::root().nodes().keys(&self.db)?;
+        let mut batch = self.db.batch();
+        let mut evicted = 0usize;
+        for key in keys {
+            let id = parse_node_key(&key)?;
+            let np = node(&id);
+            let Some(fetched_at) = np.fetched_at().get(&self.db)? else {
+                continue;
+            };
+            if fetched_at > 0 && fetched_at < cutoff_ms {
+                entity_content_clear_writes(&mut batch, &id);
+                evicted += 1;
+            }
+        }
+        if evicted > 0 {
+            batch
+                .commit_with(Durability::DisableWal)
+                .map_err(ProjectionStoreError::from)?;
+        }
+        Ok(evicted)
+    }
 }
 
 fn parse_node_key(key: &str) -> Result<ItemId, ProjectionStoreError> {
@@ -162,7 +204,7 @@ fn parse_node_key(key: &str) -> Result<ItemId, ProjectionStoreError> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{entity_store::EntityStore, events::Event, projection_apply};
+    use crate::{events::Event, projection_apply, reducer::EntityData};
 
     fn record(seq: u64, event: Event) -> crate::events::EventRecord {
         crate::events::EventRecord::new(seq, crate::events::event_timestamp(&event), event)
@@ -172,7 +214,6 @@ mod tests {
     fn applies_and_loads_reducer_nodes() {
         let tmp = tempfile::tempdir().unwrap();
         let db = Db::open(tmp.path()).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let store = ProjectionStore::from_db(&db).unwrap();
 
         let event = Event::VoteRecorded {
@@ -183,7 +224,7 @@ mod tests {
             ratio_right: 1,
             scope: String::new(),
         };
-        projection_apply::apply_records(&store, &entity_store, &[record(1, event)]).unwrap();
+        projection_apply::apply_records(&store, &[record(1, event)]).unwrap();
         assert_eq!(store.last_applied_event_count().unwrap(), 1);
 
         let loaded = store.load_tree().unwrap();
@@ -196,7 +237,6 @@ mod tests {
     fn hydrates_scope_with_child_nodes() {
         let tmp = tempfile::tempdir().unwrap();
         let db = Db::open(tmp.path()).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let store = ProjectionStore::from_db(&db).unwrap();
 
         let event = Event::VoteRecorded {
@@ -207,11 +247,36 @@ mod tests {
             ratio_right: 1,
             scope: String::new(),
         };
-        projection_apply::apply_records(&store, &entity_store, &[record(1, event)]).unwrap();
+        projection_apply::apply_records(&store, &[record(1, event)]).unwrap();
 
         let scoped = store.scope_tree(&ItemId::root()).unwrap();
         let root = scoped.get(&ItemId::root()).unwrap();
         assert_eq!(root.children.len(), 2);
         assert!(scoped.get(&ItemId::opaque("alpha")).is_some());
     }
+
+    #[test]
+    fn evicts_stale_ephemeral_content() {
+        let tmp = tempfile::tempdir().unwrap();
+        let db = Db::open(tmp.path()).unwrap();
+        let store = ProjectionStore::from_db(&db).unwrap();
+        let id = ItemId::from_url("https://reddit.com/r/rust").unwrap();
+        store
+            .put_ephemeral_content(
+                &id,
+                &EntityData {
+                    title: "Rust".into(),
+                    author: None,
+                    body_html: None,
+                    thumb_url: None,
+                    image_url: None,
+                    link_url: None,
+                },
+                1_000,
+            )
+            .unwrap();
+        assert!(store.load_node(&id).unwrap().unwrap().data.is_some());
+        assert_eq!(store.evict_content_older_than(2_000).unwrap(), 1);
+        assert!(store.load_node(&id).unwrap().unwrap().data.is_none());
+    }
 }
diff --git a/server/src/reddit.rs b/server/src/reddit.rs
index 20b7f9e3f8be39268a1767d09f5cf81eaa6ae0df..a874814f8927192ee62cab2d0db1efd27dcd57b7 100644
--- a/server/src/reddit.rs
+++ b/server/src/reddit.rs
@@ -9,10 +9,13 @@ use serde_json::Value;
 use tokio::sync::{mpsc, oneshot};
 
 use crate::{
-    entity_store::EntityStore, events::Event, fetch::now_ms, journal::JournalClient,
-    path_types::ItemId, reducer::GlobalTree,
+    events::Event, fetch::now_ms, journal::JournalClient,
+    path_types::ItemId, projection_store::ProjectionStore,
 };
 
+/// Reddit display content must not be retained longer than this (API policy).
+pub const REDDIT_CONTENT_TTL: std::time::Duration = std::time::Duration::from_secs(48 * 3600);
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum FetchJobResult {
     /// Number of entities written (1 for self, N for children).
@@ -70,7 +73,11 @@ struct OAuthToken {
 }
 
 impl RedditBroker {
-    pub fn spawn(journal: JournalClient, config: RedditApiConfig) -> Self {
+    pub fn spawn(
+        journal: JournalClient,
+        projection_store: ProjectionStore,
+        config: RedditApiConfig,
+    ) -> Self {
         let (tx, rx) = mpsc::channel(100);
 
         let mut headers = header::HeaderMap::new();
@@ -93,7 +100,7 @@ impl RedditBroker {
             "reddit worker started"
         );
 
-        tokio::spawn(reddit_worker(rx, journal, client, config));
+        tokio::spawn(reddit_worker(rx, journal, projection_store, client, config));
 
         Self { tx }
     }
@@ -189,27 +196,50 @@ pub fn entity_view_from_payload(
     None
 }
 
-pub fn apply_entity_import(
-    tree: &mut GlobalTree,
-    store: &EntityStore,
-    id: &ItemId,
-    payload: Value,
-) -> Result<(), String> {
-    let view = entity_view_from_payload(id, &payload);
-    store.put(id, &payload).map_err(|e| e.to_string())?;
-    tree.apply_entity(id, view);
-    Ok(())
-}
-
 fn notify(done: Option<oneshot::Sender<FetchJobResult>>, result: FetchJobResult) {
     if let Some(tx) = done {
         let _ = tx.send(result);
     }
 }
 
+async fn import_fetched_payload(
+    kind: FetchKind,
+    fetch_id: &ItemId,
+    payload: Value,
+    projection_store: &ProjectionStore,
+    journal: &JournalClient,
+) -> Result<usize, String> {
+    let fetched_at = now_ms();
+    let imports: Vec<(ItemId, Value)> = match kind {
+        FetchKind::SelfEntity => vec![(fetch_id.clone(), payload)],
+        FetchKind::Children => parse_children(fetch_id, &payload),
+    };
+
+    for (id, child_payload) in &imports {
+        if let Some(view) = entity_view_from_payload(id, child_payload) {
+            projection_store
+                .put_ephemeral_content(id, &view, fetched_at)
+                .map_err(|e| e.to_string())?;
+        }
+    }
+
+    let events: Vec<Event> = imports
+        .iter()
+        .map(|(id, _)| Event::NodeEnsured {
+            id: id.as_str().to_string(),
+        })
+        .collect();
+    let written = events.len();
+    if !events.is_empty() {
+        journal.append_many(events).await?;
+    }
+    Ok(written)
+}
+
 async fn reddit_worker(
     mut rx: mpsc::Receiver<RedditCommand>,
     journal: JournalClient,
+    projection_store: ProjectionStore,
     client: Client,
     config: RedditApiConfig,
 ) {
@@ -276,33 +306,20 @@ async fn reddit_worker(
 
         match outcome {
             Ok(FetchOutcome::Payload(payload)) => {
-                let imports: Vec<(ItemId, Value)> = match kind {
-                    FetchKind::SelfEntity => vec![(fetch_id.clone(), payload)],
-                    FetchKind::Children => parse_children(&fetch_id, &payload),
-                };
                 tracing::debug!(
                     item = %fetch_id,
                     ?kind,
-                    count = imports.len(),
-                    "reddit fetch got payload, importing"
+                    "reddit fetch got payload, caching ephemerally"
                 );
 
-                let events: Vec<Event> = imports
-                    .into_iter()
-                    .map(|(child_id, child_payload)| Event::EntityImported {
-                        id: child_id.as_str().to_string(),
-                        ts: now_ms(),
-                        payload: child_payload,
-                    })
-                    .collect();
-                let written = events.len();
-
-                match journal.append_many(events).await {
+                match import_fetched_payload(kind, &fetch_id, payload, &projection_store, &journal)
+                    .await
+                {
                     Err(e) => {
-                        tracing::warn!(item = %fetch_id, err = %e, "reddit import journal failed");
+                        tracing::warn!(item = %fetch_id, err = %e, "reddit import failed");
                         notify(done, FetchJobResult::Failed(e));
                     }
-                    Ok(()) => {
+                    Ok(written) => {
                         recently_fetched.insert(key.clone(), Instant::now());
                         current_delay = Duration::from_millis(600);
                         tracing::info!(item = %fetch_id, ?kind, written, "reddit import complete");
diff --git a/server/src/reducer.rs b/server/src/reducer.rs
index 1352a8f0771add3d868a1b30109b087a2a6dba6f..0c75c85150bb9e5f578bbadf58b3e43f8a80be4b 100644
--- a/server/src/reducer.rs
+++ b/server/src/reducer.rs
@@ -136,8 +136,7 @@ pub struct EntityData {
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct NodeState {
     pub id: ItemId,
-    /// Domain-specific view derived from imported payload (e.g. Reddit title/author).
-    /// Raw JSON lives in [`crate::entity_store::EntityStore`].
+    /// Ephemeral display view (Reddit title/author/etc.; not event-logged).
     pub data: Option<EntityData>,
     pub children: HashSet<ItemId>,
     pub local_ranking: GroupState,
diff --git a/server/src/state.rs b/server/src/state.rs
index e44849eec46123072b238afd40a1fdd51ce19bd9..247b9047a57956f76c4b6bef691662101e62a8f9 100644
--- a/server/src/state.rs
+++ b/server/src/state.rs
@@ -1,14 +1,14 @@
 use std::{error::Error, sync::Arc};
 
 use crate::{
-    entity_store::EntityStore,
     event_log::EventLog,
     events::Event,
+    fetch::now_ms,
     journal::JournalClient,
     path_types::ItemId,
     projection_apply,
     projection_store::ProjectionStore,
-    reddit::{RedditApiConfig, RedditBroker},
+    reddit::{RedditApiConfig, RedditBroker, REDDIT_CONTENT_TTL},
     reducer::{GlobalTree, VoteData},
     view_log::ViewLog,
     views::ViewStore,
@@ -45,7 +45,6 @@ pub fn normalize_scope(raw: &str) -> String {
 
 async fn catch_up_projection(
     event_log: &EventLog,
-    entity_store: &EntityStore,
     projection_store: &ProjectionStore,
 ) -> Result<(), crate::event_log::EventLogError> {
     let after_seq = projection_store
@@ -54,7 +53,7 @@ async fn catch_up_projection(
 
     let stats = event_log
         .replay_from(after_seq, |record| {
-            projection_apply::apply_records(projection_store, entity_store, &[record])
+            projection_apply::apply_records(projection_store, &[record])
         })
         .await?;
     if after_seq > stats.last_seq {
@@ -67,22 +66,34 @@ async fn catch_up_projection(
     Ok(())
 }
 
+fn spawn_content_evictor(projection_store: ProjectionStore) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15 * 60));
+        interval.tick().await;
+        loop {
+            interval.tick().await;
+            let cutoff = now_ms() - REDDIT_CONTENT_TTL.as_millis() as i64;
+            match projection_store.evict_content_older_than(cutoff) {
+                Ok(0) => {}
+                Ok(n) => tracing::info!(evicted = n, "reddit display content TTL eviction"),
+                Err(e) => tracing::warn!(err = %e, "reddit content TTL eviction failed"),
+            }
+        }
+    });
+}
+
 pub async fn rebuild_projection(
     cfg: &AppConfig,
 ) -> Result<ProjectionRebuildStats, Box<dyn Error + Send + Sync + 'static>> {
     let event_log = EventLog::new(cfg.event_log_path.clone());
     let store_path = format!("{}/store", cfg.data_dir);
     let db = durable::Db::open(std::path::Path::new(&store_path))?;
-    let entity_store = EntityStore::from_db(&db)?;
     let projection_store = ProjectionStore::from_db(&db)?;
 
-    entity_store.reset()?;
     projection_store.reset()?;
 
     let stats = event_log
-        .replay(|record| {
-            projection_apply::apply_records(&projection_store, &entity_store, &[record])
-        })
+        .replay(|record| projection_apply::apply_records(&projection_store, &[record]))
         .await?;
     let cursor = projection_store.last_applied_event_count()?;
     if cursor != stats.last_seq {
@@ -132,7 +143,6 @@ pub struct AppState {
     pub cfg: Arc<AppConfig>,
     pub event_log: Arc<EventLog>,
     pub view_log: Arc<ViewLog>,
-    pub entity_store: EntityStore,
     pub projection_store: ProjectionStore,
     pub views: ViewStore,
     journal: JournalClient,
@@ -145,7 +155,6 @@ impl AppState {
         let view_log = Arc::new(ViewLog::new(cfg.views_log_path.clone()));
         let store_path = format!("{}/store", cfg.data_dir);
         let db = durable::Db::open(std::path::Path::new(&store_path))?;
-        let entity_store = EntityStore::from_db(&db)?;
         let projection_store = ProjectionStore::from_db(&db)?;
         let views = ViewStore::from_db(&db)?;
 
@@ -157,22 +166,25 @@ impl AppState {
         }
         views.spawn_worker(view_log.clone());
 
-        catch_up_projection(&event_log, &entity_store, &projection_store).await?;
+        catch_up_projection(&event_log, &projection_store).await?;
         let next_seq = event_log.last_sequence().await? + 1;
 
         let journal = JournalClient::spawn(
             event_log.clone(),
-            entity_store.clone(),
             projection_store.clone(),
             next_seq,
         );
-        let reddit = RedditBroker::spawn(journal.clone(), RedditApiConfig::from_env());
+        let reddit = RedditBroker::spawn(
+            journal.clone(),
+            projection_store.clone(),
+            RedditApiConfig::from_env(),
+        );
+        spawn_content_evictor(projection_store.clone());
 
         Ok(Self {
             cfg: Arc::new(cfg),
             event_log,
             view_log,
-            entity_store,
             projection_store,
             views,
             journal,
@@ -248,54 +260,72 @@ impl AppState {
 mod tests {
     use super::{normalize_scope, parse_item_param, AppConfig, AppState};
     use crate::{
-        entity_store::EntityStore, event_log::EventLog, events::Event, path_types::ItemId,
-        projection_apply, projection_store::ProjectionStore,
+        event_log::EventLog, events::Event, path_types::ItemId, projection_apply,
+        projection_store::ProjectionStore, reducer::EntityData,
     };
-    use serde_json::json;
 
     fn event_record(seq: u64, event: Event) -> crate::events::EventRecord {
         crate::events::EventRecord::new(seq, crate::events::event_timestamp(&event), event)
     }
 
     #[tokio::test]
-    async fn replay_entity_imported_restores_view() {
+    async fn rebuild_projection_drops_ephemeral_content() {
         let tmp = tempfile::tempdir().unwrap();
-        let log_path = tmp.path().join("events.jsonl");
-        let log = EventLog::new(log_path.to_string_lossy().into_owned());
-        let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}});
-        let event = Event::EntityImported {
-            id: "https://reddit.com/r/rust".into(),
-            ts: 1,
-            payload: payload.clone(),
-        };
-        log.append(&event_record(1, event)).await.unwrap();
+        let data_dir = tmp.path().to_string_lossy().into_owned();
+        let log = EventLog::new(format!("{data_dir}/events.jsonl"));
+        log.append(&event_record(
+            1,
+            Event::NodeEnsured {
+                id: "https://reddit.com/r/rust".into(),
+            },
+        ))
+        .await
+        .unwrap();
 
         let db = durable::Db::open(tmp.path().join("store")).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let projection_store = ProjectionStore::from_db(&db).unwrap();
-        super::catch_up_projection(&log, &entity_store, &projection_store)
+        super::catch_up_projection(&log, &projection_store)
             .await
             .unwrap();
-        let tree = projection_store
-            .scope_tree(&ItemId::parse("https://reddit.com/r/rust").unwrap())
-            .unwrap();
-        let node = tree
-            .get(&ItemId::parse("https://reddit.com/r/rust").unwrap())
-            .unwrap();
-        assert_eq!(node.data.as_ref().unwrap().title, "Rust");
-        let stored = entity_store
-            .get(&ItemId::parse("https://reddit.com/r/rust").unwrap())
-            .unwrap()
+        let id = ItemId::parse("https://reddit.com/r/rust").unwrap();
+        projection_store
+            .put_ephemeral_content(
+                &id,
+                &EntityData {
+                    title: "Rust".into(),
+                    author: None,
+                    body_html: None,
+                    thumb_url: None,
+                    image_url: None,
+                    link_url: None,
+                },
+                1,
+            )
             .unwrap();
-        assert_eq!(stored["data"]["display_name"], "rust");
+        assert!(projection_store.load_node(&id).unwrap().unwrap().data.is_some());
+        drop(projection_store);
+        drop(db);
+
+        super::rebuild_projection(&AppConfig {
+            data_dir: data_dir.clone(),
+            event_log_path: format!("{data_dir}/events.jsonl"),
+            views_log_path: format!("{data_dir}/views.jsonl"),
+            port: 0,
+        })
+        .await
+        .unwrap();
+
+        let db = durable::Db::open(tmp.path().join("store")).unwrap();
+        let projection_store = ProjectionStore::from_db(&db).unwrap();
+        let node = projection_store.load_node(&id).unwrap().unwrap();
+        assert!(node.data.is_none());
     }
 
     #[tokio::test]
-    async fn rebuild_projection_restores_nodes_payloads_and_cursor_from_jsonl() {
+    async fn rebuild_projection_restores_structure_and_cursor_from_jsonl() {
         let tmp = tempfile::tempdir().unwrap();
         let data_dir = tmp.path().to_string_lossy().into_owned();
         let log = EventLog::new(format!("{data_dir}/events.jsonl"));
-        let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}});
         log.append_batch(&[
             event_record(
                 1,
@@ -305,16 +335,8 @@ mod tests {
             ),
             event_record(
                 2,
-                Event::EntityImported {
-                    id: "https://reddit.com/r/rust".into(),
-                    ts: 2,
-                    payload: payload.clone(),
-                },
-            ),
-            event_record(
-                3,
                 Event::VoteRecorded {
-                    ts: 3,
+                    ts: 2,
                     a: "alpha".into(),
                     b: "beta".into(),
                     ratio_left: 2,
@@ -328,11 +350,9 @@ mod tests {
 
         {
             let db = durable::Db::open(tmp.path().join("store")).unwrap();
-            let entity_store = EntityStore::from_db(&db).unwrap();
             let projection_store = ProjectionStore::from_db(&db).unwrap();
             projection_apply::apply_records(
                 &projection_store,
-                &entity_store,
                 &[event_record(
                     1,
                     Event::NodeEnsured {
@@ -352,13 +372,12 @@ mod tests {
         })
         .await
         .unwrap();
-        assert_eq!(stats.applied, 3);
-        assert_eq!(stats.last_seq, 3);
+        assert_eq!(stats.applied, 2);
+        assert_eq!(stats.last_seq, 2);
 
         let db = durable::Db::open(tmp.path().join("store")).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let projection_store = ProjectionStore::from_db(&db).unwrap();
-        assert_eq!(projection_store.last_applied_event_count().unwrap(), 3);
+        assert_eq!(projection_store.last_applied_event_count().unwrap(), 2);
         let tree = projection_store.scope_tree(&ItemId::root()).unwrap();
         let root = tree.get(&ItemId::root()).unwrap();
         assert!(root.children.contains(&ItemId::parse("alpha").unwrap()));
@@ -366,11 +385,6 @@ mod tests {
             .load_node(&ItemId::parse("https://reddit.com/r/stale").unwrap())
             .unwrap()
             .is_none());
-        let stored = entity_store
-            .get(&ItemId::parse("https://reddit.com/r/rust").unwrap())
-            .unwrap()
-            .unwrap();
-        assert_eq!(stored["data"]["display_name"], "rust");
     }
 
     #[tokio::test]
@@ -389,12 +403,9 @@ mod tests {
 
         {
             let db = durable::Db::open(tmp.path().join("store")).unwrap();
-            let entity_store = EntityStore::from_db(&db).unwrap();
             let projection_store = ProjectionStore::from_db(&db).unwrap();
-            // Advance the projection cursor to 2 while the log tail is only 1.
             projection_apply::apply_records(
                 &projection_store,
-                &entity_store,
                 &[event_record(
                     2,
                     Event::NodeEnsured {
@@ -403,7 +414,7 @@ mod tests {
                 )],
             )
             .unwrap();
-            let err = super::catch_up_projection(&log, &entity_store, &projection_store)
+            let err = super::catch_up_projection(&log, &projection_store)
                 .await
                 .unwrap_err();
             assert!(err
@@ -432,10 +443,9 @@ mod tests {
         .unwrap();
 
         let db = durable::Db::open(tmp.path().join("store")).unwrap();
-        let entity_store = EntityStore::from_db(&db).unwrap();
         let projection_store = ProjectionStore::from_db(&db).unwrap();
 
-        super::catch_up_projection(&log, &entity_store, &projection_store)
+        super::catch_up_projection(&log, &projection_store)
             .await
             .unwrap();
         assert_eq!(projection_store.last_applied_event_count().unwrap(), 1);
@@ -444,7 +454,7 @@ mod tests {
         let first_edge_total: f64 = first_root.local_ranking.edges.values().sum();
         assert_eq!(first_edge_total, 3.0);
 
-        super::catch_up_projection(&log, &entity_store, &projection_store)
+        super::catch_up_projection(&log, &projection_store)
             .await
             .unwrap();
         assert_eq!(projection_store.last_applied_event_count().unwrap(), 1);
@@ -556,9 +566,8 @@ mod tests {
         .unwrap();
         {
             let db = durable::Db::open(tmp.path().join("store")).unwrap();
-            let entity_store = EntityStore::from_db(&db).unwrap();
             let projection_store = ProjectionStore::from_db(&db).unwrap();
-            super::catch_up_projection(&log, &entity_store, &projection_store)
+            super::catch_up_projection(&log, &projection_store)
                 .await
                 .unwrap();
         }
@@ -605,9 +614,8 @@ mod tests {
         .unwrap();
         {
             let db = durable::Db::open(tmp.path().join("store")).unwrap();
-            let entity_store = EntityStore::from_db(&db).unwrap();
             let projection_store = ProjectionStore::from_db(&db).unwrap();
-            super::catch_up_projection(&log, &entity_store, &projection_store)
+            super::catch_up_projection(&log, &projection_store)
                 .await
                 .unwrap();
         }
diff --git a/server/src/storage_dto.rs b/server/src/storage_dto.rs
index de5ed995797ae306d3a71394a4d9d245ffd5771d..9dfb13c53efe4389277625a6ab3bfc18f566a453 100644
--- a/server/src/storage_dto.rs
+++ b/server/src/storage_dto.rs
@@ -3,17 +3,15 @@
 //! Node structure (children, edges, voted pairs, recent votes) is no longer a
 //! single blob — it lives as point-addressable durable collections (see
 //! [`crate::storage_schema`]). This module only defines the small leaf values:
-//! the derived entity view, raw entity payloads, and individual votes.
+//! ephemeral entity views and individual votes.
 
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 
 use crate::{
     path_types::ItemId,
     reducer::{EntityData, VoteData},
 };
 
-pub const ENTITY_RECORD_VERSION: u32 = 1;
 pub const VOTE_RECORD_VERSION: u32 = 1;
 pub const ENTITY_DATA_VERSION: u32 = 1;
 
@@ -29,14 +27,7 @@ impl<T> Versioned<T> {
     }
 }
 
-pub type StoredEntityRecord = Versioned<StoredEntityV1>;
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct StoredEntityV1 {
-    pub json: Value,
-}
-
-/// Derived entity view stored at a node's `data` leaf.
+/// Derived entity view stored at a node's `data` leaf (ephemeral; not logged).
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct StoredEntityDataV1 {
     pub version: u32,
@@ -63,25 +54,6 @@ pub struct StoredVoteV1 {
     pub thread_tag: String,
 }
 
-pub fn encode_entity_payload(payload: &Value) -> StoredEntityRecord {
-    Versioned::new(
-        ENTITY_RECORD_VERSION,
-        StoredEntityV1 {
-            json: payload.clone(),
-        },
-    )
-}
-
-pub fn decode_entity_payload(record: StoredEntityRecord) -> Result<Value, String> {
-    if record.version != ENTITY_RECORD_VERSION {
-        return Err(format!(
-            "unsupported entity record version: {}",
-            record.version
-        ));
-    }
-    Ok(record.payload.json)
-}
-
 pub fn encode_entity_data(data: &EntityData) -> StoredEntityDataV1 {
     StoredEntityDataV1 {
         version: ENTITY_DATA_VERSION,
diff --git a/server/src/storage_schema.rs b/server/src/storage_schema.rs
index 76bf7bc74a2c5ef4f78678a333b7661778f5b835..bd26e665e084b95b10fdfff091c31e8dc84d07b8 100644
--- a/server/src/storage_schema.rs
+++ b/server/src/storage_schema.rs
@@ -15,7 +15,7 @@ use crate::{
     reducer::{EntityData, GroupState, NodeState, VoteData},
     storage_dto::{
         decode_entity_data, decode_vote, encode_entity_data, encode_vote, parse_stored_id,
-        StoredEntityDataV1, StoredEntityRecord, StoredVoteV1,
+        StoredEntityDataV1, StoredVoteV1,
     },
 };
 
@@ -40,17 +40,17 @@ pub struct NodeSchema {
     pub voted_pairs: Map<PairKey, Leaf<bool>>,
     /// Recent votes, newest at the front (capped on write).
     pub recent_votes: Deque<Leaf<StoredVoteV1>>,
+    /// When ephemeral Reddit display content was last fetched (ms); absent after eviction.
+    pub fetched_at: Leaf<i64>,
 }
 
-/// The single database root: nodes, raw payloads, view counts, and per-concern
-/// metadata maps (cursors and schema versions).
+/// The single database root: nodes, view counts, and per-concern metadata maps
+/// (cursors and schema versions).
 #[derive(Durable)]
 #[allow(dead_code)]
 pub struct Store {
     pub nodes: Map<String, NodeSchema>,
     pub proj_meta: Map<String, Leaf<u64>>,
-    pub entities: Map<String, Leaf<StoredEntityRecord>>,
-    pub entity_meta: Map<String, Leaf<u64>>,
     pub view_counts: Map<String, Leaf<u64>>,
     pub view_meta: Map<String, Leaf<u64>>,
 }
@@ -264,12 +264,17 @@ pub fn vote_writes(
     Ok(())
 }
 
-/// Reified writes for an imported entity view (node data + path wiring).
-pub fn entity_view_writes(batch: &mut Batch, id: &ItemId, view: Option<&EntityData>) {
+/// Reified writes for ephemeral Reddit display content (not event-logged).
+pub fn entity_content_writes(batch: &mut Batch, id: &ItemId, view: &EntityData, fetched_at: i64) {
     ensure_path_writes(batch, id);
-    if let Some(view) = view {
-        batch.write(node(id).data().set(&encode_entity_data(view)));
-    }
+    batch.write(node(id).data().set(&encode_entity_data(view)));
+    batch.write(node(id).fetched_at().set(&fetched_at));
+}
+
+/// Clear cached display content for one node (structure/votes are untouched).
+pub fn entity_content_clear_writes(batch: &mut Batch, id: &ItemId) {
+    batch.write(node(id).data().delete());
+    batch.write(node(id).fetched_at().delete());
 }
 
 #[cfg(test)]
diff --git a/test/reddit_import.clj b/test/reddit_import.clj
index b476488526252c13fd73bdda76e5201678e4a714..6d8c5bca7ebad0b5abfddecd4ea48cc09d738ebe 100644
--- a/test/reddit_import.clj
+++ b/test/reddit_import.clj
@@ -59,9 +59,10 @@
                                      "curl" "-sf" browse-url))
           log (slurp (io/file log-path))]
       (is (str/includes? after "The Rust Programming Language"))
-      (is (str/includes? log "\"type\":\"entity_imported\""))
-      (is (str/includes? log "\"subscribers\":350000"))
-      (is (str/includes? log "\"display_name\":\"rust\"")))
+      (is (str/includes? log "\"type\":\"node_ensured\""))
+      (is (not (str/includes? log "\"subscribers\"")))
+      (is (not (str/includes? log "\"display_name\"")))
+      (is (not (str/includes? log "entity_imported"))))
     (let [children-sse (curl-fetch-ui-sse app-base "reddit.com/r/rust" "children")]
       (is (zero? (:exit children-sse)) "POST /ui fetch_entity (children) SSE succeeds")
       (is (str/includes? (:out children-sse) "Idiomorph.morph"))
@@ -71,10 +72,12 @@
           log2 (slurp (io/file log-path))]
       (is (str/includes? after-children "Announcing Rust 1.99"))
       (is (str/includes? after-children "Unranked"))
-      (is (str/includes? log2 "announcing_rust_199")))))
+      (is (str/includes? log2 "\"type\":\"node_ensured\""))
+      (is (str/includes? log2 "/comments/"))
+      (is (not (str/includes? log2 "\"selftext\""))))))
 
 (deftest reddit-fetch-via-mock-api
-  (testing "Fetch more queues import; event log stores full payload; page shows title"
+  (testing "Fetch caches display content ephemerally; log records structure only"
     (let [root (repo-root)
           fixtures (mock-reddit/fixtures-dir root)
           data-dir (.getAbsolutePath