diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 8ca77624cd8ce5caab104e5cd3b5035f347dc7da..abb411b724530f3d83fbb03d18afe9f4144fe973 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -480,6 +480,149 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("overwrites attachment file bytes when a message updates the same attachment index", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-")); + const now = new Date().toISOString(); + const later = new Date(Date.now() + 1_000).toISOString(); + + const serverConfig = { + mode: "web", + port: 0, + host: undefined, + cwd: "/tmp/project-attachments", + keybindingsConfigPath: path.join(stateDir, "keybindings.json"), + stateDir, + staticDir: undefined, + devUrl: undefined, + noBrowser: true, + authToken: undefined, + autoBootstrapProjectFromCwd: false, + logWebSocketEvents: false, + } satisfies ServerConfigShape; + + yield* eventStore.append({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-overwrite-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-overwrite"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-overwrite-1"), + causationEventId: null, + correlationId: CommandId.makeUnsafe("cmd-overwrite-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-overwrite"), + title: "Project Overwrite", + workspaceRoot: "/tmp/project-overwrite", + defaultModel: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-overwrite-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-overwrite"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-overwrite-2"), + causationEventId: null, + correlationId: CommandId.makeUnsafe("cmd-overwrite-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-overwrite"), + projectId: ProjectId.makeUnsafe("project-overwrite"), + title: "Thread Overwrite", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-overwrite-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-overwrite"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-overwrite-3"), + causationEventId: null, + correlationId: CommandId.makeUnsafe("cmd-overwrite-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-overwrite"), + messageId: MessageId.makeUnsafe("message-overwrite"), + role: "user", + text: "first image", + attachments: [ + { + type: "image", + name: "file.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,SGVsbG8=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-overwrite-4"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-overwrite"), + occurredAt: later, + commandId: CommandId.makeUnsafe("cmd-overwrite-4"), + causationEventId: null, + correlationId: CommandId.makeUnsafe("cmd-overwrite-4"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-overwrite"), + messageId: MessageId.makeUnsafe("message-overwrite"), + role: "user", + text: "", + attachments: [ + { + type: "image", + name: "file.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,V29ybGQ=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: later, + }, + }); + + yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig)); + + const attachmentPath = path.join( + stateDir, + "attachments", + "thread-overwrite", + "message-overwrite", + "0.png", + ); + assert.equal(fs.existsSync(attachmentPath), true); + assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64")); + fs.rmSync(stateDir, { recursive: true, force: true }); + }), + ); + it.effect("does not persist attachment files when projector transaction rolls back", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 162468b18b0947c4a517aecfb63b46be4a8e8590..5474c602bb70e40ff378950226e89d60c8ccd1ec 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -364,9 +364,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { const writesByPath = new Map(); for (const pendingWrite of input.sideEffects.writes) { - if (!writesByPath.has(pendingWrite.absolutePath)) { - writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes); - } + // Last write wins for a path so updated attachment contents replace stale bytes. + writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes); } yield* Effect.forEach( writesByPath.entries(), @@ -375,12 +374,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { yield* input.fileSystem.makeDirectory(input.path.dirname(absolutePath), { recursive: true, }); - const alreadyExists = yield* input.fileSystem - .exists(absolutePath) - .pipe(Effect.catch(() => Effect.succeed(false))); - if (!alreadyExists) { - yield* input.fileSystem.writeFile(absolutePath, bytes); - } + yield* input.fileSystem.writeFile(absolutePath, bytes); }), { concurrency: 1 }, );