[failproofai] Preserve events and redact telemetry credentials

CHANGELOG.md

@@ -4,6 +4,15 @@
 
 ### Fixes
 
+- `failproofaid` now applies `collector.redact` to externally written SDK spool
+  batches immediately before upload. SDK JSONL files previously bypassed the
+  daemon's redaction path entirely because redaction only ran while the daemon
+  created its own session and hook events. A batch written by an older SDK could
+  therefore send a captured API key verbatim even with the default `minimal`
+  setting. The uploader now scrubs every valid JSON event with the existing
+  deterministic rules, leaves malformed lines untouched for ingest to reject
+  and preserve through the failed-batch path, and still honors
+  `collector.redact: off`.
 - The two PyPI publish workflows open the next version's `CHANGELOG` section in the same `bump` commit that moves `_version.py`, via a new `scripts/changelog-open.py`. `bump` used to move the version alone, leaving `main` on a version with no section — the exact state `scripts/changelog-section.py` refuses at release time and `__tests__/ci/python-version-pipeline.test.ts` asserts against. Because bump commits carry a skip-ci marker, that never went red on itself: it went red on the next unrelated PR to run CI, which is how `main` broke after the 0.0.1b1 publish (repaired by hand in #755, which named the recurrence and left it) and again after 0.0.1b2. `sdk/python/CHANGELOG.md` gets the 0.0.1b3 section that was missing. The opener is idempotent — a re-run of `bump` against a `main` that already carries the section is a no-op rather than a second heading, which would put only the first one's body on the GitHub Release — and it matches the version with the same trailing word boundary the extractor uses, so opening `0.0.1b1` is not satisfied by an existing `0.0.1b10` (#787)
 - `fp-cloud-cli`'s Click shim survives typer 0.27.2, which moved `Abort` out of its vendored Click. `_click_compat` wrapped all six vendored imports in one `try: … except ImportError: from click import …`, so that single missing name rebound **every** symbol to pip Click — the exact silent failure the module exists to prevent. Typer catches only its own Click's exceptions, so every typed error escaped uncaught: `fp alerts show ghost` exited 1 with an empty stderr instead of 6 with a message, and the same for exits 2, 3, 4 and 5. 105 tests went red on the dependabot bump that first installed 0.27.2. The Click is now chosen once — on whether `typer._click` exists at all — and each symbol imported from that choice, so a name that goes missing raises at import (a CLI that will not start) rather than silently downgrading every error to exit 1. `Abort` alone is resolved from `typer.Abort`, which tracks the move by construction: pip Click's before typer 0.26, the vendored class through 0.27.1, `typer.exceptions.Abort` from 0.27.2 (#771)
 

crates/failproofaid/src/main.rs

@@ -789,7 +789,9 @@ fn collector_tasks() -> Vec<fpai_collect::TaskSpec> {
         ingest.url.clone(),
         ingest.key.clone(),
         cfg.failed_dir.clone(),
-    ) {
+    )
+    .map(|u| u.with_redact(cfg.settings.redact))
+    {
         Ok(u) => std::sync::Arc::new(u),
         Err(err) => {
             eprintln!("[failproofaid] collector disabled: {err}");

crates/fpai-collect/src/redact.rs

@@ -156,7 +156,7 @@ const WEAK_SECRET_NAMES: &[&str] = &["key", "token"];
 /// to be a placeholder or a flag than a credential.
 const MIN_ASSIGNMENT_VALUE: usize = 12;
 
-/// Scrub every string leaf of an event in place.
+/// Scrub credential-shaped object keys and string values in place.
 ///
 /// Returns the number of replacements, so a caller can log that redaction
 /// actually did something without logging what it removed.
@@ -165,20 +165,58 @@ pub fn scrub_value(v: &mut Value, mode: Redact) -> usize {
         return 0;
     }
     let mut n = 0;
-    scrub_in_place(v, &mut n);
+    scrub_in_place(v, None, &mut n);
     n
 }
 
-fn scrub_in_place(v: &mut Value, n: &mut usize) {
+fn is_secret_name(name: &str) -> bool {
+    let raw = name.trim_matches('-');
+    let lower = raw.to_ascii_lowercase();
+    let compound = raw.contains('_')
+        || raw.contains('-')
+        || raw.chars().skip(1).any(|c| c.is_ascii_uppercase());
+    STRONG_SECRET_NAMES.iter().any(|part| lower.ends_with(part))
+        || (compound && WEAK_SECRET_NAMES.iter().any(|part| lower.ends_with(part)))
+}
+
+fn is_literal_secret(value: &str) -> bool {
+    value.len() >= MIN_ASSIGNMENT_VALUE
+        && !value.starts_with(['{', '$', '<', '(', '`'])
+        && !value.starts_with("[redacted:")
+}
+
+fn scrub_in_place(v: &mut Value, field_name: Option<&str>, n: &mut usize) {
     match v {
         Value::String(s) => {
             if let Some(replaced) = scrub_str(s) {
                 *n += replaced.1;
                 *s = replaced.0;
+            } else if field_name.is_some_and(is_secret_name) && is_literal_secret(s) {
+                *n += 1;
+                *s = "[redacted:secret-assignment]".to_string();
+            }
+        }
+        Value::Array(a) => a.iter_mut().for_each(|e| scrub_in_place(e, None, n)),
+        Value::Object(o) => {
+            let entries = std::mem::take(o);
+            for (key, mut value) in entries {
+                scrub_in_place(&mut value, Some(&key), n);
+                let redacted_key = match scrub_str(&key) {
+                    Some((key, count)) => {
+                        *n += count;
+                        key
+                    }
+                    None => key,
+                };
+                let mut unique_key = redacted_key.clone();
+                let mut suffix = 2;
+                while o.contains_key(&unique_key) {
+                    unique_key = format!("{redacted_key}#{suffix}");
+                    suffix += 1;
+                }
+                o.insert(unique_key, value);
             }
         }
-        Value::Array(a) => a.iter_mut().for_each(|e| scrub_in_place(e, n)),
-        Value::Object(o) => o.values_mut().for_each(|e| scrub_in_place(e, n)),
         _ => {}
     }
 }
@@ -344,12 +382,8 @@ fn match_assignment(s: &str, i: usize, rest: &str) -> Option<(usize, &'static st
     if name_len == 0 {
         return None;
     }
-    let raw = name_part[name_part.len() - name_len..].to_ascii_lowercase();
-    let name = raw.trim_matches('-');
-    let compound = name.contains('_') || name.contains('-');
-    let convincing = STRONG_SECRET_NAMES.iter().any(|n| name.ends_with(n))
-        || (compound && WEAK_SECRET_NAMES.iter().any(|n| name.ends_with(n)));
-    if !convincing {
+    let name = &name_part[name_part.len() - name_len..];
+    if !is_secret_name(name) {
         return None;
     }
 
@@ -362,15 +396,18 @@ fn match_assignment(s: &str, i: usize, rest: &str) -> Option<(usize, &'static st
 
     // The value runs to the closing quote, or to whitespace / a shell
     // separator when unquoted. The closing quote is left in place.
-    let quoted = matches!(s[..i].chars().next_back(), Some('"') | Some('\''));
+    let quote = match s[..i].chars().next_back() {
+        Some(c @ ('"' | '\'')) => Some(c),
+        _ => None,
+    };
     // Bytes, not characters — see the note on `match_bearer`. This predicate
     // also accepts non-ASCII, so a char count under-reports the span and
     // `scrub_str`'s `i += len` leaves the cursor inside the value.
     let value_len: usize = rest
         .chars()
         .take_while(|c| {
-            if quoted {
-                *c != '"' && *c != '\''
+            if let Some(quote) = quote {
+                *c != quote
             } else {
                 // Quotes end an unquoted value too, matching `match_bearer`'s
                 // token run. An unquoted shell word does not contain a bare
@@ -557,6 +594,16 @@ mod tests {
             scrub(r#"--api-token=abcdefghijklmnop"trailing""#),
             r#"--api-token=[redacted:secret-assignment]"trailing""#
         );
+        // The opposite quote is valid inside a quoted shell value and must not
+        // terminate the secret early.
+        assert_eq!(
+            scrub(r#"PASSWORD='abcdefghijkL"mnopQRST'"#),
+            r#"PASSWORD='[redacted:secret-assignment]'"#
+        );
+        assert_eq!(
+            scrub(r#"PASSWORD="abcdefghijkL'mnopQRST""#),
+            r#"PASSWORD="[redacted:secret-assignment]""#
+        );
     }
 
     /// Redaction must never be a net data loss beyond the secret itself: every
@@ -643,6 +690,33 @@ mod tests {
         assert_eq!(v["type"], "tool_use");
     }
 
+    #[test]
+    fn secret_named_fields_and_credential_shaped_keys_are_scrubbed() {
+        let first = "API_KEY=abcdefghijklmnop";
+        let second = "API_KEY=qrstuvwxyzabcdef";
+        let mut nested = serde_json::Map::new();
+        nested.insert(first.to_string(), json!(1));
+        nested.insert(second.to_string(), json!(2));
+        let mut v = json!({
+            "password": "abcdefghijklmnop",
+            "client_secret": "abcdefghijklmnop",
+            "api_key": "abcdefghijklmnop",
+            "accessToken": "abcdefghijklmnop",
+            "nested": Value::Object(nested),
+        });
+
+        let n = scrub_value(&mut v, Redact::Minimal);
+        assert_eq!(v["password"], "[redacted:secret-assignment]");
+        assert_eq!(v["client_secret"], "[redacted:secret-assignment]");
+        assert_eq!(v["api_key"], "[redacted:secret-assignment]");
+        assert_eq!(v["accessToken"], "[redacted:secret-assignment]");
+        let nested = v["nested"].as_object().unwrap();
+        assert_eq!(nested.len(), 2, "redacted keys must not collapse fields");
+        assert!(!nested.contains_key(first));
+        assert!(!nested.contains_key(second));
+        assert!(n >= 6, "expected fields and keys to be scrubbed, got {n}");
+    }
+
     #[test]
     fn off_mode_changes_nothing() {
         let mut v = json!({"output": "ghp_abcdefghijklmnopqrstuvwxyz0123"});

crates/fpai-collect/src/uploader.rs

@@ -34,7 +34,9 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
+
+use crate::config::Redact;
 
 /// Suffix marking a batch that exhausted its retry budget. Deliberately NOT
 /// `.jsonl`, so every directory scan and the watcher skip it for free rather
@@ -165,6 +167,7 @@ pub struct Uploader {
     max_retries: u32,
     retry_base: Duration,
     failed_retries_max: u32,
+    redact: Redact,
     metrics: Arc<UploadMetrics>,
 }
 
@@ -197,10 +200,16 @@ impl Uploader {
             max_retries: DEFAULT_MAX_RETRIES,
             retry_base: DEFAULT_RETRY_BASE,
             failed_retries_max: DEFAULT_FAILED_RETRIES_MAX,
+            redact: Redact::default(),
             metrics: Arc::new(UploadMetrics::default()),
         })
     }
 
+    pub fn with_redact(mut self, redact: Redact) -> Self {
+        self.redact = redact;
+        self
+    }
+
     /// Shorten every delay. Tests only — without it each retry test would wait
     /// out a real multi-second backoff.
     #[doc(hidden)]
@@ -232,6 +241,7 @@ impl Uploader {
             Err(e) => return Err(UploadError::Io(e)),
         };
 
+        let bytes = redact_batch(&bytes, self.redact);
         for chunk in split_lines(&bytes, self.max_upload_bytes) {
             self.post_batch(path, chunk).await?;
         }
@@ -489,6 +499,35 @@ impl Uploader {
     }
 }
 
+fn redact_batch(bytes: &[u8], mode: Redact) -> Vec<u8> {
+    if mode == Redact::Off {
+        return bytes.to_vec();
+    }
+
+    let mut out = Vec::with_capacity(bytes.len());
+    for line in bytes.split_inclusive(|byte| *byte == b'\n') {
+        let (body, newline) = line
+            .strip_suffix(b"\n")
+            .map_or((line, false), |body| (body, true));
+        match serde_json::from_slice::<serde_json::Value>(body) {
+            Ok(mut event) => {
+                if crate::redact::scrub_value(&mut event, mode) > 0 {
+                    event
+                        .serialize(&mut serde_json::Serializer::new(&mut out))
+                        .expect("serializing JSON into Vec cannot fail");
+                } else {
+                    out.extend_from_slice(body);
+                }
+            }
+            Err(_) => out.extend_from_slice(body),
+        }
+        if newline {
+            out.push(b'\n');
+        }
+    }
+    out
+}
+
 /// Retry state carried in a parked batch's filename:
 /// `<base>.a<N>[.c<STATUS>].jsonl[.poison]`.
 ///

crates/fpai-collect/tests/uploader.rs

@@ -9,7 +9,7 @@ use std::path::{Path, PathBuf};
 use std::sync::atomic::Ordering;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use fpai_collect::{UploadError, Uploader};
+use fpai_collect::{Redact, UploadError, Uploader};
 use wiremock::matchers::{header, method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
@@ -88,6 +88,74 @@ async fn a_2xx_with_an_accepting_ack_deletes_the_batch() {
     fs::remove_dir_all(&failed).ok();
 }
 
+#[tokio::test]
+async fn sdk_batches_are_redacted_before_upload() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "accepted": 1, "skipped": 0
+        })))
+        .mount(&server)
+        .await;
+
+    let spool = tmpdir("redact-spool");
+    let failed = tmpdir("redact-failed");
+    let batch = spool.join("event-s-1-0.jsonl");
+    fs::write(
+        &batch,
+        r#"{"type":"tool_use","input":{"command":"API_KEY=abcdefghijklmnop","password":"qrstuvwxyzabcdef","API_KEY=secretvalue123456":true}}
+"#,
+    )
+    .unwrap();
+
+    uploader(&server, &failed)
+        .upload_file(&batch)
+        .await
+        .unwrap();
+
+    let requests = server.received_requests().await.unwrap();
+    let body = String::from_utf8(requests[0].body.clone()).unwrap();
+    assert!(
+        !body.contains("abcdefghijklmnop"),
+        "credential reached the wire"
+    );
+    assert!(!body.contains("qrstuvwxyzabcdef"));
+    assert!(!body.contains("API_KEY=secretvalue123456"));
+    assert!(body.contains("[redacted:secret-assignment]"));
+
+    fs::remove_dir_all(&spool).ok();
+    fs::remove_dir_all(&failed).ok();
+}
+
+#[tokio::test]
+async fn uploader_redaction_can_be_disabled() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "accepted": 1, "skipped": 0
+        })))
+        .mount(&server)
+        .await;
+
+    let spool = tmpdir("redact-off-spool");
+    let failed = tmpdir("redact-off-failed");
+    let batch = spool.join("event-s-1-0.jsonl");
+    fs::write(&batch, "{\"output\":\"API_KEY=abcdefghijklmnop\"}\n").unwrap();
+
+    uploader(&server, &failed)
+        .with_redact(Redact::Off)
+        .upload_file(&batch)
+        .await
+        .unwrap();
+
+    let requests = server.received_requests().await.unwrap();
+    let body = String::from_utf8(requests[0].body.clone()).unwrap();
+    assert!(body.contains("abcdefghijklmnop"));
+
+    fs::remove_dir_all(&spool).ok();
+    fs::remove_dir_all(&failed).ok();
+}
+
 #[tokio::test]
 async fn a_200_that_stored_nothing_is_counted_as_fully_skipped() {
     // The failure this exists for: a systematically malformed transform gets

docs/start/integrations/custom-agents.mdx

@@ -643,20 +643,24 @@ Each flush writes one batch file, `.tmp` first, then `fsync`, then an atomic ren
 The daemon only picks up `.jsonl`, so it can never read a half-written file. The stem carries a timestamp, process id and sequence number, so two processes flushing in the same millisecond cannot collide. The queue is capped at 10,000 events; past that it drops the oldest and logs.
 
 <Warning>
-  **`collector.redact` does not apply to your SDK events.** It never sees them.
+  **`collector.redact` defaults to `minimal` for SDK events too.** The SDK
+  scrubs before writing a batch to disk, and the daemon repeats the same
+  deterministic pass before upload so batches from older SDKs are protected.
 </Warning>
 
-The daemon **ships** your batches. It does not open or rewrite them.
+The daemon reads each batch and applies redaction in memory before upload. It
+does not rewrite the spool file it read.
 
-| Events | Written by | Redacted by `collector.redact`? |
+| Events | Written by | Where minimal redaction runs |
 | --- | --- | --- |
-| CLI session transcripts | The daemon | Yes |
-| Hook activity | The daemon | Yes |
-| **Everything the SDK emits** | **Your process** | **No** |
+| CLI session transcripts | The daemon | Before the daemon writes the batch |
+| Hook activity | The daemon | Before the daemon writes the batch |
+| **Everything the SDK emits** | **Your process** | **Before the SDK writes the batch and again before daemon upload** |
 
-Redaction runs where the daemon *writes* its own events — not where batches are *shipped*. So a prompt or a tool argument holding an API key still holds it on arrival.
-
-That is deliberate. These are your own instrumentation calls, and rewriting them in transit would mean the events you receive are not the events you emitted.
+Set `collector.redact` to `off` only when verbatim payloads are an explicit
+requirement; the SDK and daemon both honor that setting. Minimal redaction
+catches common API keys, bearer tokens, JWTs, and secret assignments. It cannot
+identify arbitrary sensitive prose.
 
 <Tip>
   **You control payloads at the source, in two places:**
@@ -669,7 +673,7 @@ That is deliberate. These are your own instrumentation calls, and rewriting them
     `instrument()` drops options an adapter does not read, so passing the wrong name raises nothing and changes nothing.
   - Don't hand the secret to `input=` in the first place.
 
-  `collector.redact` is not a substitute for either.
+  `collector.redact` is defence in depth, not a substitute for either.
 </Tip>
 
 <Warning>