fix(seats): scrub credentials across streamed record chunks

packages/seats/docs/record-v1.md

@@ -427,10 +427,15 @@ What's new or changed in `EVENT_TYPES`:
      and a writer refuses one that is not name-shaped.
    - Agent-authored text, such as message chunks and tool titles, belongs to
      the agent and could repeat a value the agent holds. A writer that knows
-     the run's credential values replaces every occurrence of 8 characters or
-     more with `[redacted]` before the line is hashed.
+     the run's credential values replaces every occurrence of each non-empty known value
+     with `[redacted]` before the line is hashed. For streamed message and
+     thought text, a possible credential prefix stays in memory until the
+     next chunk from the same actor and kind resolves it. At run settlement,
+     remaining harmless prefixes are written before `run.settled`. A crash
+     can lose that held suffix (at most the longest value minus one character
+     per stream), but never writes it early.
    - That replacement covers known values only. A value the writer was not
-     given, or one the agent transformed (base64, split across chunks), passes
+     given, or one the agent transformed (such as base64), passes
      through.
 2. **Readers ignore unknown fields and skip unknown kinds.** A reader that
    meets a kind it doesn't know still checks the chain over that line.

packages/seats/src/trace/store.js

@@ -12,9 +12,28 @@ import { randomUUID } from 'node:crypto';
 import { appendFileSync, mkdirSync, readFileSync, existsSync, writeFileSync } from 'node:fs';
 import { join } from 'node:path';
 import {
-  appendEntry, hashLine, kindOfType, readRecord, traceTypeOf, validateActor, verifyRecord, BOARD_ACTOR,
+  appendEntry, canonicalLine, hashLine, kindOfType, readRecord, traceTypeOf, validateActor, verifyRecord, BOARD_ACTOR,
 } from '../record/format.js';
