authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-01 01:42:02-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-01 20:06:21-08:00
logc0054cb56f1439c7bfda7771f919bdb7f948bab7
tree8148227ca4aa09e7121c259e5564650c926a9b80
parent41734dfbbfd47d2d9639dfbdd0912bb623e51403

Switch image attachments to persisted ID-based paths

- add attachment store helpers for IDs, route paths, and file resolution - stop projection pipeline from mutating/storing data URLs; keep attachment metadata unchanged - update Codex send/branch generation flows and tests to resolve images via attachment IDs

18 files changed, 499 insertions(+), 407 deletions(-)

apps/server/src/attachmentStore.ts created+95
...@@ -0,0 +1,95 @@
1import { randomUUID } from "node:crypto";
2
3import type { ChatAttachment } from "@t3tools/contracts";
4
5import {
6 ATTACHMENTS_ROUTE_PREFIX,
7 normalizeAttachmentRelativePath,
8 resolveAttachmentRelativePath,
9} from "./attachmentPaths.ts";
10import { inferImageExtension, SAFE_IMAGE_FILE_EXTENSIONS } from "./imageMime.ts";
11
12const ATTACHMENT_FILENAME_EXTENSIONS = [...SAFE_IMAGE_FILE_EXTENSIONS, ".bin"];
13
14function toSafeThreadAttachmentSegment(threadId: string): string | null {
15 const segment = encodeURIComponent(threadId);
16 if (
17 segment.length === 0 ||
18 segment === "." ||
19 segment === ".." ||
20 segment.includes("/") ||
21 segment.includes("\\") ||
22 segment.includes("\0")
23 ) {
24 return null;
25 }
26 return segment;
27}
28
29export function createAttachmentId(threadId: string): string | null {
30 const threadSegment = toSafeThreadAttachmentSegment(threadId);
31 if (!threadSegment) {
32 return null;
33 }
34 return `${threadSegment}-${randomUUID()}`;
35}
36
37export function attachmentRelativePath(attachment: ChatAttachment): string {
38 switch (attachment.type) {
39 case "image": {
40 const extension = inferImageExtension({
41 mimeType: attachment.mimeType,
42 fileName: attachment.name,
43 });
44 return `${attachment.id}${extension}`;
45 }
46 }
47}
48
49export function resolveAttachmentPath(input: {
50 readonly stateDir: string;
51 readonly attachment: ChatAttachment;
52}): string | null {
53 return resolveAttachmentRelativePath({
54 stateDir: input.stateDir,
55 relativePath: attachmentRelativePath(input.attachment),
56 });
57}
58
59export function attachmentRoutePath(attachment: ChatAttachment): string {
60 return `${ATTACHMENTS_ROUTE_PREFIX}/${encodeURIComponent(attachment.id)}`;
61}
62
63export function resolveAttachmentPathById(input: {
64 readonly stateDir: string;
65 readonly attachmentId: string;
66}): string | null {
67 const normalizedId = normalizeAttachmentRelativePath(input.attachmentId);
68 if (!normalizedId || normalizedId.includes("/") || normalizedId.includes(".")) {
69 return null;
70 }
71 for (const extension of ATTACHMENT_FILENAME_EXTENSIONS) {
72 const maybePath = resolveAttachmentRelativePath({
73 stateDir: input.stateDir,
74 relativePath: `${normalizedId}${extension}`,
75 });
76 if (maybePath) {
77 return maybePath;
78 }
79 }
80 return null;
81}
82
83export function parseAttachmentIdFromRelativePath(relativePath: string): string | null {
84 const normalized = normalizeAttachmentRelativePath(relativePath);
85 if (!normalized || normalized.includes("/")) {
86 return null;
87 }
88 const extensionIndex = normalized.lastIndexOf(".");
89 if (extensionIndex <= 0) {
90 return null;
91 }
92 const id = normalized.slice(0, extensionIndex);
93 return id.length > 0 && !id.includes(".") ? id : null;
94}
95
apps/server/src/codexAppServerManager.test.ts+2-8
...@@ -191,10 +191,7 @@ describe("sendTurn", () => {...@@ -191,10 +191,7 @@ describe("sendTurn", () => {
191 attachments: [191 attachments: [
192 {192 {
193 type: "image",193 type: "image",
194 name: "error.png",194 url: "data:image/png;base64,AAAA",
195 mimeType: "image/png",
196 sizeBytes: 1_024,
197 dataUrl: "data:image/png;base64,AAAA",
198 },195 },
199 ],196 ],
200 model: "gpt-5.3",197 model: "gpt-5.3",
...@@ -242,10 +239,7 @@ describe("sendTurn", () => {...@@ -242,10 +239,7 @@ describe("sendTurn", () => {
242 attachments: [239 attachments: [
243 {240 {
244 type: "image",241 type: "image",
245 name: "diagram.png",242 url: "data:image/png;base64,BBBB",
246 mimeType: "image/png",
247 sizeBytes: 256,
248 dataUrl: "data:image/png;base64,BBBB",
249 },243 },
250 ],244 ],
251 });245 });
apps/server/src/codexAppServerManager.ts+10-3
...@@ -14,7 +14,6 @@ import {...@@ -14,7 +14,6 @@ import {
14 normalizeModelSlug,14 normalizeModelSlug,
15 type ProviderApprovalDecision,15 type ProviderApprovalDecision,
16 type ProviderEvent,16 type ProviderEvent,
17 type ProviderSendTurnInput,
18 type ProviderSession,17 type ProviderSession,
19 type ProviderSessionStartInput,18 type ProviderSessionStartInput,
20 type ProviderTurnStartResult,19 type ProviderTurnStartResult,
...@@ -71,6 +70,14 @@ interface JsonRpcNotification {...@@ -71,6 +70,14 @@ interface JsonRpcNotification {
71 params?: unknown;70 params?: unknown;
72}71}
7372
73export interface CodexAppServerSendTurnInput {
74 readonly sessionId: ProviderSessionId;
75 readonly input?: string;
76 readonly attachments?: ReadonlyArray<{ type: "image"; url: string }>;
77 readonly model?: string;
78 readonly effort?: string;
79}
80
74export interface CodexThreadTurnSnapshot {81export interface CodexThreadTurnSnapshot {
75 id: ProviderTurnId;82 id: ProviderTurnId;
76 items: unknown[];83 items: unknown[];
...@@ -290,7 +297,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve...@@ -290,7 +297,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
290 }297 }
291 }298 }
292299
293 async sendTurn(input: ProviderSendTurnInput): Promise<ProviderTurnStartResult> {300 async sendTurn(input: CodexAppServerSendTurnInput): Promise<ProviderTurnStartResult> {
294 const context = this.requireSession(input.sessionId);301 const context = this.requireSession(input.sessionId);
295 if (!context.session.threadId) {302 if (!context.session.threadId) {
296 throw new Error("Session is missing a thread id.");303 throw new Error("Session is missing a thread id.");
...@@ -310,7 +317,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve...@@ -310,7 +317,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
310 if (attachment.type === "image") {317 if (attachment.type === "image") {
311 turnInput.push({318 turnInput.push({
312 type: "image",319 type: "image",
313 url: attachment.dataUrl,320 url: attachment.url,
314 });321 });
315 }322 }
316 }323 }
apps/server/src/git/Layers/CodexTextGeneration.test.ts+33-31
...@@ -294,28 +294,36 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -294,28 +294,36 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
294 stdinMustContain: "Attachment metadata:",294 stdinMustContain: "Attachment metadata:",
295 },295 },
296 Effect.gen(function* () {296 Effect.gen(function* () {
297 const textGeneration = yield* TextGeneration;297 const fs = yield* FileSystem.FileSystem;
298 const path = yield* Path.Path;
299 const attachmentId = `thread-branch-image-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
300 const attachmentPath = path.join(process.cwd(), "attachments", `${attachmentId}.png`);
301 yield* fs.makeDirectory(path.join(process.cwd(), "attachments"), { recursive: true });
302 yield* fs.writeFile(attachmentPath, Buffer.from("hello"));
298303
299 const generated = yield* textGeneration.generateBranchName({304 const textGeneration = yield* TextGeneration;
300 cwd: process.cwd(),305 const generated = yield* textGeneration
301 message: "Fix layout bug from screenshot.",306 .generateBranchName({
302 attachments: [307 cwd: process.cwd(),
303 {308 message: "Fix layout bug from screenshot.",
304 type: "image",309 attachments: [
305 name: "bug.png",310 {
306 mimeType: "image/png",311 type: "image",
307 sizeBytes: 5,312 id: attachmentId,
308 dataUrl: "data:image/png;base64,SGVsbG8=",313 name: "bug.png",
309 },314 mimeType: "image/png",
310 ],315 sizeBytes: 5,
311 });316 },
317 ],
318 })
319 .pipe(Effect.ensuring(fs.remove(attachmentPath).pipe(Effect.catch(() => Effect.void))));
312320
313 expect(generated.branch).toBe("fix/ui-regression");321 expect(generated.branch).toBe("fix/ui-regression");
314 }),322 }),
315 ),323 ),
316 );324 );
317325
318 it.effect("resolves persisted attachment URLs to files for codex image inputs", () =>326 it.effect("resolves persisted attachment ids to files for codex image inputs", () =>
319 withFakeCodexEnv(327 withFakeCodexEnv(
320 {328 {
321 output: JSON.stringify({329 output: JSON.stringify({
...@@ -326,12 +334,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -326,12 +334,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
326 Effect.gen(function* () {334 Effect.gen(function* () {
327 const fs = yield* FileSystem.FileSystem;335 const fs = yield* FileSystem.FileSystem;
328 const path = yield* Path.Path;336 const path = yield* Path.Path;
329 const uniqueSuffix = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;337 const attachmentId =
330 const imageFileName = `message-1-${uniqueSuffix}.png`;338 `thread-1-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
331 const imagePath = path.join(process.cwd(), "attachments", "thread-1", imageFileName);339 const imagePath = path.join(process.cwd(), "attachments", `${attachmentId}.png`);
332 yield* fs.makeDirectory(path.join(process.cwd(), "attachments", "thread-1"), {340 yield* fs.makeDirectory(path.join(process.cwd(), "attachments"), { recursive: true });
333 recursive: true,
334 });
335 yield* fs.writeFile(imagePath, Buffer.from("hello"));341 yield* fs.writeFile(imagePath, Buffer.from("hello"));
336342
337 const textGeneration = yield* TextGeneration;343 const textGeneration = yield* TextGeneration;
...@@ -342,10 +348,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -342,10 +348,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
342 attachments: [348 attachments: [
343 {349 {
344 type: "image",350 type: "image",
351 id: attachmentId,
345 name: "bug.png",352 name: "bug.png",
346 mimeType: "image/png",353 mimeType: "image/png",
347 sizeBytes: 5,354 sizeBytes: 5,
348 dataUrl: `/attachments/thread-1/${imageFileName}`,
349 },355 },
350 ],356 ],
351 })357 })
...@@ -357,8 +363,6 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -357,8 +363,6 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
357 }),363 }),
358 ),364 ),
359 ),365 ),
360 )
361 .pipe(
362 Effect.ensuring(366 Effect.ensuring(
363 fs.remove(imagePath).pipe(Effect.catch(() => Effect.void)),367 fs.remove(imagePath).pipe(Effect.catch(() => Effect.void)),
364 ),368 ),
...@@ -369,7 +373,7 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -369,7 +373,7 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
369 ),373 ),
370 );374 );
371375
372 it.effect("ignores absolute non-route attachment paths for codex image inputs", () =>376 it.effect("ignores missing attachment ids for codex image inputs", () =>
373 withFakeCodexEnv(377 withFakeCodexEnv(
374 {378 {
375 output: JSON.stringify({379 output: JSON.stringify({
...@@ -380,11 +384,9 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -380,11 +384,9 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
380 Effect.gen(function* () {384 Effect.gen(function* () {
381 const fs = yield* FileSystem.FileSystem;385 const fs = yield* FileSystem.FileSystem;
382 const path = yield* Path.Path;386 const path = yield* Path.Path;
383 const tempDir = yield* fs.makeTempDirectoryScoped({387 const missingAttachmentId = `thread-missing-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
384 prefix: "t3code-codex-attachment-bypass-",388 const missingPath = path.join(process.cwd(), "attachments", `${missingAttachmentId}.png`);
385 });389 yield* fs.remove(missingPath).pipe(Effect.catch(() => Effect.void));
386 const absoluteImagePath = path.join(tempDir, "outside.png");
387 yield* fs.writeFile(absoluteImagePath, Buffer.from("hello"));
388390
389 const textGeneration = yield* TextGeneration;391 const textGeneration = yield* TextGeneration;
390 const result = yield* textGeneration392 const result = yield* textGeneration
...@@ -394,10 +396,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {...@@ -394,10 +396,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => {
394 attachments: [396 attachments: [
395 {397 {
396 type: "image",398 type: "image",
399 id: missingAttachmentId,
397 name: "outside.png",400 name: "outside.png",
398 mimeType: "image/png",401 mimeType: "image/png",
399 sizeBytes: 5,402 sizeBytes: 5,
400 dataUrl: absoluteImagePath,
401 },403 },
402 ],404 ],
403 })405 })
apps/server/src/git/Layers/CodexTextGeneration.ts+14-75
...@@ -3,10 +3,9 @@ import { randomUUID } from "node:crypto";...@@ -3,10 +3,9 @@ import { randomUUID } from "node:crypto";
3import { Effect, FileSystem, Layer, Option, Path, Schema, Stream } from "effect";3import { Effect, FileSystem, Layer, Option, Path, Schema, Stream } from "effect";
4import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";4import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
55
6import { resolveAttachmentRoutePath } from "../../attachmentPaths.ts";6import { resolveAttachmentPath } from "../../attachmentStore.ts";
7import { ServerConfig } from "../../config.ts";7import { ServerConfig } from "../../config.ts";
8import { TextGenerationError } from "../Errors.ts";8import { TextGenerationError } from "../Errors.ts";
9import { inferImageExtension, parseBase64DataUrl } from "../../imageMime.ts";
10import {9import {
11 type BranchNameGenerationInput,10 type BranchNameGenerationInput,
12 type BranchNameGenerationResult,11 type BranchNameGenerationResult,
...@@ -109,7 +108,6 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -109,7 +108,6 @@ const makeCodexTextGeneration = Effect.gen(function* () {
109108
110 type MaterializedImageAttachments = {109 type MaterializedImageAttachments = {
111 readonly imagePaths: ReadonlyArray<string>;110 readonly imagePaths: ReadonlyArray<string>;
112 readonly tempImagePaths: ReadonlyArray<string>;
113 };111 };
114112
115 const readStreamAsString = <E>(113 const readStreamAsString = <E>(
...@@ -151,96 +149,41 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -151,96 +149,41 @@ const makeCodexTextGeneration = Effect.gen(function* () {
151 );149 );
152 };150 };
153151
154 const writeTempBinaryFile = (
155 operation: string,
156 prefix: string,
157 bytes: Uint8Array,
158 extension: string,
159 ): Effect.Effect<string, TextGenerationError> => {
160 const normalizedExtension = extension.startsWith(".") ? extension : `.${extension}`;
161 const filePath = path.join(
162 tempDir,
163 `t3code-${prefix}-${process.pid}-${randomUUID()}${normalizedExtension}`,
164 );
165 return fileSystem.writeFile(filePath, bytes).pipe(
166 Effect.mapError(
167 (cause) =>
168 new TextGenerationError({
169 operation,
170 detail: `Failed to write temp file at ${filePath}.`,
171 cause,
172 }),
173 ),
174 Effect.as(filePath),
175 );
176 };
177
178 const safeUnlink = (filePath: string): Effect.Effect<void, never> =>152 const safeUnlink = (filePath: string): Effect.Effect<void, never> =>
179 fileSystem.remove(filePath).pipe(Effect.catch(() => Effect.void));153 fileSystem.remove(filePath).pipe(Effect.catch(() => Effect.void));
180154
181 const materializeImageAttachments = (155 const materializeImageAttachments = (
182 operation: "generateCommitMessage" | "generatePrContent" | "generateBranchName",156 _operation: "generateCommitMessage" | "generatePrContent" | "generateBranchName",
183 attachments: BranchNameGenerationInput["attachments"],157 attachments: BranchNameGenerationInput["attachments"],
184 ): Effect.Effect<MaterializedImageAttachments, TextGenerationError> => {158 ): Effect.Effect<MaterializedImageAttachments, TextGenerationError> =>
185 const tempImagePaths: string[] = [];159 Effect.gen(function* () {
186
187 return Effect.gen(function* () {
188 if (!attachments || attachments.length === 0) {160 if (!attachments || attachments.length === 0) {
189 return { imagePaths: [], tempImagePaths };161 return { imagePaths: [] };
190 }162 }
191163
192 const imagePaths: string[] = [];164 const imagePaths: string[] = [];
193 for (const [index, attachment] of attachments.entries()) {165 for (const attachment of attachments) {
194 if (attachment.type !== "image") {166 if (attachment.type !== "image") {
195 continue;167 continue;
196 }168 }
197169
198 const parsed = parseBase64DataUrl(attachment.dataUrl);170 const resolvedPath = resolveAttachmentPath({
199 if (parsed && parsed.mimeType.startsWith("image/")) {
200 const bytes = Buffer.from(parsed.base64, "base64");
201 if (bytes.byteLength === 0) {
202 continue;
203 }
204
205 const extension = inferImageExtension({
206 mimeType: parsed.mimeType,
207 fileName: attachment.name,
208 });
209 const imagePath = yield* writeTempBinaryFile(
210 operation,
211 `codex-image-${index}`,
212 bytes,
213 extension,
214 );
215 imagePaths.push(imagePath);
216 tempImagePaths.push(imagePath);
217 continue;
218 }
219
220 const persistedPath = resolveAttachmentRoutePath({
221 stateDir: serverConfig.stateDir,171 stateDir: serverConfig.stateDir,
222 dataUrl: attachment.dataUrl,172 attachment,
223 });173 });
224 if (!persistedPath || !path.isAbsolute(persistedPath)) {174 if (!resolvedPath || !path.isAbsolute(resolvedPath)) {
225 continue;175 continue;
226 }176 }
227 const fileInfo = yield* fileSystem177 const fileInfo = yield* fileSystem
228 .stat(persistedPath)178 .stat(resolvedPath)
229 .pipe(Effect.catch(() => Effect.succeed(null)));179 .pipe(Effect.catch(() => Effect.succeed(null)));
230 if (!fileInfo || fileInfo.type !== "File") {180 if (!fileInfo || fileInfo.type !== "File") {
231 continue;181 continue;
232 }182 }
233 imagePaths.push(persistedPath);183 imagePaths.push(resolvedPath);
234 }184 }
235 return { imagePaths, tempImagePaths };185 return { imagePaths };
236 }).pipe(186 });
237 Effect.tapError(() =>
238 Effect.all(tempImagePaths.map((filePath) => safeUnlink(filePath)), {
239 concurrency: "unbounded",
240 }).pipe(Effect.asVoid),
241 ),
242 );
243 };
244187
245 const runCodexJson = <S extends Schema.Top>({188 const runCodexJson = <S extends Schema.Top>({
246 operation,189 operation,
...@@ -454,10 +397,7 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -454,10 +397,7 @@ const makeCodexTextGeneration = Effect.gen(function* () {
454397
455 const generateBranchName: TextGenerationShape["generateBranchName"] = (input) => {398 const generateBranchName: TextGenerationShape["generateBranchName"] = (input) => {
456 return Effect.gen(function* () {399 return Effect.gen(function* () {
457 const { imagePaths, tempImagePaths } = yield* materializeImageAttachments(400 const { imagePaths } = yield* materializeImageAttachments("generateBranchName", input.attachments);
458 "generateBranchName",
459 input.attachments,
460 );
461 const attachmentLines = (input.attachments ?? []).map(401 const attachmentLines = (input.attachments ?? []).map(
462 (attachment) =>402 (attachment) =>
463 `- ${attachment.name} (${attachment.mimeType}, ${attachment.sizeBytes} bytes)`,403 `- ${attachment.name} (${attachment.mimeType}, ${attachment.sizeBytes} bytes)`,
...@@ -492,7 +432,6 @@ const makeCodexTextGeneration = Effect.gen(function* () {...@@ -492,7 +432,6 @@ const makeCodexTextGeneration = Effect.gen(function* () {
492 branch: Schema.String,432 branch: Schema.String,
493 }),433 }),
494 imagePaths,434 imagePaths,
495 cleanupPaths: tempImagePaths,
496 });435 });
497436
498 return {437 return {
apps/server/src/main.test.ts+14-3
...@@ -49,10 +49,21 @@ const testLayer = Layer.mergeAll(...@@ -49,10 +49,21 @@ const testLayer = Layer.mergeAll(
49const runCli = (49const runCli = (
50 args: ReadonlyArray<string>,50 args: ReadonlyArray<string>,
51 env: Record<string, string> = { T3CODE_NO_BROWSER: "true" },51 env: Record<string, string> = { T3CODE_NO_BROWSER: "true" },
52) =>52) => {
53 Command.runWith(t3Cli, { version: "0.0.0-test" })(args).pipe(53 const uniqueStateDir = `/tmp/t3-cli-state-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
54 Effect.provide(ConfigProvider.layer(ConfigProvider.fromEnv({ env }))),54 return Command.runWith(t3Cli, { version: "0.0.0-test" })(args).pipe(
55 Effect.provide(
56 ConfigProvider.layer(
57 ConfigProvider.fromEnv({
58 env: {
59 T3CODE_STATE_DIR: uniqueStateDir,
60 ...env,
61 },
62 }),
63 ),
64 ),
55 );65 );
66};
5667
57beforeEach(() => {68beforeEach(() => {
58 vi.clearAllMocks();69 vi.clearAllMocks();
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+51-65
...@@ -178,7 +178,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -178,7 +178,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
178 }),178 }),
179 );179 );
180180
181 it.effect("materializes message image attachments into stateDir and stores URL references", () =>181 it.effect("stores message attachment references without mutating payloads", () =>
182 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(182 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(
183 Effect.flatMap((stateDir) =>183 Effect.flatMap((stateDir) =>
184 Effect.gen(function* () {184 Effect.gen(function* () {
...@@ -205,10 +205,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -205,10 +205,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
205 attachments: [205 attachments: [
206 {206 {
207 type: "image",207 type: "image",
208 id: "thread-attachments-att-1",
208 name: "example.png",209 name: "example.png",
209 mimeType: "image/png",210 mimeType: "image/png",
210 sizeBytes: 5,211 sizeBytes: 5,
211 dataUrl: "data:image/png;base64,SGVsbG8=",
212 },212 },
213 ],213 ],
214 turnId: null,214 turnId: null,
...@@ -232,21 +232,12 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -232,21 +232,12 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
232 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [232 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [
233 {233 {
234 type: "image",234 type: "image",
235 id: "thread-attachments-att-1",
235 name: "example.png",236 name: "example.png",
236 mimeType: "image/png",237 mimeType: "image/png",
237 sizeBytes: 5,238 sizeBytes: 5,
238 dataUrl: "/attachments/thread-attachments/message-attachments-0.png",
239 },239 },
240 ]);240 ]);
241
242 const attachmentPath = path.join(
243 stateDir,
244 "attachments",
245 "thread-attachments",
246 "message-attachments-0.png",
247 );
248 assert.equal(fs.existsSync(attachmentPath), true);
249 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("SGVsbG8=", "base64"));
250 }).pipe(241 }).pipe(
251 (effect) => runWithProjectionPipelineLayer(stateDir, effect),242 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
252 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),243 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
...@@ -255,7 +246,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -255,7 +246,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
255 ),246 ),
256 );247 );
257248
258 it.effect("materializes only image data URLs and uses a safe extension whitelist", () =>249 it.effect("preserves mixed image attachment metadata as-is", () =>
259 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(250 Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe(
260 Effect.flatMap((stateDir) =>251 Effect.flatMap((stateDir) =>
261 Effect.gen(function* () {252 Effect.gen(function* () {
...@@ -282,17 +273,17 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -282,17 +273,17 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
282 attachments: [273 attachments: [
283 {274 {
284 type: "image",275 type: "image",
276 id: "thread-attachments-safe-att-1",
285 name: "untrusted.exe",277 name: "untrusted.exe",
286 mimeType: "image/x-unknown",278 mimeType: "image/x-unknown",
287 sizeBytes: 5,279 sizeBytes: 5,
288 dataUrl: "data:image/x-unknown;base64,SGVsbG8=",
289 },280 },
290 {281 {
291 type: "image",282 type: "image",
283 id: "thread-attachments-safe-att-2",
292 name: "not-image.png",284 name: "not-image.png",
293 mimeType: "image/png",285 mimeType: "image/png",
294 sizeBytes: 5,286 sizeBytes: 5,
295 dataUrl: "data:text/plain;base64,SGVsbG8=",
296 },287 },
297 ],288 ],
298 turnId: null,289 turnId: null,
...@@ -316,36 +307,19 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -316,36 +307,19 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
316 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [307 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [
317 {308 {
318 type: "image",309 type: "image",
310 id: "thread-attachments-safe-att-1",
319 name: "untrusted.exe",311 name: "untrusted.exe",
320 mimeType: "image/x-unknown",312 mimeType: "image/x-unknown",
321 sizeBytes: 5,313 sizeBytes: 5,
322 dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin",
323 },314 },
324 {315 {
325 type: "image",316 type: "image",
317 id: "thread-attachments-safe-att-2",
326 name: "not-image.png",318 name: "not-image.png",
327 mimeType: "image/png",319 mimeType: "image/png",
328 sizeBytes: 5,320 sizeBytes: 5,
329 dataUrl: "data:text/plain;base64,SGVsbG8=",
330 },321 },
331 ]);322 ]);
332
333 const firstAttachmentPath = path.join(
334 stateDir,
335 "attachments",
336 "thread-attachments-safe",
337 "message-attachments-safe-0.bin",
338 );
339 assert.equal(fs.existsSync(firstAttachmentPath), true);
340 assert.deepEqual(fs.readFileSync(firstAttachmentPath), Buffer.from("SGVsbG8=", "base64"));
341
342 const secondAttachmentPath = path.join(
343 stateDir,
344 "attachments",
345 "thread-attachments-safe",
346 "message-attachments-safe-1.png",
347 );
348 assert.equal(fs.existsSync(secondAttachmentPath), false);
349 }).pipe(323 }).pipe(
350 (effect) => runWithProjectionPipelineLayer(stateDir, effect),324 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
351 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),325 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
...@@ -425,10 +399,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -425,10 +399,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
425 attachments: [399 attachments: [
426 {400 {
427 type: "image",401 type: "image",
402 id: "thread-clear-attachments-att-1",
428 name: "clear.png",403 name: "clear.png",
429 mimeType: "image/png",404 mimeType: "image/png",
430 sizeBytes: 5,405 sizeBytes: 5,
431 dataUrl: "data:image/png;base64,SGVsbG8=",
432 },406 },
433 ],407 ],
434 turnId: null,408 turnId: null,
...@@ -477,17 +451,18 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -477,17 +451,18 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
477 );451 );
478452
479 it.effect(453 it.effect(
480 "overwrites attachment file bytes when a message updates the same attachment index",454 "overwrites stored attachment references when a message updates attachments",
481 () =>455 () =>
482 Effect.sync(() =>456 Effect.sync(() =>
483 fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-")),457 fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-")),
484 ).pipe(458 ).pipe(
485 Effect.flatMap((stateDir) =>459 Effect.flatMap((stateDir) =>
486 Effect.gen(function* () {460 Effect.gen(function* () {
487 const projectionPipeline = yield* OrchestrationProjectionPipeline;461 const projectionPipeline = yield* OrchestrationProjectionPipeline;
488 const eventStore = yield* OrchestrationEventStore;462 const eventStore = yield* OrchestrationEventStore;
489 const now = new Date().toISOString();463 const sql = yield* SqlClient.SqlClient;
490 const later = new Date(Date.now() + 1_000).toISOString();464 const now = new Date().toISOString();
465 const later = new Date(Date.now() + 1_000).toISOString();
491466
492 yield* eventStore.append({467 yield* eventStore.append({
493 type: "project.created",468 type: "project.created",
...@@ -550,10 +525,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -550,10 +525,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
550 attachments: [525 attachments: [
551 {526 {
552 type: "image",527 type: "image",
528 id: "thread-overwrite-att-1",
553 name: "file.png",529 name: "file.png",
554 mimeType: "image/png",530 mimeType: "image/png",
555 sizeBytes: 5,531 sizeBytes: 5,
556 dataUrl: "data:image/png;base64,SGVsbG8=",
557 },532 },
558 ],533 ],
559 turnId: null,534 turnId: null,
...@@ -581,10 +556,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -581,10 +556,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
581 attachments: [556 attachments: [
582 {557 {
583 type: "image",558 type: "image",
559 id: "thread-overwrite-att-2",
584 name: "file.png",560 name: "file.png",
585 mimeType: "image/png",561 mimeType: "image/png",
586 sizeBytes: 5,562 sizeBytes: 5,
587 dataUrl: "data:image/png;base64,V29ybGQ=",
588 },563 },
589 ],564 ],
590 turnId: null,565 turnId: null,
...@@ -596,14 +571,23 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -596,14 +571,23 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
596571
597 yield* projectionPipeline.bootstrap;572 yield* projectionPipeline.bootstrap;
598573
599 const attachmentPath = path.join(574 const rows = yield* sql<{
600 stateDir,575 readonly attachmentsJson: string | null;
601 "attachments",576 }>`
602 "thread-overwrite",577 SELECT attachments_json AS "attachmentsJson"
603 "message-overwrite-0.png",578 FROM projection_thread_messages
604 );579 WHERE message_id = 'message-overwrite'
605 assert.equal(fs.existsSync(attachmentPath), true);580 `;
606 assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("V29ybGQ=", "base64"));581 assert.equal(rows.length, 1);
582 assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [
583 {
584 type: "image",
585 id: "thread-overwrite-att-2",
586 name: "file.png",
587 mimeType: "image/png",
588 sizeBytes: 5,
589 },
590 ]);
607 }).pipe(591 }).pipe(
608 (effect) => runWithProjectionPipelineLayer(stateDir, effect),592 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
609 Effect.ensuring(593 Effect.ensuring(
...@@ -701,10 +685,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -701,10 +685,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
701 attachments: [685 attachments: [
702 {686 {
703 type: "image",687 type: "image",
688 id: "thread-rollback-att-1",
704 name: "rollback.png",689 name: "rollback.png",
705 mimeType: "image/png",690 mimeType: "image/png",
706 sizeBytes: 5,691 sizeBytes: 5,
707 dataUrl: "data:image/png;base64,SGVsbG8=",
708 },692 },
709 ],693 ],
710 turnId: null,694 turnId: null,
...@@ -728,8 +712,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -728,8 +712,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
728 const attachmentPath = path.join(712 const attachmentPath = path.join(
729 stateDir,713 stateDir,
730 "attachments",714 "attachments",
731 "thread-rollback",715 "thread-rollback-att-1.png",
732 "message-rollback-0.png",
733 );716 );
734 assert.equal(fs.existsSync(attachmentPath), false);717 assert.equal(fs.existsSync(attachmentPath), false);
735 yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`;718 yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`;
...@@ -839,10 +822,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -839,10 +822,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
839 attachments: [822 attachments: [
840 {823 {
841 type: "image",824 type: "image",
825 id: "thread-revert-files-keep",
842 name: "keep.png",826 name: "keep.png",
843 mimeType: "image/png",827 mimeType: "image/png",
844 sizeBytes: 5,828 sizeBytes: 5,
845 dataUrl: "data:image/png;base64,SGVsbG8=",
846 },829 },
847 ],830 ],
848 turnId: TurnId.makeUnsafe("turn-keep"),831 turnId: TurnId.makeUnsafe("turn-keep"),
...@@ -892,10 +875,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -892,10 +875,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
892 attachments: [875 attachments: [
893 {876 {
894 type: "image",877 type: "image",
878 id: "thread-revert-files-remove",
895 name: "remove.png",879 name: "remove.png",
896 mimeType: "image/png",880 mimeType: "image/png",
897 sizeBytes: 5,881 sizeBytes: 5,
898 dataUrl: "data:image/png;base64,V29ybGQ=",
899 },882 },
900 ],883 ],
901 turnId: TurnId.makeUnsafe("turn-remove"),884 turnId: TurnId.makeUnsafe("turn-remove"),
...@@ -908,15 +891,16 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -908,15 +891,16 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
908 const keepPath = path.join(891 const keepPath = path.join(
909 stateDir,892 stateDir,
910 "attachments",893 "attachments",
911 "thread-revert-files",894 "thread-revert-files-keep.png",
912 "message-keep-0.png",
913 );895 );
914 const removePath = path.join(896 const removePath = path.join(
915 stateDir,897 stateDir,
916 "attachments",898 "attachments",
917 "thread-revert-files",899 "thread-revert-files-remove.png",
918 "message-remove-0.png",
919 );900 );
901 fs.mkdirSync(path.join(stateDir, "attachments"), { recursive: true });
902 fs.writeFileSync(keepPath, Buffer.from("keep"));
903 fs.writeFileSync(removePath, Buffer.from("remove"));
920 assert.equal(fs.existsSync(keepPath), true);904 assert.equal(fs.existsSync(keepPath), true);
921 assert.equal(fs.existsSync(removePath), true);905 assert.equal(fs.existsSync(removePath), true);
922906
...@@ -1022,10 +1006,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -1022,10 +1006,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
1022 attachments: [1006 attachments: [
1023 {1007 {
1024 type: "image",1008 type: "image",
1009 id: "thread-delete-files-att-1",
1025 name: "delete.png",1010 name: "delete.png",
1026 mimeType: "image/png",1011 mimeType: "image/png",
1027 sizeBytes: 5,1012 sizeBytes: 5,
1028 dataUrl: "data:image/png;base64,SGVsbG8=",
1029 },1013 },
1030 ],1014 ],
1031 turnId: null,1015 turnId: null,
...@@ -1035,8 +1019,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -1035,8 +1019,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
1035 },1019 },
1036 });1020 });
10371021
1038 const threadAttachmentDir = path.join(stateDir, "attachments", "thread-delete-files");1022 const threadAttachmentPath = path.join(stateDir, "attachments", "thread-delete-files-att-1.png");
1039 assert.equal(fs.existsSync(threadAttachmentDir), true);1023 fs.mkdirSync(path.join(stateDir, "attachments"), { recursive: true });
1024 fs.writeFileSync(threadAttachmentPath, Buffer.from("delete"));
1025 assert.equal(fs.existsSync(threadAttachmentPath), true);
10401026
1041 yield* appendAndProject({1027 yield* appendAndProject({
1042 type: "thread.deleted",1028 type: "thread.deleted",
...@@ -1054,7 +1040,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {...@@ -1054,7 +1040,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => {
1054 },1040 },
1055 });1041 });
10561042
1057 assert.equal(fs.existsSync(threadAttachmentDir), false);1043 assert.equal(fs.existsSync(threadAttachmentPath), false);
1058 }).pipe(1044 }).pipe(
1059 (effect) => runWithProjectionPipelineLayer(stateDir, effect),1045 (effect) => runWithProjectionPipelineLayer(stateDir, effect),
1060 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),1046 Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))),
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+33-103
...@@ -37,8 +37,7 @@ import {...@@ -37,8 +37,7 @@ import {
37 OrchestrationProjectionPipeline,37 OrchestrationProjectionPipeline,
38 type OrchestrationProjectionPipelineShape,38 type OrchestrationProjectionPipelineShape,
39} from "../Services/ProjectionPipeline.ts";39} from "../Services/ProjectionPipeline.ts";
40import { ATTACHMENTS_ROUTE_PREFIX, attachmentRouteToRelativePath } from "../../attachmentPaths.ts";40import { attachmentRelativePath } from "../../attachmentStore.ts";
41import { inferImageExtension, parseBase64DataUrl } from "../../imageMime.ts";
4241
43export const ORCHESTRATION_PROJECTOR_NAMES = {42export const ORCHESTRATION_PROJECTOR_NAMES = {
44 projects: "projection.projects",43 projects: "projection.projects",
...@@ -62,13 +61,7 @@ interface ProjectorDefinition {...@@ -62,13 +61,7 @@ interface ProjectorDefinition {
62 ) => Effect.Effect<void, ProjectionRepositoryError>;61 ) => Effect.Effect<void, ProjectionRepositoryError>;
63}62}
6463
65interface PendingAttachmentWrite {
66 readonly absolutePath: string;
67 readonly bytes: Uint8Array;
68}
69
70interface AttachmentSideEffects {64interface AttachmentSideEffects {
71 readonly writes: Array<PendingAttachmentWrite>;
72 readonly deletedThreadIds: Set<string>;65 readonly deletedThreadIds: Set<string>;
73 readonly prunedThreadRelativePaths: Map<string, Set<string>>;66 readonly prunedThreadRelativePaths: Map<string, Set<string>>;
74}67}
...@@ -89,62 +82,10 @@ function toSafeThreadAttachmentSegment(threadId: string): string | null {...@@ -89,62 +82,10 @@ function toSafeThreadAttachmentSegment(threadId: string): string | null {
89}82}
9083
91const materializeAttachmentsForProjection = Effect.fn(function* (input: {84const materializeAttachmentsForProjection = Effect.fn(function* (input: {
92 readonly threadId: string;
93 readonly messageId: string;
94 readonly attachments: ReadonlyArray<ChatAttachment>;85 readonly attachments: ReadonlyArray<ChatAttachment>;
95 readonly stageFileWrite: (pendingWrite: PendingAttachmentWrite) => void;
96}) {86}) {
97 if (input.attachments.length === 0) return [];87 if (input.attachments.length === 0) return [];
9888 return input.attachments;
99 const serverConfig = yield* Effect.service(ServerConfig);
100
101 const path = yield* Path.Path;
102 const attachmentsRootDir = path.join(serverConfig.stateDir, "attachments");
103 const threadSegment = toSafeThreadAttachmentSegment(input.threadId);
104 if (!threadSegment) {
105 yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", {
106 threadId: input.threadId,
107 messageId: input.messageId,
108 });
109 return input.attachments;
110 }
111 const messageSegment = encodeURIComponent(input.messageId);
112
113 return yield* Effect.forEach(
114 input.attachments,
115 (attachment, index) =>
116 Effect.sync(() => {
117 if (attachment.type !== "image" || !attachment.dataUrl.startsWith("data:")) {
118 return attachment;
119 }
120
121 const parsed = parseBase64DataUrl(attachment.dataUrl);
122 if (!parsed) {
123 return attachment;
124 }
125 if (!parsed.mimeType.startsWith("image/")) {
126 return attachment;
127 }
128
129 const bytes = Buffer.from(parsed.base64, "base64");
130 if (bytes.byteLength === 0) return attachment;
131
132 const fileName = `${index}${inferImageExtension({
133 mimeType: parsed.mimeType,
134 fileName: attachment.name,
135 })}`;
136 const uniqueFileName = `${messageSegment}-${fileName}`;
137 const relativePath = `${threadSegment}/${uniqueFileName}`;
138 const absolutePath = path.join(attachmentsRootDir, threadSegment, uniqueFileName);
139 input.stageFileWrite({ absolutePath, bytes });
140
141 return {
142 ...attachment,
143 dataUrl: `${ATTACHMENTS_ROUTE_PREFIX}/${relativePath}`,
144 } satisfies ChatAttachment;
145 }),
146 { concurrency: 1 },
147 );
148});89});
14990
150function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {91function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
...@@ -267,22 +208,16 @@ function collectThreadAttachmentRelativePaths(...@@ -267,22 +208,16 @@ function collectThreadAttachmentRelativePaths(
267 if (!threadSegment) {208 if (!threadSegment) {
268 return new Set();209 return new Set();
269 }210 }
270 const threadPrefix = `${threadSegment}/`;
271 const relativePaths = new Set<string>();211 const relativePaths = new Set<string>();
272 for (const message of messages) {212 for (const message of messages) {
273 for (const attachment of message.attachments ?? []) {213 for (const attachment of message.attachments ?? []) {
274 if (attachment.type !== "image") {214 if (attachment.type !== "image") {
275 continue;215 continue;
276 }216 }
277 const relativePath = attachmentRouteToRelativePath(attachment.dataUrl);217 if (!attachment.id.startsWith(`${threadSegment}-`)) {
278 if (!relativePath || !relativePath.startsWith(threadPrefix)) {
279 continue;218 continue;
280 }219 }
281 const threadRelativePath = relativePath.slice(threadPrefix.length).replace(/^[/\\]+/, "");220 relativePaths.add(attachmentRelativePath(attachment));
282 if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) {
283 continue;
284 }
285 relativePaths.add(threadRelativePath.replace(/\\/g, "/"));
286 }221 }
287 }222 }
288 return relativePaths;223 return relativePaths;
...@@ -306,8 +241,26 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid...@@ -306,8 +241,26 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid
306 });241 });
307 return;242 return;
308 }243 }
309 const threadDir = path.join(attachmentsRootDir, threadSegment);244 const entries = yield* fileSystem
310 yield* fileSystem.remove(threadDir, { recursive: true, force: true });245 .readDirectory(attachmentsRootDir, { recursive: false })
246 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));
247 yield* Effect.forEach(
248 entries,
249 (entry) =>
250 Effect.gen(function* () {
251 const normalizedEntry = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/");
252 if (normalizedEntry.length === 0 || normalizedEntry.includes("/")) {
253 return;
254 }
255 if (!normalizedEntry.startsWith(`${threadSegment}-`)) {
256 return;
257 }
258 yield* fileSystem.remove(path.join(attachmentsRootDir, normalizedEntry), {
259 force: true,
260 });
261 }),
262 { concurrency: 1 },
263 );
311 }),264 }),
312 { concurrency: 1 },265 { concurrency: 1 },
313 );266 );
...@@ -324,20 +277,22 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid...@@ -324,20 +277,22 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid
324 yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });277 yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });
325 return;278 return;
326 }279 }
327 const threadDir = path.join(attachmentsRootDir, threadSegment);
328 const entries = yield* fileSystem280 const entries = yield* fileSystem
329 .readDirectory(threadDir, { recursive: true })281 .readDirectory(attachmentsRootDir, { recursive: false })
330 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));282 .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));
331 yield* Effect.forEach(283 yield* Effect.forEach(
332 entries,284 entries,
333 (entry) =>285 (entry) =>
334 Effect.gen(function* () {286 Effect.gen(function* () {
335 const threadRelativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/");287 const relativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/");
336 if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) {288 if (relativePath.length === 0 || relativePath.includes("/")) {
289 return;
290 }
291 if (!relativePath.startsWith(`${threadSegment}-`)) {
337 return;292 return;
338 }293 }
339294
340 const absolutePath = path.join(threadDir, threadRelativePath);295 const absolutePath = path.join(attachmentsRootDir, relativePath);
341 const fileInfo = yield* fileSystem296 const fileInfo = yield* fileSystem
342 .stat(absolutePath)297 .stat(absolutePath)
343 .pipe(Effect.catch(() => Effect.succeed(null)));298 .pipe(Effect.catch(() => Effect.succeed(null)));
...@@ -345,7 +300,7 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid...@@ -345,7 +300,7 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid
345 return;300 return;
346 }301 }
347302
348 if (!keptThreadRelativePaths.has(threadRelativePath)) {303 if (!keptThreadRelativePaths.has(relativePath)) {
349 yield* fileSystem.remove(absolutePath, { force: true });304 yield* fileSystem.remove(absolutePath, { force: true });
350 }305 }
351 }),306 }),
...@@ -356,22 +311,6 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid...@@ -356,22 +311,6 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid
356 { concurrency: 1 },311 { concurrency: 1 },
357 );312 );
358313
359 const writesByPath = new Map<string, Uint8Array>();
360 for (const pendingWrite of sideEffects.writes) {
361 // Last write wins for a path so updated attachment contents replace stale bytes.
362 writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes);
363 }
364 yield* Effect.forEach(
365 writesByPath.entries(),
366 ([absolutePath, bytes]) =>
367 Effect.gen(function* () {
368 yield* fileSystem.makeDirectory(path.dirname(absolutePath), {
369 recursive: true,
370 });
371 yield* fileSystem.writeFile(absolutePath, bytes);
372 }),
373 { concurrency: 1 },
374 );
375});314});
376315
377const makeOrchestrationProjectionPipeline = Effect.gen(function* () {316const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
...@@ -589,16 +528,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -589,16 +528,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
589 const nextAttachments =528 const nextAttachments =
590 event.payload.attachments !== undefined529 event.payload.attachments !== undefined
591 ? yield* materializeAttachmentsForProjection({530 ? yield* materializeAttachmentsForProjection({
592 threadId: event.payload.threadId,
593 messageId: event.payload.messageId,
594 attachments: event.payload.attachments,531 attachments: event.payload.attachments,
595 stageFileWrite: (pendingWrite) => {532 })
596 attachmentSideEffects.writes.push(pendingWrite);
597 },
598 }).pipe(
599 Effect.provideService(ServerConfig, serverConfig),
600 Effect.provideService(Path.Path, path),
601 )
602 : existingMessage?.attachments;533 : existingMessage?.attachments;
603 yield* projectionThreadMessageRepository.upsert({534 yield* projectionThreadMessageRepository.upsert({
604 messageId: event.payload.messageId,535 messageId: event.payload.messageId,
...@@ -1089,7 +1020,6 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -1089,7 +1020,6 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
1089 const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) =>1020 const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) =>
1090 Effect.gen(function* () {1021 Effect.gen(function* () {
1091 const attachmentSideEffects: AttachmentSideEffects = {1022 const attachmentSideEffects: AttachmentSideEffects = {
1092 writes: [],
1093 deletedThreadIds: new Set<string>(),1023 deletedThreadIds: new Set<string>(),
1094 prunedThreadRelativePaths: new Map<string, Set<string>>(),1024 prunedThreadRelativePaths: new Map<string, Set<string>>(),
1095 };1025 };
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+7-21
...@@ -258,10 +258,10 @@ describe("ProviderCommandReactor", () => {...@@ -258,10 +258,10 @@ describe("ProviderCommandReactor", () => {
258 attachments: [258 attachments: [
259 {259 {
260 type: "image",260 type: "image",
261 id: "thread-1-att-rename",
261 name: "bug.png",262 name: "bug.png",
262 mimeType: "image/png",263 mimeType: "image/png",
263 sizeBytes: 5,264 sizeBytes: 5,
264 dataUrl: "data:image/png;base64,SGVsbG8=",
265 },265 },
266 ],266 ],
267 },267 },
...@@ -280,15 +280,10 @@ describe("ProviderCommandReactor", () => {...@@ -280,15 +280,10 @@ describe("ProviderCommandReactor", () => {
280 attachments: [280 attachments: [
281 {281 {
282 type: "image",282 type: "image",
283 id: "thread-1-att-rename",
283 name: "bug.png",284 name: "bug.png",
284 mimeType: "image/png",285 mimeType: "image/png",
285 sizeBytes: 5,286 sizeBytes: 5,
286 dataUrl: path.join(
287 harness.stateDir,
288 "attachments",
289 "thread-1",
290 "user-message-worktree-rename-0.png",
291 ),
292 },287 },
293 ],288 ],
294 });289 });
...@@ -307,9 +302,8 @@ describe("ProviderCommandReactor", () => {...@@ -307,9 +302,8 @@ describe("ProviderCommandReactor", () => {
307 });302 });
308 });303 });
309304
310 it("reuses persisted attachment files for branch generation and turn start", async () => {305 it("passes persisted attachment references to branch generation and turn start", async () => {
311 const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-attachments-"));306 const harness = await createHarness();
312 const harness = await createHarness({ stateDir });
313 const now = new Date().toISOString();307 const now = new Date().toISOString();
314308
315 await Effect.runPromise(309 await Effect.runPromise(
...@@ -334,10 +328,10 @@ describe("ProviderCommandReactor", () => {...@@ -334,10 +328,10 @@ describe("ProviderCommandReactor", () => {
334 attachments: [328 attachments: [
335 {329 {
336 type: "image",330 type: "image",
331 id: "thread-1-att-persisted",
337 name: "bug.png",332 name: "bug.png",
338 mimeType: "image/png",333 mimeType: "image/png",
339 sizeBytes: 5,334 sizeBytes: 5,
340 dataUrl: "data:image/png;base64,SGVsbG8=",
341 },335 },
342 ],336 ],
343 },337 },
...@@ -350,23 +344,16 @@ describe("ProviderCommandReactor", () => {...@@ -350,23 +344,16 @@ describe("ProviderCommandReactor", () => {
350 await waitFor(() => harness.generateBranchName.mock.calls.length === 1);344 await waitFor(() => harness.generateBranchName.mock.calls.length === 1);
351 await waitFor(() => harness.sendTurn.mock.calls.length === 1);345 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
352346
353 const expectedAttachmentPath = path.join(
354 stateDir,
355 "attachments",
356 "thread-1",
357 "user-message-persisted-attachments-0.png",
358 );
359
360 expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({347 expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({
361 cwd: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12",348 cwd: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12",
362 message: "Fix visual bug from screenshot",349 message: "Fix visual bug from screenshot",
363 attachments: [350 attachments: [
364 {351 {
365 type: "image",352 type: "image",
353 id: "thread-1-att-persisted",
366 name: "bug.png",354 name: "bug.png",
367 mimeType: "image/png",355 mimeType: "image/png",
368 sizeBytes: 5,356 sizeBytes: 5,
369 dataUrl: expectedAttachmentPath,
370 },357 },
371 ],358 ],
372 });359 });
...@@ -374,14 +361,13 @@ describe("ProviderCommandReactor", () => {...@@ -374,14 +361,13 @@ describe("ProviderCommandReactor", () => {
374 attachments: [361 attachments: [
375 {362 {
376 type: "image",363 type: "image",
364 id: "thread-1-att-persisted",
377 name: "bug.png",365 name: "bug.png",
378 mimeType: "image/png",366 mimeType: "image/png",
379 sizeBytes: 5,367 sizeBytes: 5,
380 dataUrl: "data:image/png;base64,SGVsbG8=",
381 },368 },
382 ],369 ],
383 });370 });
384 expect(fs.existsSync(expectedAttachmentPath)).toBe(true);
385 });371 });
386372
387 it("skips worktree branch generation after the first user turn", async () => {373 it("skips worktree branch generation after the first user turn", async () => {
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+2-71
...@@ -12,17 +12,11 @@ import {...@@ -12,17 +12,11 @@ import {
12 type ProviderThreadId,12 type ProviderThreadId,
13 type TurnId,13 type TurnId,
14} from "@t3tools/contracts";14} from "@t3tools/contracts";
15import { Cache, Cause, Duration, Effect, FileSystem, Layer, Option, Queue, Stream } from "effect";15import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect";
1616
17import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts";17import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts";
18import {
19 resolveAttachmentRelativePath,
20 resolveAttachmentRoutePath,
21} from "../../attachmentPaths.ts";
22import { ServerConfig } from "../../config.ts";
23import { GitCore } from "../../git/Services/GitCore.ts";18import { GitCore } from "../../git/Services/GitCore.ts";
24import { TextGeneration } from "../../git/Services/TextGeneration.ts";19import { TextGeneration } from "../../git/Services/TextGeneration.ts";
25import { inferImageExtension, parseBase64DataUrl } from "../../imageMime.ts";
26import { ProviderService } from "../../provider/Services/ProviderService.ts";20import { ProviderService } from "../../provider/Services/ProviderService.ts";
27import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";21import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
28import {22import {
...@@ -109,8 +103,6 @@ const make = Effect.gen(function* () {...@@ -109,8 +103,6 @@ const make = Effect.gen(function* () {
109 const providerService = yield* ProviderService;103 const providerService = yield* ProviderService;
110 const git = yield* GitCore;104 const git = yield* GitCore;
111 const textGeneration = yield* TextGeneration;105 const textGeneration = yield* TextGeneration;
112 const fileSystem = yield* FileSystem.FileSystem;
113 const serverConfig = yield* Effect.service(ServerConfig);
114 const handledTurnStartKeys = yield* Cache.make<string, true>({106 const handledTurnStartKeys = yield* Cache.make<string, true>({
115 capacity: HANDLED_TURN_START_KEY_MAX,107 capacity: HANDLED_TURN_START_KEY_MAX,
116 timeToLive: HANDLED_TURN_START_KEY_TTL,108 timeToLive: HANDLED_TURN_START_KEY_TTL,
...@@ -291,61 +283,6 @@ const make = Effect.gen(function* () {...@@ -291,61 +283,6 @@ const make = Effect.gen(function* () {
291 });283 });
292 });284 });
293285
294 const resolvePersistedAttachments = Effect.fnUntraced(function* (input: {
295 readonly threadId: ThreadId;
296 readonly messageId: string;
297 readonly attachments: ReadonlyArray<ChatAttachment> | undefined;
298 }) {
299 if (!input.attachments || input.attachments.length === 0) {
300 return input.attachments ?? [];
301 }
302
303 const threadSegment = encodeURIComponent(input.threadId);
304 const messageSegment = encodeURIComponent(input.messageId);
305
306 return yield* Effect.forEach(Array.from(input.attachments.entries()), ([index, attachment]) =>
307 Effect.gen(function* () {
308 if (attachment.type !== "image") return attachment;
309
310 const resolvedRoutePath = resolveAttachmentRoutePath({
311 stateDir: serverConfig.stateDir,
312 dataUrl: attachment.dataUrl,
313 });
314 const resolvedMaterializedPath =
315 resolvedRoutePath ??
316 (() => {
317 const parsed = parseBase64DataUrl(attachment.dataUrl);
318 if (!parsed || !parsed.mimeType.startsWith("image/")) {
319 return null;
320 }
321 const extension = inferImageExtension({
322 mimeType: parsed.mimeType,
323 fileName: attachment.name,
324 });
325 return resolveAttachmentRelativePath({
326 stateDir: serverConfig.stateDir,
327 relativePath: `${threadSegment}/${messageSegment}-${index}${extension}`,
328 });
329 })();
330
331 if (!resolvedMaterializedPath) {
332 return attachment;
333 }
334
335 const fileInfo = yield* fileSystem
336 .stat(resolvedMaterializedPath)
337 .pipe(Effect.catch(() => Effect.succeed(null)));
338 if (!fileInfo || fileInfo.type !== "File") {
339 return attachment;
340 }
341 return {
342 ...attachment,
343 dataUrl: resolvedMaterializedPath,
344 } satisfies ChatAttachment;
345 }),
346 );
347 });
348
349 const maybeGenerateAndRenameWorktreeBranchForFirstTurn = Effect.fnUntraced(function* (input: {286 const maybeGenerateAndRenameWorktreeBranchForFirstTurn = Effect.fnUntraced(function* (input: {
350 readonly threadId: ThreadId;287 readonly threadId: ThreadId;
351 readonly branch: string | null;288 readonly branch: string | null;
...@@ -440,19 +377,13 @@ const make = Effect.gen(function* () {...@@ -440,19 +377,13 @@ const make = Effect.gen(function* () {
440 return;377 return;
441 }378 }
442379
443 const branchGenerationAttachments = yield* resolvePersistedAttachments({
444 threadId: event.payload.threadId,
445 messageId: message.id,
446 attachments: message.attachments,
447 });
448
449 yield* maybeGenerateAndRenameWorktreeBranchForFirstTurn({380 yield* maybeGenerateAndRenameWorktreeBranchForFirstTurn({
450 threadId: event.payload.threadId,381 threadId: event.payload.threadId,
451 branch: thread.branch,382 branch: thread.branch,
452 worktreePath: thread.worktreePath,383 worktreePath: thread.worktreePath,
453 messageId: message.id,384 messageId: message.id,
454 messageText: message.text,385 messageText: message.text,
455 ...(message.attachments !== undefined ? { attachments: branchGenerationAttachments } : {}),386 ...(message.attachments !== undefined ? { attachments: message.attachments } : {}),
456 }).pipe(Effect.forkScoped);387 }).pipe(Effect.forkScoped);
457388
458 yield* sendTurnForThread({389 yield* sendTurnForThread({
apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts+2-2
...@@ -21,10 +21,10 @@ layer("ProjectionThreadMessageRepository", (it) => {...@@ -21,10 +21,10 @@ layer("ProjectionThreadMessageRepository", (it) => {
21 const persistedAttachments = [21 const persistedAttachments = [
22 {22 {
23 type: "image" as const,23 type: "image" as const,
24 id: "thread-preserve-attachments-att-1",
24 name: "example.png",25 name: "example.png",
25 mimeType: "image/png",26 mimeType: "image/png",
26 sizeBytes: 5,27 sizeBytes: 5,
27 dataUrl: "/attachments/thread-preserve-attachments/message-preserve-attachments/0.png",
28 },28 },
29 ];29 ];
3030
...@@ -74,10 +74,10 @@ layer("ProjectionThreadMessageRepository", (it) => {...@@ -74,10 +74,10 @@ layer("ProjectionThreadMessageRepository", (it) => {
74 attachments: [74 attachments: [
75 {75 {
76 type: "image",76 type: "image",
77 id: "thread-clear-attachments-att-1",
77 name: "example.png",78 name: "example.png",
78 mimeType: "image/png",79 mimeType: "image/png",
79 sizeBytes: 5,80 sizeBytes: 5,
80 dataUrl: "/attachments/thread-clear-attachments/message-clear-attachments/0.png",
81 },81 },
82 ],82 ],
83 isStreaming: false,83 isStreaming: false,
apps/server/src/provider/Layers/CodexAdapter.test.ts+27-8
...@@ -4,7 +4,6 @@ import {...@@ -4,7 +4,6 @@ import {
4 ProviderItemId,4 ProviderItemId,
5 type ProviderApprovalDecision,5 type ProviderApprovalDecision,
6 type ProviderEvent,6 type ProviderEvent,
7 type ProviderSendTurnInput,
8 ProviderSessionId,7 ProviderSessionId,
9 type ProviderSession,8 type ProviderSession,
10 type ProviderSessionStartInput,9 type ProviderSessionStartInput,
...@@ -12,12 +11,17 @@ import {...@@ -12,12 +11,17 @@ import {
12 ProviderTurnId,11 ProviderTurnId,
13 type ProviderTurnStartResult,12 type ProviderTurnStartResult,
14} from "@t3tools/contracts";13} from "@t3tools/contracts";
14import * as NodeServices from "@effect/platform-node/NodeServices";
15import { afterAll, assert, it, vi } from "@effect/vitest";15import { afterAll, assert, it, vi } from "@effect/vitest";
16import { assertFailure } from "@effect/vitest/utils";16import { assertFailure } from "@effect/vitest/utils";
1717
18import { Effect, Fiber, Stream } from "effect";18import { Effect, Fiber, Layer, Stream } from "effect";
1919
20import { CodexAppServerManager } from "../../codexAppServerManager.ts";20import {
21 CodexAppServerManager,
22 type CodexAppServerSendTurnInput,
23} from "../../codexAppServerManager.ts";
24import { ServerConfig } from "../../config.ts";
21import { ProviderAdapterValidationError } from "../Errors.ts";25import { ProviderAdapterValidationError } from "../Errors.ts";
22import { CodexAdapter } from "../Services/CodexAdapter.ts";26import { CodexAdapter } from "../Services/CodexAdapter.ts";
23import { makeCodexAdapterLive } from "./CodexAdapter.ts";27import { makeCodexAdapterLive } from "./CodexAdapter.ts";
...@@ -44,7 +48,7 @@ class FakeCodexManager extends CodexAppServerManager {...@@ -44,7 +48,7 @@ class FakeCodexManager extends CodexAppServerManager {
44 );48 );
4549
46 public sendTurnImpl = vi.fn(50 public sendTurnImpl = vi.fn(
47 async (_input: ProviderSendTurnInput): Promise<ProviderTurnStartResult> => ({51 async (_input: CodexAppServerSendTurnInput): Promise<ProviderTurnStartResult> => ({
48 threadId: ProviderThreadId.makeUnsafe("thread-1"),52 threadId: ProviderThreadId.makeUnsafe("thread-1"),
49 turnId: asTurnId("turn-1"),53 turnId: asTurnId("turn-1"),
50 }),54 }),
...@@ -78,7 +82,7 @@ class FakeCodexManager extends CodexAppServerManager {...@@ -78,7 +82,7 @@ class FakeCodexManager extends CodexAppServerManager {
78 return this.startSessionImpl(input);82 return this.startSessionImpl(input);
79 }83 }
8084
81 override sendTurn(input: ProviderSendTurnInput): Promise<ProviderTurnStartResult> {85 override sendTurn(input: CodexAppServerSendTurnInput): Promise<ProviderTurnStartResult> {
82 return this.sendTurnImpl(input);86 return this.sendTurnImpl(input);
83 }87 }
8488
...@@ -118,7 +122,12 @@ class FakeCodexManager extends CodexAppServerManager {...@@ -118,7 +122,12 @@ class FakeCodexManager extends CodexAppServerManager {
118}122}
119123
120const validationManager = new FakeCodexManager();124const validationManager = new FakeCodexManager();
121const validationLayer = it.layer(makeCodexAdapterLive({ manager: validationManager }));125const validationLayer = it.layer(
126 makeCodexAdapterLive({ manager: validationManager }).pipe(
127 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
128 Layer.provideMerge(NodeServices.layer),
129 ),
130);
122131
123validationLayer("CodexAdapterLive validation", (it) => {132validationLayer("CodexAdapterLive validation", (it) => {
124 it.effect("returns validation error for non-codex provider on startSession", () =>133 it.effect("returns validation error for non-codex provider on startSession", () =>
...@@ -147,7 +156,12 @@ const sessionErrorManager = new FakeCodexManager();...@@ -147,7 +156,12 @@ const sessionErrorManager = new FakeCodexManager();
147sessionErrorManager.sendTurnImpl.mockImplementation(async () => {156sessionErrorManager.sendTurnImpl.mockImplementation(async () => {
148 throw new Error("Unknown session: sess-missing");157 throw new Error("Unknown session: sess-missing");
149});158});
150const sessionErrorLayer = it.layer(makeCodexAdapterLive({ manager: sessionErrorManager }));159const sessionErrorLayer = it.layer(
160 makeCodexAdapterLive({ manager: sessionErrorManager }).pipe(
161 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
162 Layer.provideMerge(NodeServices.layer),
163 ),
164);
151165
152sessionErrorLayer("CodexAdapterLive session errors", (it) => {166sessionErrorLayer("CodexAdapterLive session errors", (it) => {
153 it.effect("maps unknown-session sendTurn errors to ProviderAdapterSessionNotFoundError", () =>167 it.effect("maps unknown-session sendTurn errors to ProviderAdapterSessionNotFoundError", () =>
...@@ -178,7 +192,12 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => {...@@ -178,7 +192,12 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => {
178});192});
179193
180const lifecycleManager = new FakeCodexManager();194const lifecycleManager = new FakeCodexManager();
181const lifecycleLayer = it.layer(makeCodexAdapterLive({ manager: lifecycleManager }));195const lifecycleLayer = it.layer(
196 makeCodexAdapterLive({ manager: lifecycleManager }).pipe(
197 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
198 Layer.provideMerge(NodeServices.layer),
199 ),
200);
182201
183lifecycleLayer("CodexAdapterLive lifecycle", (it) => {202lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
184 it.effect("maps completed agent message items to canonical message.completed events", () =>203 it.effect("maps completed agent message items to canonical message.completed events", () =>
apps/server/src/provider/Layers/CodexAdapter.ts+45-4
...@@ -17,7 +17,7 @@ import {...@@ -17,7 +17,7 @@ import {
17 ProviderThreadId,17 ProviderThreadId,
18 ProviderTurnId,18 ProviderTurnId,
19} from "@t3tools/contracts";19} from "@t3tools/contracts";
20import { Effect, Layer, Queue, Schema, Stream } from "effect";20import { Effect, FileSystem, Layer, Queue, Schema, Stream } from "effect";
2121
22import {22import {
23 ProviderAdapterProcessError,23 ProviderAdapterProcessError,
...@@ -29,6 +29,8 @@ import {...@@ -29,6 +29,8 @@ import {
29} from "../Errors.ts";29} from "../Errors.ts";
30import { CodexAdapter, type CodexAdapterShape } from "../Services/CodexAdapter.ts";30import { CodexAdapter, type CodexAdapterShape } from "../Services/CodexAdapter.ts";
31import { CodexAppServerManager } from "../../codexAppServerManager.ts";31import { CodexAppServerManager } from "../../codexAppServerManager.ts";
32import { resolveAttachmentPath } from "../../attachmentStore.ts";
33import { ServerConfig } from "../../config.ts";
32import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";34import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
3335
34const PROVIDER = "codex" as const;36const PROVIDER = "codex" as const;
...@@ -426,6 +428,8 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray<ProviderRuntime...@@ -426,6 +428,8 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray<ProviderRuntime
426428
427const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>429const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
428 Effect.gen(function* () {430 Effect.gen(function* () {
431 const fileSystem = yield* FileSystem.FileSystem;
432 const serverConfig = yield* Effect.service(ServerConfig);
429 const nativeEventLogger =433 const nativeEventLogger =
430 options?.nativeEventLogPath !== undefined434 options?.nativeEventLogPath !== undefined
431 ? makeEventNdjsonLogger(options.nativeEventLogPath)435 ? makeEventNdjsonLogger(options.nativeEventLogPath)
...@@ -475,9 +479,46 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>...@@ -475,9 +479,46 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
475 };479 };
476480
477 const sendTurn: CodexAdapterShape["sendTurn"] = (input) =>481 const sendTurn: CodexAdapterShape["sendTurn"] = (input) =>
478 Effect.tryPromise({482 Effect.gen(function* () {
479 try: () => manager.sendTurn(input),483 const codexAttachments = yield* Effect.forEach(
480 catch: (cause) => toRequestError(input.sessionId, "turn/start", cause),484 input.attachments ?? [],
485 (attachment) =>
486 Effect.gen(function* () {
487 const attachmentPath = resolveAttachmentPath({
488 stateDir: serverConfig.stateDir,
489 attachment,
490 });
491 if (!attachmentPath) {
492 return yield* toRequestError(
493 input.sessionId,
494 "turn/start",
495 new Error(`Invalid attachment id '${attachment.id}'.`),
496 );
497 }
498 const bytes = yield* fileSystem.readFile(attachmentPath).pipe(
499 Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause)),
500 );
501 return {
502 type: "image" as const,
503 url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`,
504 };
505 }),
506 { concurrency: 1 },
507 );
508
509 return yield* Effect.tryPromise({
510 try: () => {
511 const managerInput = {
512 sessionId: input.sessionId,
513 ...(input.input !== undefined ? { input: input.input } : {}),
514 ...(input.model !== undefined ? { model: input.model } : {}),
515 ...(input.effort !== undefined ? { effort: input.effort } : {}),
516 ...(codexAttachments.length > 0 ? { attachments: codexAttachments } : {}),
517 };
518 return manager.sendTurn(managerInput);
519 },
520 catch: (cause) => toRequestError(input.sessionId, "turn/start", cause),
521 });
481 });522 });
482523
483 const interruptTurn: CodexAdapterShape["interruptTurn"] = (sessionId, turnId) =>524 const interruptTurn: CodexAdapterShape["interruptTurn"] = (sessionId, turnId) =>
apps/server/src/serverLayers.ts+2-2
...@@ -1,7 +1,7 @@...@@ -1,7 +1,7 @@
1import path from "node:path";1import path from "node:path";
22
3import * as NodeServices from "@effect/platform-node/NodeServices";3import * as NodeServices from "@effect/platform-node/NodeServices";
4import { Effect, Layer } from "effect";4import { Effect, FileSystem, Layer } from "effect";
5import * as SqlClient from "effect/unstable/sql/SqlClient";5import * as SqlClient from "effect/unstable/sql/SqlClient";
66
7import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery";7import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery";
...@@ -37,7 +37,7 @@ import { NodePtyAdapterLive } from "./terminal/Layers/NodePTY";...@@ -37,7 +37,7 @@ import { NodePtyAdapterLive } from "./terminal/Layers/NodePTY";
37export function makeServerProviderLayer(): Layer.Layer<37export function makeServerProviderLayer(): Layer.Layer<
38 ProviderService,38 ProviderService,
39 ProviderUnsupportedError,39 ProviderUnsupportedError,
40 SqlClient.SqlClient | ServerConfig40 SqlClient.SqlClient | ServerConfig | FileSystem.FileSystem
41> {41> {
42 return Effect.gen(function* () {42 return Effect.gen(function* () {
43 const { stateDir } = yield* ServerConfig;43 const { stateDir } = yield* ServerConfig;
apps/server/src/wsServer.test.ts+1
...@@ -420,6 +420,7 @@ describe("WebSocket Server", () => {...@@ -420,6 +420,7 @@ describe("WebSocket Server", () => {
420 Layer.provideMerge(runtimeLayer),420 Layer.provideMerge(runtimeLayer),
421 Layer.provideMerge(openLayer),421 Layer.provideMerge(openLayer),
422 Layer.provideMerge(serverConfigLayer),422 Layer.provideMerge(serverConfigLayer),
423 Layer.provideMerge(NodeServices.layer),
423 );424 );
424 const runtimeServices = await Effect.runPromise(425 const runtimeServices = await Effect.runPromise(
425 Layer.build(dependenciesLayer).pipe(Scope.provide(scope)),426 Layer.build(dependenciesLayer).pipe(Scope.provide(scope)),
apps/server/src/wsServer.ts+99-5
...@@ -12,8 +12,11 @@ import type { Duplex } from "node:stream";...@@ -12,8 +12,11 @@ import type { Duplex } from "node:stream";
12import Mime from "@effect/platform-node/Mime";12import Mime from "@effect/platform-node/Mime";
13import {13import {
14 CommandId,14 CommandId,
15 type ClientOrchestrationCommand,
16 type OrchestrationCommand,
15 ORCHESTRATION_WS_CHANNELS,17 ORCHESTRATION_WS_CHANNELS,
16 ORCHESTRATION_WS_METHODS,18 ORCHESTRATION_WS_METHODS,
19 PROVIDER_SEND_TURN_MAX_IMAGE_BYTES,
17 ProjectId,20 ProjectId,
18 ThreadId,21 ThreadId,
19 TerminalEvent,22 TerminalEvent,
...@@ -60,6 +63,8 @@ import {...@@ -60,6 +63,8 @@ import {
60 normalizeAttachmentRelativePath,63 normalizeAttachmentRelativePath,
61 resolveAttachmentRelativePath,64 resolveAttachmentRelativePath,
62} from "./attachmentPaths";65} from "./attachmentPaths";
66import { createAttachmentId, resolveAttachmentPath, resolveAttachmentPathById } from "./attachmentStore.ts";
67import { parseBase64DataUrl } from "./imageMime.ts";
6368
64/**69/**
65 * ServerShape - Service API for server lifecycle control.70 * ServerShape - Service API for server lifecycle control.
...@@ -235,6 +240,88 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -235,6 +240,88 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
235 });240 });
236 });241 });
237242
243 const normalizeDispatchCommand = Effect.fnUntraced(function* (input: {
244 readonly command: ClientOrchestrationCommand;
245 }) {
246 if (input.command.type !== "thread.turn.start") {
247 return input.command as OrchestrationCommand;
248 }
249 const turnStartCommand = input.command;
250
251 const normalizedAttachments = yield* Effect.forEach(
252 turnStartCommand.message.attachments,
253 (attachment) =>
254 Effect.gen(function* () {
255 const parsed = parseBase64DataUrl(attachment.dataUrl);
256 if (!parsed || !parsed.mimeType.startsWith("image/")) {
257 return yield* new RouteRequestError({
258 message: `Invalid image attachment payload for '${attachment.name}'.`,
259 });
260 }
261
262 const bytes = Buffer.from(parsed.base64, "base64");
263 if (bytes.byteLength === 0 || bytes.byteLength > PROVIDER_SEND_TURN_MAX_IMAGE_BYTES) {
264 return yield* new RouteRequestError({
265 message: `Image attachment '${attachment.name}' is empty or too large.`,
266 });
267 }
268
269 const attachmentId = createAttachmentId(turnStartCommand.threadId);
270 if (!attachmentId) {
271 return yield* new RouteRequestError({
272 message: "Failed to create a safe attachment id.",
273 });
274 }
275
276 const persistedAttachment = {
277 type: "image" as const,
278 id: attachmentId,
279 name: attachment.name,
280 mimeType: parsed.mimeType.toLowerCase(),
281 sizeBytes: bytes.byteLength,
282 };
283
284 const attachmentPath = resolveAttachmentPath({
285 stateDir: serverConfig.stateDir,
286 attachment: persistedAttachment,
287 });
288 if (!attachmentPath) {
289 return yield* new RouteRequestError({
290 message: `Failed to resolve persisted path for '${attachment.name}'.`,
291 });
292 }
293
294 yield* fileSystem.makeDirectory(path.dirname(attachmentPath), { recursive: true }).pipe(
295 Effect.mapError(
296 () =>
297 new RouteRequestError({
298 message: `Failed to create attachment directory for '${attachment.name}'.`,
299 }),
300 ),
301 );
302 yield* fileSystem.writeFile(attachmentPath, bytes).pipe(
303 Effect.mapError(
304 () =>
305 new RouteRequestError({
306 message: `Failed to persist attachment '${attachment.name}'.`,
307 }),
308 ),
309 );
310
311 return persistedAttachment;
312 }),
313 { concurrency: 1 },
314 );
315
316 return {
317 ...turnStartCommand,
318 message: {
319 ...turnStartCommand.message,
320 attachments: normalizedAttachments,
321 },
322 } satisfies OrchestrationCommand;
323 });
324
238 // HTTP server — serves static files or redirects to Vite dev server325 // HTTP server — serves static files or redirects to Vite dev server
239 const httpServer = http.createServer((req, res) => {326 const httpServer = http.createServer((req, res) => {
240 const respond = (327 const respond = (
...@@ -261,10 +348,16 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -261,10 +348,16 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
261 return;348 return;
262 }349 }
263350
264 const filePath = resolveAttachmentRelativePath({351 const filePath =
265 stateDir: serverConfig.stateDir,352 !normalizedRelativePath.includes("/") && !normalizedRelativePath.includes(".")
266 relativePath: normalizedRelativePath,353 ? resolveAttachmentPathById({
267 });354 stateDir: serverConfig.stateDir,
355 attachmentId: normalizedRelativePath,
356 })
357 : resolveAttachmentRelativePath({
358 stateDir: serverConfig.stateDir,
359 relativePath: normalizedRelativePath,
360 });
268 if (!filePath) {361 if (!filePath) {
269 respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path");362 respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path");
270 return;363 return;
...@@ -508,7 +601,8 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -508,7 +601,8 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
508601
509 case ORCHESTRATION_WS_METHODS.dispatchCommand: {602 case ORCHESTRATION_WS_METHODS.dispatchCommand: {
510 const { command } = request.body;603 const { command } = request.body;
511 return yield* orchestrationEngine.dispatch(command);604 const normalizedCommand = yield* normalizeDispatchCommand({ command });
605 return yield* orchestrationEngine.dispatch(normalizedCommand);
512 }606 }
513607
514 case ORCHESTRATION_WS_METHODS.getTurnDiff: {608 case ORCHESTRATION_WS_METHODS.getTurnDiff: {
apps/web/src/store.ts+7-3
...@@ -229,6 +229,10 @@ function toAttachmentPreviewUrl(rawUrl: string): string {...@@ -229,6 +229,10 @@ function toAttachmentPreviewUrl(rawUrl: string): string {
229 return rawUrl;229 return rawUrl;
230}230}
231231
232function attachmentPreviewRoutePath(attachmentId: string): string {
233 return `/attachments/${encodeURIComponent(attachmentId)}`;
234}
235
232function normalizeTerminalIds(terminalIds: string[]): string[] {236function normalizeTerminalIds(terminalIds: string[]): string[] {
233 const ids = terminalIds.map((id) => id.trim()).filter((id) => id.length > 0);237 const ids = terminalIds.map((id) => id.trim()).filter((id) => id.length > 0);
234 const unique = [...new Set(ids)].slice(0, MAX_THREAD_TERMINAL_COUNT);238 const unique = [...new Set(ids)].slice(0, MAX_THREAD_TERMINAL_COUNT);
...@@ -487,13 +491,13 @@ export function reducer(state: AppState, action: Action): AppState {...@@ -487,13 +491,13 @@ export function reducer(state: AppState, action: Action): AppState {
487 }491 }
488 : null,492 : null,
489 messages: thread.messages.map((message) => {493 messages: thread.messages.map((message) => {
490 const attachments = message.attachments?.map((attachment, index) => ({494 const attachments = message.attachments?.map((attachment) => ({
491 type: "image" as const,495 type: "image" as const,
492 id: `${message.id}:${index}`,496 id: attachment.id,
493 name: attachment.name,497 name: attachment.name,
494 mimeType: attachment.mimeType,498 mimeType: attachment.mimeType,
495 sizeBytes: attachment.sizeBytes,499 sizeBytes: attachment.sizeBytes,
496 previewUrl: toAttachmentPreviewUrl(attachment.dataUrl),500 previewUrl: toAttachmentPreviewUrl(attachmentPreviewRoutePath(attachment.id)),
497 }));501 }));
498 const normalizedMessage: ChatMessage = {502 const normalizedMessage: ChatMessage = {
499 id: message.id,503 id: message.id,
packages/contracts/src/orchestration.ts+55-3
...@@ -60,12 +60,28 @@ export const PROVIDER_SEND_TURN_MAX_INPUT_CHARS = 120_000;...@@ -60,12 +60,28 @@ export const PROVIDER_SEND_TURN_MAX_INPUT_CHARS = 120_000;
60export const PROVIDER_SEND_TURN_MAX_ATTACHMENTS = 8;60export const PROVIDER_SEND_TURN_MAX_ATTACHMENTS = 8;
61export const PROVIDER_SEND_TURN_MAX_IMAGE_BYTES = 10 * 1024 * 1024;61export const PROVIDER_SEND_TURN_MAX_IMAGE_BYTES = 10 * 1024 * 1024;
62export const PROVIDER_SEND_TURN_MAX_IMAGE_DATA_URL_CHARS = 14_000_000;62export const PROVIDER_SEND_TURN_MAX_IMAGE_DATA_URL_CHARS = 14_000_000;
63export const CHAT_ATTACHMENT_ID_MAX_CHARS = 128;
6364
64// Correlation id is command id by design in this model.65// Correlation id is command id by design in this model.
65export const CorrelationId = CommandId;66export const CorrelationId = CommandId;
66export type CorrelationId = typeof CorrelationId.Type;67export type CorrelationId = typeof CorrelationId.Type;
6768
69export const ChatAttachmentId = TrimmedNonEmptyString.check(
70 Schema.isMaxLength(CHAT_ATTACHMENT_ID_MAX_CHARS),
71 Schema.isPattern(/^[a-z0-9_-]+$/i),
72);
73export type ChatAttachmentId = typeof ChatAttachmentId.Type;
74
68export const ChatImageAttachment = Schema.Struct({75export const ChatImageAttachment = Schema.Struct({
76 type: Schema.Literal("image"),
77 id: ChatAttachmentId,
78 name: TrimmedNonEmptyString.check(Schema.isMaxLength(255)),
79 mimeType: TrimmedNonEmptyString.check(Schema.isMaxLength(100), Schema.isPattern(/^image\//i)),
80 sizeBytes: NonNegativeInt.check(Schema.isLessThanOrEqualTo(PROVIDER_SEND_TURN_MAX_IMAGE_BYTES)),
81});
82export type ChatImageAttachment = typeof ChatImageAttachment.Type;
83
84export const UploadChatImageAttachment = Schema.Struct({
69 type: Schema.Literal("image"),85 type: Schema.Literal("image"),
70 name: TrimmedNonEmptyString.check(Schema.isMaxLength(255)),86 name: TrimmedNonEmptyString.check(Schema.isMaxLength(255)),
71 mimeType: TrimmedNonEmptyString.check(Schema.isMaxLength(100), Schema.isPattern(/^image\//i)),87 mimeType: TrimmedNonEmptyString.check(Schema.isMaxLength(100), Schema.isPattern(/^image\//i)),
...@@ -74,10 +90,12 @@ export const ChatImageAttachment = Schema.Struct({...@@ -74,10 +90,12 @@ export const ChatImageAttachment = Schema.Struct({
74 Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_IMAGE_DATA_URL_CHARS),90 Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_IMAGE_DATA_URL_CHARS),
75 ),91 ),
76});92});
77export type ChatImageAttachment = typeof ChatImageAttachment.Type;93export type UploadChatImageAttachment = typeof UploadChatImageAttachment.Type;
7894
79export const ChatAttachment = Schema.Union([ChatImageAttachment]);95export const ChatAttachment = Schema.Union([ChatImageAttachment]);
80export type ChatAttachment = typeof ChatAttachment.Type;96export type ChatAttachment = typeof ChatAttachment.Type;
97export const UploadChatAttachment = Schema.Union([UploadChatImageAttachment]);
98export type UploadChatAttachment = typeof UploadChatAttachment.Type;
8199
82export const ProjectScriptIcon = Schema.Literals([100export const ProjectScriptIcon = Schema.Literals([
83 "play",101 "play",
...@@ -307,6 +325,24 @@ export const ThreadTurnStartCommand = Schema.Struct({...@@ -307,6 +325,24 @@ export const ThreadTurnStartCommand = Schema.Struct({
307 createdAt: IsoDateTime,325 createdAt: IsoDateTime,
308});326});
309327
328export const ClientThreadTurnStartCommand = Schema.Struct({
329 type: Schema.Literal("thread.turn.start"),
330 commandId: CommandId,
331 threadId: ThreadId,
332 message: Schema.Struct({
333 messageId: MessageId,
334 role: Schema.Literal("user"),
335 text: Schema.String,
336 attachments: Schema.Array(UploadChatAttachment),
337 }),
338 model: Schema.optional(TrimmedNonEmptyString),
339 effort: Schema.optional(TrimmedNonEmptyString),
340 assistantDeliveryMode: Schema.optional(AssistantDeliveryMode),
341 approvalPolicy: ProviderApprovalPolicy,
342 sandboxMode: ProviderSandboxMode,
343 createdAt: IsoDateTime,
344});
345
310export const ThreadTurnInterruptCommand = Schema.Struct({346export const ThreadTurnInterruptCommand = Schema.Struct({
311 type: Schema.Literal("thread.turn.interrupt"),347 type: Schema.Literal("thread.turn.interrupt"),
312 commandId: CommandId,348 commandId: CommandId,
...@@ -339,7 +375,7 @@ export const ThreadSessionStopCommand = Schema.Struct({...@@ -339,7 +375,7 @@ export const ThreadSessionStopCommand = Schema.Struct({
339 createdAt: IsoDateTime,375 createdAt: IsoDateTime,
340});376});
341377
342export const ClientOrchestrationCommand = Schema.Union([378export const DispatchableClientOrchestrationCommand = Schema.Union([
343 ProjectCreateCommand,379 ProjectCreateCommand,
344 ProjectMetaUpdateCommand,380 ProjectMetaUpdateCommand,
345 ProjectDeleteCommand,381 ProjectDeleteCommand,
...@@ -352,6 +388,22 @@ export const ClientOrchestrationCommand = Schema.Union([...@@ -352,6 +388,22 @@ export const ClientOrchestrationCommand = Schema.Union([
352 ThreadCheckpointRevertCommand,388 ThreadCheckpointRevertCommand,
353 ThreadSessionStopCommand,389 ThreadSessionStopCommand,
354]);390]);
391export type DispatchableClientOrchestrationCommand =
392 typeof DispatchableClientOrchestrationCommand.Type;
393
394export const ClientOrchestrationCommand = Schema.Union([
395 ProjectCreateCommand,
396 ProjectMetaUpdateCommand,
397 ProjectDeleteCommand,
398 ThreadCreateCommand,
399 ThreadDeleteCommand,
400 ThreadMetaUpdateCommand,
401 ClientThreadTurnStartCommand,
402 ThreadTurnInterruptCommand,
403 ThreadApprovalRespondCommand,
404 ThreadCheckpointRevertCommand,
405 ThreadSessionStopCommand,
406]);
355export type ClientOrchestrationCommand = typeof ClientOrchestrationCommand.Type;407export type ClientOrchestrationCommand = typeof ClientOrchestrationCommand.Type;
356408
357export const ThreadSessionSetCommand = Schema.Struct({409export const ThreadSessionSetCommand = Schema.Struct({
...@@ -422,7 +474,7 @@ export const InternalOrchestrationCommand = Schema.Union([...@@ -422,7 +474,7 @@ export const InternalOrchestrationCommand = Schema.Union([
422export type InternalOrchestrationCommand = typeof InternalOrchestrationCommand.Type;474export type InternalOrchestrationCommand = typeof InternalOrchestrationCommand.Type;
423475
424export const OrchestrationCommand = Schema.Union([476export const OrchestrationCommand = Schema.Union([
425 ClientOrchestrationCommand,477 DispatchableClientOrchestrationCommand,
426 InternalOrchestrationCommand,478 InternalOrchestrationCommand,
427]);479]);
428export type OrchestrationCommand = typeof OrchestrationCommand.Type;480export type OrchestrationCommand = typeof OrchestrationCommand.Type;