feat(secrets): streaming CoT-redaction prototype + feasibility (roadmap-v0.6 E.4)

CHANGELOG.md

@@ -30,6 +30,21 @@ agent/PR/role/size/invariant audit trail.
 
 ### Added
 
+- **Streaming CoT-redaction prototype + feasibility (roadmap-v0.6 E.4).** Added
+  `crustcore_secrets::token_stream::TokenRedactor` and the supporting
+  `Redactor::longest_dangling_prefix` / `max_needle_len`. `TokenRedactor` buffers
+  streamed model tokens to a **redaction boundary** (newline), scans with the existing
+  `Redactor`, and emits only fully redacted chunks — so a secret split across tokens or
+  lines is caught before any byte reaches the user (invariants 2, 3). A forced
+  (no-boundary) emit retains only the **longest dangling needle-prefix** suffix (the
+  start of a not-yet-finished secret), keeping the buffer bounded without ever emitting a
+  partial secret. 6 red-team tests (split-across-tokens, straddling forced-emit,
+  full-secret-present, no-false-positives, flush-tail, bounded-worst-case). Added
+  [`docs/cot-streaming.md`](./docs/cot-streaming.md) concluding token-level CoT streaming
+  is **feasible** behind the existing `reveal_reasoning` opt-in, with the latency bound
+  (<500 ms) and the one constraint (the boundary char must not appear inside a secret).
+  Analysis + a pure core; **zero nano impact**.
+
 - **GitHub `/crustcore` slash commands (roadmap-v0.6 E.2).** Added the pure parser
   `crustcore_daemon::github_commands`: `parse_command(text) → Option<GithubCommand>`
   turns an **untrusted** PR/issue comment into a typed, bounded command —
@@ -298,6 +313,7 @@ agent/PR/role/size/invariant audit trail.
 
 | Date | Phase/Task | Change | PR / Branch | Agent / Role | Nano Δ | Invariants |
 | --- | --- | --- | --- | --- | --- | --- |
+| 2026-06-28 | v0.6/E.4 | `TokenRedactor` streaming-redaction prototype (buffer-to-boundary + dangling-prefix retention) + `docs/cot-streaming.md` feasibility (feasible, behind `reveal_reasoning`) | `claude/v06-e4-cotstream` | Claude (Implementer) | 0 kB (secrets/docs) | Enforces 2, 3, 11; no unredacted secret reaches the user mid-stream |
 | 2026-06-28 | v0.6/E.2 | `github_commands::parse_command`: untrusted PR comment → typed bounded `/crustcore` command (Run/Retry/Cancel/Explain/RiskDetected); injection stays literal; routes through the Telegram dispatch | `claude/v06-e2-ghcommands` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 4, 7, 8, 11, 16; parsed by the daemon, never model output |
 | 2026-06-28 | v0.6/A.5 | `#[ignore]`d `live_issue_to_pr_smoke` composing A.1–A.4 + D.1 end-to-end; CI decision path already covered by `golden_issue_to_pr_flow`. Completes Phase A | `claude/v06-a5-issuetopr` | Claude (Implementer) | 0 kB (eval/docs only) | Composes 6, 11, 13; verifier-owned end-to-end |
 | 2026-06-28 | v0.6/A.4 | CI monitor: `aggregate_check_runs` (failure-dominates) + `monitor_decision` (Wait/Green/SpawnRepair/StopExhausted over the budget) + bounded `repair_task_goal`; live poll `#[ignore]`d | `claude/v06-a4-cimonitor` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 4, 7, 11; repair bounded, decided by CrustCore from aggregated state |

crates/crustcore-secrets/src/lib.rs

@@ -37,6 +37,10 @@
 /// Encrypted-file vault backend (`vault-file` feature; never in nano).
 #[cfg(feature = "vault-file")]
 pub mod store;
+/// Streaming token redaction (roadmap-v0.6 E.4): buffers model tokens to a redaction
+/// boundary, scans with the existing `Redactor`, and yields only redacted chunks, with a
+/// bounded retain-window so a secret can never be split across the emit point.
+pub mod token_stream;
 
 /// OS keychain loaders (`macos-keychain` / `linux-keyring` features; never in nano).
 /// Dependency-free: they shell out to the system tool and load into an
@@ -240,6 +244,39 @@ impl Redactor {
             .sort_by_key(|(_, v)| std::cmp::Reverse(v.len()));
     }
 
