Keep secrets out of run records at cancels, cuts and flushes

backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs

@@ -1075,11 +1075,12 @@
         var writer = NewEventWriter(context.Owner, context.TeamId);   // same batched-append + flush-at-checkpoint path as the live tail
         var native = await OpenResumedCaptureAsync(context, redactor, cancellationToken).ConfigureAwait(false);   // G1: the RESUMED frame stream of the same process, continuing its source cursor and the execution's reduction
         var applicationSourceHead = context.Handle.StdoutOffset;
+        var frames = redactor.CreateFrameStream();   // a line the reader cut is redacted with its continuation, as on the live tail
 
         async Task PersistFrameAsync(SandboxOutputFrame output)
         {
             var line = output.Text;
-            var redactedLine = redactor.Redact(line);
+            var redactedLine = frames.Redact(line, output.IsComplete);
 
             // A best-effort native flush may have failed after the normalized events were durable and before this
             // source head was checkpointed. Rewind the reader to the plane head to recover those native frames, but do
@@ -4549,10 +4550,11 @@
         var facts = AgentRunFacts.For(context.Harness);   // the three facts a forced terminal reports without folding, read with THIS harness's declared spellings (or the fallback union when it declares none)
         var writer = NewEventWriter(context.Owner, context.TeamId);   // batches the DB inserts; flushed at each spool checkpoint + once at the end
         var native = await OpenNativeCaptureAsync(context, cancellationToken).ConfigureAwait(false);   // G1: the lossless frame plane, dual-written beside the log; a plane that won't open leaves this path unchanged
+        var frames = context.Redactor.CreateFrameStream();   // a line longer than one read pass arrives cut; its two frames must not each keep half of a key
 
         async Task PersistAsync(string line, SandboxOutputFrame? output)
         {
-            var redactedLine = context.Redactor.Redact(line);
+            var redactedLine = frames.Redact(line, output?.IsComplete ?? true);
 
             // Capture the faithful transcript FIRST — redact the raw line, then keep it whether or not ParseEvents
             // surfaces any event. ParseEvents drops blank/unrecognized lines; the transcript keeps them so a replay
@@ -5378,10 +5380,10 @@
     /// <summary>The same reconstruction from a payload that came from somewhere other than the row — an offloaded one fetched back out of the artifact store.</summary>
     private static AgentEvent ReplayedEvent(AgentEventKind kind, string? text, string? dataJson)
     {
         if (dataJson is not { Length: > 0 } json) return new AgentEvent { Kind = kind, Text = text };
 
         try { using var doc = JsonDocument.Parse(json); return new AgentEvent { Kind = kind, Text = text, Data = doc.RootElement.Clone() }; }
         catch (JsonException) { return new AgentEvent { Kind = kind, Text = text }; }
     }
 
     /// <summary>Ask the row, on a token of its own, whether the run actually reached a terminal state — the only honest answer to "did the landing take?" once an exception has been raised somewhere after the fenced write.</summary>
@@ -5731,11 +5733,12 @@
             if (!pump.IsCapturing) return;
 
             var delivered = 0;
+            var frames = capture.Redactor.CreateFrameStream();   // a diagnostic the reader cut must not keep the start of a key its continuation finishes
 
             async Task RecordAsync(SandboxDiagnosticLine line, CancellationToken token)
             {
                 delivered++;
-                await pump.CaptureDiagnosticAsync(line.Text, capture.Redactor.Redact(line.Text), line.IsComplete, token).ConfigureAwait(false);
+                await pump.CaptureDiagnosticAsync(line.Text, frames.Redact(line.Text, line.IsComplete), line.IsComplete, token).ConfigureAwait(false);
             }
 
             var budget = new SandboxDiagnosticBudget { MaxLines = MaxDiagnosticFrames, MaxBytes = MaxDiagnosticBytes };

backend/src/CodeSpace.Core/Services/Agents/SecretRedactor.cs

@@ -1,6 +1,7 @@
 using System.Buffers;
 using System.Security.Cryptography;
 using System.Text;
+using CodeSpace.Core.Services.Redaction;
 
 namespace CodeSpace.Core.Services.Agents;
 
@@ -80,6 +81,41 @@ public string Redact(string text)
     /// <see cref="SecretUtf8Redaction.SourceBytesConsumed"/> source bytes with each transformed result.
     /// </summary>
     public SecretUtf8RedactionStream CreateUtf8Stream() => new(_secrets);
+
+    /// <summary>
+    /// Creates a stateful redactor for decoded output frames, each persisted on its own (a native record, a transcript
+    /// line). A line longer than one read pass arrives as a frame the reader CUT — marked incomplete — and its
+    /// continuation; exact matching per frame cannot see a secret the cut split, so the end of a cut frame that a secret
+    /// could still run across is held back and redacted together with the frame that continues it.
+    /// </summary>
+    public SecretFrameRedactionStream CreateFrameStream() => new(this);
+
+    /// <summary>How much of <paramref name="text"/>, a text that may still continue, can be redacted on its own (<see cref="SecretBoundary.SettledLength"/>).</summary>
+    internal int SettledLength(string text) => SecretBoundary.SettledLength(text, _secrets);
+}
+
+public sealed class SecretFrameRedactionStream
+{
+    private readonly SecretRedactor _redactor;
+    private string _carry = "";
+
+    internal SecretFrameRedactionStream(SecretRedactor redactor) => _redactor = redactor;
+
+    /// <summary>
+    /// The redacted text to persist for one frame: the end held back from the cut frame before it, then this frame —
+    /// all of it when the frame ends its line, and when it was cut only what no secret can still run across, the rest
+    /// held for the frame that continues it. A stream that stops on a cut frame never persists what it held, so the
+    /// frames persisted always concatenate to a prefix of the line, redacted.
+    /// </summary>
+    public string Redact(string frame, bool isComplete)
+    {
+        var text = _carry + frame;
+        var settled = isComplete ? text.Length : _redactor.SettledLength(text);
+
+        _carry = text[settled..];
+
+        return _redactor.Redact(text[..settled]);
+    }
 }
 
 public sealed class SecretUtf8RedactionStream

backend/src/CodeSpace.Core/Services/Redaction/SecretBoundary.cs

@@ -0,0 +1,51 @@
+namespace CodeSpace.Core.Services.Redaction;
+
+/// <summary>
+/// Where a text that may still continue can be cut for exact-match redaction. A writer that persists a growing text in
+/// pieces — a stream flushed by size or by time, a reader's bounded pass over one long line — redacts each piece on its
+/// own, and exact matching cannot recognise a secret split between two of them. It releases the settled prefix and
+/// carries the rest into the next piece. Shared by every redactor that writes in pieces, whatever its marker.
+/// </summary>
+internal static class SecretBoundary
+{
+    /// <summary>The longest prefix of <paramref name="text"/> that no secret runs across and after which nothing could yet become one.</summary>
+    public static int SettledLength(string text, IReadOnlyList<string> secrets)
+    {
+        var settled = text.Length - LongestOpenSecretStart(text, secrets);
+
+        while (StartOfSecretAcross(text, settled, secrets) is { } start) settled = start;
+
+        return settled;
+    }
+
+    /// <summary>The length of the longest end of <paramref name="text"/> that is the start, but not the whole, of a secret.</summary>
+    private static int LongestOpenSecretStart(string text, IReadOnlyList<string> secrets)
+    {
+        var longest = 0;
+
+        foreach (var secret in secrets)
+        {
+            for (var length = Math.Min(secret.Length - 1, text.Length); length > longest; length--)
+            {
+                if (!text.AsSpan(text.Length - length).SequenceEqual(secret.AsSpan(0, length))) continue;
+
+                longest = length;
+                break;
+            }
+        }
+
+        return longest;
+    }
+
+    /// <summary>Where a whole secret that runs across <paramref name="cut"/> starts, or null when none does.</summary>
+    private static int? StartOfSecretAcross(string text, int cut, IReadOnlyList<string> secrets)
+    {
+        foreach (var secret in secrets)
+        {
+            for (var start = Math.Max(0, cut - secret.Length + 1); start < cut; start++)
+                if (text.AsSpan(start).StartsWith(secret, StringComparison.Ordinal)) return start;
+        }
+
+        return null;
+    }
+}

backend/src/CodeSpace.Core/Services/Workflows/Engine/WorkflowEngine.cs

@@ -3664,7 +3664,8 @@ private sealed record NodeExecution(WorkflowRun Run, NodeDefinition Node, INodeR
     private NodeRunContext BuildNodeRunContext(NodeExecution exec)
     {
         var nodeLogger = _loggerFactory.CreateLogger($"Workflow.{exec.Runtime.TypeKey}.{exec.Node.Id}");
-        var observability = new NodeObservability(new NodeObservationBinding(_recordLogger, _artifactStore, exec.Run.Id, exec.Node.Id, exec.Run.TeamId, exec.ParentRecordId, PersistenceSecretRedactor.FromScope(exec.Scope), _completenessWriter));
+        var redactor = PersistenceSecretRedactor.FromScope(exec.Scope);
+        var observability = new NodeObservability(new NodeObservationBinding(_recordLogger, _artifactStore, exec.Run.Id, exec.Node.Id, exec.Run.TeamId, exec.ParentRecordId, redactor, _completenessWriter));
 
         return new NodeRunContext
         {
@@ -3676,6 +3677,7 @@ private NodeRunContext BuildNodeRunContext(NodeExecution exec)
             Scope = exec.Scope,
             Logger = nodeLogger,
             Observability = observability,
+            PersistenceRedactor = redactor,
             ResumePayload = exec.ResumePayload,
             PriorAttemptPayload = exec.PriorAttemptPayload,
             NodeId = exec.Node.Id,

backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingLLMClientDecorator.cs

@@ -227,7 +227,7 @@ protected static JsonElement FailedPayload(LlmCallScope scope, string provider,
     }
 
     /// <summary>Fold one redaction into THIS call's masking observation and hand its value back, so the flag the presence delta carries is a fact about the bytes that reached storage rather than about the configuration. A scope minted outside <see cref="LlmCallScope.ForOneCall"/> observes nothing and reads back verbatim, which is the conservative answer.</summary>
-    private static T Observe<T>(LlmCallScope scope, PersistenceRedaction<T> redaction) => scope.Masking is { } masking ? masking.Observe(redaction) : redaction.Value;
+    protected static T Observe<T>(LlmCallScope scope, PersistenceRedaction<T> redaction) => scope.Masking is { } masking ? masking.Observe(redaction) : redaction.Value;
 
     /// <summary>A plain-text field (a prompt / a text completion): the inline string when small, else a content-addressed <c>$artifact_id</c> ref. Null/empty rides as-is.</summary>
     protected static async Task<object?> OffloadTextAsync(LlmCallScope scope, string? text, CancellationToken cancellationToken)

backend/src/CodeSpace.Core/Services/Workflows/Llm/RecordingStreamingStructuredLLMClientDecorator.cs

@@ -1,6 +1,7 @@
 using System.Runtime.CompilerServices;
 using System.Text;
 using System.Text.Json;
+using CodeSpace.Core.Services.Workflows.Runtime;
 using CodeSpace.Messages.Constants;
 
 namespace CodeSpace.Core.Services.Workflows.Llm;
@@ -57,9 +58,7 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat
             () => StartedPayloadAsync(scope, Provider, request.Model, request.SystemPrompt, request.UserPrompt, request.Temperature, request.MaxOutputTokens, cancellationToken), cancellationToken).ConfigureAwait(false);
 
         var accumulator = new LlmCompletionAccumulator();
-        var pending = new StringBuilder();
-        var pendingBytes = 0;
-        var deltaOrdinal = 0;
+        var pending = new PendingDelta(PersistenceSecretRedactor.Union(scope.NativeCredentialRedactor, scope.CaptureRedactor));
 
         await using var enumerator = _streamingInner.StreamAsync(request, cancellationToken).GetAsyncEnumerator(cancellationToken);
         using var timer = new PeriodicTimer(DeltaFlushInterval, _timeProvider);
@@ -71,16 +70,15 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat
             bool moved;
             try
             {
-                if (pending.Length > 0)
+                if (!pending.IsEmpty)
                 {
                     var completed = await Task.WhenAny(moveNextTask, tickTask).ConfigureAwait(false);
                     if (completed == tickTask)
                     {
                         cancellationToken.ThrowIfCancellationRequested();
                         if (!await tickTask.ConfigureAwait(false)) break;
 
-                        await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal++, cancellationToken).ConfigureAwait(false);
-                        pendingBytes = 0;
+                        await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.Settled, cancellationToken).ConfigureAwait(false);
                         tickTask = timer.WaitForNextTickAsync(cancellationToken).AsTask();
                         continue;
                     }
@@ -92,8 +90,8 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat
             {
                 // The caller's token is commonly already cancelled here. Capture uses its own non-cancelled attempt so
                 // cancellation itself remains a typed terminal fact; SafeRecord still swallows any ledger failure.
-                if (pending.Length > 0)
-                    await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal++, CancellationToken.None).ConfigureAwait(false);
+                if (!pending.IsEmpty)
+                    await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.Cut, CancellationToken.None).ConfigureAwait(false);
 
                 var failureRecorded = await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionFailed, correlationId, () => Task.FromResult(FailedPayload(scope, Provider, ex)), CancellationToken.None).ConfigureAwait(false);
                 if (failureRecorded && declared) await MarkCapturePresentAsync(scope).ConfigureAwait(false);
@@ -109,12 +107,8 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat
             if (current is LlmStreamEvent.TextDelta delta)
             {
                 pending.Append(delta.Text);
-                pendingBytes += Encoding.UTF8.GetByteCount(delta.Text);
-                if (pendingBytes >= DeltaFlushBytes)
-                {
-                    await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal++, cancellationToken).ConfigureAwait(false);
-                    pendingBytes = 0;
-                }
+                if (pending.Bytes >= DeltaFlushBytes)
+                    await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.Settled, cancellationToken).ConfigureAwait(false);
             }
 
             yield return current;
