Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions apps/sim/lib/logs/execution/trace-secret-projection.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,63 @@ describe('projectTraceSpansForSecrets', () => {
expect(result[0].output).toEqual({ value: 'prefix {{LONG}} suffix' })
})

/** Never moves backwards across tests: the slice deadline it sets outlives each one. */
let fakeClockMs = 0

/**
* Counts event-loop turns that complete while the projection is in flight, on a clock that
* advances 1ms per read so the slice expires at a fixed rate on any machine. A single yield
* before any work allows one turn — the pending immediate runs, and its successor cannot until
* the projection parks again — so two or more prove it keeps yielding mid-projection.
*/
async function countTurnsDuringProjection(spans: TraceSpan[]) {
fakeClockMs = Math.max(fakeClockMs, performance.now())
const now = vi.spyOn(performance, 'now').mockImplementation(() => {
fakeClockMs += 1
return fakeClockMs
})
let settled = false
let turns = 0
const tick = () => {
if (settled) return
turns += 1
setImmediate(tick)
}
setImmediate(tick)
try {
const projected = await projectTraceSpansForSecrets(spans, {
registry: createRegistry([{ plaintext: 'secretval', replacement: '{{SECRET}}' }]),
store: STORE,
})
return { turns, projected }
} finally {
settled = true
now.mockRestore()
}
}

it('keeps yielding to the event loop throughout a content projection', async () => {
const spans = Array.from({ length: 30 }, (_, index) =>
createSpan({ id: `span-${index}`, output: { note: `span ${index} holds secretval` } })
)

const { turns, projected } = await countTurnsDuringProjection(spans)

expect(turns).toBeGreaterThanOrEqual(2)
expect(projected[29].output).toEqual({ note: 'span 29 holds {{SECRET}}' })
})

it('keeps yielding to the event loop throughout a structure-only projection', async () => {
const children = Array.from({ length: 50 }, (_, index) => createSpan({ id: `child-${index}` }))

const { turns, projected } = await countTurnsDuringProjection([
createSpan({ id: 'root', type: 'workflow', children }),
])

expect(turns).toBeGreaterThanOrEqual(2)
expect(projected[0].children).toHaveLength(50)
})