+    /// The byte length of the **longest** registered secret, or 0 if none.
+    #[must_use]
+    pub fn max_needle_len(&self) -> usize {
+        self.needles.iter().map(|(_, v)| v.len()).max().unwrap_or(0)
+    }
+
+    /// The length (bytes) of the longest **suffix of `text` that is a proper prefix of
+    /// some registered secret** — a "dangling partial secret" that may complete in a
+    /// later chunk. The streaming [`token_stream::TokenRedactor`](crate::token_stream::TokenRedactor)
+    /// retains exactly this suffix on a forced (no-boundary) emit, so it never emits the
+    /// start of a secret that will only finish later: anything *before* the dangling
+    /// suffix has no partial-secret tail, so redacting it and emitting is safe, while the
+    /// dangling suffix stays buffered until it completes (and is caught) or is proven not
+    /// a secret. Returns 0 when the buffer's tail is not the beginning of any secret. The
+    /// length leaks nothing about the secret's contents.
+    #[must_use]
+    pub fn longest_dangling_prefix(&self, text: &str) -> usize {
+        let t = text.as_bytes();
+        let mut best = 0;
+        for (_, needle) in &self.needles {
+            let max_k = t.len().min(needle.len().saturating_sub(1));
+            let mut k = max_k;
+            while k > best {
+                if t[t.len() - k..] == needle[..k] {
+                    best = k;
+                    break;
+                }
+                k -= 1;
+            }
+        }
+        best
+    }
+
     /// Redacts every registered secret occurrence in `text`. It collects **all**
     /// match spans over the *original* text, **merges overlapping/adjacent spans**
     /// into covering intervals, and emits one marker per interval — so every byte

crates/crustcore-secrets/src/token_stream.rs

