Phase 5: P2P sync (Noise) + replay + merge + federation

gpp-core: read_raw/write_raw (verified frame transfer) + iter_ids.

gpp-sync: Noise_XX over TCP (snow), chunked encrypted transport,
state-vector exchange, set-delta object transfer, ref reconcile with
fork-preserve on divergence, add-only policy union, optional
zero-knowledge Graphex index sync, TOFU static-key pinning, repo-id
gate. Loopback integration tests (converge / fork / mismatch).

gpp-replay: content-addressed environment snapshots, re-materialization,
drift diff, dry-run.

gpp-cli: gpp sync (add/remove/status/serve/peer/all), gpp replay,
gpp merge (two-parent fork merge), gpp graphex federation (add/list).
ROADMAP Phase 5 marked done.

104 workspace tests pass; clippy + rustfmt clean. Verified end-to-end:
two repos sync over real Noise TCP, peer replays the synced changeset.

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

Author: mahabubul470

Cargo.lock

@@ -22,6 +22,8 @@ gpp-trust = { path = "../gpp-trust" }
 gpp-policy = { path = "../gpp-policy" }
 gpp-cost = { path = "../gpp-cost" }
 gpp-anomaly = { path = "../gpp-anomaly" }
+gpp-sync = { path = "../gpp-sync" }
+gpp-replay = { path = "../gpp-replay" }
 globset.workspace = true
 clap.workspace = true
 anyhow.workspace = true

crates/gpp-cli/Cargo.toml

@@ -87,6 +87,62 @@ pub enum Command {
     Anomaly(AnomalyArgs),
     /// Cross-layer audit report
     Audit(AuditArgs),
+    /// Peer-to-peer synchronization
+    Sync(SyncArgs),
+    /// Reproduce a changeset's environment
+    Replay(ReplayArgs),
+    /// Merge a divergent fork branch into the current branch
+    Merge(MergeArgs),
+}
+
+#[derive(Args)]
+pub struct SyncArgs {
+    #[command(subcommand)]
+    pub action: Option<SyncSub>,
+    /// Sync only the Graphex layer (no code objects)
+    #[arg(long, global = true)]
+    pub graph_only: bool,
+    /// Also sync the Graphex index
+    #[arg(long, global = true)]
+    pub include_graphex: bool,
+}
+
+#[derive(Subcommand)]
+pub enum SyncSub {
+    /// Register a peer
+    Add { name: String, address: String },
+    /// Remove a peer
+    Remove { name: String },
+    /// Show configured peers
+    Status,
+    /// Accept inbound syncs on an address (Ctrl-C to stop)
+    Serve { address: String },
+    /// Sync with one peer (default: all configured peers)
+    Peer { name: String },
+}
+
+#[derive(Args)]
+pub struct ReplayArgs {
+    /// Changeset to reproduce (HEAD, a branch, or a hash)
+    pub changeset: String,
+    /// Compare a reproduced dir to the snapshot instead of writing
+    #[arg(long)]
+    pub diff: bool,
+    /// Show what would be reproduced without writing
+    #[arg(long)]
+    pub dry_run: bool,
+    /// Directory to materialize into
+    #[arg(long, default_value = "replay-out")]
+    pub output: std::path::PathBuf,
+    /// Capture/override an env var (key=value, repeatable)
+    #[arg(long = "env", value_name = "K=V")]
+    pub env: Vec<String>,
+}
+
+#[derive(Args)]
+pub struct MergeArgs {
+    /// The fork ref to merge (e.g. "main.fork.office")
+    pub fork_ref: String,
 }
 
 #[derive(Args)]
@@ -323,6 +379,26 @@ pub enum GraphexAction {
     },
     /// Auto-infer proposed module nodes from the HEAD changeset
     Infer,
+    /// Manage cross-project subgraph federation
+    Federation {
+        #[command(subcommand)]
+        action: FederationAction,
+    },
+}
+
+#[derive(Subcommand)]
+pub enum FederationAction {
+    /// Register a federated source (peer project subgraph)
+    Add {
+        #[arg(long)]
+        project: String,
+        #[arg(long)]
+        address: String,
+        #[arg(long, default_value = "default")]
+        subgraph: String,
+    },
+    /// List federated sources
+    List,
 }
 
 #[derive(Args)]

crates/gpp-cli/src/cli.rs

@@ -12,6 +12,7 @@ mod phase1;
 mod phase2;
 mod phase3;
 mod phase4;
+mod phase5;
 mod repo;
 
 use std::process::ExitCode;
@@ -46,6 +47,9 @@ fn main() -> ExitCode {
         Command::Cost(a) => phase4::cost(a, repo_override, args.json),
         Command::Anomaly(a) => phase4::anomaly(a, repo_override),
         Command::Audit(a) => phase4::audit(a, repo_override),
+        Command::Sync(a) => phase5::sync(a, repo_override),
+        Command::Replay(a) => phase5::replay(a, repo_override),
+        Command::Merge(a) => phase5::merge(a, repo_override),
     };
 
     match result {

crates/gpp-cli/src/main.rs

@@ -11,7 +11,9 @@ use gpp_graphex::{
 };
 use gpp_history::{Author, RefStore};
 
-use crate::cli::{GraphexAction, GraphexArgs, KeysAction, KeysArgs, McpServerArgs};
+use crate::cli::{
+    FederationAction, GraphexAction, GraphexArgs, KeysAction, KeysArgs, McpServerArgs,
+};
 use crate::config;
 use crate::phase1::parse_time;
 use crate::repo::Repo;
@@ -390,6 +392,37 @@ pub fn graphex(args: &GraphexArgs, repo_override: Option<&Path>, json: bool) ->
             println!("\nReview with `gpp graphex pending`.");
             Ok(())
         }
+
+        GraphexAction::Federation { action } => {
+            let fed_dir = repo.gpp_dir().join("graphex").join("federation");
+            std::fs::create_dir_all(&fed_dir)?;
+            let sources = fed_dir.join("sources.toml");
+            match action {
+                FederationAction::Add {
+                    project,
+                    address,
+                    subgraph,
+                } => {
+                    let mut body = std::fs::read_to_string(&sources).unwrap_or_default();
+                    body.push_str(&format!(
+                        "[[source]]\nproject = {project:?}\naddress = {address:?}\nsubgraph = {subgraph:?}\n\n"
+                    ));
+                    std::fs::write(&sources, body)?;
+                    println!(
+                        "federated source added: {project} ({subgraph}) @ {address}\n\
+                         pull it with `gpp sync --graph-only` once added as a peer"
+                    );
+                    Ok(())
+                }
+                FederationAction::List => {
+                    match std::fs::read_to_string(&sources) {
+                        Ok(b) if !b.trim().is_empty() => print!("{b}"),
+                        _ => println!("(no federated sources)"),
+                    }
+                    Ok(())
+                }
+            }
+        }
     }
 }