@@ -1317,10 +1317,16 @@ jobs:
 
       - name: Unpack the e2e binaries
         shell: bash
+        # Decompress to a file, not into `| tar -xf -`: a tar that exits at
+        # the end-of-archive marker without draining stdin (macOS's does)
+        # leaves zstd writing into a closed pipe, and pipefail turns its
+        # "Broken pipe" into a failed leg that evicts the queue entry.
         run: |
           set -euo pipefail
           mkdir -p target/e2e-bin
-          zstd -q -d -c target/e2e-archive/e2e-bin.tar.zst | tar -xf - -C target/e2e-bin
+          zstd -q -d -f target/e2e-archive/e2e-bin.tar.zst -o target/e2e-bin.tar
+          tar -xf target/e2e-bin.tar -C target/e2e-bin
+          rm target/e2e-bin.tar
 
       - name: Stage the CLI where the test binaries expect it
         # `CARGO_BIN_EXE_socket-patch` was baked in at compile time as
@@ -2165,10 +2171,16 @@ jobs:
           path: target/e2e-archive
       - name: Unpack the e2e binaries
         shell: bash
+        # Decompress to a file, not into `| tar -xf -`: a tar that exits at
+        # the end-of-archive marker without draining stdin (macOS's does)
+        # leaves zstd writing into a closed pipe, and pipefail turns its
+        # "Broken pipe" into a failed leg that evicts the queue entry.
         run: |
           set -euo pipefail
           mkdir -p target/e2e-bin
-          zstd -q -d -c target/e2e-archive/e2e-bin.tar.zst | tar -xf - -C target/e2e-bin
+          zstd -q -d -f target/e2e-archive/e2e-bin.tar.zst -o target/e2e-bin.tar
+          tar -xf target/e2e-bin.tar -C target/e2e-bin
+          rm target/e2e-bin.tar
       - name: Install Rust
         run: rustup show
       - name: Install the cargo under test
 
@@ -1386,6 +1386,22 @@ pub(crate) async fn run_nested_apply(
     report
 }
 
