fix: redact native failures and settle buffered identity output

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/AGENT-IDENTITY.md

@@ -94,3 +94,5 @@ Ordinary disaster-recovery backups include encrypted identities. Recovery needs
 both the database and the matching wrapping key; back up the key separately.
 Do not use a disaster-recovery backup as a development clone when agents should
 have distinct identities.
+
+Native failure records and reports redact the assigned key before truncating diagnostics. Streaming output buffers settle at item or turn completion: short structural prefixes (such as a trailing dash) are preserved, longer interrupted key fragments become redaction markers, and pending buffers are cleared. Terminal events carry these settled `outputTails`; transcript projection displays them as deltas without changing the source event receipt.

packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs

@@ -534,7 +534,29 @@ impl DurableState {
         // Pending secret fragments are memory-only and never enter durable state.
         let mut identity_output = self.identity_output.clone();
         let payload = if event_type == "item.delta" {
-            identity_output.redact(&format!("{}:{}", self.turn_id, self.item_id), &payload)
+            identity_output.redact(
+                &format!("{}:{}", self.turn_id, self.item_id),
+                &payload,
+                &self.item_id,
+            )
+        } else if matches!(
+            event_type.as_str(),
+            "item.completed" | "item.failed" | "item.cancelled"
+        ) {
+            identity_output.settle(
+                &format!("{}:{}", self.turn_id, self.item_id),
+                &payload,
+                false,
+            )
+        } else if matches!(
+            event_type.as_str(),
+            "turn.completed"
+                | "turn.failed"
+                | "turn.cancelled"
+                | "turn.interrupted"
+                | "session.closed"
+        ) {
+            identity_output.settle(&format!("{}:", self.turn_id), &payload, true)
         } else {
             payload
         };
@@ -1640,7 +1662,18 @@ fn durable_semantics_changed_by_sanitization(original: &Value, sanitized: &Value
 }
 
 #[derive(Clone, Default, PartialEq)]
-struct IdentityOutputBuffer(BTreeMap<String, String>);
+struct IdentityOutputBuffer {
+    pending: BTreeMap<String, String>,
+    streams: BTreeMap<String, IdentityOutputStream>,
+}
+
+#[derive(Clone, PartialEq)]
+struct IdentityOutputStream {
+    scope: String,
+    item_id: Option<Value>,
+    payload: Value,
+    paths: Vec<String>,
+}
 
 impl std::fmt::Debug for IdentityOutputBuffer {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -1649,25 +1682,34 @@ impl std::fmt::Debug for IdentityOutputBuffer {
 }
 
 impl IdentityOutputBuffer {
-    fn redact(&mut self, stream: &str, payload: &Value) -> Value {
+    fn redact(&mut self, stream: &str, payload: &Value, item_id: &str) -> Value {
         let Ok(key) = std::env::var("PAPERCLIP_AGENT_PRIVATE_KEY") else {
             return payload.clone();
         };
         if key.is_empty() {
             return payload.clone();
         }
-        self.redact_with_key(stream, payload, &key)
+        let result = self.redact_with_key(stream, payload, &key);
+        if let Some(descriptor) = self.streams.get_mut(&Self::stream_key(stream, payload)) {
+            descriptor.item_id = Some(json!(item_id));
+        }
+        result
     }
 
-    fn redact_with_key(&mut self, stream: &str, payload: &Value, key: &str) -> Value {
-        let stream = json!([
+    fn stream_key(stream: &str, payload: &Value) -> String {
+        json!([
             stream,
             payload.get("itemId"),
             payload.get("providerItemId"),
             payload.get("channel"),
             payload.get("stream")
         ])
-        .to_string();
+        .to_string()
+    }
+
+    fn redact_with_key(&mut self, stream: &str, payload: &Value, key: &str) -> Value {
+        let scope = stream.to_owned();
+        let stream = Self::stream_key(stream, payload);
         let mut values = vec![key.to_owned(), key.trim().to_owned()];
         values.extend(
             key.lines()
@@ -1675,7 +1717,93 @@ impl IdentityOutputBuffer {
                 .map(str::to_owned),
         );
         values.sort_by_key(|value| std::cmp::Reverse(value.len()));
-        self.visit(&stream, "", payload, &values)
+        let result = self.visit(&stream, "", payload, &values);
+        let prefix = format!("{stream}.");
+        let paths: Vec<String> = self
+            .pending
+            .keys()
+            .filter(|path| path.starts_with(&prefix))
+            .cloned()
+            .collect();
+        if paths.is_empty() {
+            self.streams.remove(&stream);
+        } else {
+            let metadata = ["itemId", "providerItemId", "kind", "channel", "stream"]
+                .iter()
+                .filter_map(|key| {
+                    payload
+                        .get(*key)
+                        .filter(|value| value.is_string())
+                        .map(|value| ((*key).to_owned(), sanitize_value(value)))
+                })
+                .collect();
+            self.streams.insert(
+                stream,
+                IdentityOutputStream {
+                    scope,
+                    item_id: None,
+                    payload: Value::Object(metadata),
+                    paths,
+                },
+            );
+        }
+        result
+    }
+
+    fn settle(&mut self, scope: &str, payload: &Value, whole_turn: bool) -> Value {
+        let streams: Vec<String> = self
+            .streams
+            .iter()
+            .filter(|(_, descriptor)| {
+                (if whole_turn {
+                    descriptor.scope.starts_with(scope)
+                } else {
+                    descriptor.scope == scope
+                }) && (whole_turn
+                    || payload.get("itemId").is_none()
+                    || descriptor.payload.get("itemId").is_none()
+                    || payload.get("itemId") == descriptor.payload.get("itemId"))
+            })
+            .map(|(stream, _)| stream.clone())
+            .collect();
+        let mut tails = Vec::new();
+        for stream in streams {
+            let descriptor = self.streams.remove(&stream).unwrap();
+            let primary = descriptor
+                .paths
+                .iter()
+                .find(|path| **path == format!("{stream}.text"))
+                .or_else(|| descriptor.paths.first())
+                .cloned();
+            let mut text = String::new();
+            for path in descriptor.paths {
+                if let Some(held) = self.pending.remove(&path) {
+                    if Some(&path) == primary.as_ref() {
+                        text = if held.len() < 8 {
+                            held
+                        } else {
+                            "[REDACTED]".to_owned()
+                        };
+                    }
+                }
+            }
+            if !text.is_empty() {
+                let mut tail_payload = descriptor.payload;
+                tail_payload["text"] = json!(text);
+                tails.push(json!({"itemId": descriptor.item_id, "payload": tail_payload}));
+            }
+        }
+        let mut result = payload.clone();
+        if !tails.is_empty() {
+            let mut existing = result
+                .get("outputTails")
+                .and_then(Value::as_array)
+                .cloned()
+                .unwrap_or_default();
+            existing.extend(tails);
+            result["outputTails"] = json!(existing);
+        }
+        result
     }
 
     fn visit(&mut self, path: &str, field: &str, value: &Value, secrets: &[String]) -> Value {
@@ -1701,7 +1829,7 @@ impl IdentityOutputBuffer {
                     .collect(),
             ),
             Value::String(chunk) if matches!(field, "text" | "delta" | "output" | "patch") => {
-                let mut text = self.0.remove(path).unwrap_or_default();
+                let mut text = self.pending.remove(path).unwrap_or_default();
                 text.push_str(chunk);
                 for secret in secrets {
                     text = text.replace(secret, "[REDACTED]");
@@ -1717,7 +1845,7 @@ impl IdentityOutputBuffer {
                 }
                 if held > 0 {
                     let tail = text.split_off(text.len() - held);
-                    self.0.insert(path.to_owned(), tail);
+                    self.pending.insert(path.to_owned(), tail);
                 }
                 Value::String(text)
             }
@@ -2518,6 +2646,29 @@ mod tests {
         assert!(!format!("{state:?}").contains(&body[..20]));
     }
 
+    #[test]
+    fn identity_output_settles_item_and_turn_tails_without_retaining_fragments() {
+        let key = "-----BEGIN PRIVATE KEY-----\nMC4CAQAwBQYDK2VwBCIEITestSecretMaterialForAnAgentIdentity\n-----END PRIVATE KEY-----\n";
+        let mut output = IdentityOutputBuffer::default();
+        assert_eq!(
+            output.redact_with_key(
+                "turn:item",
+                &json!({"itemId": "one", "kind": "reasoning", "text": "Thinking-"}),
+                key
+            )["text"],
+            "Thinking"
+        );
+        let settled = output.settle("turn:item", &json!({"itemId": "one"}), false);
+        assert_eq!(settled["outputTails"][0]["payload"]["text"], "-");
+        assert!(output.pending.is_empty());
+        assert!(output.streams.is_empty());
+        output.redact_with_key("turn:other", &json!({"text": &key[..42]}), key);
+        let settled = output.settle("turn:", &json!({"status": "completed"}), true);
+        assert_eq!(settled["outputTails"][0]["payload"]["text"], "[REDACTED]");
+        assert!(output.pending.is_empty());
+        assert!(output.streams.is_empty());
+    }
+
     use std::time::Duration;
 
     use super::*;

server/src/__tests__/agent-identity.test.ts

@@ -207,6 +207,17 @@ describe("identity output redaction", () => {
       expect(redactor.redact({ output: text }).output).not.toContain(pem.split("\n")[1]);
     }
   });
+  it("settles ordinary endings and interrupted keys on item and turn completion", () => {
+    const redactor = createAgentIdentityRedactor(pem);
+    expect(redactor.delta("run:turn:item", { kind: "reasoning", text: "Thinking-" }, "item").text).toBe("Thinking");
+    const completed = redactor.settleDeltas("run:turn:item", {}, false);
+    expect(completed).toMatchObject({ outputTails: [{ itemId: "item", payload: { kind: "reasoning", text: "-" } }] });
+    expect(redactor.settleDeltas("run:turn:", {}, true)).toEqual({});
+    redactor.delta("run:turn:other", { text: pem.slice(0, 42) }, "other");
+    expect(redactor.settleDeltas("run:turn:", {}, true)).toMatchObject({ outputTails: [{ payload: { text: "***REDACTED***" } }] });
+    expect(redactor.delta("run:turn:other", { text: "hello" }).text).toBe("hello");
+  });
+
   it("redacts every split boundary and interleaved streams, including interrupted prefixes", () => {
     for (let split = 1; split < pem.length; split++) {
       const redactor = createAgentIdentityRedactor(pem);

server/src/services/agent-identity-redaction.ts

@@ -10,6 +10,9 @@ export function createAgentIdentityRedactor(privateKeyPem?: string) {
     ...privateKeyPem.split(/\r?\n/).filter(line => line && !line.startsWith("-----")),
   ])].sort((a, b) => b.length - a.length) : [];
   const pending = new Map<string, string>();
+  const deltaStreams = new Map<string, {
+    scope: string; itemId?: string; payload: Record<string, unknown>; paths: string[];
+  }>();
   const redactor = {
     values,
     redact<T>(value: T): T { return redactRegisteredSecretValues(value, values); },
@@ -32,18 +35,24 @@ export function createAgentIdentityRedactor(privateKeyPem?: string) {
     finish(stream: string): string {
       const held = pending.get(stream);
       pending.delete(stream);
-      return held ? REDACTED_EVENT_VALUE : "";
+      // Ed25519 PEM/PKCS#8 values begin with fixed structural bytes. Preserve
+      // ordinary endings such as a dash; longer interrupted fragments stay hidden.
+      return held ? (held.length < 8 ? held : REDACTED_EVENT_VALUE) : "";
     },
     /** Delta payloads can repeat output under text and provider-specific fields. */
-    delta<T>(stream: string, value: T): T {
+    delta<T>(stream: string, value: T, itemId?: string): T {
+      const scope = stream;
+      const paths: string[] = [];
       const payload = value as Record<string, unknown>;
       // Durable transports can share one envelope item ID across provider items.
       stream = JSON.stringify([stream, payload.itemId, payload.providerItemId, payload.channel, payload.stream]);
       const visit = (entry: unknown, path: string, field: string): unknown => {
         if (typeof entry === "string") {
-          return /^(text|delta|output|patch)$/.test(field)
-            ? redactor.chunk(`${stream}:${path}`, entry)
-            : redactor.redact(entry);
+          if (!/^(text|delta|output|patch)$/.test(field)) return redactor.redact(entry);
+          const key = `${stream}:${path}`;
+          const output = redactor.chunk(key, entry);
+          if (pending.has(key)) paths.push(key);
+          return output;
         }
         if (Array.isArray(entry)) return entry.map((child, index) => visit(child, `${path}.${index}`, field));
         if (entry && typeof entry === "object") return Object.fromEntries(
@@ -51,7 +60,35 @@ export function createAgentIdentityRedactor(privateKeyPem?: string) {
         );
         return entry;
       };
-      return visit(value, "", "") as T;
+      const result = visit(value, "", "") as T;
+      for (const previous of deltaStreams.get(stream)?.paths ?? []) {
+        if (pending.has(previous) && !paths.includes(previous)) paths.push(previous);
+      }
+      if (paths.length) deltaStreams.set(stream, { scope, itemId, paths,
+        payload: Object.fromEntries(["itemId", "providerItemId", "kind", "channel", "stream"]
+          .filter(key => typeof payload[key] === "string").map(key => [key, redactor.redact(payload[key])])),
+      });
+      else deltaStreams.delete(stream);
+      return result;
+    },
+    /** Settle output in the terminal event, preserving its source receipt identity. */
+    settleDeltas(scope: string, payload: Record<string, unknown>, wholeTurn: boolean) {
+      const tails: Array<{ itemId?: string; payload: Record<string, unknown> }> = [];
+      for (const [stream, descriptor] of deltaStreams) {
+        if (wholeTurn ? !descriptor.scope.startsWith(scope) : descriptor.scope !== scope) continue;
+        if (!wholeTurn && payload.itemId && descriptor.payload.itemId && payload.itemId !== descriptor.payload.itemId) continue;
+        const primary = descriptor.paths.find(path => path === `${stream}:.text`) ?? descriptor.paths[0];
+        let text = "";
+        for (const path of descriptor.paths) {
+          const tail = redactor.finish(path);
+          if (path === primary) text = tail;
+        }
+        deltaStreams.delete(stream);
+        if (text) tails.push({ itemId: descriptor.itemId, payload: { ...descriptor.payload, text } });
+      }
+      return tails.length ? { ...payload, outputTails: [
+        ...(Array.isArray(payload.outputTails) ? payload.outputTails : []), ...tails,
+      ] } : payload;
     },
   };
   return redactor;

server/src/services/native-runtime/native-session-executor.test.ts

@@ -18,6 +18,7 @@ import {
 } from "node:fs/promises";
 import { execFileSync } from "node:child_process";
 import { tmpdir } from "node:os";
+import { inspect } from "node:util";
 import { join } from "node:path";
 import {
   heartbeatRuns,
@@ -39,7 +40,7 @@ import {
   type NativeExecutionInput,
   type PrpEvent,
 } from "@paperclipai/paperclip-runner";
-import { createHash } from "node:crypto";
+import { createHash, generateKeyPairSync } from "node:crypto";
 import { DatabaseSync } from "node:sqlite";
 import { nativeSha256 } from "./canonical.js";
 import * as noLaunchProofModule from "./native-maintenance-no-launch.js";
@@ -7611,6 +7612,34 @@ describe("native session bounded recovery", () => {
     });
   });
 
+  it.each(["full", "truncated"])("redacts %s identity from native failure writes, logs, and rethrown errors", async (mode) => {
+    const privateKeyPem = generateKeyPairSync("ed25519").privateKey.export({ type: "pkcs8", format: "pem" }).toString();
+    const keyBody = privateKeyPem.split("\n")[1];
+    const failure = new Error(`native provider failed: ${mode === "full" ? privateKeyPem : privateKeyPem.slice(0, 64)}`);
+    void failure.stack; // Also cover stacks materialized before the catch boundary.
+    const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
+    const logs: string[] = [];
+    state.execute.mockReset().mockRejectedValueOnce(failure);
+    state.upsertRecoveryAction.mockReset().mockResolvedValue({});
+    const service = vi.spyOn(issueServiceModule, "issueService").mockReturnValue({
+      update: vi.fn(async () => ({ status: "blocked", statusVersion: 1 })),
+    } as unknown as ReturnType<typeof issueServiceModule.issueService>);
+    try {
+      await expect(executePaperclipNativeSession({
+        db: leaseDb(execution, {}, {}, updates), execution, runnerInstanceId: "runner",
+        runnerEnvironment: { PAPERCLIP_AGENT_PRIVATE_KEY: privateKeyPem },
+        onLog: async (_stream, text) => { logs.push(text); },
+      })).rejects.toBe(failure);
+      expect(updates.some(entry => entry.table === nativeRunFinalizations && entry.values.failureDetail)).toBe(true);
+      for (const output of [inspect(updates.map(entry => entry.values), { depth: null }), logs.join(""), failure.message, failure.stack!]) {
+        expect(output).not.toContain(keyBody.slice(0, 24));
+        expect(output).toContain("***REDACTED***");
+      }
+    } finally {
+      service.mockRestore();
+    }
+  });
+
   it("persists actionable operator recovery without an automatic cleanup wake", async () => {
     const updates: Array<{ table: unknown; values: Record<string, unknown> }> =
       [];

server/src/services/native-runtime/native-session-executor.ts

@@ -7446,6 +7446,10 @@ export async function executePaperclipNativeSession(input: {
 async function executePaperclipNativeSessionWithinScope(
   input: Parameters<typeof executePaperclipNativeSession>[0],
 ): Promise<AdapterExecutionResult> {
+  const identityRedactor = createAgentIdentityRedactor(input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY);
+  const redactIdentityText = (text: string) => redactSensitiveText(
+    identityRedactor.chunk("diagnostic", text) + identityRedactor.finish("diagnostic"),
+  );
   if (
     input.execution.provider.kind !== "codex" &&
     input.execution.provider.kind !== "opencode" &&
@@ -8424,7 +8428,7 @@ async function executePaperclipNativeSessionWithinScope(
             requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined,
             onSessionClosed: input.instructionWorkingCopy?.collectStopped,
             onCheckpoint: async (snapshot) => {
-              snapshot = createAgentIdentityRedactor(input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY).redact(snapshot);
+              snapshot = identityRedactor.redact(snapshot);
               if (warmSessionId !== null && warmConfigDigest !== null) {
                 await persistWarmNativeCheckpoint(
                   input.execution,
@@ -8441,7 +8445,7 @@ async function executePaperclipNativeSessionWithinScope(
               }
             },
             onPostCompletionEnrichmentFailure: async ({ stage, error }) => {
-              const detail = redactSensitiveText(
+              const detail = redactIdentityText(
                 error instanceof Error ? error.message : String(error),
               ).slice(-4_096);
               await input.onLog?.(
@@ -8452,7 +8456,7 @@ async function executePaperclipNativeSessionWithinScope(
             onSessionQuarantined: async (reason) => {
               await input.onLog?.(
                 "stderr",
-                `[paperclip-runner] warm native session quarantined: ${redactSensitiveText(reason).slice(-1_000)}\n`,
+                `[paperclip-runner] warm native session quarantined: ${redactIdentityText(reason).slice(-1_000)}\n`,
               );
             },
             onContinuityBreak: async (continuity) => {
@@ -8649,6 +8653,16 @@ async function executePaperclipNativeSessionWithinScope(
     clearSteeringDeliveries(input.execution.binding.runId);
     clearNativeRuntimeRequestResolutions(input.execution.binding.runId);
   } catch (error) {
+    // Preserve typed failure semantics while removing credentials before any
+    // persistence, reporting, logging, or rethrow at this boundary.
+    if (error instanceof Error) {
+      const originalMessage = error.message;
+      error.message = redactIdentityText(originalMessage);
+      if (error.stack) error.stack = redactIdentityText(originalMessage
+        ? error.stack.split(originalMessage).join(error.message) : error.stack);
+    } else {
+      error = identityRedactor.redact(error);
+    }
     if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) {
       await leaseRenewal.stop().catch(() => undefined);
       liveQuestions.close();
@@ -8687,7 +8701,7 @@ async function executePaperclipNativeSessionWithinScope(
         );
 
       const failedAtMs = Date.now();
-      const executionFailureMessage = redactSensitiveText(
+      const executionFailureMessage = redactIdentityText(
         error instanceof Error ? error.message : String(error),
       ).slice(-4_096);
       if (!taskSettleScope) {
@@ -8798,11 +8812,10 @@ async function executePaperclipNativeSessionWithinScope(
       const { exhausted } = recoveryProjection;
       const integrityFailure =
         sourceFailureCode === "native_event_replay_conflict";
-      const message =
-        error instanceof Error
-          ? error.message.slice(0, 2_000)
-          : String(error).slice(0, 2_000);
-      const sanitizedStderrTail = redactSensitiveText(message).slice(-4_096);
+      const message = redactIdentityText(
+        error instanceof Error ? error.message : String(error),
+      ).slice(0, 2_000);
+      const sanitizedStderrTail = redactIdentityText(message).slice(-4_096);
       // Set inside the transaction only when the write below genuinely
       // transitions the run into "failed". Read after the transaction
       // commits, so a rolled-back write never reports a false failure.
@@ -8833,7 +8846,7 @@ async function executePaperclipNativeSessionWithinScope(
             recoveryState:
               phase === "retryable_failure" ? "resuming_session" : "blocked",
             failureCode,
-            failureDetail: {
+            failureDetail: identityRedactor.redact({
               message,
               originalFailureCode:
                 error instanceof NativeProviderTerminalFailure
@@ -8870,7 +8883,7 @@ async function executePaperclipNativeSessionWithinScope(
                                   "bootstrap_retry"
                                 ? "Retry provider bootstrap on this same run; durable evidence proves no provider session or provider event was created."
                                 : "Resume this same run from its exact persisted native provider checkpoint after the retry delay.",
-            },
+            }),
             nextAttemptAt,
             recoveryHistory: sql`(
               select coalesce(jsonb_agg(item order by ordinal), '[]'::jsonb)

server/src/services/native-runtime/paperclip-control-plane-port.test.ts

@@ -353,6 +353,23 @@ describe("PaperclipControlPlanePort conformance", () => {
         });
       }
     }
+    await port.appendEvent({
+      schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq: sourceSeq++,
+      sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId,
+      normalizedSessionId: identity.sessionId, turnId: "tail-turn", itemId: "reasoning-item",
+      eventType: "item.delta", schemaVersion: 1, priority: 1, emittedAt: new Date().toISOString(),
+      payload: { kind: "reasoning", text: "Thinking-" },
+    });
+    const terminal: PrpEvent = {
+      schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq: sourceSeq++,
+      sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId,
+      normalizedSessionId: identity.sessionId, turnId: "tail-turn",
+      eventType: "turn.completed", schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(),
+      payload: { status: "completed" },
+    };
+    await port.appendEvent(terminal);
+    expect((await port.appendEvent(terminal)).disposition).toBe("duplicate");
+    expect(observed.at(-1)?.payload.outputTails).toEqual([{ itemId: "reasoning-item", payload: { kind: "reasoning", text: "-" } }]);
     const persistedEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId));
     const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
     for (const value of [persistedEvents, observed, run.runnerProfileJson]) {

server/src/services/native-runtime/paperclip-control-plane-port.ts

@@ -204,7 +204,9 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
     if (event.sourceInstanceId !== expectedSourceInstanceId) {
       throw new Error("native_event_source_binding_mismatch");
     }
-    if (event.eventType === "item.delta" && this.#identityRedactor.values.length > 0) {
+    const terminalItem = /^item\.(completed|failed|cancelled)$/.test(event.eventType);
+    const terminalTurn = /^(turn|session)\.(completed|failed|cancelled|interrupted|closed)$/.test(event.eventType);
+    if ((event.eventType === "item.delta" || terminalItem || terminalTurn) && this.#identityRedactor.values.length > 0) {
       const key = `${event.sourceInstanceId}:${event.sourceEventId}`;
       const cached = this.#redactedDeltas.get(key);
       const originalSha = nativeSha256(validated.event);
@@ -212,10 +214,14 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
         if (cached.originalSha !== originalSha) throw new Error("native_event_source_payload_conflict");
         event = cached.event;
       } else {
-        event = { ...event, payload: this.#identityRedactor.delta(
-          `${event.sourceInstanceId}:${event.turnId}:${event.itemId}`,
-          validated.event.payload,
-        ) };
+        const stream = `${event.sourceInstanceId}:${event.turnId}:${event.itemId}`;
+        event = { ...event, payload: event.eventType === "item.delta"
+          ? this.#identityRedactor.delta(stream, validated.event.payload, event.itemId)
+          : this.#identityRedactor.settleDeltas(
+            terminalTurn ? (event.eventType.startsWith("session.") ? `${event.sourceInstanceId}:` : `${event.sourceInstanceId}:${event.turnId}:`) : stream,
+            event.payload, terminalTurn,
+          ),
+        };
         // Retries must use the same redacted payload without consuming a delta twice.
         const bytes = Buffer.byteLength(JSON.stringify(event));
         this.#redactedDeltas.set(key, { event, originalSha, bytes });

ui/src/components/transcript/native-run-events.test.ts

@@ -307,6 +307,16 @@ describe("nativeRunEventsToTranscript", () => {
     expect(transcript.filter((entry) => entry.kind === "run_terminal")).toHaveLength(1);
   });
 
+  it("preserves buffered output tails when a native turn ends", () => {
+    const transcript = nativeRunEventsToTranscript([
+      event(1, "item.delta", { itemId: "reasoning-1", kind: "reasoning", text: "Thinking" }),
+      event(2, "turn.completed", { status: "completed", outputTails: [
+        { itemId: "reasoning-1", payload: { kind: "reasoning", text: "-" } },
+      ] }),
+    ]);
+    expect(transcript.filter(entry => entry.kind === "thinking").map(entry => "text" in entry ? entry.text : "").join("")).toBe("Thinking-");
+  });
+
   it("projects provider-neutral messages, tools, usage, and the final reply", () => {
     const transcript = nativeRunEventsToTranscript([
       event(6, "run.result.proposed", runResult("Done safely.")),

ui/src/components/transcript/native-run-events.ts

@@ -612,6 +612,24 @@ export function nativeRunEventsToTranscript(events: readonly HeartbeatRunEvent[]
       if (seenSourceEventIds.has(sourceEventId)) return false;
       seenSourceEventIds.add(sourceEventId);
       return true;
+    }).flatMap((event) => {
+      const envelope = record(event.payload?.prpEvent);
+      const payload = record(envelope?.payload);
+      if (!envelope || envelope.eventType !== event.eventType
+        || !/^(item|turn|session)\.(completed|failed|cancelled|interrupted|closed)$/.test(event.eventType)
+        || !Array.isArray(payload?.outputTails)) return [event];
+      // Redaction can hold a possible key prefix until the item/turn ends.
+      // Project its settled tail as display-only deltas before the terminal.
+      const tails = payload.outputTails.flatMap((value, index): HeartbeatRunEvent[] => {
+        const tail = record(value);
+        const tailPayload = record(tail?.payload);
+        if (!tailPayload || typeof tailPayload.text !== "string") return [];
+        return [{ ...event, eventType: "item.delta", payload: { prpEvent: {
+          ...envelope, eventType: "item.delta", itemId: tail?.itemId,
+          sourceEventId: `${envelope.sourceEventId}:output-tail:${index}`, payload: tailPayload,
+        } } }];
+      });
+      return [...tails, event];
     });
   const hasAcceptedResult = orderedEvents.some(
     (event) => event.eventType === "run.result.accepted",