Skip to content

Commit aa652bd

Browse files
committed
fix(executor): read files in Copilot-started runs and replay repeated memory attachments once
Workflow runs Chat starts execute as Chat's principal, and content delivery classified every read under it: queued runs had no gate and failed every workspace-file read, and in-process runs applied Chat's per-item policy and accounted to Chat's request. executeWorkflowCore now runs each execution under runWorkflowOutsideContentDelivery, where that principal reads like any other; the run's result stays gated where Chat started it. Agent conversation memory counted every attachment reference against its replay limit, so re-attaching the same files each turn failed the conversation after a few runs and re-sent identical files. Replay now sends each storage key once, where it first appears, and the limit counts what replay sends.
1 parent 2d0ac79 commit aa652bd

10 files changed

Lines changed: 290 additions & 20 deletions

File tree

‎.claude/rules/sim-content-delivery.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,11 @@ Only Copilot principals (`delegated` / `organization_delegated` with `serviceId:
2828
classified. Every other caller is admitted without a provenance read, so the UI, API keys, MCP and
2929
the workflow executor see byte-identical results.
3030

31+
A run Chat starts executes as Chat's principal, so `executeWorkflowCore` runs every execution under
32+
`runWorkflowOutsideContentDelivery`, where that principal reads like any other. What the run returns
33+
is gated where Chat started it: run execute/resume withhold unadmitted inline file bytes
34+
(`withholdUnadmittedInlineFiles`); Chat's run tools return no file bytes.
35+
3136
## The pieces
3237

3338
- `lib/core/application/content-delivery.ts`: the primitive.

‎.cursor/rules/sim-content-delivery.mdc‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,11 @@ Only Copilot principals (`delegated` / `organization_delegated` with `serviceId:
1919
classified. Every other caller is admitted without a provenance read, so the UI, API keys, MCP and
2020
the workflow executor see byte-identical results.
2121

22+
A run Chat starts executes as Chat's principal, so `executeWorkflowCore` runs every execution under
23+
`runWorkflowOutsideContentDelivery`, where that principal reads like any other. What the run returns
24+
is gated where Chat started it: run execute/resume withhold unadmitted inline file bytes
25+
(`withholdUnadmittedInlineFiles`); Chat's run tools return no file bytes.
26+
2227
## The pieces
2328

2429
- `lib/core/application/content-delivery.ts`: the primitive.

‎apps/sim/executor/handlers/agent/handler.test.ts‎

Lines changed: 107 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -574,20 +574,20 @@ describe('AgentBlockHandler', () => {
574574
context: 'execution',
575575
base64: 'iVBORw0KGgo=',
576576
}
577+
const hydrateWithBytes = () =>
578+
vi.spyOn(userFileBase64, 'hydrateUserFilesWithBase64').mockImplementation(
579+
async (value) =>
580+
(value as (typeof file)[]).map((attachment) => ({
581+
...attachment,
582+
base64: file.base64,
583+
})) as typeof value
584+
)
577585

578586
it.each(['files', 'messages', 'userPrompt'] as const)(
579587
'replays a previous turn from %s with a fresh provider attachment',
580588
async (source) => {
581589
mockGetProviderFromModel.mockReturnValue('openai')
582-
const hydrate = vi
583-
.spyOn(userFileBase64, 'hydrateUserFilesWithBase64')
584-
.mockImplementation(async (value) => {
585-
const files = value as (typeof file)[]
586-
return files.map((attachment) => ({
587-
...attachment,
588-
base64: file.base64,
589-
})) as typeof value
590-
})
590+
const hydrate = hydrateWithBytes()
591591
const inputs: AgentInputs = {
592592
model: 'gpt-4o',
593593
memoryType: 'conversation',
@@ -682,6 +682,104 @@ describe('AgentBlockHandler', () => {
682682
)
683683
).toBe(false)
684684
})
685+
686+
const numbered = (index: number) => ({
687+
...file,
688+
id: `file-${index}`,
689+
name: `page-${index}.png`,
690+
key: `execution/test-workspace/test-workflow/exec-1/page-${index}.png`,
691+
})
692+
693+
it('replays files every turn re-attached once each, where they first appear', async () => {
694+
mockGetProviderFromModel.mockReturnValue('openai')
695+
const hydrate = hydrateWithBytes()
696+
const attachments = Array.from({ length: 7 }, (_, index) => numbered(index))
697+
queueTableRows(schemaMock.memory, [
698+
{
699+
secretProvenanceVersion: null,
700+
data: [1, 2, 3].flatMap((turn) => [
701+
{ role: 'user', content: `Turn ${turn}`, files: attachments },
702+
{ role: 'assistant', content: `Answer ${turn}` },
703+
]),
704+
},
705+
])
706+
await handler.execute({ ...mockContext, executionId: 'exec-4' }, mockBlock, {
707+
model: 'gpt-4o',
708+
memoryType: 'conversation',
709+
conversationId: 'conversation-1',
710+
messages: [{ role: 'user', content: 'Turn 4' }],
711+
files: attachments,
712+
})
713+
const sent = mockExecuteProviderRequest.mock.calls.at(-1)?.[1].messages as Message[]
714+
const userTurns = sent.filter((message) => message.role === 'user')
715+
expect(userTurns.map((message) => message.content)).toEqual([
716+
'Turn 1',
717+
'Turn 2',
718+
'Turn 3',
719+
'Turn 4',
720+
])
721+
expect(userTurns[0].files?.map((attachment) => attachment.key)).toEqual(
722+
attachments.map((attachment) => attachment.key)
723+
)
724+
expect(userTurns.slice(1).map((message) => message.files)).toEqual([
725+
undefined,
726+
undefined,
727+
undefined,
728+
])
729+
expect(hydrate.mock.calls.flatMap(([value]) => value as unknown[])).toHaveLength(7)
730+
hydrate.mockRestore()
731+
})
732+
733+
it('still refuses a replay of more distinct files than the memory limit', async () => {
734+
mockGetProviderFromModel.mockReturnValue('openai')
735+
const hydrate = hydrateWithBytes()
736+
queueTableRows(schemaMock.memory, [
737+
{
738+
secretProvenanceVersion: null,
739+
data: [0, 1, 2].map((turn) => ({
740+
role: 'user',
741+
content: `Turn ${turn}`,
742+
files: Array.from({ length: 7 }, (_, index) => numbered(turn * 7 + index)),
743+
})),
744+
},
745+
])
746+
await expect(
747+
handler.execute({ ...mockContext, executionId: 'exec-4' }, mockBlock, {
748+
model: 'gpt-4o',
749+
memoryType: 'conversation',
750+
conversationId: 'conversation-1',
751+
messages: [{ role: 'user', content: 'Turn 3' }],
752+
})
753+
).rejects.toThrow('Conversation memory exceeds 20 file attachments')
754+
expect(mockExecuteProviderRequest).not.toHaveBeenCalled()
755+
hydrate.mockRestore()
756+
})
757+
758+
it('keeps a repeated file on a turn that has no text to stand on without it', async () => {
759+
mockGetProviderFromModel.mockReturnValue('openai')
760+
const hydrate = hydrateWithBytes()
761+
queueTableRows(schemaMock.memory, [
762+
{
763+
secretProvenanceVersion: null,
764+
data: [
765+
{ role: 'user', content: 'Read this', files: [file] },
766+
{ role: 'assistant', content: 'Read it' },
767+
{ role: 'user', content: '', files: [file] },
768+
{ role: 'assistant', content: 'Same file' },
769+
],
770+
},
771+
])
772+
await handler.execute({ ...mockContext, executionId: 'exec-3' }, mockBlock, {
773+
model: 'gpt-4o',
774+
memoryType: 'conversation',
775+
conversationId: 'conversation-1',
776+
messages: [{ role: 'user', content: 'And now?' }],
777+
})
778+
const sent = mockExecuteProviderRequest.mock.calls.at(-1)?.[1].messages as Message[]
779+
const userTurns = sent.filter((message) => message.role === 'user')
780+
expect(userTurns.map((message) => message.files?.length)).toEqual([1, 1, undefined])
781+
hydrate.mockRestore()
782+
})
685783
})
686784

687785
describe('model fallback', () => {

‎apps/sim/executor/handlers/agent/handler.ts‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,7 @@ import {
6868
getMemoryMessageAppendKey,
6969
getMemoryMessageTurnId,
7070
memoryService,
71+
selectReplayAttachments,
7172
} from '@/executor/handlers/agent/memory'
7273
import {
7374
buildLoadSkillTool,
@@ -1754,22 +1755,30 @@ export class AgentBlockHandler implements BlockHandler {
17541755

17551756
const inlineMaxBytes = getInlineHydrationMaxBytes(providerId)
17561757

1758+
const replayAttachments = selectReplayAttachments(messages)
1759+
17571760
for (let messageIndex = 0; messageIndex < messages.length; messageIndex++) {
17581761
const message = messages[messageIndex]
17591762
if (!message.files?.length) {
17601763
continue
17611764
}
1765+
const files = replayAttachments[messageIndex]
1766+
if (files.length === 0) {
1767+
nextMessages[messageIndex] = omit(message, ['files'])
1768+
copyNativeConversationMessage(message, nextMessages[messageIndex])
1769+
continue
1770+
}
17621771

17631772
const unsafeGeneratedDocumentFiles = new Set<string>()
17641773
const groups = new Map<boolean, Array<{ file: UserFile; index: number }>>()
1765-
message.files.forEach((file, index) => {
1774+
files.forEach((file, index) => {
17661775
const workspaceFile =
17671776
ctx.principal?.kind === 'system' && tryInferContextFromKey(file.key) === 'workspace'
17681777
const group = groups.get(workspaceFile) ?? []
17691778
group.push({ file, index })
17701779
groups.set(workspaceFile, group)
17711780
})
1772-
const hydratedFiles = [...message.files]
1781+
const hydratedFiles = [...files]
17731782
await Promise.all(
17741783
[...groups.values()].map(async (group) => {
17751784
const hydrated = await hydrateUserFilesWithBase64(
@@ -1805,7 +1814,7 @@ export class AgentBlockHandler implements BlockHandler {
18051814
const modelSafeHydratedFiles = hydratedFiles.flatMap((file, fileIndex) => {
18061815
if (unsafeGeneratedDocumentFiles.has(`${file.key}:${file.id}`)) return []
18071816

1808-
const sourceFile = message.files?.[fileIndex]
1817+
const sourceFile = files[fileIndex]
18091818
const nameProjection = sourceFile ? projectedNameByFile.get(sourceFile) : undefined
18101819
if (
18111820
!nameProjection ||

‎apps/sim/executor/handlers/agent/memory.ts‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ import {
4343
readBoundMemorySecretProvenance,
4444
} from '@/lib/memory/secret-provenance'
4545
import type { AgentInputs, FileNameProjection, Message } from '@/executor/handlers/agent/types'
46-
import type { ExecutionContext } from '@/executor/types'
46+
import type { ExecutionContext, UserFile } from '@/executor/types'
4747
import {
4848
projectResolvedSecretModelContent,
4949
projectResolvedSecretModelJsonStrings,
@@ -118,6 +118,25 @@ function isOptionalMemoryStorageFailure(error: unknown): boolean {
118118
)
119119
}
120120

121+
/**
122+
* The attachments each message sends to the model. A storage key pins its bytes, so a file attached
123+
* on several turns is sent once, where it first appears, keeping the replayed prefix stable across
124+
* runs; a message with no text keeps its own so it is never sent empty.
125+
*/
126+
export function selectReplayAttachments(messages: readonly Message[]): UserFile[][] {
127+
const sent = new Set<string>()
128+
return messages.map((message) => {
129+
const files = message.files ?? []
130+
const unsent = files.filter((file) => {
131+
const identity = file.key || file.id
132+
if (sent.has(identity)) return false
133+
sent.add(identity)
134+
return true
135+
})
136+
return unsent.length === 0 && !message.content?.trim() ? files : unsent
137+
})
138+
}
139+
121140
export class Memory {
122141
async fetchMemoryMessages(
123142
ctx: ExecutionContext,
@@ -177,7 +196,10 @@ export class Memory {
177196
}
178197

179198
/** Bound historical downloads independently of text-token windows and per-file byte caps. */
180-
const fileCount = messages.reduce((count, message) => count + (message.files?.length ?? 0), 0)
199+
const fileCount = selectReplayAttachments(messages).reduce(
200+
(count, files) => count + files.length,
201+
0
202+
)
181203
if (fileCount > MEMORY.MAX_REPLAY_FILE_REFERENCES) {
182204
throw new Error(
183205
`Conversation memory exceeds ${MEMORY.MAX_REPLAY_FILE_REFERENCES} file attachments. Use a smaller memory window.`

‎apps/sim/lib/core/application/content-delivery.test.ts‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ import {
1919
observeContentOutsideModel,
2020
requireAdmittedContent,
2121
runDeclaredContentOperation,
22+
runWorkflowOutsideContentDelivery,
2223
trackContentDelivery,
2324
} from '@/lib/core/application/content-delivery'
2425
import { defineWorkspaceOperation } from '@/lib/core/application/workspace-operation'
@@ -499,6 +500,37 @@ describe('observeContentDelivery', () => {
499500
})
500501
})
501502

503+
describe('runWorkflowOutsideContentDelivery', () => {
504+
it('keeps an in-process run from consulting or accounting to the Chat request around it', async () => {
505+
const { gate, batches } = recordingGate()
506+
const record = await observeContentDelivery(gate, () =>
507+
trackContentDelivery(async () => {
508+
noteContentDeliveryOperation(SOURCED, CHAT)
509+
await runWorkflowOutsideContentDelivery(() =>
510+
runDeclaredContentOperation(SOURCED, CHAT, () =>
511+
admitContent(CHAT, [source('tainted', 'unknown')])
512+
)
513+
)
514+
})
515+
)
516+
expect(batches).toEqual([])
517+
expect(record.unaccounted).toBe(true)
518+
})
519+
520+
it('classifies again under a gate installed inside the run, and after the run returns', async () => {
521+
const outer = recordingGate()
522+
const inner = recordingGate()
523+
await observeContentDelivery(outer.gate, async () => {
524+
await runWorkflowOutsideContentDelivery(() =>
525+
observeContentDelivery(inner.gate, () => admitContent(CHAT, [source('inside')]))
526+
)
527+
await admitContent(CHAT, [source('result')])
528+
})
529+
expect(inner.batches).toEqual([['inside']])
530+
expect(outer.batches).toEqual([['result']])
531+
})
532+
})
533+
502534
describe('observeContentOutsideModel', () => {
503535
it('admits every item, tainted ones included, inside a gate that would refuse them', async () => {
504536
const { gate, batches } = recordingGate()

‎apps/sim/lib/core/application/content-delivery.ts‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,8 @@ interface ContentLedger {
9999

100100
const gates = new AsyncLocalStorage<ContentDeliveryGate>()
101101
const ledgers = new AsyncLocalStorage<ContentLedger>()
102+
/** Set inside {@link runWorkflowOutsideContentDelivery}; any gate installed within clears it. */
103+
const workflowRuns = new AsyncLocalStorage<true>()
102104

103105
const MODEL_UNSAFE_CONTENT_MESSAGES: Record<ContentSourceKind, string> = {
104106
'workspace-file': 'File cannot be sent to a model because its secret provenance is unavailable',
@@ -155,19 +157,32 @@ class ContentDeliveryUndeclaredError extends Error {
155157
}
156158
}
157159

158-
/** Chat's own workspace, organization, and resource delegations. */
160+
/**
161+
* Chat's own workspace, organization, and resource delegations, outside a workflow run (see
162+
* {@link runWorkflowOutsideContentDelivery}).
163+
*/
159164
export function isCopilotPrincipal(principal: Principal): boolean {
160165
return (
161166
(principal.kind === 'delegated' ||
162167
principal.kind === 'organization_delegated' ||
163168
principal.kind === 'resource_delegated') &&
164-
principal.serviceId === 'copilot'
169+
principal.serviceId === 'copilot' &&
170+
workflowRuns.getStore() === undefined
165171
)
166172
}
167173

168174
/** Installs the gate every item reported inside `execute` is decided by; the innermost gate wins. */
169175
export function observeContentDelivery<T>(gate: ContentDeliveryGate, execute: () => T): T {
170-
return gates.run(gate, execute)
176+
return workflowRuns.exit(() => gates.run(gate, execute))
177+
}
178+
179+
/**
180+
* Runs a workflow execution as any caller, even when Chat started it: its blocks read for the run,
181+
* and what the run returns is gated where Chat started it. It neither consults nor accounts to a
182+
* Chat request it runs in-process under; a gate installed inside it classifies again.
183+
*/
184+
export function runWorkflowOutsideContentDelivery<T>(execute: () => T): T {
185+
return ledgers.exit(() => workflowRuns.run(true, execute))
171186
}
172187

173188
/**
@@ -202,7 +217,7 @@ const OUTSIDE_MODEL_GATE: ContentDeliveryGate = {
202217
* so a tainted source still yields a derived file that carries its taint. Each use names why.
203218
*/
204219
export function observeContentOutsideModel<T>(execute: () => T): T {
205-
return gates.run(OUTSIDE_MODEL_GATE, execute)
220+
return observeContentDelivery(OUTSIDE_MODEL_GATE, execute)
206221
}
207222

208223
/**

‎apps/sim/lib/core/application/index.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ export {
8585
redactBeforeCutting,
8686
requireAdmittedContent,
8787
runDeclaredContentOperation,
88+
runWorkflowOutsideContentDelivery,
8889
trackContentDelivery,
8990
type WithheldContent,
9091
} from './content-delivery'

‎apps/sim/lib/workflows/executor/execution-core.ts‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import type { Edge } from '@xyflow/react'
1414
import { eq } from 'drizzle-orm'
1515
import { z } from 'zod'
1616
import { type EffectivePiiRedaction, resolveEffectivePiiRedaction } from '@/lib/billing/retention'
17+
import { runWorkflowOutsideContentDelivery } from '@/lib/core/application/content-delivery'
1718
import { ExecutionScopeDeletedError } from '@/lib/core/errors/execution-scope-deleted'
1819
import { logFailureOnce } from '@/lib/core/errors/failure-log'
1920
import { UserFailure } from '@/lib/core/errors/user-failure'
@@ -604,8 +605,14 @@ function prefetchExecutionReads(options: ExecuteWorkflowCoreOptions): ExecutionR
604605
* paid inside its readiness budget. Connecting on intent rather than at worker
605606
* start keeps the tasks that never execute a workflow, most of the fleet by
606607
* volume, from opening a connection they would never use.
608+
*
609+
* Runs under `runWorkflowOutsideContentDelivery`, so a run Chat started reads what any run reads.
607610
*/
608-
export async function executeWorkflowCore(
611+
export function executeWorkflowCore(options: ExecuteWorkflowCoreOptions): Promise<ExecutionResult> {
612+
return runWorkflowOutsideContentDelivery(() => executeWorkflowCoreInRun(options))
613+
}
614+
615+
async function executeWorkflowCoreInRun(
609616
options: ExecuteWorkflowCoreOptions
610617
): Promise<ExecutionResult> {
611618
connectExecutionSignalHub()

0 commit comments

Comments
 (0)