From 1c4687baf397328793a6df7572fce29ba97e5124 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 6 Oct 2026 12:02:18 -0500 Subject: [PATCH] fix: bound identity redaction state and isolate interleaved output Co-Authored-By: Paperclip --- .../crates/runner-core/src/durable/state.rs | 25 +++++++++++++------ .../src/services/agent-identity-redaction.ts | 6 ++++- .../paperclip-control-plane-port.test.ts | 10 +++++--- .../paperclip-control-plane-port.ts | 19 +++++++++++--- 4 files changed, 46 insertions(+), 14 deletions(-) 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 b2559876c41..9259ec22301 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 @@ -1660,6 +1660,14 @@ impl IdentityOutputBuffer { } fn redact_with_key(&mut self, stream: &str, payload: &Value, key: &str) -> Value { + let stream = json!([ + stream, + payload.get("itemId"), + payload.get("providerItemId"), + payload.get("channel"), + payload.get("stream") + ]) + .to_string(); let mut values = vec![key.to_owned(), key.trim().to_owned()]; values.extend( key.lines() @@ -1667,7 +1675,7 @@ impl IdentityOutputBuffer { .map(str::to_owned), ); values.sort_by_key(|value| std::cmp::Reverse(value.len())); - self.visit(stream, "", payload, &values) + self.visit(&stream, "", payload, &values) } fn visit(&mut self, path: &str, field: &str, value: &Value, secrets: &[String]) -> Value { @@ -2483,16 +2491,19 @@ mod tests { for split in 1..body.len() { let mut buffer = IdentityOutputBuffer::default(); let first = buffer.redact_with_key( - "item-1", - &json!({"text": &body[..split], "update": {"delta": &body[..split]}}), + "envelope-item", + &json!({"itemId": "item-1", "text": &body[..split], "update": {"delta": &body[..split]}}), + key, + ); + let other = buffer.redact_with_key( + "envelope-item", + &json!({"itemId": "item-2", "text": "ordinary output\n"}), key, ); - let other = - buffer.redact_with_key("item-2", &json!({"text": "ordinary output\n"}), key); assert_eq!(other["text"], "ordinary output\n"); let second = buffer.redact_with_key( - "item-1", - &json!({"text": &body[split..], "update": {"delta": &body[split..]}}), + "envelope-item", + &json!({"itemId": "item-1", "text": &body[split..], "update": {"delta": &body[split..]}}), key, ); assert_eq!(first["text"], ""); diff --git a/server/src/services/agent-identity-redaction.ts b/server/src/services/agent-identity-redaction.ts index 5c4947e37b4..bbc89af5d41 100644 --- a/server/src/services/agent-identity-redaction.ts +++ b/server/src/services/agent-identity-redaction.ts @@ -24,7 +24,8 @@ export function createAgentIdentityRedactor(privateKeyPem?: string) { if (text.endsWith(value.slice(0, size))) { held = size; break; } } } - pending.set(stream, held ? text.slice(-held) : ""); + if (held) pending.set(stream, text.slice(-held)); + else pending.delete(stream); return held ? text.slice(0, -held) : text; }, // An interrupted partial secret must not be flushed as plaintext. @@ -35,6 +36,9 @@ export function createAgentIdentityRedactor(privateKeyPem?: string) { }, /** Delta payloads can repeat output under text and provider-specific fields. */ delta(stream: string, value: T): T { + 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) 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 117e37991e2..dfc98969a66 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 @@ -337,16 +337,20 @@ describe("PaperclipControlPlanePort conformance", () => { ["halves", [keyBody.slice(0, 31), keyBody.slice(31)]], ["bytes", [...keyBody]], ] as const) { - for (const chunk of chunks) { + for (const [index, chunk] of chunks.entries()) { const event: PrpEvent = { schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq: sourceSeq++, sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "identity-turn", itemId, eventType: "item.delta", schemaVersion: 1, priority: 1, emittedAt: new Date().toISOString(), - payload: { kind: "commandExecution", text: chunk, update: { delta: chunk } }, + payload: { itemId, kind: "commandExecution", text: chunk, update: { delta: chunk } }, }; await port.appendEvent(event); expect((await port.appendEvent(event)).disposition).toBe("duplicate"); + if (index === 0) await port.appendEvent({ + ...event, sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq: sourceSeq++, + payload: { itemId: "another-provider-item", kind: "commandExecution", text: "Unrelated output\n", update: { delta: "Unrelated output\n" } }, + }); } } const persistedEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)); @@ -356,7 +360,7 @@ describe("PaperclipControlPlanePort conformance", () => { expect(JSON.stringify(value)).toContain("***REDACTED***"); } for (const itemId of ["halves", "bytes"]) { - const deltas = observed.filter(event => event.itemId === itemId).map(event => event.payload); + const deltas = observed.filter(event => event.payload.itemId === itemId).map(event => event.payload); expect(deltas.map(delta => delta.text).join("")).toBe("***REDACTED***"); expect(deltas.map(delta => (delta.update as { delta: string }).delta).join("")).toBe("***REDACTED***"); } 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 d92b4c973a2..344c340f970 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -40,6 +40,9 @@ export interface PaperclipControlPlaneBinding { controlPlaneSourceInstanceId: string; } +const IDENTITY_DELTA_RETRY_WINDOW = 1_024; +const IDENTITY_DELTA_RETRY_BYTES = 4 * 1024 * 1024; + function isPrpEvent(value: NativeRunEvent | PrpEvent): value is PrpEvent { return "schema" in value && [ "paperclip.prp.event.v1", @@ -73,7 +76,8 @@ function assertTerminal(value: unknown): asserts value is PrpTerminalState { export class PaperclipControlPlanePort implements ControlPlanePort { readonly #db: Db; readonly #identityRedactor: ReturnType; - readonly #redactedDeltas = new Map(); + readonly #redactedDeltas = new Map(); + #redactedDeltaBytes = 0; readonly #binding: PaperclipControlPlaneBinding; #sessionId: string | null = null; readonly #onCommittedEvent?: (event: PrpEvent) => Promise; @@ -200,7 +204,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (event.sourceInstanceId !== expectedSourceInstanceId) { throw new Error("native_event_source_binding_mismatch"); } - if (event.eventType === "item.delta") { + if (event.eventType === "item.delta" && this.#identityRedactor.values.length > 0) { const key = `${event.sourceInstanceId}:${event.sourceEventId}`; const cached = this.#redactedDeltas.get(key); const originalSha = nativeSha256(validated.event); @@ -213,7 +217,16 @@ export class PaperclipControlPlanePort implements ControlPlanePort { validated.event.payload, ) }; // Retries must use the same redacted payload without consuming a delta twice. - this.#redactedDeltas.set(key, { event, originalSha }); + const bytes = Buffer.byteLength(JSON.stringify(event)); + this.#redactedDeltas.set(key, { event, originalSha, bytes }); + this.#redactedDeltaBytes += bytes; + // Match the bounded runner transcript retry window; large tool output + // also has a byte cap so long runs cannot retain another full transcript. + while (this.#redactedDeltas.size > IDENTITY_DELTA_RETRY_WINDOW || this.#redactedDeltaBytes > IDENTITY_DELTA_RETRY_BYTES) { + const oldest = this.#redactedDeltas.keys().next().value!; + this.#redactedDeltaBytes -= this.#redactedDeltas.get(oldest)!.bytes; + this.#redactedDeltas.delete(oldest); + } } } const persisted = await appendHeartbeatRunEvent(this.#db, {