-import { redactKnownCredentials } from '../upstream-refusal.js';
+// Match original input once: replacement markers must not be scrubbed again
+// when a second credential happens to be a substring of the marker.
+const redactText = (input, secrets, final = true) => {
+  let text = '';
+  let offset = 0;
+  while (offset < input.length) {
+    const rest = input.slice(offset);
+    if (!final && secrets.some(secret => secret.length > rest.length && secret.startsWith(rest))) break;
+    const match = secrets.find(secret => input.startsWith(secret, offset));
+    if (match) {
+      text += '[redacted]';
+      offset += match.length;
+    } else {
+      text += input[offset];
+      offset += 1;
+    }
+  }
+  return { text, tail: input.slice(offset) };
+};
+const STREAM_KINDS = new Set(['agent.message_chunk', 'agent.thought_chunk']);
 
 const RUN_ID = /^[A-Za-z0-9_-]{6,64}$/;
 const readJsonl = (path) => (existsSync(path)
@@ -26,7 +45,7 @@ const readJsonl = (path) => (existsSync(path)
 /** Known credential values never reach a line, even inside agent text. */
 const scrub = (value, secrets) => {
   if (secrets.length === 0) return value;
-  if (typeof value === 'string') return redactKnownCredentials(value, secrets);
+  if (typeof value === 'string') return redactText(value, secrets).text;
   if (Array.isArray(value)) return value.map((item) => scrub(item, secrets));
   if (value && typeof value === 'object') {
     return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, scrub(item, secrets)]));
@@ -71,12 +90,22 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => {
     let settled = false;
     let broken = false;
     const secrets = [];
+    const streams = new Map(); // actor + kind → at most maxCredentialLength - 1 held characters
     const grants = new Map(); // open grants: grantId → { parentGrantId, binding, expiresAt }
     const startedAt = now();
 
-    const write = (kind, data, actor) => {
+    const write = (kind, data, actor, final = false) => {
+      let streamChange;
+      let safeData = scrub(data, secrets);
+      if (STREAM_KINDS.has(kind) && typeof data.text === 'string' && secrets.length) {
+        const key = canonicalLine({ kind, actor: actor ?? runActor ?? null });
+        const previous = streams.get(key);
+        const { text, tail } = redactText((previous?.tail || '') + data.text, secrets, final);
+        safeData = { ...safeData, text };
+        streamChange = { key, tail, kind, actor };
+      }
       const line = appendEntry(last, {
-        runId, taskId, podId, ts: now().toISOString(), kind, data: scrub(data, secrets), actor,
+        runId, taskId, podId, ts: now().toISOString(), kind, data: safeData, actor,
       }, { runActor });
       try {
         appendFileSync(path, `${line}\n`, { mode: 0o600 });
@@ -85,6 +114,11 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => {
         broken = true;
         throw error;
       }
+      if (streamChange) {
+        const { key, tail } = streamChange;
+        if (tail) streams.set(key, streamChange);
+        else streams.delete(key);
+      }
       last = line;
       seq += 1;
       const entry = JSON.parse(line);
@@ -104,6 +138,15 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => {
       if (!kind) throw new Error(`unknown trace event type: ${type}`);
       if (settled) return null;
       if (broken) throw new Error(`run ${runId}: the record is broken after a failed write`);
+      // Settle is the last possible chunk. Release harmless unfinished prefixes
+      // before sealing the chain; unrelated events never flush those prefixes.
+      if (kind === 'run.settled') {
+        // A rejected settlement must not release a prefix before more chunks.
+        appendEntry(last, {
+          runId, taskId, podId, ts: startedAt.toISOString(), kind, data: scrub(data, secrets), actor: options?.actor,
+        }, { runActor });
+        for (const stream of [...streams.values()]) write(stream.kind, { text: '' }, stream.actor, true);
+      }
       const event = write(kind, data, options?.actor);
       if (kind === 'task.done') {
         // The board closed the task: every grant bound to it ends, in one write.
@@ -143,7 +186,8 @@ export const createTraceStore = ({ root, now = () => new Date() } = {}) => {
     };
 
     const redact = (values = []) => {
-      for (const value of values) if (typeof value === 'string' && !secrets.includes(value)) secrets.push(value);
+      for (const value of values) if (typeof value === 'string' && value.length > 0 && !secrets.includes(value)) secrets.push(value);
+      secrets.sort((a, b) => b.length - a.length);
     };
 
     append(indexPath, {

packages/seats/test/record-redact-supervised.test.js

@@ -84,7 +84,8 @@ const echoingAdapter = ({ baseURL, envRoot, fail, seen }) => ({
       });
       const deltas = await sseDeltas(response);
       seen.reply = deltas.join('');
-      for (const text of deltas) ctx.trace.emit('agent.message_chunk', { text }, as);
+      // Force the known value across record writes, including one-character deltas.
+      for (const delta of deltas) for (const text of delta) ctx.trace.emit('agent.message_chunk', { text }, as);
       ctx.trace.emit('tool.call', {
         toolCallId: 'tc-1', title: `printenv ${KEY_NAME}`, kind: 'execute', status: 'pending', surface: 'native',
       }, as);

packages/seats/test/record-store.test.js

@@ -193,3 +193,71 @@ test('the store records session.opened and context usage', () => withStore((stor
   assert.deepEqual(store.readRun('run-st-ses1').map((e) => e.kind), ['attempt.started', 'session.opened', 'usage', 'run.settled']);
   assert.deepEqual(verifyRecord(readFileSync(fileOf(root, 'run-st-ses1'), 'utf8')).errors, []);
 }));
+
+for (const secret of ['sk-test-stream-canary-123456', 'abc123']) {
+  test(`redact: every split of a ${secret.length}-character credential stays private`, () => withStore((store, root) => {
+    for (let cut = 1; cut < secret.length; cut += 1) {
+      const runId = `run-split-${secret.length}-${cut}`;
+      const run = store.startRun({ runId, actor: AGENT });
+      run.redact([secret]);
+      run.emit('agent.message_chunk', { text: `before ${secret.slice(0, cut)}` });
+      // An unrelated event must not flush a potentially secret suffix.
+      run.emit('usage', { inputTokens: 1 });
+      const during = store.readRun(runId).filter(e => e.kind === 'agent.message_chunk').map(e => e.data.text).join('');
+      assert.equal(during, 'before ', 'a potential credential prefix stays in memory');
+      run.emit('agent.message_chunk', { text: `${secret.slice(cut)} after` });
+      run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 });
+      const events = store.readRun(runId);
+      const text = events.filter(e => e.kind === 'agent.message_chunk').map(e => e.data.text).join('');
+      assert.equal(text, 'before [redacted] after');
+      assert.equal(readFileSync(fileOf(root, runId), 'utf8').includes(secret), false);
+      assert.equal(store.verifyRun(runId).ok, true);
+      for (const e of events.filter(e => e.kind === 'agent.message_chunk')) assert.deepEqual(e.actor, AGENT);
+      assert.equal(events.at(-1).kind, 'run.settled');
+    }
+  }));
+}
+
+test('redact: short credentials in metadata and single-character streamed deltas are scrubbed', () => withStore((store) => {
+  const run = store.startRun({ runId: 'run-short-stream' });
+  run.redact(['abc123', '']);
+  for (const text of 'safe abc123 suffix') run.emit('agent.message_chunk', { text });
+  run.emit('tool.call', { toolCallId: 't', title: 'token abc123' });
+  run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 });
+  const events = store.readRun(run.runId);
+  assert.equal(events.filter(e => e.kind === 'agent.message_chunk').map(e => e.data.text).join(''), 'safe [redacted] suffix');
+  assert.equal(events.find(e => e.kind === 'tool.call').data.title, 'token [redacted]');
+}));
+
+test('redact: a harmless trailing prefix is retained at settle, without crossing runs', () => withStore((store) => {
+  const a = store.startRun({ runId: 'run-tail-one', actor: AGENT });
+  const b = store.startRun({ runId: 'run-tail-two', actor: AGENT });
+  a.redact(['secret123']);
+  b.redact(['secret123']);
+  a.emit('agent.message_chunk', { text: 'ordinary sec' });
+  b.emit('agent.message_chunk', { text: 'ret123' });
+  a.emit('run.settled', { outcome: 'failed', failure: { kind: 'exited', detail: 'stopped' }, durationMs: 1, attempts: 1 });
+  b.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 });
+  const texts = runId => store.readRun(runId).filter(e => e.kind === 'agent.message_chunk').map(e => e.data.text).join('');
+  assert.equal(texts(a.runId), 'ordinary sec');
+  assert.equal(texts(b.runId), 'ret123');
+  assert.equal(store.verifyRun(a.runId).ok, true);
+}));
+
+test('redact: actors have separate suffixes, and a rejected settle cannot flush a secret prefix', () => withStore((store) => {
+  const run = store.startRun({ runId: 'run-actor-stream', actor: AGENT });
+  const other = { type: 'agent', id: 'u-other', owner: 'u-ana' };
+  run.redact(['secret123', 'secret']);
+  run.emit('agent.message_chunk', { text: 'secret' }, { actor: AGENT });
+  run.emit('agent.message_chunk', { text: '123' }, { actor: other });
+  assert.throws(() => run.emit('run.settled', { outcome: 'ok' }, { actor: { type: 'human', id: 'invalid id' } }));
+  const pending = store.readRun(run.runId).filter(e => e.kind === 'agent.message_chunk' && e.actor.id === AGENT.id).map(e => e.data.text).join('');
+  assert.equal(pending, '', 'a rejected settle must leave the prefix in memory');
+  run.emit('agent.message_chunk', { text: '123 done' }, { actor: AGENT });
+  run.emit('run.settled', { outcome: 'ok', durationMs: 1, attempts: 1 });
+  const entries = store.readRun(run.runId);
+  const textFor = id => entries.filter(e => e.kind === 'agent.message_chunk' && e.actor.id === id).map(e => e.data.text).join('');
+  assert.equal(textFor(AGENT.id), '[redacted] done');
+  assert.equal(textFor(other.id), '123');
+  assert.equal(store.verifyRun(run.runId).ok, true);
+}));