Skip to content

Commit 2fe1e88

Browse files
authored
fix(executor): read files in Copilot-started runs and replay repeated memory attachments once (#9179)
Workflow runs Chat starts now execute under runWorkflowOutsideContentDelivery, so their reads behave like any run's while the result stays gated where Chat started it. Agent memory replay sends each repeated user attachment once, and the replay limit counts what is sent.
1 parent f79d322 commit 2fe1e88

10 files changed

Lines changed: 316 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 keep inline file bytes only for admitted files
34+
(`withholdUnadmittedInlineFiles`), and Chat's in-app run tools strip `base64` from their result.
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 keep inline file bytes only for admitted files
25+
(`withholdUnadmittedInlineFiles`), and Chat's in-app run tools strip `base64` from their result.
26+
2227
## The pieces
2328

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

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

Lines changed: 130 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,127 @@ 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+
})
783+
784+
it('keeps a user turn’s file when only an earlier assistant turn carried it', async () => {
785+
mockGetProviderFromModel.mockReturnValue('openai')
786+
const hydrate = hydrateWithBytes()
787+
queueTableRows(schemaMock.memory, [
788+
{
789+
secretProvenanceVersion: null,
790+
data: [
791+
{ role: 'assistant', content: 'Here is a file', files: [file] },
792+
{ role: 'user', content: 'Read it', files: [file] },
793+
],
794+
},
795+
])
796+
await handler.execute({ ...mockContext, executionId: 'exec-3' }, mockBlock, {
797+
model: 'gpt-4o',
798+
memoryType: 'conversation',
799+
conversationId: 'conversation-1',
800+
messages: [{ role: 'user', content: 'And now?' }],
801+
})
802+
const sent = mockExecuteProviderRequest.mock.calls.at(-1)?.[1].messages as Message[]
803+
expect(sent.find((message) => message.content === 'Read it')?.files).toHaveLength(1)
804+
hydrate.mockRestore()
805+
})
685806
})
686807

687808
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: 26 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,27 @@ 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 a user
123+
* attached on several turns is sent once, where it first appears, keeping the replayed prefix
124+
* stable across runs; a user turn with no text keeps its own so it is never sent empty. Only user
125+
* turns are deduplicated: every provider sends their attachments, while some drop other roles'.
126+
*/
127+
export function selectReplayAttachments(messages: readonly Message[]): UserFile[][] {
128+
const sent = new Set<string>()
129+
return messages.map((message) => {
130+
const files = message.files ?? []
131+
if (message.role !== 'user') return files
132+
const unsent = files.filter((file) => {
133+
const identity = file.key || file.id
134+
if (sent.has(identity)) return false
135+
sent.add(identity)
136+
return true
137+
})
138+
return unsent.length === 0 && !message.content?.trim() ? files : unsent
139+
})
140+
}
141+
121142
export class Memory {
122143
async fetchMemoryMessages(
123144
ctx: ExecutionContext,
@@ -177,7 +198,10 @@ export class Memory {
177198
}
178199

179200
/** 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)
201+
const fileCount = selectReplayAttachments(messages).reduce(
202+
(count, files) => count + files.length,
203+
0
204+
)
181205
if (fileCount > MEMORY.MAX_REPLAY_FILE_REFERENCES) {
182206
throw new Error(
183207
`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'

0 commit comments

Comments
 (0)