@@ -123,21 +117,79 @@ await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionStarted, correlat
 
         // Keep the final tail in the ordered live projection. This is at most one extra row per streamed call, preserves
         // exact concatenation, and does not change authority: completed below remains the canonical whole output.
-        if (pending.Length > 0)
-            await RecordPendingDeltaAsync(scope, correlationId, pending, deltaOrdinal, cancellationToken).ConfigureAwait(false);
+        if (!pending.IsEmpty)
+            await RecordPendingDeltaAsync(scope, correlationId, pending, DeltaFlush.End, cancellationToken).ConfigureAwait(false);
 
         var recorded = await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionCompleted, correlationId,
             async () => CompletedPayload(scope, Provider, accumulator.ResolveModel(request.Model), accumulator.Usage, await OffloadTextAsync(scope, accumulator.Text, CancellationToken.None).ConfigureAwait(false)), CancellationToken.None).ConfigureAwait(false);
         if (recorded && declared) await MarkCapturePresentAsync(scope).ConfigureAwait(false);
     }
 
-    private async Task RecordPendingDeltaAsync(LlmCallScope scope, Guid correlationId, StringBuilder pending, int ordinal, CancellationToken cancellationToken)
+    private async Task RecordPendingDeltaAsync(LlmCallScope scope, Guid correlationId, PendingDelta pending, DeltaFlush flush, CancellationToken cancellationToken)
     {
-        var fragment = pending.ToString();
-        pending.Clear();
+        if (pending.Take(flush) is not { } fragment) return;
+
+        var text = Observe(scope, fragment.Text);
 
         await SafeRecordAsync(scope, WorkflowRunRecordTypes.InteractionDelta, correlationId,
-            async () => DeltaPayload(scope, Provider, ordinal, await OffloadTextAsync(scope, fragment, cancellationToken).ConfigureAwait(false)), cancellationToken).ConfigureAwait(false);
+            async () => DeltaPayload(scope, Provider, fragment.Ordinal, await OffloadTextAsync(scope, text, cancellationToken).ConfigureAwait(false)), cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>What a flush may release of the held text.</summary>
+    private enum DeltaFlush
+    {
+        /// <summary>Mid-stream: only the settled text; the rest is carried into the next flush.</summary>
+        Settled,
+
+        /// <summary>The stream ended: everything, verbatim — a held tail nothing followed never became a secret.</summary>
+        End,
+
+        /// <summary>
+        /// The stream was cut — a cancel or a fault: everything, with a held tail that could start a secret masked. The
+        /// carry holds exactly that text, the rest of it was never streamed, and exact matching never sees it whole.
+        /// </summary>
+        Cut,
+    }
+
+    /// <summary>
+    /// The text streamed since the last delta row, and the ordinal the next row takes. Each row is redacted on its own, so
+    /// a flush mid-stream releases only the SETTLED text — none that a secret could still run across — and carries the
+    /// rest into the next flush; otherwise a secret the model prints across a flush (by size or by the one-second
+    /// cadence) is stored as two halves no exact match recognises, and the live view concatenates them back. The end
+    /// of the stream releases everything, so the rows still concatenate to exactly the streamed text, masked; a cut
+    /// releases everything too, but masks the end it was holding back.
+    /// </summary>
+    private sealed class PendingDelta(PersistenceSecretRedactor boundary)
+    {
+        private readonly StringBuilder _text = new();
+        private int _ordinal;
+
+        /// <summary>UTF-8 size of the held text — the size bound is in bytes, not UTF-16 chars.</summary>
+        public int Bytes { get; private set; }
+
+        public bool IsEmpty => _text.Length == 0;
+
+        public void Append(string text)
+        {
+            _text.Append(text);
+            Bytes += Encoding.UTF8.GetByteCount(text);
+        }
+
+        /// <summary>The text one row may carry and that row's ordinal, or null when nothing is settled yet.</summary>
+        public (PersistenceRedaction<string?> Text, int Ordinal)? Take(DeltaFlush flush)
+        {
+            var text = _text.ToString();
+            var length = flush == DeltaFlush.Settled ? boundary.SettledLength(text) : text.Length;
+
+            if (length == 0) return null;
+
+            _text.Remove(0, length);
+            Bytes = Encoding.UTF8.GetByteCount(text.AsSpan(length));
+
+            var released = flush == DeltaFlush.Cut ? boundary.RedactCut(text) : new PersistenceRedaction<string?>(text[..length], false);
+
+            return (released, _ordinal++);
+        }
     }
 
     /// <summary>The <c>interaction.delta</c> payload: the ordinal (monotonic within the call) + the coalesced text fragment (inline-or-<c>$artifact_id</c>).</summary>