diff --git a/doc/AGENT-IDENTITY.md b/doc/AGENT-IDENTITY.md index 5bff3917ddd..bc4d081543f 100644 --- a/doc/AGENT-IDENTITY.md +++ b/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. diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index 9259ec22301..1201c84a4c7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/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); +struct IdentityOutputBuffer { + pending: BTreeMap, + streams: BTreeMap, +} + +#[derive(Clone, PartialEq)] +struct IdentityOutputStream { + scope: String, + item_id: Option, + payload: Value, + paths: Vec, +} 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 = 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 = 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::*; diff --git a/server/src/__tests__/agent-identity.test.ts b/server/src/__tests__/agent-identity.test.ts index 289bd5641c7..97dbdb5daf6 100644 --- a/server/src/__tests__/agent-identity.test.ts +++ b/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); diff --git a/server/src/services/agent-identity-redaction.ts b/server/src/services/agent-identity-redaction.ts index bbc89af5d41..6bcd1e7d32a 100644 --- a/server/src/services/agent-identity-redaction.ts +++ b/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(); + const deltaStreams = new Map; paths: string[]; + }>(); const redactor = { values, redact(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(stream: string, value: T): T { + delta(stream: string, value: T, itemId?: string): T { + const scope = stream; + const paths: string[] = []; const payload = value as Record; // 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, wholeTurn: boolean) { + const tails: Array<{ itemId?: string; payload: Record }> = []; + 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; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index cb5d0ef8bb1..88a0da0c2a7 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/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 }> = []; + 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); + 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 }> = []; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6a2bfd0a708..0fb5a5ab50b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -7446,6 +7446,10 @@ export async function executePaperclipNativeSession(input: { async function executePaperclipNativeSessionWithinScope( input: Parameters[0], ): Promise { + 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) diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index dfc98969a66..81bcbfdad86 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/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]) { diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 344c340f970..e41fd522c8e 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/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 }); diff --git a/ui/src/components/transcript/native-run-events.test.ts b/ui/src/components/transcript/native-run-events.test.ts index 7fba1ff24ca..7fb01b5b371 100644 --- a/ui/src/components/transcript/native-run-events.test.ts +++ b/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.")), diff --git a/ui/src/components/transcript/native-run-events.ts b/ui/src/components/transcript/native-run-events.ts index 954a3ce7856..85f5426a16a 100644 --- a/ui/src/components/transcript/native-run-events.ts +++ b/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",