feat(aws): add serverless filters for DynamoDB, CloudWatch, S3 transfer, Secrets Manager

src/config.rs

@@ -16,6 +16,8 @@ pub struct Config {
     pub telemetry: TelemetryConfig,
     #[serde(default)]
     pub hooks: HooksConfig,
+    #[serde(default)]
+    pub limits: LimitsConfig,
 }
 
 #[derive(Debug, Serialize, Deserialize, Default)]
@@ -94,6 +96,37 @@ impl Default for TelemetryConfig {
     }
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub struct LimitsConfig {
+    /// Max total grep results to show (default: 200)
+    pub grep_max_results: usize,
+    /// Max matches per file in grep output (default: 25)
+    pub grep_max_per_file: usize,
+    /// Max staged/modified files shown in git status (default: 15)
+    pub status_max_files: usize,
+    /// Max untracked files shown in git status (default: 10)
+    pub status_max_untracked: usize,
+    /// Max chars for parser passthrough fallback (default: 2000)
+    pub passthrough_max_chars: usize,
+}
+
+impl Default for LimitsConfig {
+    fn default() -> Self {
+        Self {
+            grep_max_results: 200,
+            grep_max_per_file: 25,
+            status_max_files: 15,
+            status_max_untracked: 10,
+            passthrough_max_chars: 2000,
+        }
+    }
+}
+
+/// Get limits config. Falls back to defaults if config can't be loaded.
+pub fn limits() -> LimitsConfig {
+    Config::load().map(|c| c.limits).unwrap_or_default()
+}
+
 /// Check if telemetry is enabled in config. Returns None if config can't be loaded.
 pub fn telemetry_enabled() -> Option<bool> {
     Config::load().ok().map(|c| c.telemetry.enabled)

src/container.rs

@@ -183,6 +183,19 @@ fn docker_logs(args: &[String], _verbose: u8) -> Result<()> {
     let stderr = String::from_utf8_lossy(&output.stderr);
     let raw = format!("{}\n{}", stdout, stderr);
 
+    if !output.status.success() {
+        if !stderr.trim().is_empty() {
+            eprint!("{}", stderr);
+        }
+        timer.track(
+            &format!("docker logs {}", container),
+            "rtk docker logs",
+            &raw,
+            &raw,
+        );
+        std::process::exit(output.status.code().unwrap_or(1));
+    }
+
     let analyzed = crate::log_cmd::run_stdin_str(&raw);
     let rtk = format!("🐳 Logs for {}:\n{}", container, analyzed);
     println!("{}", rtk);
@@ -208,6 +221,15 @@ fn kubectl_pods(args: &[String], _verbose: u8) -> Result<()> {
     let raw = String::from_utf8_lossy(&output.stdout).to_string();
     let mut rtk = String::new();
 
+    if !output.status.success() {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        if !stderr.trim().is_empty() {
+            eprint!("{}", stderr);
+        }
+        timer.track("kubectl get pods", "rtk kubectl pods", &raw, &raw);
+        std::process::exit(output.status.code().unwrap_or(1));
+    }
+
     let json: serde_json::Value = match serde_json::from_str(&raw) {
         Ok(v) => v,
         Err(_) => {
@@ -306,6 +328,15 @@ fn kubectl_services(args: &[String], _verbose: u8) -> Result<()> {
     let raw = String::from_utf8_lossy(&output.stdout).to_string();
     let mut rtk = String::new();
 
+    if !output.status.success() {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        if !stderr.trim().is_empty() {
+            eprint!("{}", stderr);
+        }
+        timer.track("kubectl get svc", "rtk kubectl svc", &raw, &raw);
+        std::process::exit(output.status.code().unwrap_or(1));
+    }
+
     let json: serde_json::Value = match serde_json::from_str(&raw) {
         Ok(v) => v,
         Err(_) => {
@@ -381,6 +412,21 @@ fn kubectl_logs(args: &[String], _verbose: u8) -> Result<()> {
 
     let output = cmd.output().context("Failed to run kubectl logs")?;
     let raw = String::from_utf8_lossy(&output.stdout).to_string();
+
+    if !output.status.success() {
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        if !stderr.trim().is_empty() {
+            eprint!("{}", stderr);
+        }
+        timer.track(
+            &format!("kubectl logs {}", pod),
+            "rtk kubectl logs",
+            &raw,
+            &raw,
+        );
+        std::process::exit(output.status.code().unwrap_or(1));
+    }
+
     let analyzed = crate::log_cmd::run_stdin_str(&raw);
     let rtk = format!("☸️  Logs for {}:\n{}", pod, analyzed);
     println!("{}", rtk);

src/discover/registry.rs

@@ -353,15 +353,22 @@ fn rewrite_compound(cmd: &str, excluded: &[String]) -> Option<String> {
                     }
                     seg_start = i;
                 } else {
-                    // `|` pipe — rewrite first segment only, pass through the rest unchanged
+                    // `|` pipe — rewrite first segment only if the command's
+                    // output format is pipe-safe (text-based). Commands that
+                    // transform structured output (JSON → compressed text)
+                    // must NOT be rewritten, as downstream programs (jq,
+                    // python json.load, etc.) expect the original format.
                     let seg = cmd[seg_start..i].trim();
-                    let rewritten =
-                        rewrite_segment(seg, excluded).unwrap_or_else(|| seg.to_string());
-                    if rewritten != seg {
-                        any_changed = true;
+                    if is_pipe_safe(seg) {
+                        let rewritten = rewrite_segment(seg, excluded)
+                            .unwrap_or_else(|| seg.to_string());
+                        if rewritten != seg {
+                            any_changed = true;
+                        }
+                        result.push_str(&rewritten);
+                    } else {
+                        result.push_str(seg);
                     }
-                    result.push_str(&rewritten);
-                    // Preserve the space before the pipe that was lost by trim()
                     result.push(' ');
                     result.push_str(cmd[i..].trim_start());
                     return if any_changed { Some(result) } else { None };
@@ -500,6 +507,23 @@ fn rewrite_tail_lines(cmd: &str) -> Option<String> {
     None
 }
 
+/// Commands whose RTK-filtered output changes the format in ways that break
+/// downstream pipe consumers (e.g., `aws ... | jq` expects JSON, but RTK
+/// compresses it to summary text). These must NOT be rewritten before a pipe.
+const PIPE_UNSAFE_PREFIXES: &[&str] = &["aws ", "aws\t"];
+
+/// Returns true if a command segment is safe to rewrite before a pipe.
+/// Most RTK filters produce text output compatible with grep/head/tail.
+/// Commands that transform structured output (JSON → compressed text) are not.
+fn is_pipe_safe(seg: &str) -> bool {
+    let trimmed = seg.trim();
+    // Strip env prefixes (sudo, env VAR=val) to get the actual command
+    let stripped = ENV_PREFIX.replace(trimmed, "");
+    let cmd = stripped.trim();
+    !PIPE_UNSAFE_PREFIXES.iter().any(|p| cmd.starts_with(p))
+        && cmd != "aws"
+}
+
 /// Rewrite a single (non-compound) command segment.
 /// Returns `Some(rewritten)` if matched (including already-RTK pass-through).
 /// Returns `None` if no match (caller uses original segment).
@@ -1857,6 +1881,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_rewrite_pipe_aws_skips_rewrite() {
+        // AWS piped to jq/python must NOT be rewritten — RTK's compressed
+        // output would break downstream JSON consumers
+        assert_eq!(
+            rewrite_command(
+                "aws dynamodb scan --table-name foo | python -c 'import json; json.load(sys.stdin)'",
+                &[]
+            ),
+            None
+        );
+    }
+
+    #[test]
+    fn test_rewrite_pipe_aws_in_compound() {
+        // `&&` segments before a pipe should still be rewritten, but the
+        // aws segment before the pipe should NOT
+        assert_eq!(
+            rewrite_command("git status && aws s3 ls | python parse.py", &[]),
+            Some("rtk git status && aws s3 ls | python parse.py".into())
+        );
+    }
+
     #[test]
     fn test_rewrite_compound_four_segments() {
         assert_eq!(

src/git.rs

@@ -1,3 +1,4 @@
+use crate::config;
 use crate::tracking;
 use crate::utils::resolved_command;
 use anyhow::{Context, Result};
@@ -592,33 +593,46 @@ fn format_status_output(porcelain: &str) -> String {
     }
 
     // Build summary
+    let limits = config::limits();
+    let max_files = limits.status_max_files;
+    let max_untracked = limits.status_max_untracked;
+
     if staged > 0 {
         output.push_str(&format!("✅ Staged: {} files\n", staged));
-        for f in staged_files.iter().take(5) {
+        for f in staged_files.iter().take(max_files) {
             output.push_str(&format!("   {}\n", f));
         }
-        if staged_files.len() > 5 {
-            output.push_str(&format!("   ... +{} more\n", staged_files.len() - 5));
+        if staged_files.len() > max_files {
+            output.push_str(&format!(
+                "   ... +{} more\n",
+                staged_files.len() - max_files
+            ));
         }
     }
 
     if modified > 0 {
         output.push_str(&format!("📝 Modified: {} files\n", modified));
-        for f in modified_files.iter().take(5) {
+        for f in modified_files.iter().take(max_files) {
             output.push_str(&format!("   {}\n", f));
         }
-        if modified_files.len() > 5 {
-            output.push_str(&format!("   ... +{} more\n", modified_files.len() - 5));
+        if modified_files.len() > max_files {
+            output.push_str(&format!(
+                "   ... +{} more\n",
+                modified_files.len() - max_files
+            ));
         }
     }
 
     if untracked > 0 {
         output.push_str(&format!("❓ Untracked: {} files\n", untracked));
-        for f in untracked_files.iter().take(3) {
+        for f in untracked_files.iter().take(max_untracked) {
             output.push_str(&format!("   {}\n", f));
         }
-        if untracked_files.len() > 3 {
-            output.push_str(&format!("   ... +{} more\n", untracked_files.len() - 3));
+        if untracked_files.len() > max_untracked {
+            output.push_str(&format!(
+                "   ... +{} more\n",
+                untracked_files.len() - max_untracked
+            ));
         }
     }
 
@@ -1690,23 +1704,48 @@ A  added.rs
 
     #[test]
     fn test_format_status_output_truncation() {
-        // Test that >5 staged files show "... +N more"
-        let porcelain = r#"## main
-M  file1.rs
-M  file2.rs
-M  file3.rs
-M  file4.rs
-M  file5.rs
-M  file6.rs
-M  file7.rs
-"#;
-        let result = format_status_output(porcelain);
-        assert!(result.contains("✅ Staged: 7 files"));
+        // Test that >15 staged files show "... +N more"
+        let mut porcelain = String::from("## main\n");
+        for i in 1..=20 {
+            porcelain.push_str(&format!("M  file{}.rs\n", i));
+        }
+        let result = format_status_output(&porcelain);
+        assert!(result.contains("✅ Staged: 20 files"));
+        assert!(result.contains("file1.rs"));
+        assert!(result.contains("file15.rs"));
+        assert!(result.contains("... +5 more"));
+        assert!(!result.contains("file16.rs"));
+        assert!(!result.contains("file20.rs"));
+    }
+
+    #[test]
+    fn test_format_status_modified_truncation() {
+        // Test that >15 modified files show "... +N more"
+        let mut porcelain = String::from("## main\n");
+        for i in 1..=20 {
+            porcelain.push_str(&format!(" M file{}.rs\n", i));
+        }
+        let result = format_status_output(&porcelain);
+        assert!(result.contains("📝 Modified: 20 files"));
+        assert!(result.contains("file1.rs"));
+        assert!(result.contains("file15.rs"));
+        assert!(result.contains("... +5 more"));
+        assert!(!result.contains("file16.rs"));
+    }
+
+    #[test]
+    fn test_format_status_untracked_truncation() {
+        // Test that >10 untracked files show "... +N more"
+        let mut porcelain = String::from("## main\n");
+        for i in 1..=15 {
+            porcelain.push_str(&format!("?? file{}.rs\n", i));
+        }
+        let result = format_status_output(&porcelain);
+        assert!(result.contains("❓ Untracked: 15 files"));
         assert!(result.contains("file1.rs"));
-        assert!(result.contains("file5.rs"));
-        assert!(result.contains("... +2 more"));
-        assert!(!result.contains("file6.rs"));
-        assert!(!result.contains("file7.rs"));
+        assert!(result.contains("file10.rs"));
+        assert!(result.contains("... +5 more"));
+        assert!(!result.contains("file11.rs"));
     }
 
     #[test]

src/golangci_cmd.rs

@@ -1,3 +1,4 @@
+use crate::config;
 use crate::tracking;
 use crate::utils::{resolved_command, truncate};
 use anyhow::{Context, Result};
@@ -106,7 +107,7 @@ fn filter_golangci_json(output: &str) -> String {
             return format!(
                 "golangci-lint (JSON parse failed: {})\n{}",
                 e,
-                truncate(output, 500)
+                truncate(output, config::limits().passthrough_max_chars)
             );
         }
     };