diff --git a/docs/contributing/TECHNICAL.md b/docs/contributing/TECHNICAL.md
index 54f15422ae..6edfe6a6e0 100644
--- a/docs/contributing/TECHNICAL.md
+++ b/docs/contributing/TECHNICAL.md
@@ -143,8 +143,10 @@ rewrite_compound(cmd, excluded)                    [src/discover/registry.rs]
   |
   |  Step 2 — Split on operators, rewrite each segment
   |  Operator (&&, ||, ;) → rewrite both sides
-  |  Pipe (|) → keep producers/intermediate stages raw
-  |             rewrite only a pipeline-safe final stage
+  |  Pipe (|) → keep intermediate stages raw
+  |             rewrite a pipeline-safe final stage, or the
+  |             producer when its rule allows it and all
+  |             consumers are display-only (#3171)
   |  Stderr pipe (|&) → keep the complete pipeline raw
   |  Shellism (&) → rewrite both sides (background)
   |
diff --git a/src/discover/README.md b/src/discover/README.md
index 384497c5bc..ea60da1eb3 100644
--- a/src/discover/README.md
+++ b/src/discover/README.md
@@ -21,7 +21,7 @@ When a hook sends `cargo fmt --all && cargo test 2>&1 | tail -20`:
 → [Arg("cargo"), Arg("test"), Redirect("2>&1"), Operator("&&"), Arg("git"), Arg("status")]
 ```
 
-**Compound splitting** — The rewrite engine walks the tokens, splitting on `Operator` (`&&`, `||`, `;`) and typed `Pipe` tokens (`|`, `|&`). For normal pipelines, producers and intermediate stages stay raw, and only an argument-safe final stage marked `pipeline_final_safe` is rewritten. The initial safe set is ordinary `grep` and `rg` invocations; search pattern-file forms (`-f`/`--file`) defer because they can consume pipeline stdin as configuration. Stderr pipelines (`|&`) and pipelines containing opaque shell groups remain raw.
+**Compound splitting** — The rewrite engine walks the tokens, splitting on `Operator` (`&&`, `||`, `;`) and typed `Pipe` tokens (`|`, `|&`). For normal pipelines, intermediate stages stay raw. A final stage whose rule's `pipeline_safety` allows it (ordinary `grep` and `rg` invocations) is rewritten; search pattern-file forms (`-f`/`--file`) defer because they can consume pipeline stdin as configuration. The producer stage is rewritten when its rule's `pipeline_safety` allows it and every downstream stage is a display-only consumer (`cat`, `head`, non-following `tail`) with no file-target redirect (#3171). Stderr pipelines (`|&`) and pipelines containing opaque shell groups remain raw.
 
 **Per-segment rewriting** — Each segment goes through:
 
diff --git a/src/discover/lexer.rs b/src/discover/lexer.rs
index ae8603d7a9..19b0228f40 100644
--- a/src/discover/lexer.rs
+++ b/src/discover/lexer.rs
@@ -400,7 +400,7 @@ fn contains_substitution(cmd: &str) -> bool {
 
 // `>&N`/`>&-` (and `N>&M`) is fd-dup/close; bare `>&` before a word is
 // `>word 2>&1` — a file target.
-fn redirect_has_file_target(tokens: &[ParsedToken], i: usize) -> bool {
+pub(crate) fn redirect_has_file_target(tokens: &[ParsedToken], i: usize) -> bool {
     let value = &tokens[i].value;
     if let Some(pos) = value.find(">&") {
         let tail = &value[pos + 2..];
diff --git a/src/discover/registry.rs b/src/discover/registry.rs
index ae5991130a..f3413aa664 100644
--- a/src/discover/registry.rs
+++ b/src/discover/registry.rs
@@ -6,8 +6,8 @@ use std::path::Path;
 use std::sync::LazyLock;
 
 use super::lexer::{
-    advance_quote_state, coalesce_words, is_crlf_at, shell_split, split_on_operators, tokenize,
-    tokenize_with_newlines, ParsedToken, PipeKind, TokenKind,
+    advance_quote_state, coalesce_words, is_crlf_at, redirect_has_file_target, shell_split,
+    split_on_operators, tokenize, tokenize_with_newlines, ParsedToken, PipeKind, TokenKind,
 };
 use super::rules::{RtkRule, IGNORED_EXACT, IGNORED_PREFIXES, RULES};
 
@@ -974,6 +974,7 @@ struct PipelineAnalysis {
     end_offset: usize,
     next_clause_offset: Option<usize>,
     final_stage_start: Option<usize>,
+    all_consumers_safe: bool,
 }
 
 fn analyze_pipeline(
@@ -995,14 +996,21 @@ fn analyze_pipeline(
     let mut stage_start = segment_start;
     let mut final_stage_start = None;
     let mut has_supported_structure = true;
+    let mut consumers_all_safe = true;
 
-    for token in tokens {
+    for (i, token) in tokens.iter().enumerate() {
         if token.offset >= end_offset {
             break;
         }
         if token.offset < first_pipe_offset {
             continue;
         }
+        if token.kind == TokenKind::Redirect {
+            if redirect_has_file_target(tokens, i) {
+                consumers_all_safe = false;
+            }
+            continue;
+        }
         let TokenKind::Pipe(kind) = token.kind else {
             continue;
         };
@@ -1010,6 +1018,11 @@ fn analyze_pipeline(
         if cmd[stage_start..token.offset].trim().is_empty() || kind == PipeKind::StdoutAndStderr {
             has_supported_structure = false;
         }
+        if token.offset > first_pipe_offset
+            && !is_safe_pipe_consumer(cmd[stage_start..token.offset].trim())
+        {
+            consumers_all_safe = false;
+        }
 
         stage_start = token.offset + token.value.len();
         final_stage_start = Some(stage_start);
@@ -1017,6 +1030,8 @@ fn analyze_pipeline(
 
     if cmd[stage_start..end_offset].trim().is_empty() {
         has_supported_structure = false;
+    } else if !is_safe_pipe_consumer(cmd[stage_start..end_offset].trim()) {
+        consumers_all_safe = false;
     }
 
     PipelineAnalysis {
@@ -1027,9 +1042,24 @@ fn analyze_pipeline(
         } else {
             None
         },
+        all_consumers_safe: has_supported_structure && consumers_all_safe,
     }
 }
 
+fn rewrite_pipeline_stage(
+    cmd: &str,
+    stage_start: usize,
+    stage_end: usize,
+    context: RewriteContext,
+    excluded: &[ExcludePattern],
+    transparent_prefixes: &[String],
+) -> Option<String> {
+    let stage = cmd[stage_start..stage_end].trim();
+
+    rewrite_segment_inner(stage, excluded, transparent_prefixes, context, 0)
+        .filter(|rewritten| rewritten != stage)
+}
+
 fn rewrite_pipeline_final_stage(
     cmd: &str,
     segment_start: usize,
@@ -1038,16 +1068,15 @@ fn rewrite_pipeline_final_stage(
     transparent_prefixes: &[String],
 ) -> Option<String> {
     let final_stage_start = analysis.final_stage_start?;
-    let final_stage = cmd[final_stage_start..analysis.end_offset].trim();
 
-    rewrite_segment_inner(
-        final_stage,
+    rewrite_pipeline_stage(
+        cmd,
+        final_stage_start,
+        analysis.end_offset,
+        RewriteContext::PipelineFinal,
         excluded,
         transparent_prefixes,
-        RewriteContext::PipelineFinal,
-        0,
     )
-    .filter(|rewritten| rewritten != final_stage)
     .map(|rewritten| {
         format!(
             "{} {}",
@@ -1057,6 +1086,36 @@ fn rewrite_pipeline_final_stage(
     })
 }
 
+// #3171
+fn rewrite_pipeline_producer(
+    cmd: &str,
+    segment_start: usize,
+    first_pipe_offset: usize,
+    analysis: PipelineAnalysis,
+    excluded: &[ExcludePattern],
+    transparent_prefixes: &[String],
+) -> Option<String> {
+    if !analysis.all_consumers_safe {
+        return None;
+    }
+
+    rewrite_pipeline_stage(
+        cmd,
+        segment_start,
+        first_pipe_offset,
+        RewriteContext::PipelineProducer,
+        excluded,
+        transparent_prefixes,
+    )
+    .map(|rewritten| {
+        format!(
+            "{} {}",
+            rewritten,
+            cmd[first_pipe_offset..analysis.end_offset].trim()
+        )
+    })
+}
+
 /// Rewrite a compound command (with `&&`, `||`, `;`, `|`) by rewriting each
 /// segment. Third of three compound-command segmenters — see the comparison
 /// table on [`crate::discover::lexer::split_for_permissions`]. Deliberately
@@ -1120,7 +1179,17 @@ fn rewrite_compound(
                     analysis,
                     excluded,
                     transparent_prefixes,
-                );
+                )
+                .or_else(|| {
+                    rewrite_pipeline_producer(
+                        cmd,
+                        seg_start,
+                        tok.offset,
+                        analysis,
+                        excluded,
+                        transparent_prefixes,
+                    )
+                });
 
                 if let Some(rewritten) = rewritten_pipeline {
                     any_changed = true;
@@ -1206,6 +1275,58 @@ const ROUTABLE_WRAPPER_PREFIXES: &[&str] = &["uv run"];
 /// not spawnable, so they must never fall through: `rtk exec foo` cannot run.
 const SHELL_KEYWORD_PREFIXES: &[&str] = &["noglob", "command", "builtin", "exec", "nocorrect"];
 
+struct SafePipeConsumer {
+    name: &'static str,
+    unsafe_flags: &'static [&'static str],
+    unsafe_flag_chars: &'static [char],
+}
+
+const SAFE_PIPE_CONSUMERS: &[SafePipeConsumer] = &[
+    SafePipeConsumer {
+        name: "cat",
+        unsafe_flags: &[],
+        unsafe_flag_chars: &[],
+    },
+    SafePipeConsumer {
+        name: "head",
+        unsafe_flags: &[],
+        unsafe_flag_chars: &[],
+    },
+    // #3171: only non-following tail is display-only
+    SafePipeConsumer {
+        name: "tail",
+        unsafe_flags: &["--follow"],
+        unsafe_flag_chars: &['f', 'F'],
+    },
+];
+
+fn arg_matches_unsafe_flag(consumer: &SafePipeConsumer, arg: &str) -> bool {
+    if let Some(rest) = arg.strip_prefix("--") {
+        let name = rest.split_once('=').map_or(rest, |(name, _)| name);
+        return !name.is_empty()
+            && consumer.unsafe_flags.iter().any(|flag| {
+                flag.strip_prefix("--")
+                    .is_some_and(|full| full.starts_with(name))
+            });
+    }
+    arg.strip_prefix('-').is_some_and(|rest| {
+        rest.chars()
+            .any(|c| consumer.unsafe_flag_chars.contains(&c))
+    })
+}
+
+fn is_safe_pipe_consumer(stage: &str) -> bool {
+    let words = shell_split(stage);
+    let mut words = words.iter();
+    let Some(head) = words.next() else {
+        return false;
+    };
+    let Some(consumer) = SAFE_PIPE_CONSUMERS.iter().find(|c| c.name == head.as_str()) else {
+        return false;
+    };
+    !words.any(|arg| arg_matches_unsafe_flag(consumer, arg))
+}
+
 /// Every built-in transparent wrapper, paired with whether it may fall through.
 /// Derived from the two lists above so they cannot drift apart.
 fn builtin_transparent_prefixes() -> impl Iterator<Item = (&'static str, bool)> {
@@ -1221,6 +1342,7 @@ const MAX_PREFIX_DEPTH: usize = 10;
 enum RewriteContext {
     Normal,
     PipelineFinal,
+    PipelineProducer,
 }
 
 /// Checks whether grep or rg reads patterns from a file.
@@ -1239,7 +1361,7 @@ fn search_uses_pattern_file(cmd: &str) -> bool {
         })
 }
 
-fn pipeline_final_command_is_safe(rtk_cmd: &str, cmd: &str) -> bool {
+fn pipeline_command_is_safe(rtk_cmd: &str, cmd: &str) -> bool {
     !matches!(rtk_cmd, "rtk grep" | "rtk rg") || !search_uses_pattern_file(cmd)
 }
 
@@ -1434,7 +1556,7 @@ fn rewrite_segment_inner(
         }
         // TOML-only commands: consult the registry so the hook filters them too (#2179).
         Classification::Unsupported { .. } => {
-            if context == RewriteContext::PipelineFinal {
+            if context != RewriteContext::Normal {
                 return None;
             }
             if crate::core::toml_filter::toml_disabled() {
@@ -1459,7 +1581,14 @@ fn rewrite_segment_inner(
     // Find the matching rule (rtk_cmd values are unique across all rules)
     let rule = RULES.iter().find(|r| r.rtk_cmd == rtk_equivalent)?;
     if context == RewriteContext::PipelineFinal
-        && (!rule.pipeline_final_safe || !pipeline_final_command_is_safe(rule.rtk_cmd, cmd_part))
+        && (!rule.pipeline_safety.final_safe() || !pipeline_command_is_safe(rule.rtk_cmd, cmd_part))
+    {
+        return None;
+    }
+    // #3171
+    if context == RewriteContext::PipelineProducer
+        && (!rule.pipeline_safety.producer_safe()
+            || !pipeline_command_is_safe(rule.rtk_cmd, cmd_part))
     {
         return None;
     }
@@ -2104,11 +2233,96 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_analyze_pipeline_all_consumers_safe() {
+        for cmd in ["git log | tail -5", "git log | head | cat"] {
+            assert!(analyze_test_pipeline(cmd).all_consumers_safe, "{cmd}");
+        }
+        for cmd in [
+            "git log | wc -l",
+            "git log | tail > f",
+            "cargo test |& tail",
+            "git log | FOO=1 tail",
+        ] {
+            assert!(!analyze_test_pipeline(cmd).all_consumers_safe, "{cmd}");
+        }
+    }
+
+    #[test]
+    fn test_pipeline_producer_safe_rule_set() {
+        let mut safe_rules: Vec<_> = RULES
+            .iter()
+            .filter(|rule| rule.pipeline_safety.producer_safe())
+            .map(|rule| rule.rtk_cmd)
+            .collect();
+        safe_rules.sort_unstable();
+        safe_rules.dedup();
+
+        assert_eq!(
+            safe_rules,
+            vec![
+                "rtk brew",
+                "rtk bundle",
+                "rtk cargo",
+                "rtk composer",
+                "rtk df",
+                "rtk diff",
+                "rtk dotnet",
+                "rtk du",
+                "rtk ecs",
+                "rtk find",
+                "rtk git",
+                "rtk go",
+                "rtk golangci-lint run",
+                "rtk grep",
+                "rtk hadolint",
+                "rtk helm",
+                "rtk iptables",
+                "rtk lint",
+                "rtk liquibase",
+                "rtk ls",
+                "rtk markdownlint",
+                "rtk mix",
+                "rtk mvn",
+                "rtk mypy",
+                "rtk next",
+                "rtk paratest",
+                "rtk pest",
+                "rtk phpstan",
+                "rtk phpunit",
+                "rtk pint",
+                "rtk pio",
+                "rtk pip",
+                "rtk poetry",
+                "rtk pre-commit",
+                "rtk prettier",
+                "rtk ps",
+                "rtk pytest",
+                "rtk quarto",
+                "rtk rake",
+                "rtk rg",
+                "rtk rspec",
+                "rtk rubocop",
+                "rtk ruff",
+                "rtk shellcheck",
+                "rtk shopify",
+                "rtk swift",
+                "rtk systemctl",
+                "rtk terraform",
+                "rtk tofu",
+                "rtk tree",
+                "rtk trunk",
+                "rtk wc",
+                "rtk yamllint",
+            ]
+        );
+    }
+
     #[test]
     fn test_pipeline_final_safe_rule_set() {
         let safe_rules: Vec<_> = RULES
             .iter()
-            .filter(|rule| rule.pipeline_final_safe)
+            .filter(|rule| rule.pipeline_safety.final_safe())
             .map(|rule| rule.rtk_cmd)
             .collect();
 
@@ -2886,10 +3100,6 @@ mod tests {
 
     #[test]
     fn test_rewrite_pipe_unsafe_final_stage_stays_raw() {
-        assert_eq!(
-            rewrite_command_no_prefixes("cargo test | tail -50", &[]),
-            None
-        );
         assert_eq!(
             rewrite_command_no_prefixes("find . | xargs grep TODO", &[]),
             None
@@ -2920,6 +3130,265 @@ mod tests {
         );
     }
 
+    // --- Safe pipe consumers: producer rewrite ---
+
+    #[test]
+    fn test_rewrite_pipe_safe_consumers_producer_rewritten() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -5", &[]),
+            Some("rtk git log | tail -5".into())
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("cargo test | tail -50", &[]),
+            Some("rtk cargo test | tail -50".into())
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git diff | cat", &[]),
+            Some("rtk git diff | cat".into())
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("RUST_BACKTRACE=1 cargo test 2>&1 | tail -50", &[]),
+            Some("RUST_BACKTRACE=1 rtk cargo test 2>&1 | tail -50".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_multi_pipe_all_safe_consumers() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | head -20 | tail -5", &[]),
+            Some("rtk git log | head -20 | tail -5".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_safe_consumer_with_next_clause() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -5 && git status", &[]),
+            Some("rtk git log | tail -5 && rtk git status".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_mixed_consumers_stay_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | head | wc -l", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail | xargs echo", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | grep feat | wc -l", &[]),
+            None
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_producer_no_rule_or_excluded_stays_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("unknowncmd | tail -5", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -5", &["git log".into()]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("rtk git log | tail -5", &[]),
+            None
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_consumer_decorations_stay_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | FOO=1 tail -5", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | /usr/bin/tail -5", &[]),
+            None
+        );
+        assert_eq!(rewrite_command_no_prefixes("git log |& tail -5", &[]), None);
+    }
+
+    #[test]
+    fn test_rewrite_pipe_consumer_fd_dup_redirect_rewritten() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -5 2>&1", &[]),
+            Some("rtk git log | tail -5 2>&1".into())
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -5 2>/dev/null", &[]),
+            Some("rtk git log | tail -5 2>/dev/null".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_consumer_redirect_stays_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -5 > out.txt", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | cat > file.txt", &[]),
+            None
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_read_producer_stays_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("head -20 file.txt | tail -5", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("cat file.txt | tail -5", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("tail -20 file.txt | head -5", &[]),
+            None
+        );
+    }
+
+    fn assert_consumer_flag_blocks_rewrite(consumer: &str, spelling: &str) {
+        let cmd = format!("git log | {consumer} {spelling}");
+        assert_eq!(rewrite_command_no_prefixes(&cmd, &[]), None, "{cmd}");
+    }
+
+    /// Every shell spelling of a consumer's unsafe flag must keep the producer raw.
+    /// Driven off `SAFE_PIPE_CONSUMERS` so a consumer added later is covered on arrival:
+    /// `getopt_long` accepts any unambiguous prefix of a long option, and the shell strips
+    /// quotes and backslashes before the flag ever reaches the consumer.
+    #[test]
+    fn test_unsafe_consumer_flag_spellings_stay_raw() {
+        for consumer in SAFE_PIPE_CONSUMERS {
+            for flag in consumer.unsafe_flags {
+                let name = flag
+                    .strip_prefix("--")
+                    .expect("unsafe_flags entries are long options");
+                for len in 1..=name.len() {
+                    let abbrev = &name[..len];
+                    for spelling in [
+                        format!("--{abbrev}"),
+                        format!("\"--{abbrev}\""),
+                        format!("'--{abbrev}'"),
+                        format!("\\-\\-{abbrev}"),
+                        format!("--{abbrev}=x"),
+                    ] {
+                        assert_consumer_flag_blocks_rewrite(consumer.name, &spelling);
+                    }
+                }
+            }
+
+            for ch in consumer.unsafe_flag_chars {
+                for spelling in [
+                    format!("-{ch}"),
+                    format!("\"-{ch}\""),
+                    format!("'-{ch}'"),
+                    format!("\\-{ch}"),
+                    format!("-{ch}q"),
+                    format!("-q{ch}"),
+                    format!("-{ch}n20"),
+                ] {
+                    assert_consumer_flag_blocks_rewrite(consumer.name, &spelling);
+                }
+            }
+        }
+    }
+
+    /// Guards the test above against passing vacuously if the consumer table empties.
+    #[test]
+    fn test_safe_consumer_spellings_still_rewrite() {
+        for cmd in [
+            "git log | cat",
+            "git log | head -20",
+            "git log | tail -20",
+            "git log | tail -n 20",
+        ] {
+            assert!(
+                rewrite_command_no_prefixes(cmd, &[]).is_some(),
+                "{cmd} should rewrite"
+            );
+        }
+    }
+
+    #[test]
+    fn test_rewrite_pipe_following_tail_stays_raw() {
+        for cmd in [
+            "git log | tail -f",
+            "git log | tail -F",
+            "git log | tail --follow",
+            "git log | tail --follow=name",
+            "git log | tail --foll",
+            "git log | tail --f",
+            "git log | tail -fn20",
+            "git log | tail \"-f\"",
+            "git log | tail \\-f",
+        ] {
+            assert_eq!(rewrite_command_no_prefixes(cmd, &[]), None, "{cmd}");
+        }
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | tail -n 20", &[]),
+            Some("rtk git log | tail -n 20".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_producer_unsafe_rules_stay_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("ping 127.0.0.1 | head -5", &[]),
+            None
+        );
+        assert_eq!(rewrite_command_no_prefixes("vitest | head", &[]), None);
+        assert_eq!(
+            rewrite_command_no_prefixes("npm run dev | head -5", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("docker logs app | tail -20", &[]),
+            None
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_producer_pattern_file_stays_raw() {
+        assert_eq!(
+            rewrite_command_no_prefixes("grep -f patterns.txt input.txt | cat", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("rg --file=patterns.txt input.txt | cat", &[]),
+            None
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("grep foo src/main.rs | head -5", &[]),
+            Some("rtk grep foo src/main.rs | head -5".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_producer_batch_rules_rewritten() {
+        assert_eq!(
+            rewrite_command_no_prefixes("pytest | tail -20", &[]),
+            Some("rtk pytest | tail -20".into())
+        );
+        assert_eq!(
+            rewrite_command_no_prefixes("terraform plan | head -40", &[]),
+            Some("rtk terraform plan | head -40".into())
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_final_grep_beats_producer_path() {
+        assert_eq!(
+            rewrite_command_no_prefixes("git log | grep feat", &[]),
+            Some("git log | rtk grep feat".into())
+        );
+    }
+
     #[test]
     fn test_rewrite_find_no_pipe_still_rewritten() {
         // find WITHOUT a pipe should still be rewritten
@@ -5917,7 +6386,7 @@ mod tests {
     fn test_rewrite_pipe_then_and() {
         assert_eq!(
             rewrite_command_no_prefixes("git log | head -5 && git stash", &[]),
-            Some("git log | head -5 && rtk git stash".into())
+            Some("rtk git log | head -5 && rtk git stash".into())
         );
     }
 
@@ -5925,7 +6394,7 @@ mod tests {
     fn test_rewrite_pipe_then_semicolon() {
         assert_eq!(
             rewrite_command_no_prefixes("cargo test | head; git status", &[]),
-            Some("cargo test | head; rtk git status".into())
+            Some("rtk cargo test | head; rtk git status".into())
         );
     }
 
@@ -5960,7 +6429,7 @@ mod tests {
     fn test_rewrite_multi_pipe_then_and() {
         assert_eq!(
             rewrite_command_no_prefixes("git log | head | tail && git status", &[]),
-            Some("git log | head | tail && rtk git status".into())
+            Some("rtk git log | head | tail && rtk git status".into())
         );
     }
 
diff --git a/src/discover/rules.rs b/src/discover/rules.rs
index 85dc23af3d..b0ae4f7a0e 100644
--- a/src/discover/rules.rs
+++ b/src/discover/rules.rs
@@ -1,10 +1,29 @@
 use super::report::RtkStatus;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PipelineSafety {
+    None,
+    ProducerOnly,
+    #[allow(dead_code)]
+    FinalOnly,
+    Both,
+}
+
+impl PipelineSafety {
+    pub fn producer_safe(self) -> bool {
+        matches!(self, Self::ProducerOnly | Self::Both)
+    }
+
+    pub fn final_safe(self) -> bool {
+        matches!(self, Self::FinalOnly | Self::Both)
+    }
+}
+
 pub struct RtkRule {
     pub pattern: &'static str,
     pub rtk_cmd: &'static str,
-    /// Whether this command may be rewritten as the final pipeline stage.
-    pub pipeline_final_safe: bool,
+    /// Pipeline stage positions this command may be rewritten in (#3171).
+    pub pipeline_safety: PipelineSafety,
     pub rewrite_prefixes: &'static [&'static str],
     pub category: &'static str,
     pub savings_pct: f64,
@@ -20,7 +39,7 @@ impl RtkRule {
     pub const DEFAULT: RtkRule = RtkRule {
         pattern: "",
         rtk_cmd: "",
-        pipeline_final_safe: false,
+        pipeline_safety: PipelineSafety::None,
         rewrite_prefixes: &[],
         category: "",
         savings_pct: 60.0,
@@ -39,6 +58,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:git|yadm)\s+(?:-[Cc]\s+\S+\s+)*(status|log|diff|show|add|commit|checkout|push|pull|branch|fetch|stash|worktree)",
         rtk_cmd: "rtk git",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["git", "yadm"],
         category: "Git",
         savings_pct: 70.0,
@@ -71,12 +91,12 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^cargo\s+(build|test|clippy|check|fmt|install)",
         rtk_cmd: "rtk cargo",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["cargo"],
         category: "Cargo",
         savings_pct: 80.0,
         subcmd_savings: &[("test", 90.0), ("check", 80.0)],
         subcmd_status: &[("fmt", RtkStatus::Passthrough)],
-        ..RtkRule::DEFAULT
     },
     RtkRule {
         pattern: r"^pnpm\s+(exec|i|install|list|ls|outdated|run|run-script)",
@@ -112,7 +132,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^grep\s+",
         rtk_cmd: "rtk grep",
-        pipeline_final_safe: true,
+        pipeline_safety: PipelineSafety::Both,
         rewrite_prefixes: &["grep"],
         category: "Files",
         savings_pct: 75.0,
@@ -121,7 +141,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^rg\s+",
         rtk_cmd: "rtk rg",
-        pipeline_final_safe: true,
+        pipeline_safety: PipelineSafety::Both,
         rewrite_prefixes: &["rg"],
         category: "Files",
         savings_pct: 75.0,
@@ -130,6 +150,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^ls(\s|$)",
         rtk_cmd: "rtk ls",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["ls"],
         category: "Files",
         savings_pct: 65.0,
@@ -138,6 +159,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^find\s+",
         rtk_cmd: "rtk find",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["find"],
         category: "Files",
         savings_pct: 70.0,
@@ -170,6 +192,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^((p?np(m|x)|p?npm\s+(exec|run|run-script)|npm\s+(rum|urn|x)|pnpm\s+dlx)\s+)?(biome|eslint|lint)(\s|$)",
         rtk_cmd: "rtk lint",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &[
             "biome",
             "eslint",
@@ -220,6 +243,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^((p?np(m|x)|p?npm\s+(exec|run|run-script)|npm\s+(rum|urn|x)|pnpm\s+dlx)\s+)?prettier",
         rtk_cmd: "rtk prettier",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &[
             "npm exec prettier",
             "npm prettier",
@@ -244,6 +268,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^((p?np(m|x)|p?npm\s+(exec|run|run-script)|npm\s+(rum|urn|x)|pnpm\s+dlx)\s+)?next\s+build",
         rtk_cmd: "rtk next",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &[
             "next build",
             "npm exec next build",
@@ -426,6 +451,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^tree(\s|$)",
         rtk_cmd: "rtk tree",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["tree"],
         category: "Files",
         savings_pct: 70.0,
@@ -434,6 +460,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^diff\s+",
         rtk_cmd: "rtk diff",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["diff"],
         category: "Files",
         ..RtkRule::DEFAULT
@@ -457,6 +484,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(python3?\s+-m\s+)?mypy(\s|$)",
         rtk_cmd: "rtk mypy",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["python3 -m mypy", "python -m mypy", "mypy"],
         category: "Build",
         savings_pct: 80.0,
@@ -465,6 +493,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^ruff\s+(check|format)",
         rtk_cmd: "rtk ruff",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["ruff"],
         category: "Python",
         savings_pct: 80.0,
@@ -474,6 +503,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(python[0-9.]*\s+-m\s+)?pytest(\s|$)",
         rtk_cmd: "rtk pytest",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["python3 -m pytest", "python -m pytest", "pytest"],
         category: "Python",
         savings_pct: 90.0,
@@ -482,6 +512,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(pip3?|uv\s+pip)\s+(list|outdated|install|show)",
         rtk_cmd: "rtk pip",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["pip3", "pip", "uv pip"],
         category: "Python",
         savings_pct: 75.0,
@@ -499,6 +530,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^go\s+(test|build|vet)",
         rtk_cmd: "rtk go",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["go"],
         category: "Go",
         savings_pct: 85.0,
@@ -508,6 +540,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:golangci-lint|golangci)\s+(run)(?:\s|$)",
         rtk_cmd: "rtk golangci-lint run",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["golangci-lint run", "golangci run"],
         category: "Go",
         savings_pct: 85.0,
@@ -526,6 +559,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^bundle\s+(install|update)\b",
         rtk_cmd: "rtk bundle",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["bundle"],
         category: "Ruby",
         savings_pct: 70.0,
@@ -534,6 +568,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:bundle\s+exec\s+)?(?:bin/)?(?:rake|rails)\s+test",
         rtk_cmd: "rtk rake",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &[
             "bundle exec rails",
             "bundle exec rake",
@@ -549,6 +584,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:bundle\s+exec\s+)?rspec(?:\s|$)",
         rtk_cmd: "rtk rspec",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["bundle exec rspec", "bin/rspec", "rspec"],
         category: "Tests",
         savings_pct: 65.0,
@@ -557,6 +593,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:bundle\s+exec\s+)?rubocop(?:\s|$)",
         rtk_cmd: "rtk rubocop",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["bundle exec rubocop", "rubocop"],
         category: "Build",
         savings_pct: 65.0,
@@ -589,6 +626,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:php\s+)?(?:\./)?(?:(?:vendor/)?bin/)?phpunit(?:\s|$)",
         rtk_cmd: "rtk phpunit",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         // rewrite_segment_inner normalizes the php wrapper, `./`, vendor/bin and
         // composer bin-dir before matching, so only the residual forms remain:
         // a plain `bin/` (not a Composer dir, so it survives normalization) and
@@ -601,6 +639,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:php\s+)?(?:\./)?(?:(?:vendor/)?bin/)?phpstan\s+analy[sz]e\b",
         rtk_cmd: "rtk phpstan",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["bin/phpstan", "phpstan"],
         category: "Build",
         savings_pct: 65.0,
@@ -610,6 +649,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:\./)?(?:vendor/bin/)?pest(?:\s|$)",
         rtk_cmd: "rtk pest",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["pest"],
         category: "Tests",
         savings_pct: 80.0,
@@ -618,6 +658,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:\./)?(?:vendor/bin/)?paratest(?:\s|$)",
         rtk_cmd: "rtk paratest",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["paratest"],
         category: "Tests",
         savings_pct: 80.0,
@@ -626,6 +667,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:\./)?(?:vendor/bin/)?ecs(?:\s|$)",
         rtk_cmd: "rtk ecs",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["ecs"],
         category: "Build",
         savings_pct: 70.0,
@@ -634,6 +676,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:\./)?(?:vendor/bin/)?pint(?:\s|$)",
         rtk_cmd: "rtk pint",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["pint"],
         category: "Build",
         savings_pct: 70.0,
@@ -731,6 +774,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^brew\s+(install|upgrade)\b",
         rtk_cmd: "rtk brew",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["brew"],
         category: "PackageManager",
         savings_pct: 65.0,
@@ -739,6 +783,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^composer\s+(install|update|require)\b",
         rtk_cmd: "rtk composer",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["composer"],
         category: "PackageManager",
         savings_pct: 65.0,
@@ -747,6 +792,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^df(\s|$)",
         rtk_cmd: "rtk df",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["df"],
         category: "System",
         ..RtkRule::DEFAULT
@@ -754,6 +800,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^dotnet\s+build\b",
         rtk_cmd: "rtk dotnet",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["dotnet"],
         category: "Build",
         savings_pct: 70.0,
@@ -762,6 +809,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^du\b",
         rtk_cmd: "rtk du",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["du"],
         category: "System",
         ..RtkRule::DEFAULT
@@ -793,6 +841,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^hadolint\b",
         rtk_cmd: "rtk hadolint",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["hadolint"],
         category: "Build",
         savings_pct: 65.0,
@@ -801,6 +850,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^helm\b",
         rtk_cmd: "rtk helm",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["helm"],
         category: "Infra",
         savings_pct: 65.0,
@@ -809,6 +859,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^iptables\b",
         rtk_cmd: "rtk iptables",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["iptables"],
         category: "Infra",
         ..RtkRule::DEFAULT
@@ -824,6 +875,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^markdownlint\b",
         rtk_cmd: "rtk markdownlint",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["markdownlint"],
         category: "Build",
         savings_pct: 65.0,
@@ -832,6 +884,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^mix\s+(compile|format)(\s|$)",
         rtk_cmd: "rtk mix",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["mix"],
         category: "Build",
         savings_pct: 65.0,
@@ -840,6 +893,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^(?:\./mvnw|mvnw\.cmd|mvnw|mvn)\b(?:\s+\S+)*?\s+(compile|test|integration-test|package|install|verify|deploy)\b",
         rtk_cmd: "rtk mvn",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["./mvnw", "mvnw.cmd", "mvnw", "mvn"],
         category: "Build",
         savings_pct: 82.0,
@@ -870,6 +924,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^pio\s+run",
         rtk_cmd: "rtk pio",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["pio"],
         category: "Build",
         savings_pct: 65.0,
@@ -878,6 +933,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^poetry\s+(install|lock|update)\b",
         rtk_cmd: "rtk poetry",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["poetry"],
         category: "Python",
         savings_pct: 65.0,
@@ -886,6 +942,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^pre-commit\b",
         rtk_cmd: "rtk pre-commit",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["pre-commit"],
         category: "Build",
         savings_pct: 65.0,
@@ -894,6 +951,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^ps(\s|$)",
         rtk_cmd: "rtk ps",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["ps"],
         category: "System",
         ..RtkRule::DEFAULT
@@ -916,6 +974,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^quarto\s+render",
         rtk_cmd: "rtk quarto",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["quarto"],
         category: "Build",
         savings_pct: 65.0,
@@ -932,6 +991,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^shellcheck\b",
         rtk_cmd: "rtk shellcheck",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["shellcheck"],
         category: "Build",
         savings_pct: 65.0,
@@ -940,6 +1000,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^shopify\s+theme\s+(push|pull)",
         rtk_cmd: "rtk shopify",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["shopify"],
         category: "Build",
         savings_pct: 65.0,
@@ -955,6 +1016,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^swift\s+(build|test)\b",
         rtk_cmd: "rtk swift",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["swift"],
         category: "Build",
         savings_pct: 65.0,
@@ -964,6 +1026,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^systemctl\s+status\b",
         rtk_cmd: "rtk systemctl",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["systemctl"],
         category: "System",
         savings_pct: 65.0,
@@ -972,6 +1035,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^terraform\s+plan",
         rtk_cmd: "rtk terraform",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["terraform"],
         category: "Infra",
         savings_pct: 70.0,
@@ -980,6 +1044,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^tofu\s+(fmt|init|plan|validate)(\s|$)",
         rtk_cmd: "rtk tofu",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["tofu"],
         category: "Infra",
         savings_pct: 70.0,
@@ -988,6 +1053,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^trunk\s+build",
         rtk_cmd: "rtk trunk",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["trunk"],
         category: "Build",
         savings_pct: 65.0,
@@ -1004,6 +1070,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^yamllint\b",
         rtk_cmd: "rtk yamllint",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["yamllint"],
         category: "Build",
         savings_pct: 65.0,
@@ -1012,6 +1079,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^wc(\s|$)",
         rtk_cmd: "rtk wc",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["wc"],
         category: "Files",
         ..RtkRule::DEFAULT
@@ -1027,6 +1095,7 @@ pub const RULES: &[RtkRule] = &[
     RtkRule {
         pattern: r"^liquibase(?:\s|$)",
         rtk_cmd: "rtk liquibase",
+        pipeline_safety: PipelineSafety::ProducerOnly,
         rewrite_prefixes: &["liquibase"],
         category: "Infra",
         savings_pct: 65.0,
diff --git a/src/hooks/hook_cmd.rs b/src/hooks/hook_cmd.rs
index ed1353bfd6..5fbd000166 100644
--- a/src/hooks/hook_cmd.rs
+++ b/src/hooks/hook_cmd.rs
@@ -1750,6 +1750,17 @@ mod tests {
         assert_eq!(cmd, "cargo test | rtk grep FAILED");
     }
 
+    #[test]
+    fn test_claude_pipeline_rewrites_producer_when_consumers_safe() {
+        let result = run_claude_inner(&claude_input("git log | tail -5")).unwrap();
+        let v: Value = serde_json::from_str(&result).unwrap();
+        let cmd = v
+            .pointer("/hookSpecificOutput/updatedInput/command")
+            .and_then(|c| c.as_str())
+            .unwrap();
+        assert_eq!(cmd, "rtk git log | tail -5");
+    }
+
     #[test]
     fn test_claude_json_output_structure() {
         let result = run_claude_inner(&claude_input("git status")).unwrap();