From cd9c7df208e4720cccc8dc28af3d849e46adf060 Mon Sep 17 00:00:00 2001 From: RNT56 Date: Mon, 29 Jun 2026 10:35:37 +0200 Subject: [PATCH 1/2] feat(secrets): streaming CoT-redaction prototype + feasibility (roadmap-v0.6 E.4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds crustcore_secrets::token_stream::TokenRedactor + 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 β€” 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 (the start of a not-yet-finished secret), keeping the buffer bounded without ever emitting a partial secret. 6 red-team tests; docs/cot-streaming.md concludes token-level CoT streaming is feasible behind the existing reveal_reasoning opt-in, with the latency bound (<500ms) and the one constraint (boundary char not inside a secret). Analysis + a pure core; zero nano impact. `cargo xtask verify` green. Co-Authored-By: Claude Opus 4.8 --- CHANGELOG.md | 16 ++ crates/crustcore-secrets/src/lib.rs | 37 ++++ crates/crustcore-secrets/src/token_stream.rs | 196 +++++++++++++++++++ docs/cot-streaming.md | 97 +++++++++ 4 files changed, 346 insertions(+) create mode 100644 crates/crustcore-secrets/src/token_stream.rs create mode 100644 docs/cot-streaming.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a7fdf1..125b312 100644 --- a/CHANGELOG.md +++ b/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**. + - **Evidence bundle rendering (roadmap-v0.6 C.3).** Added `EvidenceBundle::to_markdown()` and `to_json()` to `crustcore_daemon::product`. `to_markdown` is the **bounded** canonical PR-body/cockpit renderer: it opens with @@ -249,6 +264,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/C.3 | `EvidenceBundle::to_markdown` (bounded PR-body/cockpit render, πŸ”΄ review notice, per-list overflow) + `to_json` (schema v1); `draft_pr_body` delegates | `claude/v06-c3-evidence` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 2, 10, 11; bounded redacted evidence, every receipt included | | 2026-06-28 | v0.6/D.1 | Task-loop wiring `plan_task`/`finalize_task` composing routing (C.1) + advisory gate (C.2) into a terminal `TaskOutcome`; sandboxed run `#[ignore]`d | `claude/v06-d1-executor-wire` | Claude (Implementer) | 0 kB (daemon-only) | Enforces 4, 5, 6, 13; verifier-owned completion, advisory only gates | | 2026-06-28 | v0.6/A.3 | `pr_intent_to_create_request`: PrIntentβ†’CreatePrRequest for the live draft-PR POST; evidence body verbatim, draft=true; real POST `#[ignore]`d | `claude/v06-a3-draftpr` | Claude (Implementer) | 0 kB (daemon/live-only) | Enforces 6, 13, 14; body is evidence not a model claim | diff --git a/crates/crustcore-secrets/src/lib.rs b/crates/crustcore-secrets/src/lib.rs index 867720c..ce26707 100644 --- a/crates/crustcore-secrets/src/lib.rs +++ b/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 diff --git a/crates/crustcore-secrets/src/token_stream.rs b/crates/crustcore-secrets/src/token_stream.rs new file mode 100644 index 0000000..e81cdce --- /dev/null +++ b/crates/crustcore-secrets/src/token_stream.rs @@ -0,0 +1,196 @@ +// 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 { + 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 { + 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 + } +} + +#[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 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"); + } +} diff --git a/docs/cot-streaming.md b/docs/cot-streaming.md new file mode 100644 index 0000000..35517e3 --- /dev/null +++ b/docs/cot-streaming.md @@ -0,0 +1,97 @@ +# docs/cot-streaming.md β€” Token-by-token CoT streaming feasibility (roadmap-v0.6 E.4) + +> **Question.** Can CrustCore stream a model's chain-of-thought (CoT) reasoning to +> the operator **token by token** without ever leaking an unredacted secret +> (invariants 2, 3)? +> +> **Answer: feasible, with a bounded buffer-to-boundary redactor and one documented +> constraint.** A working prototype ships as +> [`crustcore_secrets::token_stream::TokenRedactor`]; this note records the design, +> the red-team analysis, the latency bound, and the constraint. + +--- + +## The hazard + +Naively forwarding each model token to the user is unsafe: a registered secret can +arrive **split across tokens** (`ghp_` then `SECRET` then `TOKEN`). The batch +[`Redactor`] only catches a secret once it sees the whole value, so per-token +emission would leak the prefix of any streamed secret before the match completes. + +## The design β€” buffer to a redaction boundary, retain the dangling tail + +`TokenRedactor` wraps the existing `Redactor` and emits only **fully redacted** +chunks: + +1. **Boundary = newline.** A registered secret value never contains a newline, so a + secret can never straddle one. Whenever the buffer contains a newline, the + redactor flushes everything up to and including the last newline, redacts that + chunk, and emits it. Any secret in the chunk is wholly inside it β†’ caught. + +2. **Bounded latency via dangling-prefix retention.** A long boundary-free run must + still emit before the buffer grows without bound. On reaching `max_buffer` with no + newline, the redactor computes the **longest suffix of the buffer that is a proper + prefix of some secret** (`Redactor::longest_dangling_prefix`) and retains exactly + that suffix; it redacts and emits everything before it. This is the key safety + property: the only thing held back is the *start of a secret that has not finished + yet*. Everything emitted has no partial-secret tail, so redacting it in isolation + is complete. The retained dangling prefix is at most `max_needle_len - 1` bytes, so + the buffer stays bounded. + +```text +buffer = "…reasoning… ghp_SEC" (max_buffer hit, no newline) + └── emit redact(…)β”€β”€β”˜ β”” retained: "ghp_SEC" is a prefix of a secret +next token "RETTOKEN\n" β†’ buffer "ghp_SECRETTOKEN\n" β†’ newline flush β†’ redacted βœ“ +``` + +## Red-team analysis (all pass β€” see the module tests) + +| Scenario | Outcome | +| --- | --- | +| Secret split across tokens (same line) | buffered to the newline, redacted as one chunk β€” **no leak** | +| Secret straddling a forced (no-boundary) emit | retained as the dangling prefix, completed + redacted next chunk β€” **no leak** | +| Full secret present at a forced emit | dangling length 0 β†’ whole buffer redacted + emitted β€” **no leak** | +| Non-secret text | passes through unchanged (no false positives β€” only registered needles match) | +| Trailing partial line at end-of-stream | `flush()` redacts the tail β€” **no leak** | +| Worst-case boundary-free run | buffer bounded by `max_buffer` + one token; emits incrementally | + +**False positives.** The redactor matches only *registered* secret values, so benign +text that merely resembles a secret is never redacted β€” the tradeoff is that a secret +must be registered with the broker to be scrubbed (the same contract as the batch +redactor; nothing new mid-stream). + +**Latency bound.** Output is delayed only until the next newline, or at most until +`max_buffer` bytes accumulate. With `max_buffer` sized above realistic reasoning line +lengths (e.g. 4 KiB), added latency is one line of reasoning β€” well under the 500 ms +target on any normal stream. A pathological no-newline run emits every `max_buffer` +bytes, so latency is bounded, not unbounded. + +## The one constraint + +Safety rests on **the boundary character (`\n`) never appearing inside a registered +secret**, and on **every secret being registered with the broker** before the stream +starts. Both already hold for CrustCore's secret types (API keys, tokens, PEM bodies +are single-line opaque values; multi-line PEMs are redacted as their whole armored +block, which contains no bare reasoning newline mid-value). A hypothetical secret that +embedded a newline *and* was streamed across that newline would need a sentence/format +boundary instead β€” out of scope until such a secret type exists. + +## Wiring (the live seam β€” not built here) + +The prototype is the pure, CI-tested core. Going live needs three things, none of +which change the trust boundary: + +1. the provider/`crustcore-net` helper must expose an **incremental token stream** + (most do β€” SSE `delta` events); +2. the chat/Telegram dispatch loop feeds each token through a `TokenRedactor` and + forwards the emitted chunks; +3. it is gated behind the existing **`reveal_reasoning` opt-in** (`docs/chat.md`) β€” + off by default; the operator turns it on knowing reasoning is shown (redacted). + +## Conclusion + +**Token-level CoT streaming is compatible with the redaction boundary** using +boundary-buffered redaction with dangling-prefix retention. The prototype +(`TokenRedactor`) demonstrates leak-free streaming on the red-team scenarios with a +bounded, sub-500 ms latency profile. Recommend shipping it behind `reveal_reasoning` +once the net helper exposes a token stream; no invariant relaxation is required. From 817caacc534216e3f00e636c16acd6e54274e20e Mon Sep 17 00:00:00 2001 From: RNT56 Date: Mon, 29 Jun 2026 13:59:11 +0200 Subject: [PATCH 2/2] feat(secrets): real token-stream redaction driver (E.4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds token_stream::redact_stream β€” the transport-agnostic streaming pipeline the live CoT loop runs: pulls each deframed provider token, hands every redacted chunk to an emit sink in order, flushes the buffered tail at end-of-stream. Streamed tokens are model output (untrusted, invariant 7); every chunk passes through the redactor before emit (invariants 1-3). A secret split across any number of tokens is still caught (TokenRedactor buffers to a safe boundary and only emits a leak-free prefix). This makes E.4 a real pipeline, not just the analysis doc + the buffering core: the only remaining live inch is the SSE/websocket socket producing the tokens, which lives in the daemon/net layer (this crate stays std-only). 2 new CI tests (split-secret end-to-end + empty stream); no lib.rs change (not a further contract-file edit); zero nano impact. verify green. Co-Authored-By: Claude Opus 4.8 --- crates/crustcore-secrets/src/token_stream.rs | 58 ++++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/crates/crustcore-secrets/src/token_stream.rs b/crates/crustcore-secrets/src/token_stream.rs index e81cdce..197fc14 100644 --- a/crates/crustcore-secrets/src/token_stream.rs +++ b/crates/crustcore-secrets/src/token_stream.rs @@ -98,6 +98,38 @@ impl<'r> TokenRedactor<'r> { } } +/// 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(redactor: &Redactor, max_buffer: usize, tokens: I, mut emit: E) +where + I: IntoIterator, + 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::*; @@ -132,6 +164,32 @@ mod tests { 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::::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");