fix: bound identity redaction state and isolate interleaved output

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

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"], "");

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<T>(stream: string, value: T): T {
+      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)

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***");
     }

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<typeof createAgentIdentityRedactor>;
-  readonly #redactedDeltas = new Map<string, { event: PrpEvent; originalSha: string }>();
+  readonly #redactedDeltas = new Map<string, { event: PrpEvent; originalSha: string; bytes: number }>();
+  #redactedDeltaBytes = 0;
   readonly #binding: PaperclipControlPlaneBinding;
   #sessionId: string | null = null;
   readonly #onCommittedEvent?: (event: PrpEvent) => Promise<void>;
@@ -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, {