diff --git a/src/commands/publish/mod.rs b/src/commands/publish/mod.rs
index cc17311..44bca0c 100644
--- a/src/commands/publish/mod.rs
+++ b/src/commands/publish/mod.rs
@@ -19,7 +19,8 @@ use crate::store::{load_active_bundle_for_update, save_active_bundle};
 use anyhow::{bail, Context, Result};
 use remote::{
     apply_artifact_publish_result, apply_publish_remote_result, publish_repo_remote,
-    publish_repo_remote_from_artifact, ArtifactPublishResult, PublishJob, PublishRemoteResult,
+    publish_repo_remote_from_artifact, report_publish_remote_result, report_pushed, PublishEvent,
+    PublishJob, PublishRemoteResult,
 };
 pub(crate) use scope::publish_scope_repo_ids;
 use scope::{
@@ -70,45 +71,80 @@ pub fn create_publications(
         })
         .collect();
 
-    let results: Vec<(String, Result<PublishRemoteResult>)> = std::thread::scope(|scope| {
+    let total = jobs.len();
+    if total > 1 {
+        println!("{}", out::muted(format!("publishing {total} repo(s)…")));
+    }
+
+    // Workers stream their steps over a channel so every repo's push and
+    // review object are printed the moment they exist, not after the slowest
+    // worker joined. The bundle is updated afterwards, once the workers have
+    // released their borrow of it.
+    let (tx, rx) = std::sync::mpsc::channel();
+    let outcomes: Vec<PublishRemoteResult> = std::thread::scope(|scope| {
         let active = &active;
         let bundle = &bundle_snapshot;
-        let handles: Vec<_> = jobs
-            .iter()
-            .map(|job| {
-                let job = job.clone();
+        for job in &jobs {
+            let job = job.clone();
+            let tx = tx.clone();
+            scope.spawn(move || {
                 let repo_id = job.repo.id.clone();
-                scope.spawn(move || {
-                    (
-                        repo_id,
-                        publish_repo_remote(active, bundle, &job, draft, renew, set_upstream),
-                    )
-                })
-            })
-            .collect();
-
-        handles
-            .into_iter()
-            .map(|handle| handle.join().expect("publish worker thread panicked"))
-            .collect()
-    });
-
-    for (repo_id, result) in results {
-        match result {
-            Ok(outcome) => {
-                if apply_publish_remote_result(&mut active, &outcome)? {
-                    bundle_changed = true;
-                }
-            }
-            Err(error) => {
-                println!(
-                    "{}: {}",
-                    out::repo(&repo_id),
-                    out::danger("PR create failed")
+                let on_pushed = |pushed| {
+                    let _ = tx.send(PublishEvent::Pushed {
+                        repo_id: repo_id.clone(),
+                        pushed,
+                    });
+                };
+                let result = publish_repo_remote(
+                    active,
+                    bundle,
+                    &job,
+                    draft,
+                    renew,
+                    set_upstream,
+                    &on_pushed,
                 );
-                failures.push(format!("{repo_id}: {error:#}"));
+                // The receiver outlives every worker; a send cannot fail.
+                let _ = tx.send(PublishEvent::Done {
+                    repo_id,
+                    result: Box::new(result),
+                });
+            });
+        }
+        drop(tx);
+
+        let mut outcomes = Vec::new();
+        let mut done = 0;
+        for event in rx {
+            match event {
+                PublishEvent::Pushed { repo_id, pushed } => report_pushed(&repo_id, &pushed),
+                PublishEvent::Done { repo_id, result } => {
+                    done += 1;
+                    let progress = out::progress(done, total);
+                    match *result {
+                        Ok(outcome) => {
+                            report_publish_remote_result(&outcome, &progress);
+                            outcomes.push(outcome);
+                        }
+                        Err(error) => {
+                            println!(
+                                "{}: {}{progress}",
+                                out::repo(&repo_id),
+                                out::danger("PR create failed")
+                            );
+                            failures.push(format!("{repo_id}: {error:#}"));
+                        }
+                    }
+                }
             }
         }
+        outcomes
+    });
+
+    for outcome in &outcomes {
+        if apply_publish_remote_result(&mut active, outcome)? {
+            bundle_changed = true;
+        }
     }
 
     if bundle_changed {
@@ -196,46 +232,44 @@ pub fn create_publications_from_artifact(
         })
         .collect();
 
-    let results: Vec<(String, Result<ArtifactPublishResult>)> = std::thread::scope(|scope| {
+    let total = jobs.len();
+    if total > 1 {
+        println!("{}", out::muted(format!("publishing {total} repo(s)…")));
+    }
+
+    // Same streaming shape as the worktree path: workers publish against the
+    // snapshot while the live artifact is updated as each result arrives.
+    let (tx, rx) = std::sync::mpsc::channel();
+    std::thread::scope(|scope| {
         let cwd = cwd.as_ref();
-        let bundle = &bundle_snapshot;
-        let handles: Vec<_> = jobs
-            .iter()
-            .map(|job| {
-                let job = job.clone();
+        let snapshot = &bundle_snapshot;
+        for job in &jobs {
+            let job = job.clone();
+            let tx = tx.clone();
+            scope.spawn(move || {
                 let repo_id = job.repo.id.clone();
-                scope.spawn(move || {
-                    (
-                        repo_id,
-                        publish_repo_remote_from_artifact(cwd, bundle, &job, draft, renew),
-                    )
-                })
-            })
-            .collect();
-
-        handles
-            .into_iter()
-            .map(|handle| {
-                handle
-                    .join()
-                    .expect("artifact publish worker thread panicked")
-            })
-            .collect()
-    });
-
-    for (repo_id, result) in results {
-        match result {
-            Ok(outcome) => apply_artifact_publish_result(&mut bundle, &outcome),
-            Err(error) => {
-                println!(
-                    "{}: {}",
-                    out::repo(&repo_id),
-                    out::danger("PR create failed")
-                );
-                failures.push(format!("{repo_id}: {error:#}"));
+                let result = publish_repo_remote_from_artifact(cwd, snapshot, &job, draft, renew);
+                // The receiver outlives every worker; a send cannot fail.
+                let _ = tx.send((repo_id, result));
+            });
+        }
+        drop(tx);
+
+        for (done, (repo_id, result)) in rx.into_iter().enumerate() {
+            let progress = out::progress(done + 1, total);
+            match result {
+                Ok(outcome) => apply_artifact_publish_result(&mut bundle, &outcome, &progress),
+                Err(error) => {
+                    println!(
+                        "{}: {}{progress}",
+                        out::repo(&repo_id),
+                        out::danger("PR create failed")
+                    );
+                    failures.push(format!("{repo_id}: {error:#}"));
+                }
             }
         }
-    }
+    });
 
     if failures.is_empty() && sync {
         failures.extend(sync_publications_for_indexes_from_artifact(
diff --git a/src/commands/publish/remote.rs b/src/commands/publish/remote.rs
index 760de54..67bed3f 100644
--- a/src/commands/publish/remote.rs
+++ b/src/commands/publish/remote.rs
@@ -21,11 +21,26 @@ pub(super) struct PublishJob {
     pub(super) base_branch: String,
 }
 
+#[derive(Clone)]
 pub(super) struct PushedInfo {
     sha: String,
     branch: String,
 }
 
+/// What a publish worker reports while it runs. `Pushed` fires as soon as the
+/// branch is on origin, before the review object is looked up or created, so
+/// the caller can print progress per step instead of per joined worker.
+pub(super) enum PublishEvent {
+    Pushed {
+        repo_id: String,
+        pushed: PushedInfo,
+    },
+    Done {
+        repo_id: String,
+        result: Box<Result<PublishRemoteResult>>,
+    },
+}
+
 pub(super) enum PublishStatus {
     ExistsRecorded(String),
     FoundExisting(PullRequest),
@@ -35,7 +50,6 @@ pub(super) enum PublishStatus {
 pub(super) struct PublishRemoteResult {
     pub(super) repo_index: usize,
     repo_id: String,
-    pushed: PushedInfo,
     status: PublishStatus,
 }
 
@@ -52,6 +66,7 @@ pub(super) fn publish_repo_remote(
     draft: bool,
     renew: bool,
     set_upstream: bool,
+    on_pushed: &dyn Fn(PushedInfo),
 ) -> Result<PublishRemoteResult> {
     let repo = &job.repo;
     let base_branch = &job.base_branch;
@@ -77,6 +92,7 @@ pub(super) fn publish_repo_remote(
         sha,
         branch: format!("origin/{branch}"),
     };
+    on_pushed(pushed.clone());
 
     if let Some(existing) = publication_for_repo(bundle, &repo.id) {
         if existing.base_branch != *base_branch {
@@ -90,7 +106,6 @@ pub(super) fn publish_repo_remote(
             return Ok(PublishRemoteResult {
                 repo_index: job.repo_index,
                 repo_id: repo.id.clone(),
-                pushed,
                 status: PublishStatus::ExistsRecorded(existing.url.clone()),
             });
         }
@@ -110,7 +125,6 @@ pub(super) fn publish_repo_remote(
             return Ok(PublishRemoteResult {
                 repo_index: job.repo_index,
                 repo_id: repo.id.clone(),
-                pushed,
                 status: PublishStatus::FoundExisting(existing),
             });
         }
@@ -136,7 +150,6 @@ pub(super) fn publish_repo_remote(
     Ok(PublishRemoteResult {
         repo_index: job.repo_index,
         repo_id: repo.id.clone(),
-        pushed,
         status: PublishStatus::Created(summary),
     })
 }
@@ -228,54 +241,63 @@ pub(super) fn publish_repo_remote_from_artifact(
     })
 }
 
-pub(super) fn apply_publish_remote_result(
-    active: &mut ActiveBundle,
-    outcome: &PublishRemoteResult,
-) -> Result<bool> {
+/// Print the `pushed` line for a repo whose branch just reached origin.
+pub(super) fn report_pushed(repo_id: &str, pushed: &PushedInfo) {
     println!(
         "{}: {} {} {}",
-        out::repo(&outcome.repo_id),
+        out::repo(repo_id),
         out::movement("pushed"),
-        out::branch(&outcome.pushed.branch),
-        out::sha(short_sha(&outcome.pushed.sha))
+        out::branch(&pushed.branch),
+        out::sha(short_sha(&pushed.sha))
     );
+}
+
+/// Print the review-object line for one repo. `progress` is the streamed
+/// ` (done/total)` tail from [`out::progress`].
+fn report_status(repo_id: &str, status: &PublishStatus, progress: &str) {
+    match status {
+        PublishStatus::ExistsRecorded(url) => println!(
+            "{}: {} {}{progress}",
+            out::repo(repo_id),
+            out::movement("exists"),
+            url
+        ),
+        PublishStatus::FoundExisting(summary) => println!(
+            "{}: {} {}{progress}",
+            out::repo(repo_id),
+            out::movement("exists"),
+            summary.url
+        ),
+        PublishStatus::Created(summary) => println!(
+            "{}: {} #{} {}{progress}",
+            out::repo(repo_id),
+            out::movement("created"),
+            summary.number,
+            summary.url
+        ),
+    }
+}
+
+pub(super) fn report_publish_remote_result(outcome: &PublishRemoteResult, progress: &str) {
+    report_status(&outcome.repo_id, &outcome.status, progress);
+}
 
+/// Record a worker's outcome on the bundle. Reporting is separate
+/// ([`report_publish_remote_result`]) because the bundle is still borrowed by
+/// the other workers while results stream in.
+pub(super) fn apply_publish_remote_result(
+    active: &mut ActiveBundle,
+    outcome: &PublishRemoteResult,
+) -> Result<bool> {
     let repo = active.bundle.repos[outcome.repo_index].clone();
-    let mut changed = false;
     match &outcome.status {
-        PublishStatus::ExistsRecorded(url) => {
-            println!(
-                "{}: {} {}",
-                out::repo(&outcome.repo_id),
-                out::movement("exists"),
-                url
-            );
-        }
+        PublishStatus::ExistsRecorded(_) => Ok(false),
         PublishStatus::FoundExisting(summary) | PublishStatus::Created(summary) => {
             let forge = providers::for_repo(&repo)?;
             providers::upsert_publication(&mut active.bundle, &repo, forge.as_ref(), summary);
-            let pr = publication_for_repo(&active.bundle, &outcome.repo_id)
-                .expect("publication was just inserted");
-            match &outcome.status {
-                PublishStatus::FoundExisting(_) => println!(
-                    "{}: {} {}",
-                    out::repo(&outcome.repo_id),
-                    out::movement("exists"),
-                    pr.url
-                ),
-                PublishStatus::Created(_) => println!(
-                    "{}: {} #{} {}",
-                    out::repo(&outcome.repo_id),
-                    out::movement("created"),
-                    pr.number,
-                    pr.url
-                ),
-                PublishStatus::ExistsRecorded(_) => unreachable!(),
-            }
-            changed = true;
+            Ok(true)
         }
     }
-    Ok(changed)
 }
 
 fn ensure_review_can_be_renewed(
@@ -333,39 +355,14 @@ fn review_is_terminal(review: &PullRequest) -> bool {
 pub(super) fn apply_artifact_publish_result(
     bundle: &mut ChangeGroup,
     outcome: &ArtifactPublishResult,
+    progress: &str,
 ) {
+    report_status(&outcome.repo_id, &outcome.status, progress);
     let repo = bundle.repos[outcome.repo_index].clone();
-    match &outcome.status {
-        PublishStatus::ExistsRecorded(url) => {
-            println!(
-                "{}: {} {}",
-                out::repo(&outcome.repo_id),
-                out::movement("exists"),
-                url
-            );
-        }
-        PublishStatus::FoundExisting(summary) | PublishStatus::Created(summary) => {
-            let forge = providers::for_repo(&repo).expect("forge resolves for published repo");
-            providers::upsert_publication(bundle, &repo, forge.as_ref(), summary);
-            let pr = publication_for_repo(bundle, &outcome.repo_id)
-                .expect("publication was just inserted");
-            match &outcome.status {
-                PublishStatus::FoundExisting(_) => println!(
-                    "{}: {} {}",
-                    out::repo(&outcome.repo_id),
-                    out::movement("exists"),
-                    pr.url
-                ),
-                PublishStatus::Created(_) => println!(
-                    "{}: {} #{} {}",
-                    out::repo(&outcome.repo_id),
-                    out::movement("created"),
-                    pr.number,
-                    pr.url
-                ),
-                PublishStatus::ExistsRecorded(_) => unreachable!(),
-            }
-        }
+    if let PublishStatus::FoundExisting(summary) | PublishStatus::Created(summary) = &outcome.status
+    {
+        let forge = providers::for_repo(&repo).expect("forge resolves for published repo");
+        providers::upsert_publication(bundle, &repo, forge.as_ref(), summary);
     }
 }
 
diff --git a/src/commands/push.rs b/src/commands/push.rs
index 34c588c..bc3f12e 100644
--- a/src/commands/push.rs
+++ b/src/commands/push.rs
@@ -72,41 +72,54 @@ pub fn push_repos(
     }
 
     let indexes = resolve_repo_indexes(&active, selectors, all)?;
-    let results: Vec<(String, Result<PushSuccess>)> = std::thread::scope(|scope| {
-        let handles: Vec<_> = indexes
-            .iter()
-            .map(|&index| {
-                let active = &active;
-                let repo = &active.bundle.repos[index];
-                let repo_id = repo.id.clone();
-                scope.spawn(move || (repo_id, push_repo(active, repo, set_upstream, force)))
-            })
-            .collect();
-
-        handles
-            .into_iter()
-            .map(|handle| handle.join().expect("push worker thread panicked"))
-            .collect()
-    });
+    let total = indexes.len();
+    if total > 1 {
+        println!("{}", out::muted(format!("pushing {total} repo(s)…")));
+    }
 
-    let mut failures = Vec::new();
-    for (repo_id, result) in results {
-        match result {
-            Ok(success) => {
-                println!(
-                    "{}: {} {} {}",
-                    out::repo(&repo_id),
-                    out::movement("pushed"),
-                    out::branch(success.upstream),
-                    out::sha(short_sha(&success.sha))
-                );
-            }
-            Err(error) => {
-                println!("{}: {}", out::repo(&repo_id), out::danger("push failed"));
-                failures.push(format!("{repo_id}: {error:#}"));
+    // Report each repo the moment its push finishes rather than after the
+    // slowest one: with many repos and a slow origin, a report batched after
+    // the last join reads as a hang.
+    let (tx, rx) = std::sync::mpsc::channel();
+    let failures: Vec<String> = std::thread::scope(|scope| {
+        for &index in &indexes {
+            let active = &active;
+            let repo = &active.bundle.repos[index];
+            let repo_id = repo.id.clone();
+            let tx = tx.clone();
+            scope.spawn(move || {
+                let result = push_repo(active, repo, set_upstream, force);
+                // The receiver outlives every worker; a send cannot fail.
+                let _ = tx.send((repo_id, result));
+            });
+        }
+        drop(tx);
+
+        let mut failures = Vec::new();
+        for (done, (repo_id, result)) in rx.into_iter().enumerate() {
+            let progress = out::progress(done + 1, total);
+            match result {
+                Ok(success) => {
+                    println!(
+                        "{}: {} {} {}{progress}",
+                        out::repo(&repo_id),
+                        out::movement("pushed"),
+                        out::branch(success.upstream),
+                        out::sha(short_sha(&success.sha))
+                    );
+                }
+                Err(error) => {
+                    println!(
+                        "{}: {}{progress}",
+                        out::repo(&repo_id),
+                        out::danger("push failed")
+                    );
+                    failures.push(format!("{repo_id}: {error:#}"));
+                }
             }
         }
-    }
+        failures
+    });
 
     if !failures.is_empty() {
         bail!("push failed:\n{}", failures.join("\n"));
diff --git a/src/commands/remote/history.rs b/src/commands/remote/history.rs
index 1065482..5d35f5e 100644
--- a/src/commands/remote/history.rs
+++ b/src/commands/remote/history.rs
@@ -130,12 +130,22 @@ pub(super) fn push_project_history_events(
                     let path = path.as_str();
                     scope.spawn(move || {
                         let payload = json!({ "events": batch });
-                        request_json::<RemoteHistoryPush>(remote, token, "POST", path, Some(&payload))
+                        request_json::<RemoteHistoryPush>(
+                            remote,
+                            token,
+                            "POST",
+                            path,
+                            Some(&payload),
+                        )
                     })
                 })
                 .collect();
             for handle in started {
-                handles.push(handle.join().unwrap_or_else(|_| Err(anyhow::anyhow!("history push thread panicked"))));
+                handles.push(
+                    handle
+                        .join()
+                        .unwrap_or_else(|_| Err(anyhow::anyhow!("history push thread panicked"))),
+                );
             }
         }
         handles
@@ -210,7 +220,9 @@ pub(super) fn plan_history_push(
     encoded_lines: &[String],
     cursor: Option<&HistorySyncCursor>,
 ) -> HistoryPushPlan {
-    let Some(cursor) = cursor else { return HistoryPushPlan::Full };
+    let Some(cursor) = cursor else {
+        return HistoryPushPlan::Full;
+    };
     if cursor.event_count == 0 || cursor.event_count > encoded_lines.len() {
         return HistoryPushPlan::Full;
     }
@@ -225,7 +237,8 @@ pub(super) fn plan_history_push(
 }
 
 fn history_sync_state_path(root: &Path, project_id: &str) -> std::path::PathBuf {
-    crate::store::history_path(root, project_id).with_file_name(format!("{project_id}.history-sync.json"))
+    crate::store::history_path(root, project_id)
+        .with_file_name(format!("{project_id}.history-sync.json"))
 }
 
 pub(super) fn load_history_sync_state(root: &Path, project_id: &str) -> Result<HistorySyncState> {
@@ -259,7 +272,8 @@ fn record_history_sync(
         std::fs::create_dir_all(parent)
             .with_context(|| format!("failed to create {}", parent.display()))?;
     }
-    let body = serde_json::to_string_pretty(&state).context("failed to encode history sync state")?;
+    let body =
+        serde_json::to_string_pretty(&state).context("failed to encode history sync state")?;
     std::fs::write(&path, format!("{body}\n"))
         .with_context(|| format!("failed to write {}", path.display()))
 }
@@ -328,7 +342,9 @@ mod push_plan_tests {
     use super::*;
 
     fn lines(count: usize) -> Vec<String> {
-        (0..count).map(|index| format!("{{\"eventId\":\"e{index}\"}}")).collect()
+        (0..count)
+            .map(|index| format!("{{\"eventId\":\"e{index}\"}}"))
+            .collect()
     }
 
     #[test]
@@ -339,18 +355,36 @@ mod push_plan_tests {
     #[test]
     fn an_unchanged_ledger_sends_nothing_and_an_appended_one_sends_the_tail() {
         let pushed = lines(3);
-        let cursor = HistorySyncCursor { event_count: 3, fingerprint: history_fingerprint(&pushed) };
-        assert_eq!(plan_history_push(&pushed, Some(&cursor)), HistoryPushPlan::UpToDate);
-        assert_eq!(plan_history_push(&lines(5), Some(&cursor)), HistoryPushPlan::Tail(3));
+        let cursor = HistorySyncCursor {
+            event_count: 3,
+            fingerprint: history_fingerprint(&pushed),
+        };
+        assert_eq!(
+            plan_history_push(&pushed, Some(&cursor)),
+            HistoryPushPlan::UpToDate
+        );
+        assert_eq!(
+            plan_history_push(&lines(5), Some(&cursor)),
+            HistoryPushPlan::Tail(3)
+        );
     }
 
     #[test]
     fn a_rewritten_prefix_or_a_shrunken_ledger_sends_everything() {
         let pushed = lines(3);
-        let cursor = HistorySyncCursor { event_count: 3, fingerprint: history_fingerprint(&pushed) };
+        let cursor = HistorySyncCursor {
+            event_count: 3,
+            fingerprint: history_fingerprint(&pushed),
+        };
         let mut rewritten = lines(4);
         rewritten[1] = "{\"eventId\":\"e1\",\"message\":\"enriched\"}".to_string();
-        assert_eq!(plan_history_push(&rewritten, Some(&cursor)), HistoryPushPlan::Full);
-        assert_eq!(plan_history_push(&lines(2), Some(&cursor)), HistoryPushPlan::Full);
+        assert_eq!(
+            plan_history_push(&rewritten, Some(&cursor)),
+            HistoryPushPlan::Full
+        );
+        assert_eq!(
+            plan_history_push(&lines(2), Some(&cursor)),
+            HistoryPushPlan::Full
+        );
     }
 }
diff --git a/src/commands/remote/push.rs b/src/commands/remote/push.rs
index 654ca97..3fc7ea2 100644
--- a/src/commands/remote/push.rs
+++ b/src/commands/remote/push.rs
@@ -680,7 +680,10 @@ fn push_active_bundle_to_remote(
         .context("No project selected. Pass --project or run `knit init <name>`.")?;
     // Say that the sync has started: the branch pushes above finish in
     // seconds, and a silent minute after them read as a hang.
-    println!("{}", out::muted(format!("syncing {} to {remote_name}…", active.bundle.id)));
+    println!(
+        "{}",
+        out::muted(format!("syncing {} to {remote_name}…", active.bundle.id))
+    );
     let remote = resolve_remote(&config, remote_name)?;
     let token = resolve_token(remote_name, remote)?;
     let local_project = load_project_if_present(&active.root, &project_id)?;
diff --git a/src/output.rs b/src/output.rs
index 1726a09..04cf2f7 100644
--- a/src/output.rs
+++ b/src/output.rs
@@ -49,6 +49,17 @@ pub fn muted(text: impl Display) -> String {
     paint(text, Style::Dim)
 }
 
+/// Progress tail for streamed multi-repo output: ` (done/total)` in muted
+/// style, or nothing when a single repo is reported and a count adds no
+/// information.
+pub fn progress(done: usize, total: usize) -> String {
+    if total <= 1 {
+        String::new()
+    } else {
+        muted(format!(" ({done}/{total})"))
+    }
+}
+
 pub fn path(text: impl Display) -> String {
     paint(text, Style::Dim)
 }
@@ -158,3 +169,17 @@ fn code(style: Style) -> &'static str {
         Style::Magenta => "\x1b[35m",
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::progress;
+
+    #[test]
+    fn progress_tail_is_empty_for_a_single_repo_and_counts_otherwise() {
+        std::env::set_var("NO_COLOR", "1");
+        assert_eq!(progress(1, 1), "");
+        assert_eq!(progress(1, 0), "");
+        assert_eq!(progress(1, 3), " (1/3)");
+        assert_eq!(progress(3, 3), " (3/3)");
+    }
+}
diff --git a/tests/bundle.rs b/tests/bundle.rs
index a7d5998..4d8e664 100644
--- a/tests/bundle.rs
+++ b/tests/bundle.rs
@@ -663,14 +663,20 @@ fn no_workspace_fallback_names_the_open_bundles_instead_of_asking_for_a_new_one(
         let mut config: Value =
             serde_json::from_str(&fs::read_to_string(&config_path).unwrap()).unwrap();
         config.as_object_mut().unwrap().remove("activeBundle");
-        fs::write(&config_path, format!("{}\n", serde_json::to_string_pretty(&config).unwrap()))
-            .unwrap();
+        fs::write(
+            &config_path,
+            format!("{}\n", serde_json::to_string_pretty(&config).unwrap()),
+        )
+        .unwrap();
     };
     clear_fallback();
     let one = knit_fails(&workspace, ["status"]);
     assert!(one.contains("Bundle `fix-a` is open"), "{one}");
     assert!(one.contains("--bundle fix-a"), "{one}");
-    assert!(!one.contains("feature title"), "told to create a bundle while one is open: {one}");
+    assert!(
+        !one.contains("feature title"),
+        "told to create a bundle while one is open: {one}"
+    );
 
     // Several open: list them all, ask for --bundle, never "create one".
     knit(&workspace, ["bundle", "fix b"]);
@@ -686,7 +692,10 @@ fn no_workspace_fallback_names_the_open_bundles_instead_of_asking_for_a_new_one(
     fs::create_dir_all(&empty).unwrap();
     knit(&empty, ["init", "demo"]);
     let none = knit_fails(&empty, ["status"]);
-    assert!(none.contains("Run `knit bundle \"feature title\"` first"), "{none}");
+    assert!(
+        none.contains("Run `knit bundle \"feature title\"` first"),
+        "{none}"
+    );
 
     fs::remove_dir_all(root).unwrap();
 }
diff --git a/tests/publish.rs b/tests/publish.rs
index af08bd7..883509e 100644
--- a/tests/publish.rs
+++ b/tests/publish.rs
@@ -43,6 +43,12 @@ fn pr_create_pushes_creates_records_and_syncs_cross_links() {
     assert!(create.contains("frontend"));
     assert!(create.contains("created"));
     assert!(create.contains("synced"));
+    // Multi-repo publishing streams: a header up front, one `pushed` line per
+    // repo as its branch reaches origin, and a done/total tail per review.
+    assert!(create.contains("publishing 2 repo(s)"), "{create}");
+    assert_eq!(create.matches(": pushed ").count(), 2, "{create}");
+    assert!(create.contains("(1/2)"), "{create}");
+    assert!(create.contains("(2/2)"), "{create}");
 
     assert_eq!(
         git(
diff --git a/tests/sync.rs b/tests/sync.rs
index 556469a..9085606 100644
--- a/tests/sync.rs
+++ b/tests/sync.rs
@@ -465,6 +465,11 @@ fn push_sends_selected_feature_branches_in_parallel() {
     let push = knit(&workspace, ["push", "backend", "frontend"]);
     assert!(push.contains("backend"));
     assert!(push.contains("frontend"));
+    // Multi-repo pushes stream: a header up front and a done/total tail on
+    // each repo's line as it finishes, so a slow origin never looks like a hang.
+    assert!(push.contains("pushing 2 repo(s)"), "{push}");
+    assert!(push.contains("(1/2)"), "{push}");
+    assert!(push.contains("(2/2)"), "{push}");
     assert_eq!(
         git(
             &backend_remote,
@@ -2401,8 +2406,15 @@ fn sync_push_history_sends_only_what_the_remote_does_not_have_yet() {
     // Nothing changed: the second push makes no history request at all, and
     // still reports what the remote holds.
     let second = knit_with_env(&workspace, ["sync", "push", "--history"], &env);
-    assert!(second.contains(&format!("{} event(s)", initial.len())), "{second}");
-    assert_eq!(common::recorded_history_pushes(&fake_dir).len(), 1, "an unchanged ledger was pushed again");
+    assert!(
+        second.contains(&format!("{} event(s)", initial.len())),
+        "{second}"
+    );
+    assert_eq!(
+        common::recorded_history_pushes(&fake_dir).len(),
+        1,
+        "an unchanged ledger was pushed again"
+    );
 
     // One more commit: only the events it added ride in the third push.
     append_line(&feature.join("app.txt"), "second");
@@ -2426,7 +2438,11 @@ fn sync_push_history_sends_only_what_the_remote_does_not_have_yet() {
     let other_dir = root.join("other-remote");
     let other_url = spawn_fake_remote_push_api(&other_dir);
     knit(&workspace, ["remote", "add", "mirror", &other_url]);
-    knit_with_env(&workspace, ["sync", "push", "--history", "--remote", "mirror"], &env);
+    knit_with_env(
+        &workspace,
+        ["sync", "push", "--history", "--remote", "mirror"],
+        &env,
+    );
     let mirror = common::recorded_history_pushes(&other_dir);
     assert_eq!(mirror.len(), 1, "{mirror:?}");
     assert_eq!(mirror[0].len(), initial.len() + pushes[1].len());