feat: Phase C-2 log-splitting — per-key logs + sync topics

Core of docs/prd/granular-topic-subscription.md: derive p2panda log_id
from interp_key (zodia_ops::log_id_for_key) instead of the hardcoded
INTERP_LOG_ID=0, and add a per-key topic (topic_key_for_interp) so
subscribing to a key's topic actually scopes what replicates.

ZodiaSyncNode now holds a topic->SyncHandle map instead of one fixed
handle; subscribe/unsubscribe open and drop per-key topics on demand.
publish_doc (DocOp, the active write path) routes through the derived
log/topic; publish (legacy InterpOp) stays on log 0 over the global
topic, since existing signed ops can't be re-homed to a new log_id.

App wires the always-subscribed set: every key in the user's own
natal chart is subscribed right after sync spawns, on both cold-start
paths. On-demand subscribe for arbitrary aspect pages, the
grace-period unsubscribe timer, and mock-backed ZodiaSyncNode tests
are not yet implemented — tracked in the PRD's progress notes.

Cargo.lock

@@ -7423,6 +7423,7 @@ dependencies = [
 name = "zodia-ops"
 version = "0.7.1"
 dependencies = [
+ "blake3",
  "ciborium",
  "p2panda-core",
  "serde",

app/src/app.rs

@@ -581,6 +581,9 @@ impl AsyncComponent for AppModel {
                 info!("network up, node ···{}", model.node_id_text);
                 let _ = net.publish_announce().await;
                 model.sync_publish_tx = try_spawn_sync(&model.config, &net, &sender).await;
+                if let (Some(tx), Some(chart)) = (&model.sync_publish_tx, &model.chart) {
+                    subscribe_own_chart_keys(chart, tx);
+                }
                 model.network = Some(Arc::new(net));
                 start_network_command(&sender, rx);
                 sender.input(AppMsg::NetworkReady);
@@ -646,6 +649,9 @@ impl AsyncComponent for AppModel {
                     };
                     let _ = net.publish_announce().await;
                     self.sync_publish_tx = try_spawn_sync(&self.config, &net, &sender).await;
+                    if let (Some(tx), Some(chart)) = (&self.sync_publish_tx, &self.chart) {
+                        subscribe_own_chart_keys(chart, tx);
+                    }
                     self.network = Some(Arc::new(net));
                     start_network_command(&sender, rx);
                     sender.input(AppMsg::NetworkReady);
@@ -2406,10 +2412,31 @@ async fn try_spawn_network(
     }
 }
 
+/// Always-subscribed set (Phase C-2): the keys in the user's own natal
+/// chart, so the home aspect list and Sky feed stay live without needing a
+/// per-page subscribe on every cold start. Fire-and-forget over `try_send`
+/// like every other publish-channel send in this file — subscribing is not
+/// latency-critical and the channel has slack.
+fn subscribe_own_chart_keys(chart: &Chart, tx: &tokio::sync::mpsc::Sender<SyncPublishMsg>) {
+    for aspect in chart.natal_aspects() {
+        let key = zodia_core::InterpKey::from_natal(&aspect).to_sig();
+        let _ = tx.try_send(SyncPublishMsg::Subscribe(key));
+    }
+}
+
 /// Message type for sending publish requests to the background sync task.
 pub(crate) enum SyncPublishMsg {
     Publish(InterpOp),
     PublishDoc(zodia_ops::DocOp),
+    /// Open a key's per-key sync topic (Phase C-2). Idempotent.
+    Subscribe(String),
+    /// Close a key's per-key sync topic. Idempotent. Not yet sent by any
+    /// caller — the grace-period-unsubscribe UI wiring (open on aspect-page
+    /// visit, unsubscribe after N idle minutes) is unimplemented; this
+    /// variant exists so `zodia_sync::ZodiaSyncNode::unsubscribe` is already
+    /// reachable once that lands. See docs/prd/granular-topic-subscription.md.
+    #[allow(dead_code)]
+    Unsubscribe(String),
 }
 
 /// Spawn the LogSync background pump and return a channel for publishing.
@@ -2465,6 +2492,14 @@ async fn try_spawn_sync(
                                 warn!("sync publish_doc: {e}");
                             }
                         }
+                        SyncPublishMsg::Subscribe(key) => {
+                            if let Err(e) = node.subscribe(&key).await {
+                                warn!("sync subscribe {key}: {e}");
+                            }
+                        }
+                        SyncPublishMsg::Unsubscribe(key) => {
+                            node.unsubscribe(&key);
+                        }
                     }
                 }
                 Some(sync_event) = node.inbound.recv() => {

core/src/lib.rs

@@ -21,6 +21,7 @@ pub use ephemeris::{EphemerisError, compute_positions};
 pub use houses::{HouseError, HouseKind, HouseSystem};
 pub use interp::{Angle, InterpKey, InterpKind, humanize_key, parse_interp_sig};
 pub use planet::{Planet, PlanetPositions};
-pub use topic::{TopicKey, solar_longitude, solar_month, topic_key_global, topic_keys_for_chart};
+pub use topic::{TopicKey, solar_longitude, solar_month, topic_key_for_interp,
+                topic_key_global, topic_keys_for_chart};
 pub use transit::{HouseTransit, TransitAspect, TransitSet, build_transit_set,
                   compute_transit_aspects, house_transit_window, transit_window};

core/src/topic.rs

@@ -93,6 +93,14 @@ pub fn topic_key_global() -> TopicKey {
     hash_topic("zodia:v1:global")
 }
 
+/// Per-`interp_key` sync topic (Phase C-2 — granular subscription).  Scopes
+/// LogSync replication to exactly the key a subscriber cares about, paired
+/// with a `log_id` derived the same way (see `zodia_ops::log_id_for_key`)
+/// so the log synced over this topic actually contains only this key's ops.
+pub fn topic_key_for_interp(interp_key: &str) -> TopicKey {
+    hash_topic(&format!("zodia:v1:interp:{interp_key}"))
+}
+
 // ── solar position helpers (kept for Tier-0 blob metadata) ───────────────────
 
 

docs/prd/granular-topic-subscription.md

@@ -1,7 +1,7 @@
 # PRD: Granular per-key sync topics with lazy subscription (Phase C-2)
 
-**Status:** needs-triage
-**Branch:** TBD (will be `feat/granular-topics`)
+**Status:** partially shipped — log-splitting + always-subscribe done; on-demand page subscribe, grace-period unsubscribe, and mock-backed tests not started
+**Branch:** `main`
 **Foundation already landed:** 0.7.0 (`zodia-ops` + `zodia-pipeline`, single global sync topic) and 0.7.1 (activity feed, collaborative interpretations — both still ride the single global topic).
 **Supersedes:** the "Topic granularity and lazy subscription" sketch in `docs/prd/operations-and-streams-rearchitecture.md` (§ Implementation Decisions). That sketch assumed per-key topics were a routing-layer change; this PRD corrects that assumption against what actually shipped and defines the real migration.
 
@@ -76,6 +76,12 @@ Before this phase, opening any aspect page works identically regardless of wheth
 
 This is a real, user-visible regression during the transition window — a page that used to show community interpretations instantly may show fewer of them right after upgrade, recovering as more of the network's *active* keys get touched again post-migration. This needs a line in the eventual release notes; see Migration story below.
 
+## Progress notes
+
+Shipped: `log_id_for_key` (`ops/src/lib.rs`), `DocOp::interp_key()`, `topic_key_for_interp` (`core/src/topic.rs`), and the `ZodiaSyncNode` multi-topic refactor (`sync/src/lib.rs` — `HashMap<Topic, SyncHandle>`, `subscribe`/`unsubscribe`, `publish_doc` routed through the derived log/topic, legacy `publish` untouched on log 0). `app/src/app.rs` sends `Subscribe` for every natal-chart key right after sync spawns (both cold-start paths).
+
+Not shipped: opening a key's topic when its aspect page is visited outside the always-subscribed set, the grace-period unsubscribe timer (`SyncPublishMsg::Unsubscribe` exists but nothing sends it yet), and the mock-backed `ZodiaSyncNode` tests from Testing Decisions below — this crate has no mock `LogSync`/`SyncHandle` harness yet, unlike `zodia-pipeline`'s fake-stream tests.
+
 ## Testing Decisions
 
 Following the existing pattern (`zodia-pipeline` processors tested against a fake `Stream<InterpOp>`, no live-iroh integration tests):

ops/Cargo.toml

@@ -10,3 +10,4 @@ ciborium.workspace = true
 serde.workspace = true
 thiserror.workspace = true
 p2panda-core = "0.6"
+blake3.workspace = true

ops/src/lib.rs

@@ -125,6 +125,17 @@ pub enum DocOp {
 }
 
 impl DocOp {
+    /// The key this op targets. Every variant carries one — used to route
+    /// publishes to the per-key log (see `log_id_for_key`).
+    pub fn interp_key(&self) -> &str {
+        match self {
+            DocOp::Edit { interp_key, .. }
+            | DocOp::Veto { interp_key, .. }
+            | DocOp::AffirmRev { interp_key, .. }
+            | DocOp::EditorPresence { interp_key, .. } => interp_key,
+        }
+    }
+
     pub fn encode(&self) -> Vec<u8> {
         let mut buf = Vec::new();
         ciborium::into_writer(self, &mut buf).expect("ciborium encode infallible for owned data");
@@ -143,6 +154,23 @@ mod doc_op_tests {
         Hash::from_bytes([7u8; 32])
     }
 
+    #[test]
+    fn interp_key_accessor_covers_every_variant() {
+        assert_eq!(DocOp::Edit {
+            interp_key: "natal:sun_trine_moon".into(), base_rev: sample_hash(),
+            crdt_update: vec![], affected_blocks: vec![],
+        }.interp_key(), "natal:sun_trine_moon");
+        assert_eq!(DocOp::Veto {
+            interp_key: "natal:x".into(), target_edit_op_id: sample_hash(),
+        }.interp_key(), "natal:x");
+        assert_eq!(DocOp::AffirmRev {
+            interp_key: "natal:y".into(), target_rev: [0u8; 32],
+        }.interp_key(), "natal:y");
+        assert_eq!(DocOp::EditorPresence {
+            interp_key: "natal:z".into(), joined: true,
+        }.interp_key(), "natal:z");
+    }
+
     #[test]
     fn edit_roundtrip() {
         let op = DocOp::Edit {
@@ -201,6 +229,16 @@ impl InterpOp {
     }
 }
 
+// ── per-key log routing (Phase C-2) ─────────────────────────────────────────────
+
+/// Derive a p2panda `log_id` from an `interp_key` so each key gets its own
+/// per-author log — required for per-key sync topics to actually scope
+/// what they replicate (see `docs/prd/granular-topic-subscription.md`).
+pub fn log_id_for_key(interp_key: &str) -> u64 {
+    let hash = blake3::hash(format!("interp-log:v1:{interp_key}").as_bytes());
+    u64::from_le_bytes(hash.as_bytes()[..8].try_into().unwrap())
+}
+
 // ── error ─────────────────────────────────────────────────────────────────────
 
 #[derive(Debug, Error)]
@@ -261,6 +299,43 @@ mod tests {
         assert_eq!(op, InterpOp::decode(&bytes).unwrap());
     }
 
+    #[test]
+    fn log_id_for_key_deterministic() {
+        assert_eq!(log_id_for_key("natal:sun_trine_moon"), log_id_for_key("natal:sun_trine_moon"));
+    }
+
+    #[test]
+    fn log_id_for_key_no_collisions_over_synthetic_keyspace() {
+        // Every planet pair x every aspect kind x natal/transit/sky/house
+        // prefix — a realistic upper bound on real key volume, larger than
+        // any single author will plausibly publish to.
+        const PLANETS: &[&str] = &[
+            "sun", "moon", "mercury", "venus", "mars",
+            "jupiter", "saturn", "uranus", "neptune", "pluto",
+        ];
+        const ASPECTS: &[&str] = &[
+            "conjunction", "opposition", "square", "trine", "sextile",
+            "quincunx", "semi_sextile", "semi_square", "sesquiquadrate",
+        ];
+        const PREFIXES: &[&str] = &["natal", "transit", "sky"];
+
+        let mut seen = std::collections::HashSet::new();
+        let mut count = 0usize;
+        for prefix in PREFIXES {
+            for a in PLANETS {
+                for b in PLANETS {
+                    if a == b { continue; }
+                    for aspect in ASPECTS {
+                        let key = format!("{prefix}:{a}_{aspect}_{b}");
+                        count += 1;
+                        assert!(seen.insert(log_id_for_key(&key)), "collision on {key}");
+                    }
+                }
+            }
+        }
+        assert!(count > 2000, "sanity: expected a large synthetic keyspace, got {count}");
+    }
+
     #[test]
     fn decode_rejects_garbage() {
         assert!(InterpOp::decode(&[0xff; 8]).is_err());

sync/src/lib.rs

@@ -22,11 +22,12 @@
 //! `LogSync` uses for both `LogStore` and `TopicStore` duties.  Crashing
 //! mid-sync no longer loses the local log; the new node will reuse it.
 
+use std::collections::HashMap;
 use std::path::Path;
 
 use futures_util::StreamExt;
 use p2panda_core::{Body, Header, Operation, SigningKey, Timestamp, Topic, VerifyingKey};
-use p2panda_net::sync::LogSync;
+use p2panda_net::sync::{LogSync, SyncHandle};
 use p2panda_net::{Endpoint, Gossip};
 use p2panda_store::logs::LogStore;
 use p2panda_store::operations::OperationStore;
@@ -36,7 +37,8 @@ use thiserror::Error;
 use tokio::sync::mpsc;
 use tracing::{debug, warn};
 
-use zodia_ops::{DocOp, InterpOp};
+use zodia_core::topic_key_for_interp;
+use zodia_ops::{DocOp, InterpOp, log_id_for_key};
 
 // ── sync event ────────────────────────────────────────────────────────────────
 
@@ -69,7 +71,12 @@ pub enum SyncEvent {
 
 // ── log id ────────────────────────────────────────────────────────────────────
 
-/// Each author has exactly one log containing all their interpretations.
+/// Legacy log: every pre-Phase-C-2 `InterpOp` an author ever published.
+/// Signed operations can't be re-homed to a derived `log_id` (the p2panda
+/// header signature covers `log_id`), so this stays the permanent address
+/// of pre-migration history — `InterpOp::publish` still targets it.  New
+/// `DocOp` writes use `zodia_ops::log_id_for_key` instead (see
+/// `docs/prd/granular-topic-subscription.md`).
 const INTERP_LOG_ID: u64 = 0;
 
 // ── errors ────────────────────────────────────────────────────────────────────
@@ -86,13 +93,26 @@ pub enum SyncError {
 
 /// The live sync handle.  Keeps the p2panda LogSync machinery alive and
 /// exposes a raw `Operation<()>` channel for the app's pipeline to consume.
+///
+/// Phase C-2: holds one `SyncHandle` per subscribed topic rather than a
+/// single fixed one.  `global_topic` (legacy `InterpOp` traffic, log 0)
+/// is always subscribed; per-key topics for `DocOp` traffic come and go
+/// as the app calls `subscribe`/`unsubscribe`.
 pub struct ZodiaSyncNode {
     /// p2panda signing key — same bytes as the Zodia identity `SigningKey`.
-    signing_key: SigningKey,
+    signing_key:  SigningKey,
     /// File-backed p2panda operation store.
-    sync_store:  SqliteStore,
-    /// LogSync handle for our single sync topic.
-    handle:      p2panda_net::sync::SyncHandle<Operation<()>, TopicLogSyncEvent<()>>,
+    sync_store:   SqliteStore,
+    /// Shared LogSync engine — `.stream()` opens a new topic subscription
+    /// without needing a fresh endpoint/gossip pair.
+    log_sync:     LogSync<SqliteStore, u64, ()>,
+    /// The always-on legacy topic; `publish` (InterpOp) targets this one.
+    global_topic: Topic,
+    /// Forwarder-task sender, cloned into each newly opened topic's pump.
+    ev_tx:        mpsc::Sender<SyncEvent>,
+    /// Live handles keyed by topic. Dropping an entry ends that topic's
+    /// sync session (`SyncHandle::drop` sends `ToSyncManager::Close`).
+    handles:      HashMap<Topic, SyncHandle<Operation<()>, TopicLogSyncEvent<()>>>,
     /// Mixed-purpose channel: operation arrivals plus lifecycle events
     /// (session start / finish / failure).  Operations feed the pipeline;
     /// lifecycle events drive UI sync-status indicators.
@@ -100,12 +120,13 @@ pub struct ZodiaSyncNode {
 }
 
 impl ZodiaSyncNode {
-    /// Spawn the sync node.
+    /// Spawn the sync node, opening `sync_topic` (the legacy global topic)
+    /// immediately.
     ///
     /// * `signing_key` — the local identity p2panda `SigningKey`
     /// * `endpoint`    — clone of `ZodiaNetwork`'s iroh endpoint
     /// * `gossip`      — clone of `ZodiaNetwork`'s gossip engine
-    /// * `sync_topic`  — the sync topic (use `Topic::from(topic_key_global().0)`)
+    /// * `sync_topic`  — the legacy sync topic (use `Topic::from(topic_key_global().0)`)
     /// * `store_dir`   — directory in which the sync-store SQLite file lives
     pub async fn spawn(
         signing_key:  SigningKey,
@@ -129,15 +150,49 @@ impl ZodiaSyncNode {
             .await
             .map_err(|e| SyncError::Sync(format!("{e:?}")))?;
 
-        let handle = log_sync
-            .stream(sync_topic, true)
+        let (ev_tx, ev_rx) = mpsc::channel::<SyncEvent>(256);
+
+        let mut node = Self {
+            signing_key,
+            sync_store,
+            log_sync,
+            global_topic: sync_topic,
+            ev_tx,
+            handles: HashMap::new(),
+            inbound: ev_rx,
+        };
+        node.open_topic(sync_topic).await?;
+
+        Ok(node)
+    }
+
+    /// Subscribe to a key's per-key topic (Phase C-2).  Idempotent — a
+    /// key already subscribed is a no-op.  `DocOp` traffic for `interp_key`
+    /// only reaches this device while subscribed.
+    pub async fn subscribe(&mut self, interp_key: &str) -> Result<(), SyncError> {
+        self.open_topic(Topic::from(topic_key_for_interp(interp_key).0)).await
+    }
+
+    /// Unsubscribe from a key's per-key topic.  No-op if not subscribed.
+    /// Dropping the handle ends the sync session (`SyncHandle::drop`).
+    pub fn unsubscribe(&mut self, interp_key: &str) {
+        let topic = Topic::from(topic_key_for_interp(interp_key).0);
+        self.handles.remove(&topic);
+    }
+
+    /// Open (if not already) a LogSync stream for `topic` and start its
+    /// forwarder task.  Shared by `spawn`'s global-topic bootstrap and
+    /// `subscribe`'s per-key topics.
+    async fn open_topic(&mut self, topic: Topic) -> Result<(), SyncError> {
+        if self.handles.contains_key(&topic) {
+            return Ok(());
+        }
+
+        let handle = self.log_sync
+            .stream(topic, true)
             .await
             .map_err(|e| SyncError::Sync(format!("{e:?}")))?;
 
-        let (ev_tx, ev_rx) = mpsc::channel::<SyncEvent>(256);
-
-        // ── subscription background task ──────────────────────────────────────
-        //
         // Thin forwarder: every `OperationReceived` becomes a SyncEvent::
         // OperationReceived; lifecycle events become SyncEvent variants so
         // the app can drive sync-status UI off them.
@@ -145,6 +200,7 @@ impl ZodiaSyncNode {
             .subscribe()
             .await
             .map_err(|e| SyncError::Sync(format!("{e:?}")))?;
+        let ev_tx = self.ev_tx.clone();
 
         tokio::spawn(async move {
             while let Some(result) = subscription.next().await {
@@ -188,27 +244,37 @@ impl ZodiaSyncNode {
             }
         });
 
-        Ok(Self {
-            signing_key,
-            sync_store,
-            handle,
-            inbound: ev_rx,
-        })
+        self.handles.insert(topic, handle);
+        Ok(())
     }
 
-    /// Publish a locally authored `InterpOp` to the p2panda log.  See
-    /// [`Self::publish_doc`] for the Phase F-collab `DocOp` equivalent.
+    /// Publish a locally authored `InterpOp` to the legacy global log
+    /// (log 0). See [`Self::publish_doc`] for the Phase F-collab `DocOp`
+    /// equivalent, which routes to a per-key log/topic instead.
     pub async fn publish(&mut self, op: InterpOp) -> Result<(), SyncError> {
-        self.publish_bytes(op.encode()).await
+        let topic = self.global_topic;
+        self.publish_bytes(op.encode(), INTERP_LOG_ID, topic).await
     }
 
-    /// Publish a locally authored `DocOp` (Phase F-collab) to the same
-    /// log.  Same backlink/seq/sign mechanics as [`Self::publish`].
+    /// Publish a locally authored `DocOp` (Phase F-collab) to its key's
+    /// per-key log/topic (Phase C-2), subscribing first if not already —
+    /// publishing into a topic you're not on isn't meaningful, so this
+    /// implicitly opens it, mirroring "the page you're editing is already
+    /// subscribed" from the app-layer lifecycle policy.
     pub async fn publish_doc(&mut self, op: DocOp) -> Result<(), SyncError> {
-        self.publish_bytes(op.encode()).await
+        let interp_key = op.interp_key().to_string();
+        let log_id = log_id_for_key(&interp_key);
+        let topic = Topic::from(topic_key_for_interp(&interp_key).0);
+        self.open_topic(topic).await?;
+        self.publish_bytes(op.encode(), log_id, topic).await
     }
 
-    async fn publish_bytes(&mut self, payload_bytes: Vec<u8>) -> Result<(), SyncError> {
+    async fn publish_bytes(
+        &mut self,
+        payload_bytes: Vec<u8>,
+        log_id:        u64,
+        topic:         Topic,
+    ) -> Result<(), SyncError> {
         // p2panda-store's `insert_operation` runs inside a transaction
         // started by `begin()`.  Without it, the store returns
         // `TransactionMissing` ("tried to interact with inexistant
@@ -222,7 +288,7 @@ impl ZodiaSyncNode {
 
         // Determine the next sequence number + backlink from our log tip.
         let latest: Option<Operation<()>> = self.sync_store
-            .get_latest_entry(&self.signing_key.verifying_key(), &INTERP_LOG_ID)
+            .get_latest_entry(&self.signing_key.verifying_key(), &log_id)
             .await
             .map_err(|e| SyncError::PandaStore(e.to_string()))?;
 
@@ -254,7 +320,7 @@ impl ZodiaSyncNode {
         };
 
         if let Err(e) = self.sync_store
-            .insert_operation(&op_hash, &operation, &INTERP_LOG_ID)
+            .insert_operation(&op_hash, &operation, &log_id)
             .await
         {
             // Rollback drops the permit and frees the semaphore so the
@@ -268,7 +334,11 @@ impl ZodiaSyncNode {
             .await
             .map_err(|e| SyncError::PandaStore(e.to_string()))?;
 
-        self.handle
+        // `open_topic` (called by every publish path above) guarantees an
+        // entry exists for `topic` by the time we get here.
+        self.handles
+            .get(&topic)
+            .expect("publish_bytes called after open_topic")
             .publish(operation)
             .await
             .map_err(|e| SyncError::Sync(format!("{e:?}")))?;