@@ -0,0 +1,254 @@
+// SPDX-License-Identifier: Apache-2.0
+//! Streaming token redaction (roadmap-v0.6 E.4 — feasibility prototype).
+//!
+//! Token-by-token chain-of-thought streaming is *only* safe if no unredacted secret can
+//! reach the user mid-stream (invariants 2, 3). Naively emitting each model token would
+//! leak a secret that arrives split across tokens (`ghp_` then `SECRET…`). [`TokenRedactor`]
+//! makes streaming safe by **buffering to a redaction boundary**, scanning the buffer with
+//! the existing [`Redactor`], and emitting only redacted chunks:
+//!
+//! - **Boundary = newline.** A registered secret value never contains a newline, so a
+//!   secret can never straddle a newline. Emitting (and redacting) everything up to and
+//!   including a newline is therefore safe — any secret in that chunk is wholly inside it.
+//! - **Bounded latency via dangling-prefix retention.** A long line with no newline must
+//!   still emit eventually (no unbounded buffering). When the buffer reaches `max_buffer`,
+//!   the redactor retains only the longest suffix that is a *prefix of some secret*
+//!   ([`Redactor::longest_dangling_prefix`]) — the start of a not-yet-finished secret —
+//!   and redacts+emits everything before it. The emitted part has no partial-secret tail,
+//!   so a secret straddling the buffer end is retained whole and caught when it completes.
+//!
+//! See `docs/cot-streaming.md` for the full analysis, constraints, and the
+//! `reveal_reasoning` opt-in.
+
+use crate::Redactor;
+
+/// Buffers streamed tokens and yields only **fully redacted** chunks. Borrow a
+/// [`Redactor`] holding the live secret set; the redactor is never mutated.
+pub struct TokenRedactor<'r> {
+    redactor: &'r Redactor,
+    buffer: String,
+    /// Force an emit once the buffer reaches this many bytes (bounds latency).
+    max_buffer: usize,
+}
+
+impl<'r> TokenRedactor<'r> {
+    /// Builds a streaming redactor. `max_buffer` bounds how long a boundary-free run may
+    /// grow before a forced emit (latency cap).
+    #[must_use]
+    pub fn new(redactor: &'r Redactor, max_buffer: usize) -> Self {
+        TokenRedactor {
+            redactor,
+            buffer: String::new(),
+            max_buffer: max_buffer.max(1),
+        }
+    }
+
+    /// Accepts one streamed token. Returns a redacted chunk when a newline boundary is
+    /// reached, or when the buffer hits `max_buffer` (the bounded-latency forced emit);
+    /// otherwise buffers and returns `None`. The returned chunk is fully redacted; any
+    /// **dangling partial secret** at the tail stays buffered for the next chunk, so a
+    /// secret is never split across the emit point (the start of a not-yet-complete
+    /// secret is never emitted — invariants 2, 3).
+    #[must_use]
+    pub fn accept_token(&mut self, token: &str) -> Option<String> {
+        self.buffer.push_str(token);
+
+        // Boundary emit: flush through the last newline (a secret cannot span a newline).
+        if let Some(nl) = self.buffer.rfind('\n') {
+            let emit: String = self.buffer.drain(..=nl).collect();
+            return Some(self.redactor.redact(&emit));
+        }
+
+        // Forced emit: the boundary-free run is too long. Retain only the longest
+        // dangling needle-prefix suffix; emit (redacted) everything before it. A secret
+        // straddling the buffer end is, by construction, that dangling suffix, so it is
+        // retained whole — never emitted partially.
+        if self.buffer.len() >= self.max_buffer {
+            let d = self.redactor.longest_dangling_prefix(&self.buffer);
+            let cut = self.suffix_boundary(d);
+            if cut == 0 {
+                // The whole buffer is a dangling partial secret — keep waiting.
+                return None;
+            }
+            let emit: String = self.buffer.drain(..cut).collect();
+            return Some(self.redactor.redact(&emit));
+        }
+        None
+    }
+
+    /// Flushes the buffered tail at end-of-stream, redacted. Call once the model stops.
+    #[must_use]
+    pub fn flush(&mut self) -> Option<String> {
+        if self.buffer.is_empty() {
+            return None;
+        }
+        let emit: String = self.buffer.drain(..).collect();
+        Some(self.redactor.redact(&emit))
+    }
+
+    /// The byte offset that retains `window` bytes as a suffix, rounded **down** to a char
+    /// boundary so we never split a UTF-8 sequence.
+    fn suffix_boundary(&self, window: usize) -> usize {
+        let len = self.buffer.len();
+        let mut cut = len.saturating_sub(window);
+        while cut > 0 && !self.buffer.is_char_boundary(cut) {
+            cut -= 1;
+        }
+        cut
+    }
+}
+
+/// Drives a [`TokenRedactor`] over a provider token stream end-to-end: pulls each token
+/// from `tokens` (the deframed provider chunks — an SSE `data:` delta, a websocket frame,
+/// or a test vector) and hands every redacted chunk to `emit` in arrival order, flushing
+/// the buffered tail at end-of-stream.
+///
+/// This is the **real streaming pipeline** the live loop runs; it is transport-agnostic,
+/// so the *only* remaining live inch is the SSE/websocket socket that produces `tokens`,
+/// which lives in the daemon/net layer (this crate stays std-only). The redaction itself
+/// is fully CI-tested here.
+///
+/// Trust boundary: streamed tokens are model output — untrusted (invariant 7) — so every
+/// chunk passes through the redactor before `emit` (invariants 1–3). This driver never
+/// *decides* to stream (the caller opens a stream only when `reveal_reasoning` is set —
+/// `docs/cot-streaming.md`); it only redacts what it is handed. A secret split across any
+/// number of tokens is still caught: [`TokenRedactor`] buffers to a safe boundary and
+/// only ever emits a leak-free prefix.
+pub fn redact_stream<I, E>(redactor: &Redactor, max_buffer: usize, tokens: I, mut emit: E)
+where
+    I: IntoIterator<Item = String>,
+    E: FnMut(String),
+{
+    let mut tr = TokenRedactor::new(redactor, max_buffer);
+    for token in tokens {
+        if let Some(chunk) = tr.accept_token(&token) {
+            emit(chunk);
+        }
+    }
+    if let Some(chunk) = tr.flush() {
+        emit(chunk);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn redactor_with(secret: &[u8]) -> Redactor {
+        let mut r = Redactor::new();
+        r.register("secret", secret);
+        r
+    }
+
+    /// Feeds tokens, collecting every emitted chunk + the flush; returns the full output.
+    fn stream(tr: &mut TokenRedactor, tokens: &[&str]) -> String {
+        let mut out = String::new();
+        for t in tokens {
+            if let Some(chunk) = tr.accept_token(t) {
+                out.push_str(&chunk);
+            }
+        }
+        if let Some(chunk) = tr.flush() {
+            out.push_str(&chunk);
+        }
+        out
+    }
+
+    #[test]
+    fn secret_split_across_tokens_is_caught() {
+        let r = redactor_with(b"ghp_SECRETTOKEN");
+        let mut tr = TokenRedactor::new(&r, 4096);
+        // The secret arrives split across three tokens, then a newline.
+        let out = stream(&mut tr, &["thinking ghp_", "SECRET", "TOKEN done\n"]);
+        assert!(!out.contains("ghp_SECRETTOKEN"), "secret leaked: {out}");
+        assert!(out.contains("thinking"));
+    }
+
+    #[test]
+    fn redact_stream_driver_redacts_a_split_secret_end_to_end() {
+        // The public driver IS the streaming pipeline: feed the deframed provider tokens,
+        // collect everything `emit` yields, assert the split secret never reaches the sink.
+        let r = redactor_with(b"ghp_SECRETTOKEN");
+        let tokens = vec![
+            "reasoning: ghp_".to_string(),
+            "SECRET".to_string(),
+            "TOKEN looks right\n".to_string(),
+            "continuing\n".to_string(),
+        ];
+        let mut sink = String::new();
+        redact_stream(&r, 4096, tokens, |chunk| sink.push_str(&chunk));
+        assert!(!sink.contains("ghp_SECRETTOKEN"), "secret leaked: {sink}");
+        assert!(sink.contains("reasoning:"));
+        assert!(sink.contains("continuing"));
+    }
+
+    #[test]
+    fn redact_stream_over_an_empty_stream_emits_nothing() {
+        let r = redactor_with(b"ghp_SECRETTOKEN");
+        let mut emitted = 0usize;
+        redact_stream(&r, 4096, Vec::<String>::new(), |_| emitted += 1);
+        assert_eq!(emitted, 0);
+    }
+
+    #[test]
+    fn secret_at_end_of_buffer_with_no_newline_is_retained_not_leaked() {
+        let r = redactor_with(b"ghp_SECRETTOKEN");
+        // Tiny max_buffer to force the no-boundary emit path.
+        let mut tr = TokenRedactor::new(&r, 8);
+        // No newline ever; the secret straddles the forced-emit point.
+        let mut out = String::new();
+        for t in ["aaaaaa ghp_SE", "CRETTOKEN bbbb"] {
+            if let Some(c) = tr.accept_token(t) {
+                out.push_str(&c);
+            }
+        }
+        if let Some(c) = tr.flush() {
+            out.push_str(&c);
+        }
+        assert!(
+            !out.contains("ghp_SECRETTOKEN"),
+            "secret leaked on forced emit: {out}"
+        );
+    }
+
+    #[test]
+    fn worst_case_no_boundary_stays_bounded() {
+        let r = redactor_with(b"zzzz");
+        let max = 64;
+        let mut tr = TokenRedactor::new(&r, max);
+        // Feed a long boundary-free run; the internal buffer must never exceed max.
+        for _ in 0..1000 {
+            let _ = tr.accept_token("abcdefghij"); // 10 bytes each
+            assert!(
+                tr.buffer.len() <= max + 10,
+                "buffer grew unbounded: {}",
+                tr.buffer.len()
+            );
+        }
+    }
+
+    #[test]
+    fn non_secret_text_passes_through_unchanged() {
+        let r = redactor_with(b"ghp_SECRETTOKEN");
+        let mut tr = TokenRedactor::new(&r, 4096);
+        let out = stream(&mut tr, &["hello world\n", "no secrets here\n"]);
+        assert_eq!(out, "hello world\nno secrets here\n");
+    }
+
+    #[test]
+    fn flush_emits_the_trailing_partial_line_redacted() {
+        let r = redactor_with(b"ghp_TAILSECRET");
+        let mut tr = TokenRedactor::new(&r, 4096);
+        // No trailing newline — the tail only emits on flush.
+        let out = stream(&mut tr, &["leftover ghp_TAILSECRET"]);
+        assert!(!out.contains("ghp_TAILSECRET"));
+        assert!(out.contains("leftover"));
+    }
+
+    #[test]
+    fn empty_redactor_is_a_passthrough() {
+        let r = Redactor::new();
+        let mut tr = TokenRedactor::new(&r, 4096);
+        assert_eq!(stream(&mut tr, &["any ", "text\n"]), "any text\n");
+    }
+}