From 1f9976e8b9a0a6aac60b6c47588b8ee0efdba2b1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 6 Oct 2026 11:39:27 -0500 Subject: [PATCH] fix: redact native identity fragments and preserve run credentials Co-Authored-By: Paperclip --- doc/AGENT-IDENTITY.md | 3 + .../src/acpx-engine/execute.test.ts | 15 +++ .../adapter-utils/src/acpx-engine/execute.ts | 2 +- .../codex-local/src/server/execute.test.ts | 21 ++- .../codex-local/src/server/execute.ts | 6 +- .../crates/runner-core/src/codex_provider.rs | 9 +- .../crates/runner-core/src/durable/mod.rs | 5 +- .../crates/runner-core/src/durable/state.rs | 124 ++++++++++++++++++ .../provider-trace-file-sink.test.ts | 18 +++ .../src/contracts/provider-trace-file-sink.ts | 5 + .../src/drivers/acpx/runtime-sandbox.ts | 2 +- .../src/services/agent-identity-redaction.ts | 19 ++- .../paperclip-control-plane-port.test.ts | 22 ++++ .../paperclip-control-plane-port.ts | 19 ++- 14 files changed, 258 insertions(+), 12 deletions(-) diff --git a/doc/AGENT-IDENTITY.md b/doc/AGENT-IDENTITY.md index cebdf9c8216..5bff3917ddd 100644 --- a/doc/AGENT-IDENTITY.md +++ b/doc/AGENT-IDENTITY.md @@ -43,6 +43,9 @@ compatibility includes the public key ID, so an older warm process is replaced before its next turn. Private values are excluded from invocation metadata and execution configuration; output and failure diagnostics redact known private material. The run-log redactor handles split stdout/stderr chunks independently. +Native output deltas also buffer potential key fragments before persistence. +Raw provider trace contents are suppressed for identity-enabled runs; trace +routing, timing, and interpretation metadata remain available. Managed remote providers, including Cursor Cloud, receive the same private key as managed local processes. Selecting a runtime therefore trusts its host with diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 4296a110e79..491eda13b17 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -172,6 +172,7 @@ async function runExecutor( runtime?: Record; executionTransport?: Record; authToken?: string; + agentIdentity?: AdapterExecutionContext["agentIdentity"]; executionTarget?: Record; runtimeMcp?: AdapterRuntimeMcpAccess; prepareRemoteManagedHome?: AcpxEngineExecutorOptions["prepareRemoteManagedHome"]; @@ -210,6 +211,7 @@ async function runExecutor( context: options.context ?? {}, executionTransport: options.executionTransport, authToken: options.authToken, + agentIdentity: options.agentIdentity, executionTarget: options.executionTarget, runtimeMcp: options.runtimeMcp, startupTraceContext: options.startupTraceContext, @@ -548,6 +550,19 @@ describe("shared ACPX engine runtime behavior", () => { ); }); + it("keeps assigned API and tool credentials in the identity-enabled Codex shell environment", async () => { + const { meta } = await runExecutor({ agent: "codex", env: { MY_SERVICE_TOKEN: "assigned-tool-token" } }, { + authToken: "assigned-run-token", + agentIdentity: { keyId: "sha256:test", publicKeyPem: "public", privateKeyPem: "private" }, + }); + const config = JSON.parse(String((meta[0]?.env as Record).CODEX_CONFIG)); + expect(config.shell_environment_policy.include_only).toEqual(expect.arrayContaining([ + "PAPERCLIP_API_KEY", "MY_SERVICE_TOKEN", "PAPERCLIP_AGENT_PRIVATE_KEY", + ])); + expect(JSON.stringify(config)).not.toContain("assigned-run-token"); + expect(JSON.stringify(config)).not.toContain("assigned-tool-token"); + }); + it("forwards arbitrary Codex model IDs verbatim without picker-dependent session config", async () => { const arbitraryModel = "gpt-999-test-does-not-exist"; const { configOptions, meta } = await runExecutor({ diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index a08e17bbea6..a777a9b3867 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -2051,7 +2051,7 @@ async function buildRuntime(input: { ...(input.ctx.agentIdentity ? { identityEnvironmentKeys: [...new Set([ "PATH", "HOME", "LANG", "TMPDIR", "CODEX_HOME", "PAPERCLIP_AGENT_KEY_ID", "PAPERCLIP_AGENT_PUBLIC_KEY", "PAPERCLIP_AGENT_PRIVATE_KEY", - ...Object.keys(env).filter(key => !/key|secret|token/i.test(key)), + ...Object.keys(env), ])] } : {}), requestedModel, requestedThinkingEffort, diff --git a/packages/adapters/codex-local/src/server/execute.test.ts b/packages/adapters/codex-local/src/server/execute.test.ts index 2c1284a9fd2..23a0b5b333c 100644 --- a/packages/adapters/codex-local/src/server/execute.test.ts +++ b/packages/adapters/codex-local/src/server/execute.test.ts @@ -117,7 +117,7 @@ describe("codex execute — outbound auth copy-back restore contribution", () => ); } - async function runTeardown(input: { sandboxAuth: string; hostAuth: string; onProviderStopped?: () => Promise }) { + async function runTeardown(input: { sandboxAuth: string; hostAuth: string; onProviderStopped?: () => Promise; withIdentity?: boolean }) { const rootDir = await mkdtemp( path.join(os.tmpdir(), "paperclip-codex-copyback-e2e-"), ); @@ -136,8 +136,14 @@ describe("codex execute — outbound auth copy-back restore contribution", () => process.env.CODEX_HOME = sharedHostHome; sandboxAuthFixture.bytes = Buffer.from(input.sandboxAuth, "utf8"); + const commandArgs: string[] = []; const executionResult = await execute({ runId: "run-copyback-e2e", + ...(input.withIdentity ? { + authToken: "assigned-run-token", + agentIdentity: { keyId: "sha256:test", publicKeyPem: "public", privateKeyPem: "private" }, + } : {}), + onMeta: async meta => { commandArgs.push(...(meta.commandArgs ?? [])); }, onProviderStopped: input.onProviderStopped, agent: { id: "agent-1", @@ -152,7 +158,7 @@ describe("codex execute — outbound auth copy-back restore contribution", () => engine: "cli", // External CODEX_HOME (outside the managed company tree) so no managed // seeding rewrites auth.json before teardown; equals the shared host home. - env: { CODEX_HOME: sharedHostHome }, + env: { CODEX_HOME: sharedHostHome, ...(input.withIdentity ? { MY_SERVICE_TOKEN: "assigned-tool-token" } : {}) }, }, context: { paperclipWorkspace: { @@ -176,12 +182,23 @@ describe("codex execute — outbound auth copy-back restore contribution", () => }); return { + commandArgs, finalHostAuth: await readFile(hostAuthPath, "utf8"), finalHostMode: (await lstat(hostAuthPath)).mode & 0o777, executionResult, }; } + it("keeps assigned API and tool credentials in the identity-enabled CLI shell", async () => { + const { commandArgs } = await runTeardown({ sandboxAuth: "{}", hostAuth: "{}", withIdentity: true }); + const policy = commandArgs.find(arg => arg.startsWith("shell_environment_policy.include_only=")); + expect(JSON.parse(policy!.slice(policy!.indexOf("=") + 1))).toEqual(expect.arrayContaining([ + "PAPERCLIP_API_KEY", "MY_SERVICE_TOKEN", "PAPERCLIP_AGENT_PRIVATE_KEY", + ])); + expect(commandArgs.join(" ")).not.toContain("assigned-run-token"); + expect(commandArgs.join(" ")).not.toContain("assigned-tool-token"); + }); + it("collects stopped-provider instruction edits before a throwing remote restore", async () => { const order: string[] = []; prepareAdapterExecutionTargetRuntime.mockImplementationOnce(async () => ({ diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index c85e8b3a628..e767643723f 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1230,8 +1230,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise identityNames.includes(key) || !/key|secret|token/i.test(key)); + const shellKeys = [...new Set([ + ...Object.keys(process.env).filter(key => !/key|secret|token/i.test(key)), + ...Object.keys(env), ...identityNames, + ])]; args.unshift("-c", "features.shell_snapshot=false", "-c", 'shell_environment_policy.inherit="all"', "-c", "shell_environment_policy.ignore_default_excludes=true", "-c", `shell_environment_policy.include_only=${JSON.stringify(shellKeys)}`); } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs index 09e490c992f..81c607b8c8f 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs @@ -245,7 +245,14 @@ impl ProviderTraceSink { } fn frame(&mut self, direction: &str, raw: &[u8]) -> Option { - let redacted = crate::durable::redact_agent_identity_text(&String::from_utf8_lossy(raw)); + // Raw protocol frames can contain arbitrary private-key fragments. + // Preserve trace metadata without retaining raw content for identity runs. + let redacted = + if std::env::var("PAPERCLIP_AGENT_PRIVATE_KEY").is_ok_and(|key| !key.is_empty()) { + "[REDACTED: agent identity runtime]".to_owned() + } else { + String::from_utf8_lossy(raw).into_owned() + }; let raw = redacted.as_bytes(); if self.captured_bytes.saturating_add(raw.len()) > self.max_bytes { self.truncated = true; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs index 1a47897bd0c..7cbe493ba05 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/mod.rs @@ -16,9 +16,8 @@ pub use runner::{ TerminalDeliveryReconciliation, }; pub(crate) use state::{ - create_private_temporary_file, open_private_regular_file, redact_agent_identity_text, - redact_sensitive_text_values, redact_text, sanitize_value, validate_semantic_tool_input, - verify_private_directory, + create_private_temporary_file, open_private_regular_file, redact_sensitive_text_values, + redact_text, sanitize_value, validate_semantic_tool_input, verify_private_directory, }; pub use state::{ Command, CommandDisposition, DurableState, DurableStateStore, EventPriority, 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 8ae20ab8046..b2559876c41 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 @@ -312,6 +312,8 @@ pub enum CommandDisposition { #[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct DurableState { + #[serde(skip)] + identity_output: IdentityOutputBuffer, pub schema: String, pub runner_instance_id: String, pub environment_lease_id: String, @@ -352,6 +354,7 @@ pub struct DurableState { impl DurableState { pub(crate) fn new(config: &DurableRunnerConfig) -> Self { Self { + identity_output: IdentityOutputBuffer::default(), schema: STATE_SCHEMA.to_owned(), runner_instance_id: config.runner_instance_id.clone(), environment_lease_id: config.environment_lease_id.clone(), @@ -527,6 +530,15 @@ impl DurableState { } validate_semantic_tool_input_digest(event_type.as_str(), &payload)?; + // Clone until enqueue succeeds: rejected writes must not consume output. + // 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) + } else { + payload + }; + let sanitized_payload = preserve_bounded_display_content( event_type.as_str(), &payload, @@ -634,6 +646,7 @@ impl DurableState { ); } self.peak_outbox_bytes = self.peak_outbox_bytes.max(projected); + self.identity_output = identity_output; Ok(source_seq) } @@ -1626,6 +1639,85 @@ fn durable_semantics_changed_by_sanitization(original: &Value, sanitized: &Value } } +#[derive(Clone, Default, PartialEq)] +struct IdentityOutputBuffer(BTreeMap); + +impl std::fmt::Debug for IdentityOutputBuffer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("IdentityOutputBuffer { [REDACTED] }") + } +} + +impl IdentityOutputBuffer { + fn redact(&mut self, stream: &str, payload: &Value) -> 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) + } + + fn redact_with_key(&mut self, stream: &str, payload: &Value, key: &str) -> Value { + let mut values = vec![key.to_owned(), key.trim().to_owned()]; + values.extend( + key.lines() + .filter(|line| !line.is_empty() && !line.starts_with("-----")) + .map(str::to_owned), + ); + values.sort_by_key(|value| std::cmp::Reverse(value.len())); + self.visit(stream, "", payload, &values) + } + + fn visit(&mut self, path: &str, field: &str, value: &Value, secrets: &[String]) -> Value { + match value { + Value::Object(object) => Value::Object( + object + .iter() + .map(|(key, child)| { + ( + key.clone(), + self.visit(&format!("{path}.{key}"), key, child, secrets), + ) + }) + .collect(), + ), + Value::Array(array) => Value::Array( + array + .iter() + .enumerate() + .map(|(index, child)| { + self.visit(&format!("{path}.{index}"), field, child, secrets) + }) + .collect(), + ), + Value::String(chunk) if matches!(field, "text" | "delta" | "output" | "patch") => { + let mut text = self.0.remove(path).unwrap_or_default(); + text.push_str(chunk); + for secret in secrets { + text = text.replace(secret, "[REDACTED]"); + } + let mut held = 0; + for secret in secrets { + for size in (held + 1..secret.len().min(text.len() + 1)).rev() { + if secret.is_char_boundary(size) && text.ends_with(&secret[..size]) { + held = size; + break; + } + } + } + if held > 0 { + let tail = text.split_off(text.len() - held); + self.0.insert(path.to_owned(), tail); + } + Value::String(text) + } + other => other.clone(), + } + } +} + pub(crate) fn sanitize_value(value: &Value) -> Value { match value { Value::Object(object) => Value::Object( @@ -2383,6 +2475,38 @@ fn current_timestamp() -> Result { #[cfg(test)] mod tests { + #[test] + fn identity_delta_redaction_buffers_every_split_and_never_serializes_pending_material() { + use super::*; + let key = "-----BEGIN PRIVATE KEY-----\nMC4CAQAwBQYDK2VwBCIEIExamplePrivateMaterial\n-----END PRIVATE KEY-----\n"; + let body = key.lines().nth(1).unwrap(); + 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]}}), + 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..]}}), + key, + ); + assert_eq!(first["text"], ""); + assert_eq!(second["text"], "[REDACTED]"); + assert_eq!(second["update"]["delta"], "[REDACTED]"); + } + let mut state = DurableState::new(&config(PathBuf::from("/unused"))); + state + .identity_output + .redact_with_key("item", &json!({"text": &body[..20]}), key); + assert!(!serde_json::to_string(&state).unwrap().contains(&body[..20])); + assert!(!format!("{state:?}").contains(&body[..20])); + } + use std::time::Duration; use super::*; diff --git a/packages/paperclip-runner/src/contracts/provider-trace-file-sink.test.ts b/packages/paperclip-runner/src/contracts/provider-trace-file-sink.test.ts index 405a28edd77..fc30e120364 100644 --- a/packages/paperclip-runner/src/contracts/provider-trace-file-sink.test.ts +++ b/packages/paperclip-runner/src/contracts/provider-trace-file-sink.test.ts @@ -14,6 +14,24 @@ afterEach(async () => { }); describe("ProviderTraceFileSink", () => { + it("omits raw identity-run frames even when a key is split into single-byte deltas", async () => { + const root = await mkdtemp(join(tmpdir(), "paperclip-provider-trace-identity-")); + roots.push(root); + const tracePath = join(root, "provider.ndjson"); + const sink = await createProviderTraceFileSink({ path: tracePath, provider: "opencode", channel: "test" }); + const key = "-----BEGIN PRIVATE KEY-----\nMC4CAQAwBQYDK2VwBCIEIExamplePrivateMaterial\n-----END PRIVATE KEY-----\n"; + sink!.addSensitiveValues([key]); + for (const delta of key) sink!.frame({ + direction: "provider_to_client", raw: JSON.stringify({ delta }), transport: "sse", + }); + await sink!.finish(); + const records = (await readFile(tracePath, "utf8")).trim().split("\n").map(line => JSON.parse(line)); + const frames = records.filter(record => record.kind === "frame"); + expect(frames).toHaveLength(key.length); + for (const frame of frames) expect(Buffer.from(frame.rawBase64, "base64").toString()) + .toBe("[REDACTED: agent identity runtime]"); + }); + it("hardens an existing file before it appends provider frames", async () => { const root = await mkdtemp(join(tmpdir(), "paperclip-provider-trace-mode-")); roots.push(root); diff --git a/packages/paperclip-runner/src/contracts/provider-trace-file-sink.ts b/packages/paperclip-runner/src/contracts/provider-trace-file-sink.ts index aff5932900b..666ad36f036 100644 --- a/packages/paperclip-runner/src/contracts/provider-trace-file-sink.ts +++ b/packages/paperclip-runner/src/contracts/provider-trace-file-sink.ts @@ -33,6 +33,7 @@ export class ProviderTraceFileSink { readonly #channel: string; readonly #maxBytes: number; readonly #sensitiveValues = new Set(); + #hasPrivateIdentity = false; readonly #queue: string[] = []; #nextFrameId: number; #nextDebugSequence: number; @@ -62,6 +63,7 @@ export class ProviderTraceFileSink { addSensitiveValues(values: readonly (string | undefined)[]): void { for (const value of values) { + if (value?.includes("-----BEGIN PRIVATE KEY-----")) this.#hasPrivateIdentity = true; if (value) this.#sensitiveValues.add(value); } } @@ -181,6 +183,9 @@ export class ProviderTraceFileSink { } #redact(raw: Buffer): Buffer { + // Raw frames may split a private key at any byte boundary. Keep trace + // routing/timing metadata, but never spool raw content for identity runs. + if (this.#hasPrivateIdentity) return Buffer.from("[REDACTED: agent identity runtime]"); if (this.#sensitiveValues.size === 0) return raw; let value = raw.toString("utf8"); for (const sensitive of [...this.#sensitiveValues].sort( diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts index b75c21ecf6a..78e6bdb3d9b 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts @@ -458,7 +458,7 @@ export async function prepareAcpxRuntimeSandbox(input: { `include_only = ${JSON.stringify([...new Set([ "PATH", "HOME", "LANG", "LANGUAGE", "TZ", "TMPDIR", "TEMP", "TMP", "CODEX_HOME", "PAPERCLIP_AGENT_KEY_ID", "PAPERCLIP_AGENT_PUBLIC_KEY", "PAPERCLIP_AGENT_PRIVATE_KEY", - ...Object.keys(input.environment).filter(key => !/key|secret|token/i.test(key)), + ...Object.keys(input.environment), ])])}`, ] : []), "", diff --git a/server/src/services/agent-identity-redaction.ts b/server/src/services/agent-identity-redaction.ts index 35b581dc4c5..5c4947e37b4 100644 --- a/server/src/services/agent-identity-redaction.ts +++ b/server/src/services/agent-identity-redaction.ts @@ -10,7 +10,7 @@ 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(); - return { + const redactor = { values, redact(value: T): T { return redactRegisteredSecretValues(value, values); }, // Hold only a suffix that could begin a secret. Separate buffers prevent @@ -33,5 +33,22 @@ export function createAgentIdentityRedactor(privateKeyPem?: string) { pending.delete(stream); return held ? REDACTED_EVENT_VALUE : ""; }, + /** Delta payloads can repeat output under text and provider-specific fields. */ + delta(stream: string, value: T): T { + 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 (Array.isArray(entry)) return entry.map((child, index) => visit(child, `${path}.${index}`, field)); + if (entry && typeof entry === "object") return Object.fromEntries( + Object.entries(entry).map(([key, child]) => [key, visit(child, `${path}.${key}`, key)]), + ); + return entry; + }; + return visit(value, "", "") as T; + }, }; + return redactor; } 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 892e1f363b1..117e37991e2 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 @@ -332,12 +332,34 @@ describe("PaperclipControlPlanePort conformance", () => { await port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity, semanticResult: { ...CONTROL_PLANE_CONFORMANCE_RESULT, summary: privateKeyPem }, }); + let sourceSeq = 2; + for (const [itemId, chunks] of [ + ["halves", [keyBody.slice(0, 31), keyBody.slice(31)]], + ["bytes", [...keyBody]], + ] as const) { + for (const chunk of chunks) { + 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 } }, + }; + await port.appendEvent(event); + expect((await port.appendEvent(event)).disposition).toBe("duplicate"); + } + } 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]) { expect(JSON.stringify(value)).not.toContain(keyBody); expect(JSON.stringify(value)).toContain("***REDACTED***"); } + for (const itemId of ["halves", "bytes"]) { + const deltas = observed.filter(event => event.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***"); + } }); it("runs the unchanged package conformance suite against Paperclip persistence", async () => { 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 0afdea0261d..d92b4c973a2 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -73,6 +73,7 @@ function assertTerminal(value: unknown): asserts value is PrpTerminalState { export class PaperclipControlPlanePort implements ControlPlanePort { readonly #db: Db; readonly #identityRedactor: ReturnType; + readonly #redactedDeltas = new Map(); readonly #binding: PaperclipControlPlaneBinding; #sessionId: string | null = null; readonly #onCommittedEvent?: (event: PrpEvent) => Promise; @@ -189,7 +190,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported"); const validated = validatePrpEvent(value); if (!validated.ok) throw new Error(`native_event_schema_invalid:${validated.issues[0]?.message ?? "unknown"}`); - const event = this.#identityRedactor.redact(validated.event); + let event = this.#identityRedactor.redact(validated.event); if (event.runId !== this.#binding.runId || (this.#sessionId && event.normalizedSessionId !== this.#sessionId)) { throw new Error("native_event_binding_mismatch"); } @@ -199,6 +200,22 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (event.sourceInstanceId !== expectedSourceInstanceId) { throw new Error("native_event_source_binding_mismatch"); } + if (event.eventType === "item.delta") { + const key = `${event.sourceInstanceId}:${event.sourceEventId}`; + const cached = this.#redactedDeltas.get(key); + const originalSha = nativeSha256(validated.event); + if (cached) { + 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, + ) }; + // Retries must use the same redacted payload without consuming a delta twice. + this.#redactedDeltas.set(key, { event, originalSha }); + } + } const persisted = await appendHeartbeatRunEvent(this.#db, { companyId: this.#binding.companyId, runId: this.#binding.runId,