it('projects a successfully resolved numeric Function result without mutating runtime output', async () => {
const runtimeOutput = { result: 12345678, unchanged: 5678 }
const source = createSpan({ output: runtimeOutput })
Expand Down
82 changes: 72 additions & 10 deletions apps/sim/lib/logs/execution/trace-secret-projection.ts
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ const MAX_LARGE_VALUES = 1_024
const MAX_LARGE_VALUE_CHAIN_DEPTH = 32
const MAX_LARGE_MANIFEST_CHUNKS = MAX_LARGE_VALUES
const MAX_LARGE_MANIFEST_ITEMS = MAX_CONTENT_NODES
const PROJECTION_SLICE_MS = 10
const OMIT = Symbol('omit-trace-content')
const LARGE_VALUE_REF_KEYS = new Set([
LARGE_VALUE_REF_MARKER,
Expand Down Expand Up @@ -167,6 +168,38 @@ function createProjectionContext(
}
}

let projectionSliceEndsAt = 0
let projectionTurnScheduled = false
const projectionTurnQueue: Array<() => void> = []

/**
* Projection is CPU-bound and its awaits settle as microtasks, which never let the event loop
* service I/O, so without a macrotask yield one large trace — or a page of them projected
* concurrently — holds the whole process until it finishes.
*
* The slice is process-wide and handed to one waiting projection per event-loop turn: releasing
* them together would let each finish its in-flight field before I/O runs again, so a turn would
* grow with concurrency instead of staying at one slice plus one field.
*/
async function yieldWhenSliceExpires(): Promise<void> {
if (performance.now() < projectionSliceEndsAt) return
await new Promise<void>((resolve) => {
projectionTurnQueue.push(resolve)
scheduleProjectionTurn()
})
}

function scheduleProjectionTurn(): void {
if (projectionTurnScheduled) return
projectionTurnScheduled = true
setImmediate(() => {
projectionTurnScheduled = false
projectionSliceEndsAt = performance.now() + PROJECTION_SLICE_MS
projectionTurnQueue.shift()?.()
if (projectionTurnQueue.length > 0) scheduleProjectionTurn()
})
}

function visitNode(state: TraversalState, depth: number): void {
state.nodes += 1
if (state.nodes > MAX_CONTENT_NODES) {
Expand Down Expand Up @@ -800,6 +833,7 @@ async function sanitizeContentField(
value: unknown,
context: ProjectionContext
): Promise<unknown | typeof OMIT> {
await yieldWhenSliceExpires()
Comment thread
waleedlatif1 marked this conversation as resolved.
try {
return await sanitizeMaterializedValue(value, context)
} catch (error) {
Expand Down Expand Up @@ -829,6 +863,7 @@ async function sanitizeIterationToolCalls(
assertTraceStructureArrayFits(calls, state)
const sanitized: IterationToolCall[] = []
for (const call of calls) {
await yieldWhenSliceExpires()
if (!takeTraceStructureNode(state, depth)) {
throw new TraceSecretProjectionError('Trace tool-call structure limit exceeded')
}
Expand All @@ -851,6 +886,7 @@ async function sanitizeLegacyToolCalls(
assertTraceStructureArrayFits(calls, state)
const sanitized: ToolCall[] = []
for (const call of calls) {
await yieldWhenSliceExpires()
if (!takeTraceStructureNode(state, depth)) {
throw new TraceSecretProjectionError('Trace tool-call structure limit exceeded')
}
Expand Down Expand Up @@ -881,6 +917,7 @@ async function sanitizeProviderSegment(
state: BoundedTraceStructureState,
depth: number
): Promise<ProviderTimingSegment> {
await yieldWhenSliceExpires()
if (!takeTraceStructureNode(state, depth)) {
throw new TraceSecretProjectionError('Trace provider structure limit exceeded')
}
Expand Down Expand Up @@ -913,8 +950,10 @@ async function sanitizeTraceSpan(
span: TraceSpan,
context: ProjectionContext,
depth: number,
state: BoundedTraceStructureState
state: BoundedTraceStructureState,
assertContentIsSafe: SpanContentAssertion
): Promise<TraceSpan> {
await yieldWhenSliceExpires()
if (!takeTraceStructureNode(state, depth)) {
throw new TraceSecretProjectionError('Trace span structure limit exceeded')
}
Expand Down Expand Up @@ -954,11 +993,13 @@ async function sanitizeTraceSpan(
projected.providerTiming = { ...span.providerTiming, segments }
}

assertContentIsSafe(projected, depth)

if (span.children !== undefined) {
assertTraceStructureArrayFits(span.children, state)
const children: TraceSpan[] = []
for (const child of span.children) {
children.push(await sanitizeTraceSpan(child, context, depth + 1, state))
children.push(await sanitizeTraceSpan(child, context, depth + 1, state, assertContentIsSafe))
}
projected.children = children
}
Expand Down Expand Up @@ -1278,11 +1319,12 @@ function assertNoPlaintext(

type TraceContentVisitor = (value: unknown) => void

function visitTraceSpanContent(
/** Visits one span's content, not its children's; `visitTraceSpanContent` descends from here. */
function visitTraceSpanOwnContent(
span: TraceSpan,
visit: TraceContentVisitor,
structureState: BoundedTraceStructureState,
depth = 0
depth: number
): void {
if (!takeTraceStructureNode(structureState, depth)) {
throw new TraceSecretProjectionError('Trace invariant structure limit exceeded')
Expand Down Expand Up @@ -1335,6 +1377,15 @@ function visitTraceSpanContent(
}
}
assertTraceStructureArrayFits(span.children ?? [], structureState)
}

function visitTraceSpanContent(
span: TraceSpan,
visit: TraceContentVisitor,
structureState: BoundedTraceStructureState,
depth = 0
): void {
visitTraceSpanOwnContent(span, visit, structureState, depth)
for (const child of span.children ?? []) {
visitTraceSpanContent(child, visit, structureState, depth + 1)
}
Expand All @@ -1348,12 +1399,23 @@ function visitTraceSpansContent(traceSpans: TraceSpan[], visit: TraceContentVisi
}
}

function assertTraceSpansContentIsSafe(
traceSpans: TraceSpan[],
type SpanContentAssertion = (span: TraceSpan, depth: number) => void

/**
* The whole-trace content check, applied to each projected span as soon as its own content is
* final. Called in projection order (pre-order, a parent before its children), it walks exactly
* what one walk of the finished trace would, so an over-budget trace fails at the span that
* crosses the budget instead of after the whole trace was projected and then discarded.
*/
function createSpanContentAssertion(
traceSpans: readonly TraceSpan[],
context: PlaintextInvariantContext
): void {
): SpanContentAssertion {
const structureState: BoundedTraceStructureState = { nodes: 0, truncated: false }
const contentState: TraversalState = { nodes: 0, ancestors: new WeakSet<object>() }
visitTraceSpansContent(traceSpans, (value) => assertNoPlaintext(value, context, contentState))
assertTraceStructureArrayFits(traceSpans, structureState)
const visit = (value: unknown) => assertNoPlaintext(value, context, contentState)
return (span, depth) => visitTraceSpanOwnContent(span, visit, structureState, depth)
}

function collectPostTransformLargeValues(
Expand Down Expand Up @@ -1577,10 +1639,10 @@ export async function projectTraceSpansForSecrets(
const projected: TraceSpan[] = []
const state: BoundedTraceStructureState = { nodes: 0, truncated: false }
assertTraceStructureArrayFits(traceSpans, state)
const assertContentIsSafe = createSpanContentAssertion(traceSpans, context)
for (const span of traceSpans) {
projected.push(await sanitizeTraceSpan(span, context, 0, state))
projected.push(await sanitizeTraceSpan(span, context, 0, state, assertContentIsSafe))
}
assertTraceSpansContentIsSafe(projected, context)
return projected
} catch (error) {
logger.warn('Trace secret projection failed; retaining structural spans only', {
Expand Down
Loading