+/// The nested apply's non-fatal warnings (today the default policy's
+/// `content_mismatch_overwritten` overwrites, #1004) as `get`'s string
+/// `warnings[]` entries, code-prefixed like `get`'s `fold_narrowing_into_result`.
+/// A JSON caller's nested apply is silent, so the envelope is the only
+/// place these surface; a human caller's apply already printed them.
+pub(crate) fn apply_warning_lines(report: Option<&ApplyRunReport>) -> Vec<String> {
+    report
+        .map(|r| {
+            r.warnings
+                .iter()
+                .map(|w| format!("({}) {}", w.code, w.detail))
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
 /// Whether apply's package key `key` covers the patch record purl
 /// `record`: the same purl, or `key` is the unqualified base of a
 /// qualified record (apply keys a release-variant base by its base purl).
@@ -1662,7 +1678,9 @@ pub async fn download_and_apply_patches_with(
     }
     // Surface release-narrowing fallbacks (uninstalled package / no
     // matching variant) so JSON consumers can see why all variants were
-    // kept. Omitted entirely when narrowing was clean.
+    // kept, and the apply's mismatch overwrites. Omitted entirely when
+    // both were clean.
+    warnings.extend(apply_warning_lines(apply_report.as_ref()));
     if !warnings.is_empty() {
         result_json["warnings"] = serde_json::json!(warnings);
     }
 
@@ -85,6 +85,25 @@ fn mismatch_event_detail(file: &str, dry_run: bool) -> String {
     )
 }
 
+/// One `content_mismatch_overwritten` run warning per mismatch-overwritten
+/// file across `results`, in the event detail's words prefixed with the
+/// package purl — what a nested apply hands back to `get` / `scan --mode
+/// agent` (#1004).
+fn mismatch_overwrite_warnings(results: &[ApplyResult], dry_run: bool) -> Vec<RunWarning> {
+    results
+        .iter()
+        .flat_map(|r| {
+            let purl = normalize_purl(&r.package_key);
+            mismatch_overwritten_files(r)
+                .into_iter()
+                .map(move |file| RunWarning {
+                    code: "content_mismatch_overwritten".to_string(),
+                    detail: format!("{purl}: {}", mismatch_event_detail(&file, dry_run)),
+                })
+        })
+        .collect()
+}
+
 /// The mismatch policy this run applies with: `--force` ⊃ default
 /// (adds the missing-file skip), `--strict` restores fail-closed.
 fn mismatch_policy(force: bool, strict: bool) -> MismatchPolicy {
@@ -963,15 +982,20 @@ pub(crate) struct ApplyRunReport {
     /// failed run's caller can count exactly what applied. Filled only
     /// when `code != 0`.
     pub applied: Vec<String>,
+    /// Non-fatal per-file warnings the caller's envelope must carry: one
+    /// `content_mismatch_overwritten` per file the default mismatch policy
+    /// overwrote (#1004). A nested apply never prints JSON and is silent
+    /// for a JSON caller, so this is their only channel. Filled whenever
+    /// the apply loop ran, whatever the exit code.
+    pub warnings: Vec<RunWarning>,
 }
 
 impl ApplyRunReport {
     fn run_failure(code: i32, error_code: &str, error: impl Into<String>) -> Self {
         Self {
             code,
-            failures: Vec::new(),
             run_error: Some((error_code.to_string(), error.into())),
-            applied: Vec::new(),
+            ..Self::default()
         }
     }
 }
@@ -1371,10 +1395,16 @@ pub(crate) async fn run_locked(
                 .await;
             }
 
+            // The mismatch overwrites, for a nested caller's envelope (the
+            // JSON events above are the standalone apply's copy).
+            let warnings = mismatch_overwrite_warnings(&results, args.common.dry_run);
             // A requested-but-failed VEX flips an otherwise-successful
             // apply to a non-zero exit (fail-the-command contract).
             if success && !vex_failed {
-                return ApplyRunReport::default();
+                return ApplyRunReport {
+                    warnings,
+                    ..ApplyRunReport::default()
+                };
             }
             let failures = if success {
                 Vec::new()
@@ -1412,6 +1442,7 @@ pub(crate) async fn run_locked(
                 failures,
                 run_error,
                 applied,
+                warnings,
             }
         }
         Err(e) => {
 
@@ -30,11 +30,12 @@ use crate::args::{apply_env_toggles, GlobalArgs};
 // `commands::get` paths (the in-process tests and embedders call them);
 // the engine itself lives in the shared `agent_download` helper.
 use crate::commands::agent_download::{
-    decide_patch_action, download_patch_records_preflighted, download_patch_records_reusing,
-    filter_to_installed_releases, fold_apply_failures, max_vuln_severity, merge_metadata,
-    nested_apply_args, patch_event_metadata, report_error, report_lock_failure, run_nested_apply,
-    run_outcome, unwind_new_blobs, warn_on_vendored_uuid_drift, write_all_patch_blobs,
-    DetachedDownload, PatchAction, VendorRefusals,
+    apply_warning_lines, decide_patch_action, download_patch_records_preflighted,
+    download_patch_records_reusing, filter_to_installed_releases, fold_apply_failures,
+    max_vuln_severity, merge_metadata, nested_apply_args, patch_event_metadata, report_error,
+    report_lock_failure, run_nested_apply, run_outcome, unwind_new_blobs,
+    warn_on_vendored_uuid_drift, write_all_patch_blobs, DetachedDownload, PatchAction,
+    VendorRefusals,
 };
 pub use crate::commands::agent_download::{
     download_and_apply_patches_with, DownloadParams, DownloadRun,
@@ -2060,6 +2061,7 @@ async fn save_and_apply_patch(args: &GetArgs, client: &ApiClient, patch: &PatchR
             result_json["applied"] = serde_json::json!(applied);
         }
         // Same contract as `download_and_apply_patches_with`: omitted when clean.
+        warnings.extend(apply_warning_lines(apply_report.as_ref()));
         if !warnings.is_empty() {
             result_json["warnings"] = serde_json::json!(warnings);
         }
@@ -3165,6 +3167,7 @@ mod tests {
             failures,
             run_error: None,
             applied: applied.iter().map(|p| p.to_string()).collect(),
+            warnings: Vec::new(),
         }
     }
 
@@ -3321,6 +3324,7 @@ mod tests {
             failures: Vec::new(),
             run_error: Some(("yarn_pnp_unsupported".to_string(), "pnp".to_string())),
             applied: Vec::new(),
+            warnings: Vec::new(),
         };
         assert_eq!(fold_apply_failures(&mut env, &report, |_| None), 0);
         assert!(env.get("errorCode").is_none(), "{env}");
 
@@ -96,9 +96,18 @@ pub(crate) async fn discover_wiring(
     common: &crate::args::GlobalArgs,
     root: &Path,
 ) -> socket_patch_core::vex::discover::Discovery {
+    #[cfg(test)]
+    DISCOVERIES.with(|n| n.set(n.get() + 1));
     socket_patch_core::vex::discover_patched_refs_with(root, &discover_options(common)).await
 }
 
+#[cfg(test)]
+thread_local! {
+    /// How many times this thread ran [`discover_wiring`]: discovery walks
+    /// every lockfile, so tests pin the paths that must not repeat it.
+    pub(crate) static DISCOVERIES: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
+}
+
 /// [`discover_wiring`] of the snapshot's root, reading through `snapshot`.
 pub(crate) async fn discover_wiring_in(
     common: &crate::args::GlobalArgs,
@@ -128,13 +137,8 @@ pub(crate) async fn hosted_inventory(
     )
 }
 
-/// The project's hosted state, v5-style: v5 hosted mode keeps no ledger,
-/// so the hosted pins [`discover_wiring`] finds in the lockfiles are the
-/// whole record. Shaped as a [`RedirectState`] for the readers that classify
-/// hosted against vendored state (one uuid-only record per pinned purl, no
-/// edits) — it is never persisted.
-///
-/// [`RedirectState`]: socket_patch_core::patch::redirect::RedirectState
+/// [`hosted_state_from_pins`] over a fresh [`discover_wiring`] of `root`
+/// (the unit tests' load-then-derive entry point).
 #[cfg(test)]
 pub(crate) async fn hosted_state_from_lockfiles(
     common: &crate::args::GlobalArgs,
@@ -147,8 +151,14 @@ pub(crate) async fn hosted_state_from_lockfiles(
     )
 }
 
-/// [`hosted_state_from_lockfiles`] over already-discovered pins. A purl
-/// pinned to several uuids (different lockfiles) keeps the first.
+/// The project's hosted state, v5-style: v5 hosted mode keeps no ledger,
+/// so the hosted pins [`discover_wiring`] finds in the lockfiles are the
+/// whole record. Shaped as a [`RedirectState`] for the readers that classify
+/// hosted against vendored state (one uuid-only record per pinned purl, no
+/// edits) — it is never persisted. A purl pinned to several uuids
+/// (different lockfiles) keeps the first.
+///
+/// [`RedirectState`]: socket_patch_core::patch::redirect::RedirectState
 pub(crate) fn hosted_state_from_pins(
     pins: &[socket_patch_core::patch::redirect::upstream::HostedPin],
 ) -> socket_patch_core::patch::redirect::RedirectState {
 
@@ -102,6 +102,19 @@ fn print_hosted_leg_warnings(common: &GlobalArgs, warnings: &[(String, String)])
     }
 }
 
+/// `--preserve-state` restored hosted pins anyway (hosted has no
+/// preservable local state): say so on stderr (`Note: …`, never under
+/// `--silent` / `--json`) and return the `hosted_state_not_preservable`
+/// run warning for the envelope's `warnings[]` — the same code
+/// `rollback --preserve-state` reports.
+fn note_hosted_state_not_preservable(common: &GlobalArgs) -> (String, String) {
+    let warning = super::rollback::hosted_state_not_preservable_warning();
+    if !common.silent && !common.json {
+        eprintln!("Note: {}.", warning.1);
+    }
+    warning
+}
+
 /// Emit a `remove` error envelope and return. Used by the many error
 /// paths in `run` so they all share the same JSON shape. `dry_run` rides
 /// the envelope so preview failures report `dryRun: true`.
@@ -785,11 +798,8 @@ pub async fn run(args: RemoveArgs) -> i32 {
                     return 1;
                 }
             };
-            if args.preserve_state && !leg.reverted.is_empty() && loud {
-                eprintln!(
-                    "Note: hosted wiring has no preservable local state; its lockfile pins \
-                     now resolve upstream."
-                );
+            if args.preserve_state && !leg.reverted.is_empty() {
+                hosted_leg_warnings.push(note_hosted_state_not_preservable(&args.common));
             }
             // `run_hosted_leg` printed one line per restored purl.
             printed_progress |= loud && !leg.reverted.is_empty();
@@ -1431,7 +1441,11 @@ async fn remove_hosted_only(
     } else {
         PatchAction::Removed
     };
-    for (code, detail) in &leg.warnings {
+    let mut warnings = leg.warnings.clone();
+    if args.preserve_state && !leg.reverted.is_empty() {
+        warnings.push(note_hosted_state_not_preservable(&args.common));
+    }
+    for (code, detail) in &warnings {
         env.warnings.push(crate::json_envelope::RunWarning {
             code: code.clone(),
             detail: detail.clone(),
 
@@ -76,6 +76,19 @@ pub(crate) fn join_clauses(clauses: &[String]) -> String {
     }
 }
 
+/// The `hosted_state_not_preservable` run warning: a `--preserve-state`
+/// run (rollback or remove) restored hosted pins to upstream anyway — the
+/// lockfile pins are hosted mode's only record, so there is no local
+/// state to keep.
+pub(crate) fn hosted_state_not_preservable_warning() -> (String, String) {
+    (
+        "hosted_state_not_preservable".into(),
+        "hosted wiring has no preservable local state: the lockfile pins are the only \
+         record, and they now resolve upstream; re-run `scan --mode hosted` to re-wire"
+            .into(),
+    )
+}
+
 /// Capitalize the first character and end with `?`.
 pub(crate) fn as_question(text: &str) -> String {
     if text.is_empty() {
@@ -1490,13 +1503,7 @@ pub async fn run(args: RollbackArgs) -> i32 {
                 ));
             }
             if args.preserve_state && !hosted_leg.reverted.is_empty() {
-                run_warnings.push((
-                    "hosted_state_not_preservable".into(),
-                    "hosted wiring has no preservable local state: the lockfile pins are \
-                     the only record, and they now resolve upstream; re-run \
-                     `scan --mode hosted` to re-wire"
-                        .into(),
-                ));
+                run_warnings.push(hosted_state_not_preservable_warning());
             }
             if !path_scope.is_empty() {
                 let scope = path_scope.bind(&cwd);