Overwrite attachment bytes for repeated message attachment index

- Make projection attachment writes use last-write-wins per file path - Always write attachment files so updated message attachments replace stale bytes - Add regression test covering same message/attachment index update behavior

Julius Marminge committed 8 months ago

Commit 7d4b0261 parent 9962f572 files · +146 −9Browse files

Unified · Split

2 files

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +143 −0
apps/server/src/orchestration/Layers/ProjectionPipeline.ts +3 −9

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+143−0

@@ -480,6 +480,149 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { · 479 unmodified lines
480	}),	480	}),
481	);	481	);
482		482
		483	it.effect("overwrites attachment file bytes when a message updates the same attachment index", () =>
		484	Effect.gen(function* () {
		485	const projectionPipeline = yield* OrchestrationProjectionPipeline;
		486	const eventStore = yield* OrchestrationEventStore;
		487	const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-"));
		488	const now = new Date().toISOString();
		489	const later = new Date(Date.now() + 1_000).toISOString();
		490
		491	const serverConfig = {
		492	mode: "web",
		493	port: 0,
		494	host: undefined,
		495	cwd: "/tmp/project-attachments",
		496	keybindingsConfigPath: path.join(stateDir, "keybindings.json"),
		497	stateDir,
		498	staticDir: undefined,
		499	devUrl: undefined,
		500	noBrowser: true,
		501	authToken: undefined,
		502	autoBootstrapProjectFromCwd: false,
		503	logWebSocketEvents: false,
		504	} satisfies ServerConfigShape;
		505
		506	yield* eventStore.append({
		507	type: "project.created",
		508	eventId: EventId.makeUnsafe("evt-overwrite-1"),
		509	aggregateKind: "project",
		510	aggregateId: ProjectId.makeUnsafe("project-overwrite"),
		511	occurredAt: now,
		512	commandId: CommandId.makeUnsafe("cmd-overwrite-1"),
		513	causationEventId: null,
		514	correlationId: CommandId.makeUnsafe("cmd-overwrite-1"),
		515	metadata: {},
		516	payload: {
		517	projectId: ProjectId.makeUnsafe("project-overwrite"),
		518	title: "Project Overwrite",
		519	workspaceRoot: "/tmp/project-overwrite",
		520	defaultModel: null,
		521	scripts: [],
		522	createdAt: now,
		523	updatedAt: now,
		524	},
		525	});
		526
		527	yield* eventStore.append({
		528	type: "thread.created",
		529	eventId: EventId.makeUnsafe("evt-overwrite-2"),
		530	aggregateKind: "thread",
		531	aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
		532	occurredAt: now,
		533	commandId: CommandId.makeUnsafe("cmd-overwrite-2"),
		534	causationEventId: null,
		535	correlationId: CommandId.makeUnsafe("cmd-overwrite-2"),
		536	metadata: {},
		537	payload: {
		538	threadId: ThreadId.makeUnsafe("thread-overwrite"),
		539	projectId: ProjectId.makeUnsafe("project-overwrite"),
		540	title: "Thread Overwrite",
		541	model: "gpt-5-codex",
		542	branch: null,
		543	worktreePath: null,
		544	createdAt: now,
		545	updatedAt: now,
		546	},
		547	});
		548
		549	yield* eventStore.append({
		550	type: "thread.message-sent",
		551	eventId: EventId.makeUnsafe("evt-overwrite-3"),
		552	aggregateKind: "thread",
		553	aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
		554	occurredAt: now,
		555	commandId: CommandId.makeUnsafe("cmd-overwrite-3"),
		556	causationEventId: null,
		557	correlationId: CommandId.makeUnsafe("cmd-overwrite-3"),
		558	metadata: {},
		559	payload: {
		560	threadId: ThreadId.makeUnsafe("thread-overwrite"),
		561	messageId: MessageId.makeUnsafe("message-overwrite"),
		562	role: "user",
		563	text: "first image",
		564	attachments: [
		565	{
		566	type: "image",
		567	name: "file.png",
		568	mimeType: "image/png",
		569	sizeBytes: 5,
		570	dataUrl: "data:image/png;base64,SGVsbG8=",
		571	},
		572	],
		573	turnId: null,
		574	streaming: false,
		575	createdAt: now,
		576	updatedAt: now,
		577	},
		578	});
		579
		580	yield* eventStore.append({
		581	type: "thread.message-sent",
		582	eventId: EventId.makeUnsafe("evt-overwrite-4"),
		583	aggregateKind: "thread",
		584	aggregateId: ThreadId.makeUnsafe("thread-overwrite"),
		585	occurredAt: later,
		586	commandId: CommandId.makeUnsafe("cmd-overwrite-4"),
		587	causationEventId: null,
		588	correlationId: CommandId.makeUnsafe("cmd-overwrite-4"),
		589	metadata: {},
		590	payload: {
		591	threadId: ThreadId.makeUnsafe("thread-overwrite"),
		592	messageId: MessageId.makeUnsafe("message-overwrite"),
		593	role: "user",
		594	text: "",
		595	attachments: [
		596	{
		597	type: "image",
		598	name: "file.png",
		599	mimeType: "image/png",
		600	sizeBytes: 5,
		601	dataUrl: "data:image/png;base64,V29ybGQ=",
		602	},
		603	],
		604	turnId: null,
		605	streaming: false,
		606	createdAt: now,
		607	updatedAt: later,
		608	},
		609	});
		610
		611	yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig));
		612
		613	const attachmentPath = path.join(
		614	stateDir,
		615	"attachments",
		616	"thread-overwrite",
		617	"message-overwrite",
		618	"0.png",
		619	);
		620	assert.equal(fs.existsSync(attachmentPath), true);
		621	assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64"));
		622	fs.rmSync(stateDir, { recursive: true, force: true });
		623	}),
		624	);
		625
483	it.effect("does not persist attachment files when projector transaction rolls back", () =>	626	it.effect("does not persist attachment files when projector transaction rolls back", () =>
484	Effect.gen(function* () {	627	Effect.gen(function* () {
485	const projectionPipeline = yield* OrchestrationProjectionPipeline;	628	const projectionPipeline = yield* OrchestrationProjectionPipeline;

apps/server/src/orchestration/Layers/ProjectionPipeline.ts+3−9

   const writesByPath = new Map<string, Uint8Array>();
   for (const pendingWrite of input.sideEffects.writes) {
-    if (!writesByPath.has(pendingWrite.absolutePath)) {
+    // Last write wins for a path so updated attachment contents replace stale bytes.
-      writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes);
+    writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes);
-    }
   }
   yield* Effect.forEach(
     writesByPath.entries(),
         yield* input.fileSystem.makeDirectory(input.path.dirname(absolutePath), {
           recursive: true,
         });
-        const alreadyExists = yield* input.fileSystem
+        yield* input.fileSystem.writeFile(absolutePath, bytes);
-          .exists(absolutePath)
-          .pipe(Effect.catch(() => Effect.succeed(false)));
-        if (!alreadyExists) {
-          yield* input.fileSystem.writeFile(absolutePath, bytes);
-        }
       }),
     { concurrency: 1 },
   );