diff --git a/apps/server/src/attachmentStore.ts b/apps/server/src/attachmentStore.ts new file mode 100644 index 0000000000000000000000000000000000000000..39f80a3cab2dee081a8dc926117e40a889472804 --- /dev/null +++ b/apps/server/src/attachmentStore.ts @@ -0,0 +1,95 @@ +import { randomUUID } from "node:crypto"; + +import type { ChatAttachment } from "@t3tools/contracts"; + +import { + ATTACHMENTS_ROUTE_PREFIX, + normalizeAttachmentRelativePath, + resolveAttachmentRelativePath, +} from "./attachmentPaths.ts"; +import { inferImageExtension, SAFE_IMAGE_FILE_EXTENSIONS } from "./imageMime.ts"; + +const ATTACHMENT_FILENAME_EXTENSIONS = [...SAFE_IMAGE_FILE_EXTENSIONS, ".bin"]; + +function toSafeThreadAttachmentSegment(threadId: string): string | null { + const segment = encodeURIComponent(threadId); + if ( + segment.length === 0 || + segment === "." || + segment === ".." || + segment.includes("/") || + segment.includes("\\") || + segment.includes("\0") + ) { + return null; + } + return segment; +} + +export function createAttachmentId(threadId: string): string | null { + const threadSegment = toSafeThreadAttachmentSegment(threadId); + if (!threadSegment) { + return null; + } + return `${threadSegment}-${randomUUID()}`; +} + +export function attachmentRelativePath(attachment: ChatAttachment): string { + switch (attachment.type) { + case "image": { + const extension = inferImageExtension({ + mimeType: attachment.mimeType, + fileName: attachment.name, + }); + return `${attachment.id}${extension}`; + } + } +} + +export function resolveAttachmentPath(input: { + readonly stateDir: string; + readonly attachment: ChatAttachment; +}): string | null { + return resolveAttachmentRelativePath({ + stateDir: input.stateDir, + relativePath: attachmentRelativePath(input.attachment), + }); +} + +export function attachmentRoutePath(attachment: ChatAttachment): string { + return `${ATTACHMENTS_ROUTE_PREFIX}/${encodeURIComponent(attachment.id)}`; +} + +export function resolveAttachmentPathById(input: { + readonly stateDir: string; + readonly attachmentId: string; +}): string | null { + const normalizedId = normalizeAttachmentRelativePath(input.attachmentId); + if (!normalizedId || normalizedId.includes("/") || normalizedId.includes(".")) { + return null; + } + for (const extension of ATTACHMENT_FILENAME_EXTENSIONS) { + const maybePath = resolveAttachmentRelativePath({ + stateDir: input.stateDir, + relativePath: `${normalizedId}${extension}`, + }); + if (maybePath) { + return maybePath; + } + } + return null; +} + +export function parseAttachmentIdFromRelativePath(relativePath: string): string | null { + const normalized = normalizeAttachmentRelativePath(relativePath); + if (!normalized || normalized.includes("/")) { + return null; + } + const extensionIndex = normalized.lastIndexOf("."); + if (extensionIndex <= 0) { + return null; + } + const id = normalized.slice(0, extensionIndex); + return id.length > 0 && !id.includes(".") ? id : null; +} + diff --git a/apps/server/src/codexAppServerManager.test.ts b/apps/server/src/codexAppServerManager.test.ts index b7b6c91237c89fe924e1cca5c21a4c04493ab1b8..7ec669d4c68b905d059bf605586d58a7bc3dc3f3 100644 --- a/apps/server/src/codexAppServerManager.test.ts +++ b/apps/server/src/codexAppServerManager.test.ts @@ -191,10 +191,7 @@ describe("sendTurn", () => { attachments: [ { type: "image", - name: "error.png", - mimeType: "image/png", - sizeBytes: 1_024, - dataUrl: "data:image/png;base64,AAAA", + url: "data:image/png;base64,AAAA", }, ], model: "gpt-5.3", @@ -242,10 +239,7 @@ describe("sendTurn", () => { attachments: [ { type: "image", - name: "diagram.png", - mimeType: "image/png", - sizeBytes: 256, - dataUrl: "data:image/png;base64,BBBB", + url: "data:image/png;base64,BBBB", }, ], }); diff --git a/apps/server/src/codexAppServerManager.ts b/apps/server/src/codexAppServerManager.ts index c2b5945048494d471b4cf0b71e3dc7e5f38beffc..cb50a302f9389101cd940fc56d460c321e51d10a 100644 --- a/apps/server/src/codexAppServerManager.ts +++ b/apps/server/src/codexAppServerManager.ts @@ -14,7 +14,6 @@ import { normalizeModelSlug, type ProviderApprovalDecision, type ProviderEvent, - type ProviderSendTurnInput, type ProviderSession, type ProviderSessionStartInput, type ProviderTurnStartResult, @@ -71,6 +70,14 @@ interface JsonRpcNotification { params?: unknown; } +export interface CodexAppServerSendTurnInput { + readonly sessionId: ProviderSessionId; + readonly input?: string; + readonly attachments?: ReadonlyArray<{ type: "image"; url: string }>; + readonly model?: string; + readonly effort?: string; +} + export interface CodexThreadTurnSnapshot { id: ProviderTurnId; items: unknown[]; @@ -290,7 +297,7 @@ export class CodexAppServerManager extends EventEmitter { + async sendTurn(input: CodexAppServerSendTurnInput): Promise { const context = this.requireSession(input.sessionId); if (!context.session.threadId) { throw new Error("Session is missing a thread id."); @@ -310,7 +317,7 @@ export class CodexAppServerManager extends EventEmitter { stdinMustContain: "Attachment metadata:", }, Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const attachmentId = `thread-branch-image-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`; + const attachmentPath = path.join(process.cwd(), "attachments", `${attachmentId}.png`); + yield* fs.makeDirectory(path.join(process.cwd(), "attachments"), { recursive: true }); + yield* fs.writeFile(attachmentPath, Buffer.from("hello")); + const textGeneration = yield* TextGeneration; - - const generated = yield* textGeneration.generateBranchName({ - cwd: process.cwd(), - message: "Fix layout bug from screenshot.", - attachments: [ - { - type: "image", - name: "bug.png", - mimeType: "image/png", - sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", - }, - ], - }); + const generated = yield* textGeneration + .generateBranchName({ + cwd: process.cwd(), + message: "Fix layout bug from screenshot.", + attachments: [ + { + type: "image", + id: attachmentId, + name: "bug.png", + mimeType: "image/png", + sizeBytes: 5, + }, + ], + }) + .pipe(Effect.ensuring(fs.remove(attachmentPath).pipe(Effect.catch(() => Effect.void)))); expect(generated.branch).toBe("fix/ui-regression"); }), ), ); - it.effect("resolves persisted attachment URLs to files for codex image inputs", () => + it.effect("resolves persisted attachment ids to files for codex image inputs", () => withFakeCodexEnv( { output: JSON.stringify({ @@ -326,12 +334,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; - const uniqueSuffix = `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`; - const imageFileName = `message-1-${uniqueSuffix}.png`; - const imagePath = path.join(process.cwd(), "attachments", "thread-1", imageFileName); - yield* fs.makeDirectory(path.join(process.cwd(), "attachments", "thread-1"), { - recursive: true, - }); + const attachmentId = + `thread-1-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`; + const imagePath = path.join(process.cwd(), "attachments", `${attachmentId}.png`); + yield* fs.makeDirectory(path.join(process.cwd(), "attachments"), { recursive: true }); yield* fs.writeFile(imagePath, Buffer.from("hello")); const textGeneration = yield* TextGeneration; @@ -342,10 +348,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { attachments: [ { type: "image", + id: attachmentId, name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: `/attachments/thread-1/${imageFileName}`, }, ], }) @@ -357,8 +363,6 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { }), ), ), - ) - .pipe( Effect.ensuring( fs.remove(imagePath).pipe(Effect.catch(() => Effect.void)), ), @@ -369,7 +373,7 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { ), ); - it.effect("ignores absolute non-route attachment paths for codex image inputs", () => + it.effect("ignores missing attachment ids for codex image inputs", () => withFakeCodexEnv( { output: JSON.stringify({ @@ -380,11 +384,9 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; - const tempDir = yield* fs.makeTempDirectoryScoped({ - prefix: "t3code-codex-attachment-bypass-", - }); - const absoluteImagePath = path.join(tempDir, "outside.png"); - yield* fs.writeFile(absoluteImagePath, Buffer.from("hello")); + const missingAttachmentId = `thread-missing-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`; + const missingPath = path.join(process.cwd(), "attachments", `${missingAttachmentId}.png`); + yield* fs.remove(missingPath).pipe(Effect.catch(() => Effect.void)); const textGeneration = yield* TextGeneration; const result = yield* textGeneration @@ -394,10 +396,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { attachments: [ { type: "image", + id: missingAttachmentId, name: "outside.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: absoluteImagePath, }, ], }) diff --git a/apps/server/src/git/Layers/CodexTextGeneration.ts b/apps/server/src/git/Layers/CodexTextGeneration.ts index 93416ea3dc0417cf9de0be15357db811355fbb6b..064d4da9c8d81c6429046af36447788f0e87c60c 100644 --- a/apps/server/src/git/Layers/CodexTextGeneration.ts +++ b/apps/server/src/git/Layers/CodexTextGeneration.ts @@ -3,10 +3,9 @@ import { randomUUID } from "node:crypto"; import { Effect, FileSystem, Layer, Option, Path, Schema, Stream } from "effect"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; -import { resolveAttachmentRoutePath } from "../../attachmentPaths.ts"; +import { resolveAttachmentPath } from "../../attachmentStore.ts"; import { ServerConfig } from "../../config.ts"; import { TextGenerationError } from "../Errors.ts"; -import { inferImageExtension, parseBase64DataUrl } from "../../imageMime.ts"; import { type BranchNameGenerationInput, type BranchNameGenerationResult, @@ -109,7 +108,6 @@ const makeCodexTextGeneration = Effect.gen(function* () { type MaterializedImageAttachments = { readonly imagePaths: ReadonlyArray; - readonly tempImagePaths: ReadonlyArray; }; const readStreamAsString = ( @@ -151,96 +149,41 @@ const makeCodexTextGeneration = Effect.gen(function* () { ); }; - const writeTempBinaryFile = ( - operation: string, - prefix: string, - bytes: Uint8Array, - extension: string, - ): Effect.Effect => { - const normalizedExtension = extension.startsWith(".") ? extension : `.${extension}`; - const filePath = path.join( - tempDir, - `t3code-${prefix}-${process.pid}-${randomUUID()}${normalizedExtension}`, - ); - return fileSystem.writeFile(filePath, bytes).pipe( - Effect.mapError( - (cause) => - new TextGenerationError({ - operation, - detail: `Failed to write temp file at ${filePath}.`, - cause, - }), - ), - Effect.as(filePath), - ); - }; - const safeUnlink = (filePath: string): Effect.Effect => fileSystem.remove(filePath).pipe(Effect.catch(() => Effect.void)); const materializeImageAttachments = ( - operation: "generateCommitMessage" | "generatePrContent" | "generateBranchName", + _operation: "generateCommitMessage" | "generatePrContent" | "generateBranchName", attachments: BranchNameGenerationInput["attachments"], - ): Effect.Effect => { - const tempImagePaths: string[] = []; - - return Effect.gen(function* () { + ): Effect.Effect => + Effect.gen(function* () { if (!attachments || attachments.length === 0) { - return { imagePaths: [], tempImagePaths }; + return { imagePaths: [] }; } const imagePaths: string[] = []; - for (const [index, attachment] of attachments.entries()) { + for (const attachment of attachments) { if (attachment.type !== "image") { continue; } - const parsed = parseBase64DataUrl(attachment.dataUrl); - if (parsed && parsed.mimeType.startsWith("image/")) { - const bytes = Buffer.from(parsed.base64, "base64"); - if (bytes.byteLength === 0) { - continue; - } - - const extension = inferImageExtension({ - mimeType: parsed.mimeType, - fileName: attachment.name, - }); - const imagePath = yield* writeTempBinaryFile( - operation, - `codex-image-${index}`, - bytes, - extension, - ); - imagePaths.push(imagePath); - tempImagePaths.push(imagePath); - continue; - } - - const persistedPath = resolveAttachmentRoutePath({ + const resolvedPath = resolveAttachmentPath({ stateDir: serverConfig.stateDir, - dataUrl: attachment.dataUrl, + attachment, }); - if (!persistedPath || !path.isAbsolute(persistedPath)) { + if (!resolvedPath || !path.isAbsolute(resolvedPath)) { continue; } const fileInfo = yield* fileSystem - .stat(persistedPath) + .stat(resolvedPath) .pipe(Effect.catch(() => Effect.succeed(null))); if (!fileInfo || fileInfo.type !== "File") { continue; } - imagePaths.push(persistedPath); + imagePaths.push(resolvedPath); } - return { imagePaths, tempImagePaths }; - }).pipe( - Effect.tapError(() => - Effect.all(tempImagePaths.map((filePath) => safeUnlink(filePath)), { - concurrency: "unbounded", - }).pipe(Effect.asVoid), - ), - ); - }; + return { imagePaths }; + }); const runCodexJson = ({ operation, @@ -454,10 +397,7 @@ const makeCodexTextGeneration = Effect.gen(function* () { const generateBranchName: TextGenerationShape["generateBranchName"] = (input) => { return Effect.gen(function* () { - const { imagePaths, tempImagePaths } = yield* materializeImageAttachments( - "generateBranchName", - input.attachments, - ); + const { imagePaths } = yield* materializeImageAttachments("generateBranchName", input.attachments); const attachmentLines = (input.attachments ?? []).map( (attachment) => `- ${attachment.name} (${attachment.mimeType}, ${attachment.sizeBytes} bytes)`, @@ -492,7 +432,6 @@ const makeCodexTextGeneration = Effect.gen(function* () { branch: Schema.String, }), imagePaths, - cleanupPaths: tempImagePaths, }); return { diff --git a/apps/server/src/main.test.ts b/apps/server/src/main.test.ts index a74b6283dc6034d5b84d7a4e5ab1de47dcb0001e..be2b810394d337aafa87b7340d3404ed68a90508 100644 --- a/apps/server/src/main.test.ts +++ b/apps/server/src/main.test.ts @@ -49,10 +49,21 @@ const testLayer = Layer.mergeAll( const runCli = ( args: ReadonlyArray, env: Record = { T3CODE_NO_BROWSER: "true" }, -) => - Command.runWith(t3Cli, { version: "0.0.0-test" })(args).pipe( - Effect.provide(ConfigProvider.layer(ConfigProvider.fromEnv({ env }))), +) => { + const uniqueStateDir = `/tmp/t3-cli-state-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`; + return Command.runWith(t3Cli, { version: "0.0.0-test" })(args).pipe( + Effect.provide( + ConfigProvider.layer( + ConfigProvider.fromEnv({ + env: { + T3CODE_STATE_DIR: uniqueStateDir, + ...env, + }, + }), + ), + ), ); +}; beforeEach(() => { vi.clearAllMocks(); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 8cbef3862b2b0fd8c24d0f26264eca9186362b4e..4cb34606250e9c77d4ef10d39630700359b328c0 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -178,7 +178,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }), ); - it.effect("materializes message image attachments into stateDir and stores URL references", () => + it.effect("stores message attachment references without mutating payloads", () => Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe( Effect.flatMap((stateDir) => Effect.gen(function* () { @@ -205,10 +205,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-attachments-att-1", name: "example.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], turnId: null, @@ -232,21 +232,12 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ { type: "image", + id: "thread-attachments-att-1", name: "example.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "/attachments/thread-attachments/message-attachments-0.png", }, ]); - - const attachmentPath = path.join( - stateDir, - "attachments", - "thread-attachments", - "message-attachments-0.png", - ); - assert.equal(fs.existsSync(attachmentPath), true); - assert.deepEqual(fs.readFileSync(attachmentPath), Buffer.from("SGVsbG8=", "base64")); }).pipe( (effect) => runWithProjectionPipelineLayer(stateDir, effect), Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), @@ -255,7 +246,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { ), ); - it.effect("materializes only image data URLs and uses a safe extension whitelist", () => + it.effect("preserves mixed image attachment metadata as-is", () => Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe( Effect.flatMap((stateDir) => Effect.gen(function* () { @@ -282,17 +273,17 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-attachments-safe-att-1", name: "untrusted.exe", mimeType: "image/x-unknown", sizeBytes: 5, - dataUrl: "data:image/x-unknown;base64,SGVsbG8=", }, { type: "image", + id: "thread-attachments-safe-att-2", name: "not-image.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:text/plain;base64,SGVsbG8=", }, ], turnId: null, @@ -316,36 +307,19 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ { type: "image", + id: "thread-attachments-safe-att-1", name: "untrusted.exe", mimeType: "image/x-unknown", sizeBytes: 5, - dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin", }, { type: "image", + id: "thread-attachments-safe-att-2", name: "not-image.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:text/plain;base64,SGVsbG8=", }, ]); - - const firstAttachmentPath = path.join( - stateDir, - "attachments", - "thread-attachments-safe", - "message-attachments-safe-0.bin", - ); - assert.equal(fs.existsSync(firstAttachmentPath), true); - assert.deepEqual(fs.readFileSync(firstAttachmentPath), Buffer.from("SGVsbG8=", "base64")); - - const secondAttachmentPath = path.join( - stateDir, - "attachments", - "thread-attachments-safe", - "message-attachments-safe-1.png", - ); - assert.equal(fs.existsSync(secondAttachmentPath), false); }).pipe( (effect) => runWithProjectionPipelineLayer(stateDir, effect), Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), @@ -425,10 +399,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-clear-attachments-att-1", name: "clear.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], turnId: null, @@ -477,17 +451,18 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { ); it.effect( - "overwrites attachment file bytes when a message updates the same attachment index", + "overwrites stored attachment references when a message updates attachments", () => Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-overwrite-")), ).pipe( - Effect.flatMap((stateDir) => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const now = new Date().toISOString(); - const later = new Date(Date.now() + 1_000).toISOString(); + Effect.flatMap((stateDir) => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const now = new Date().toISOString(); + const later = new Date(Date.now() + 1_000).toISOString(); yield* eventStore.append({ type: "project.created", @@ -550,10 +525,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-overwrite-att-1", name: "file.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], turnId: null, @@ -581,10 +556,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-overwrite-att-2", name: "file.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,V29ybGQ=", }, ], turnId: null, @@ -596,14 +571,23 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { yield* projectionPipeline.bootstrap; - 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")); + const rows = yield* sql<{ + readonly attachmentsJson: string | null; + }>` + SELECT attachments_json AS "attachmentsJson" + FROM projection_thread_messages + WHERE message_id = 'message-overwrite' + `; + assert.equal(rows.length, 1); + assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ + { + type: "image", + id: "thread-overwrite-att-2", + name: "file.png", + mimeType: "image/png", + sizeBytes: 5, + }, + ]); }).pipe( (effect) => runWithProjectionPipelineLayer(stateDir, effect), Effect.ensuring( @@ -701,10 +685,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-rollback-att-1", name: "rollback.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], turnId: null, @@ -728,8 +712,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { const attachmentPath = path.join( stateDir, "attachments", - "thread-rollback", - "message-rollback-0.png", + "thread-rollback-att-1.png", ); assert.equal(fs.existsSync(attachmentPath), false); yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`; @@ -839,10 +822,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-revert-files-keep", name: "keep.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], turnId: TurnId.makeUnsafe("turn-keep"), @@ -892,10 +875,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-revert-files-remove", name: "remove.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,V29ybGQ=", }, ], turnId: TurnId.makeUnsafe("turn-remove"), @@ -908,15 +891,16 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { const keepPath = path.join( stateDir, "attachments", - "thread-revert-files", - "message-keep-0.png", + "thread-revert-files-keep.png", ); const removePath = path.join( stateDir, "attachments", - "thread-revert-files", - "message-remove-0.png", + "thread-revert-files-remove.png", ); + fs.mkdirSync(path.join(stateDir, "attachments"), { recursive: true }); + fs.writeFileSync(keepPath, Buffer.from("keep")); + fs.writeFileSync(removePath, Buffer.from("remove")); assert.equal(fs.existsSync(keepPath), true); assert.equal(fs.existsSync(removePath), true); @@ -1022,10 +1006,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { attachments: [ { type: "image", + id: "thread-delete-files-att-1", name: "delete.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], turnId: null, @@ -1035,8 +1019,10 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }, }); - const threadAttachmentDir = path.join(stateDir, "attachments", "thread-delete-files"); - assert.equal(fs.existsSync(threadAttachmentDir), true); + const threadAttachmentPath = path.join(stateDir, "attachments", "thread-delete-files-att-1.png"); + fs.mkdirSync(path.join(stateDir, "attachments"), { recursive: true }); + fs.writeFileSync(threadAttachmentPath, Buffer.from("delete")); + assert.equal(fs.existsSync(threadAttachmentPath), true); yield* appendAndProject({ type: "thread.deleted", @@ -1054,7 +1040,7 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }, }); - assert.equal(fs.existsSync(threadAttachmentDir), false); + assert.equal(fs.existsSync(threadAttachmentPath), false); }).pipe( (effect) => runWithProjectionPipelineLayer(stateDir, effect), Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 989b5d4dd418a235675162b144a3c84c8c3d12a3..57e1b81cca091ad8b9d9edde125f8235d4030406 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -37,8 +37,7 @@ import { OrchestrationProjectionPipeline, type OrchestrationProjectionPipelineShape, } from "../Services/ProjectionPipeline.ts"; -import { ATTACHMENTS_ROUTE_PREFIX, attachmentRouteToRelativePath } from "../../attachmentPaths.ts"; -import { inferImageExtension, parseBase64DataUrl } from "../../imageMime.ts"; +import { attachmentRelativePath } from "../../attachmentStore.ts"; export const ORCHESTRATION_PROJECTOR_NAMES = { projects: "projection.projects", @@ -62,13 +61,7 @@ interface ProjectorDefinition { ) => Effect.Effect; } -interface PendingAttachmentWrite { - readonly absolutePath: string; - readonly bytes: Uint8Array; -} - interface AttachmentSideEffects { - readonly writes: Array; readonly deletedThreadIds: Set; readonly prunedThreadRelativePaths: Map>; } @@ -89,62 +82,10 @@ function toSafeThreadAttachmentSegment(threadId: string): string | null { } const materializeAttachmentsForProjection = Effect.fn(function* (input: { - readonly threadId: string; - readonly messageId: string; readonly attachments: ReadonlyArray; - readonly stageFileWrite: (pendingWrite: PendingAttachmentWrite) => void; }) { if (input.attachments.length === 0) return []; - - const serverConfig = yield* Effect.service(ServerConfig); - - const path = yield* Path.Path; - const attachmentsRootDir = path.join(serverConfig.stateDir, "attachments"); - const threadSegment = toSafeThreadAttachmentSegment(input.threadId); - if (!threadSegment) { - yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", { - threadId: input.threadId, - messageId: input.messageId, - }); - return input.attachments; - } - const messageSegment = encodeURIComponent(input.messageId); - - return yield* Effect.forEach( - input.attachments, - (attachment, index) => - Effect.sync(() => { - if (attachment.type !== "image" || !attachment.dataUrl.startsWith("data:")) { - return attachment; - } - - const parsed = parseBase64DataUrl(attachment.dataUrl); - if (!parsed) { - return attachment; - } - if (!parsed.mimeType.startsWith("image/")) { - return attachment; - } - - const bytes = Buffer.from(parsed.base64, "base64"); - if (bytes.byteLength === 0) return attachment; - - const fileName = `${index}${inferImageExtension({ - mimeType: parsed.mimeType, - fileName: attachment.name, - })}`; - const uniqueFileName = `${messageSegment}-${fileName}`; - const relativePath = `${threadSegment}/${uniqueFileName}`; - const absolutePath = path.join(attachmentsRootDir, threadSegment, uniqueFileName); - input.stageFileWrite({ absolutePath, bytes }); - - return { - ...attachment, - dataUrl: `${ATTACHMENTS_ROUTE_PREFIX}/${relativePath}`, - } satisfies ChatAttachment; - }), - { concurrency: 1 }, - ); + return input.attachments; }); function extractActivityRequestId(payload: unknown): ApprovalRequestId | null { @@ -267,22 +208,16 @@ function collectThreadAttachmentRelativePaths( if (!threadSegment) { return new Set(); } - const threadPrefix = `${threadSegment}/`; const relativePaths = new Set(); for (const message of messages) { for (const attachment of message.attachments ?? []) { if (attachment.type !== "image") { continue; } - const relativePath = attachmentRouteToRelativePath(attachment.dataUrl); - if (!relativePath || !relativePath.startsWith(threadPrefix)) { + if (!attachment.id.startsWith(`${threadSegment}-`)) { continue; } - const threadRelativePath = relativePath.slice(threadPrefix.length).replace(/^[/\\]+/, ""); - if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) { - continue; - } - relativePaths.add(threadRelativePath.replace(/\\/g, "/")); + relativePaths.add(attachmentRelativePath(attachment)); } } return relativePaths; @@ -306,8 +241,26 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid }); return; } - const threadDir = path.join(attachmentsRootDir, threadSegment); - yield* fileSystem.remove(threadDir, { recursive: true, force: true }); + const entries = yield* fileSystem + .readDirectory(attachmentsRootDir, { recursive: false }) + .pipe(Effect.catch(() => Effect.succeed([] as Array))); + yield* Effect.forEach( + entries, + (entry) => + Effect.gen(function* () { + const normalizedEntry = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); + if (normalizedEntry.length === 0 || normalizedEntry.includes("/")) { + return; + } + if (!normalizedEntry.startsWith(`${threadSegment}-`)) { + return; + } + yield* fileSystem.remove(path.join(attachmentsRootDir, normalizedEntry), { + force: true, + }); + }), + { concurrency: 1 }, + ); }), { concurrency: 1 }, ); @@ -324,20 +277,22 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId }); return; } - const threadDir = path.join(attachmentsRootDir, threadSegment); const entries = yield* fileSystem - .readDirectory(threadDir, { recursive: true }) + .readDirectory(attachmentsRootDir, { recursive: false }) .pipe(Effect.catch(() => Effect.succeed([] as Array))); yield* Effect.forEach( entries, (entry) => Effect.gen(function* () { - const threadRelativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); - if (threadRelativePath.length === 0 || threadRelativePath.startsWith("..")) { + const relativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/"); + if (relativePath.length === 0 || relativePath.includes("/")) { + return; + } + if (!relativePath.startsWith(`${threadSegment}-`)) { return; } - const absolutePath = path.join(threadDir, threadRelativePath); + const absolutePath = path.join(attachmentsRootDir, relativePath); const fileInfo = yield* fileSystem .stat(absolutePath) .pipe(Effect.catch(() => Effect.succeed(null))); @@ -345,7 +300,7 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid return; } - if (!keptThreadRelativePaths.has(threadRelativePath)) { + if (!keptThreadRelativePaths.has(relativePath)) { yield* fileSystem.remove(absolutePath, { force: true }); } }), @@ -356,22 +311,6 @@ const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSid { concurrency: 1 }, ); - const writesByPath = new Map(); - for (const pendingWrite of sideEffects.writes) { - // Last write wins for a path so updated attachment contents replace stale bytes. - writesByPath.set(pendingWrite.absolutePath, pendingWrite.bytes); - } - yield* Effect.forEach( - writesByPath.entries(), - ([absolutePath, bytes]) => - Effect.gen(function* () { - yield* fileSystem.makeDirectory(path.dirname(absolutePath), { - recursive: true, - }); - yield* fileSystem.writeFile(absolutePath, bytes); - }), - { concurrency: 1 }, - ); }); const makeOrchestrationProjectionPipeline = Effect.gen(function* () { @@ -589,16 +528,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const nextAttachments = event.payload.attachments !== undefined ? yield* materializeAttachmentsForProjection({ - threadId: event.payload.threadId, - messageId: event.payload.messageId, attachments: event.payload.attachments, - stageFileWrite: (pendingWrite) => { - attachmentSideEffects.writes.push(pendingWrite); - }, - }).pipe( - Effect.provideService(ServerConfig, serverConfig), - Effect.provideService(Path.Path, path), - ) + }) : existingMessage?.attachments; yield* projectionThreadMessageRepository.upsert({ messageId: event.payload.messageId, @@ -1089,7 +1020,6 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const runProjectorForEvent = (projector: ProjectorDefinition, event: OrchestrationEvent) => Effect.gen(function* () { const attachmentSideEffects: AttachmentSideEffects = { - writes: [], deletedThreadIds: new Set(), prunedThreadRelativePaths: new Map>(), }; diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 65dc7f8755682d0b07225679192bc1170de7c29d..3abe82b5440cf2c98fa179be106f1b84f95e93d7 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -258,10 +258,10 @@ describe("ProviderCommandReactor", () => { attachments: [ { type: "image", + id: "thread-1-att-rename", name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], }, @@ -280,15 +280,10 @@ describe("ProviderCommandReactor", () => { attachments: [ { type: "image", + id: "thread-1-att-rename", name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: path.join( - harness.stateDir, - "attachments", - "thread-1", - "user-message-worktree-rename-0.png", - ), }, ], }); @@ -307,9 +302,8 @@ describe("ProviderCommandReactor", () => { }); }); - it("reuses persisted attachment files for branch generation and turn start", async () => { - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-attachments-")); - const harness = await createHarness({ stateDir }); + it("passes persisted attachment references to branch generation and turn start", async () => { + const harness = await createHarness(); const now = new Date().toISOString(); await Effect.runPromise( @@ -334,10 +328,10 @@ describe("ProviderCommandReactor", () => { attachments: [ { type: "image", + id: "thread-1-att-persisted", name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], }, @@ -350,23 +344,16 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.generateBranchName.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); - const expectedAttachmentPath = path.join( - stateDir, - "attachments", - "thread-1", - "user-message-persisted-attachments-0.png", - ); - expect(harness.generateBranchName.mock.calls[0]?.[0]).toEqual({ cwd: "/tmp/provider-project/.t3/worktrees/t3code-abcdef12", message: "Fix visual bug from screenshot", attachments: [ { type: "image", + id: "thread-1-att-persisted", name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: expectedAttachmentPath, }, ], }); @@ -374,14 +361,13 @@ describe("ProviderCommandReactor", () => { attachments: [ { type: "image", + id: "thread-1-att-persisted", name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", }, ], }); - expect(fs.existsSync(expectedAttachmentPath)).toBe(true); }); it("skips worktree branch generation after the first user turn", async () => { diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 18ce1c648c59a6380d56886cc8e8da8ac050bcc7..bf88e221b91c2ecec8e5fe9cd04aa0859b4ce8cc 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -12,17 +12,11 @@ import { type ProviderThreadId, type TurnId, } from "@t3tools/contracts"; -import { Cache, Cause, Duration, Effect, FileSystem, Layer, Option, Queue, Stream } from "effect"; +import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; -import { - resolveAttachmentRelativePath, - resolveAttachmentRoutePath, -} from "../../attachmentPaths.ts"; -import { ServerConfig } from "../../config.ts"; import { GitCore } from "../../git/Services/GitCore.ts"; import { TextGeneration } from "../../git/Services/TextGeneration.ts"; -import { inferImageExtension, parseBase64DataUrl } from "../../imageMime.ts"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { @@ -109,8 +103,6 @@ const make = Effect.gen(function* () { const providerService = yield* ProviderService; const git = yield* GitCore; const textGeneration = yield* TextGeneration; - const fileSystem = yield* FileSystem.FileSystem; - const serverConfig = yield* Effect.service(ServerConfig); const handledTurnStartKeys = yield* Cache.make({ capacity: HANDLED_TURN_START_KEY_MAX, timeToLive: HANDLED_TURN_START_KEY_TTL, @@ -291,61 +283,6 @@ const make = Effect.gen(function* () { }); }); - const resolvePersistedAttachments = Effect.fnUntraced(function* (input: { - readonly threadId: ThreadId; - readonly messageId: string; - readonly attachments: ReadonlyArray | undefined; - }) { - if (!input.attachments || input.attachments.length === 0) { - return input.attachments ?? []; - } - - const threadSegment = encodeURIComponent(input.threadId); - const messageSegment = encodeURIComponent(input.messageId); - - return yield* Effect.forEach(Array.from(input.attachments.entries()), ([index, attachment]) => - Effect.gen(function* () { - if (attachment.type !== "image") return attachment; - - const resolvedRoutePath = resolveAttachmentRoutePath({ - stateDir: serverConfig.stateDir, - dataUrl: attachment.dataUrl, - }); - const resolvedMaterializedPath = - resolvedRoutePath ?? - (() => { - const parsed = parseBase64DataUrl(attachment.dataUrl); - if (!parsed || !parsed.mimeType.startsWith("image/")) { - return null; - } - const extension = inferImageExtension({ - mimeType: parsed.mimeType, - fileName: attachment.name, - }); - return resolveAttachmentRelativePath({ - stateDir: serverConfig.stateDir, - relativePath: `${threadSegment}/${messageSegment}-${index}${extension}`, - }); - })(); - - if (!resolvedMaterializedPath) { - return attachment; - } - - const fileInfo = yield* fileSystem - .stat(resolvedMaterializedPath) - .pipe(Effect.catch(() => Effect.succeed(null))); - if (!fileInfo || fileInfo.type !== "File") { - return attachment; - } - return { - ...attachment, - dataUrl: resolvedMaterializedPath, - } satisfies ChatAttachment; - }), - ); - }); - const maybeGenerateAndRenameWorktreeBranchForFirstTurn = Effect.fnUntraced(function* (input: { readonly threadId: ThreadId; readonly branch: string | null; @@ -440,19 +377,13 @@ const make = Effect.gen(function* () { return; } - const branchGenerationAttachments = yield* resolvePersistedAttachments({ - threadId: event.payload.threadId, - messageId: message.id, - attachments: message.attachments, - }); - yield* maybeGenerateAndRenameWorktreeBranchForFirstTurn({ threadId: event.payload.threadId, branch: thread.branch, worktreePath: thread.worktreePath, messageId: message.id, messageText: message.text, - ...(message.attachments !== undefined ? { attachments: branchGenerationAttachments } : {}), + ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), }).pipe(Effect.forkScoped); yield* sendTurnForThread({ diff --git a/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts b/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts index 194c2943fec25b71a26b2d1394e91973121f25b9..b761387d474c202823478c49d11c22268dab82f5 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts @@ -21,10 +21,10 @@ layer("ProjectionThreadMessageRepository", (it) => { const persistedAttachments = [ { type: "image" as const, + id: "thread-preserve-attachments-att-1", name: "example.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "/attachments/thread-preserve-attachments/message-preserve-attachments/0.png", }, ]; @@ -74,10 +74,10 @@ layer("ProjectionThreadMessageRepository", (it) => { attachments: [ { type: "image", + id: "thread-clear-attachments-att-1", name: "example.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "/attachments/thread-clear-attachments/message-clear-attachments/0.png", }, ], isStreaming: false, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index c51e6963aa49baa8eacd24fab48128664ab4ac50..047adc6e4efd5b409105f058499c73e0e9e28fc3 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -4,7 +4,6 @@ import { ProviderItemId, type ProviderApprovalDecision, type ProviderEvent, - type ProviderSendTurnInput, ProviderSessionId, type ProviderSession, type ProviderSessionStartInput, @@ -12,12 +11,17 @@ import { ProviderTurnId, type ProviderTurnStartResult, } from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { afterAll, assert, it, vi } from "@effect/vitest"; import { assertFailure } from "@effect/vitest/utils"; -import { Effect, Fiber, Stream } from "effect"; +import { Effect, Fiber, Layer, Stream } from "effect"; -import { CodexAppServerManager } from "../../codexAppServerManager.ts"; +import { + CodexAppServerManager, + type CodexAppServerSendTurnInput, +} from "../../codexAppServerManager.ts"; +import { ServerConfig } from "../../config.ts"; import { ProviderAdapterValidationError } from "../Errors.ts"; import { CodexAdapter } from "../Services/CodexAdapter.ts"; import { makeCodexAdapterLive } from "./CodexAdapter.ts"; @@ -44,7 +48,7 @@ class FakeCodexManager extends CodexAppServerManager { ); public sendTurnImpl = vi.fn( - async (_input: ProviderSendTurnInput): Promise => ({ + async (_input: CodexAppServerSendTurnInput): Promise => ({ threadId: ProviderThreadId.makeUnsafe("thread-1"), turnId: asTurnId("turn-1"), }), @@ -78,7 +82,7 @@ class FakeCodexManager extends CodexAppServerManager { return this.startSessionImpl(input); } - override sendTurn(input: ProviderSendTurnInput): Promise { + override sendTurn(input: CodexAppServerSendTurnInput): Promise { return this.sendTurnImpl(input); } @@ -118,7 +122,12 @@ class FakeCodexManager extends CodexAppServerManager { } const validationManager = new FakeCodexManager(); -const validationLayer = it.layer(makeCodexAdapterLive({ manager: validationManager })); +const validationLayer = it.layer( + makeCodexAdapterLive({ manager: validationManager }).pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), + ), +); validationLayer("CodexAdapterLive validation", (it) => { it.effect("returns validation error for non-codex provider on startSession", () => @@ -147,7 +156,12 @@ const sessionErrorManager = new FakeCodexManager(); sessionErrorManager.sendTurnImpl.mockImplementation(async () => { throw new Error("Unknown session: sess-missing"); }); -const sessionErrorLayer = it.layer(makeCodexAdapterLive({ manager: sessionErrorManager })); +const sessionErrorLayer = it.layer( + makeCodexAdapterLive({ manager: sessionErrorManager }).pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), + ), +); sessionErrorLayer("CodexAdapterLive session errors", (it) => { it.effect("maps unknown-session sendTurn errors to ProviderAdapterSessionNotFoundError", () => @@ -178,7 +192,12 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { }); const lifecycleManager = new FakeCodexManager(); -const lifecycleLayer = it.layer(makeCodexAdapterLive({ manager: lifecycleManager })); +const lifecycleLayer = it.layer( + makeCodexAdapterLive({ manager: lifecycleManager }).pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), + ), +); lifecycleLayer("CodexAdapterLive lifecycle", (it) => { it.effect("maps completed agent message items to canonical message.completed events", () => diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 35f517a55a9104e10c6e9588eef22e38d70fdc1f..768af2bb2123917f00a106fa4bcece479298fd92 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -17,7 +17,7 @@ import { ProviderThreadId, ProviderTurnId, } from "@t3tools/contracts"; -import { Effect, Layer, Queue, Schema, Stream } from "effect"; +import { Effect, FileSystem, Layer, Queue, Schema, Stream } from "effect"; import { ProviderAdapterProcessError, @@ -29,6 +29,8 @@ import { } from "../Errors.ts"; import { CodexAdapter, type CodexAdapterShape } from "../Services/CodexAdapter.ts"; import { CodexAppServerManager } from "../../codexAppServerManager.ts"; +import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = "codex" as const; @@ -426,6 +428,8 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const serverConfig = yield* Effect.service(ServerConfig); const nativeEventLogger = options?.nativeEventLogPath !== undefined ? makeEventNdjsonLogger(options.nativeEventLogPath) @@ -475,9 +479,46 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => }; const sendTurn: CodexAdapterShape["sendTurn"] = (input) => - Effect.tryPromise({ - try: () => manager.sendTurn(input), - catch: (cause) => toRequestError(input.sessionId, "turn/start", cause), + Effect.gen(function* () { + const codexAttachments = yield* Effect.forEach( + input.attachments ?? [], + (attachment) => + Effect.gen(function* () { + const attachmentPath = resolveAttachmentPath({ + stateDir: serverConfig.stateDir, + attachment, + }); + if (!attachmentPath) { + return yield* toRequestError( + input.sessionId, + "turn/start", + new Error(`Invalid attachment id '${attachment.id}'.`), + ); + } + const bytes = yield* fileSystem.readFile(attachmentPath).pipe( + Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause)), + ); + return { + type: "image" as const, + url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`, + }; + }), + { concurrency: 1 }, + ); + + return yield* Effect.tryPromise({ + try: () => { + const managerInput = { + sessionId: input.sessionId, + ...(input.input !== undefined ? { input: input.input } : {}), + ...(input.model !== undefined ? { model: input.model } : {}), + ...(input.effort !== undefined ? { effort: input.effort } : {}), + ...(codexAttachments.length > 0 ? { attachments: codexAttachments } : {}), + }; + return manager.sendTurn(managerInput); + }, + catch: (cause) => toRequestError(input.sessionId, "turn/start", cause), + }); }); const interruptTurn: CodexAdapterShape["interruptTurn"] = (sessionId, turnId) => diff --git a/apps/server/src/serverLayers.ts b/apps/server/src/serverLayers.ts index 45ebda3b5b4dce07c85253ab307aace5c8644db8..632c0bb3e9392c6a6d4337de45fc2b72d407f14a 100644 --- a/apps/server/src/serverLayers.ts +++ b/apps/server/src/serverLayers.ts @@ -1,7 +1,7 @@ import path from "node:path"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { Effect, Layer } from "effect"; +import { Effect, FileSystem, Layer } from "effect"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery"; @@ -37,7 +37,7 @@ import { NodePtyAdapterLive } from "./terminal/Layers/NodePTY"; export function makeServerProviderLayer(): Layer.Layer< ProviderService, ProviderUnsupportedError, - SqlClient.SqlClient | ServerConfig + SqlClient.SqlClient | ServerConfig | FileSystem.FileSystem > { return Effect.gen(function* () { const { stateDir } = yield* ServerConfig; diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 3f18e8929ecfd10d2676a401cfd44072760d87ad..1a278e2d327282fcfd6f9a7597afedf42bffb0e8 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -420,6 +420,7 @@ describe("WebSocket Server", () => { Layer.provideMerge(runtimeLayer), Layer.provideMerge(openLayer), Layer.provideMerge(serverConfigLayer), + Layer.provideMerge(NodeServices.layer), ); const runtimeServices = await Effect.runPromise( Layer.build(dependenciesLayer).pipe(Scope.provide(scope)), diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index c40ef39a4c99c9b4c8dae58cac1844ac560fed92..b15677a68c81c9d3bf14cab4661762a0462a04ff 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -12,8 +12,11 @@ import type { Duplex } from "node:stream"; import Mime from "@effect/platform-node/Mime"; import { CommandId, + type ClientOrchestrationCommand, + type OrchestrationCommand, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, + PROVIDER_SEND_TURN_MAX_IMAGE_BYTES, ProjectId, ThreadId, TerminalEvent, @@ -60,6 +63,8 @@ import { normalizeAttachmentRelativePath, resolveAttachmentRelativePath, } from "./attachmentPaths"; +import { createAttachmentId, resolveAttachmentPath, resolveAttachmentPathById } from "./attachmentStore.ts"; +import { parseBase64DataUrl } from "./imageMime.ts"; /** * ServerShape - Service API for server lifecycle control. @@ -235,6 +240,88 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< }); }); + const normalizeDispatchCommand = Effect.fnUntraced(function* (input: { + readonly command: ClientOrchestrationCommand; + }) { + if (input.command.type !== "thread.turn.start") { + return input.command as OrchestrationCommand; + } + const turnStartCommand = input.command; + + const normalizedAttachments = yield* Effect.forEach( + turnStartCommand.message.attachments, + (attachment) => + Effect.gen(function* () { + const parsed = parseBase64DataUrl(attachment.dataUrl); + if (!parsed || !parsed.mimeType.startsWith("image/")) { + return yield* new RouteRequestError({ + message: `Invalid image attachment payload for '${attachment.name}'.`, + }); + } + + const bytes = Buffer.from(parsed.base64, "base64"); + if (bytes.byteLength === 0 || bytes.byteLength > PROVIDER_SEND_TURN_MAX_IMAGE_BYTES) { + return yield* new RouteRequestError({ + message: `Image attachment '${attachment.name}' is empty or too large.`, + }); + } + + const attachmentId = createAttachmentId(turnStartCommand.threadId); + if (!attachmentId) { + return yield* new RouteRequestError({ + message: "Failed to create a safe attachment id.", + }); + } + + const persistedAttachment = { + type: "image" as const, + id: attachmentId, + name: attachment.name, + mimeType: parsed.mimeType.toLowerCase(), + sizeBytes: bytes.byteLength, + }; + + const attachmentPath = resolveAttachmentPath({ + stateDir: serverConfig.stateDir, + attachment: persistedAttachment, + }); + if (!attachmentPath) { + return yield* new RouteRequestError({ + message: `Failed to resolve persisted path for '${attachment.name}'.`, + }); + } + + yield* fileSystem.makeDirectory(path.dirname(attachmentPath), { recursive: true }).pipe( + Effect.mapError( + () => + new RouteRequestError({ + message: `Failed to create attachment directory for '${attachment.name}'.`, + }), + ), + ); + yield* fileSystem.writeFile(attachmentPath, bytes).pipe( + Effect.mapError( + () => + new RouteRequestError({ + message: `Failed to persist attachment '${attachment.name}'.`, + }), + ), + ); + + return persistedAttachment; + }), + { concurrency: 1 }, + ); + + return { + ...turnStartCommand, + message: { + ...turnStartCommand.message, + attachments: normalizedAttachments, + }, + } satisfies OrchestrationCommand; + }); + // HTTP server — serves static files or redirects to Vite dev server const httpServer = http.createServer((req, res) => { const respond = ( @@ -261,10 +348,16 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< return; } - const filePath = resolveAttachmentRelativePath({ - stateDir: serverConfig.stateDir, - relativePath: normalizedRelativePath, - }); + const filePath = + !normalizedRelativePath.includes("/") && !normalizedRelativePath.includes(".") + ? resolveAttachmentPathById({ + stateDir: serverConfig.stateDir, + attachmentId: normalizedRelativePath, + }) + : resolveAttachmentRelativePath({ + stateDir: serverConfig.stateDir, + relativePath: normalizedRelativePath, + }); if (!filePath) { respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path"); return; @@ -508,7 +601,8 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< case ORCHESTRATION_WS_METHODS.dispatchCommand: { const { command } = request.body; - return yield* orchestrationEngine.dispatch(command); + const normalizedCommand = yield* normalizeDispatchCommand({ command }); + return yield* orchestrationEngine.dispatch(normalizedCommand); } case ORCHESTRATION_WS_METHODS.getTurnDiff: { diff --git a/apps/web/src/store.ts b/apps/web/src/store.ts index cd7ed5f98562a0e325b85811ea12a4e946faea7f..6d3a2fff05c3172af1a1074640e4173552a9d13e 100644 --- a/apps/web/src/store.ts +++ b/apps/web/src/store.ts @@ -229,6 +229,10 @@ function toAttachmentPreviewUrl(rawUrl: string): string { return rawUrl; } +function attachmentPreviewRoutePath(attachmentId: string): string { + return `/attachments/${encodeURIComponent(attachmentId)}`; +} + function normalizeTerminalIds(terminalIds: string[]): string[] { const ids = terminalIds.map((id) => id.trim()).filter((id) => id.length > 0); const unique = [...new Set(ids)].slice(0, MAX_THREAD_TERMINAL_COUNT); @@ -487,13 +491,13 @@ export function reducer(state: AppState, action: Action): AppState { } : null, messages: thread.messages.map((message) => { - const attachments = message.attachments?.map((attachment, index) => ({ + const attachments = message.attachments?.map((attachment) => ({ type: "image" as const, - id: `${message.id}:${index}`, + id: attachment.id, name: attachment.name, mimeType: attachment.mimeType, sizeBytes: attachment.sizeBytes, - previewUrl: toAttachmentPreviewUrl(attachment.dataUrl), + previewUrl: toAttachmentPreviewUrl(attachmentPreviewRoutePath(attachment.id)), })); const normalizedMessage: ChatMessage = { id: message.id, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 1de117eb66b56d7c8c3f86f09393050e4ddee8fc..a6b1f821415b1b9f904069a4793187ff66ac51d9 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -60,12 +60,28 @@ export const PROVIDER_SEND_TURN_MAX_INPUT_CHARS = 120_000; export const PROVIDER_SEND_TURN_MAX_ATTACHMENTS = 8; export const PROVIDER_SEND_TURN_MAX_IMAGE_BYTES = 10 * 1024 * 1024; export const PROVIDER_SEND_TURN_MAX_IMAGE_DATA_URL_CHARS = 14_000_000; +export const CHAT_ATTACHMENT_ID_MAX_CHARS = 128; // Correlation id is command id by design in this model. export const CorrelationId = CommandId; export type CorrelationId = typeof CorrelationId.Type; +export const ChatAttachmentId = TrimmedNonEmptyString.check( + Schema.isMaxLength(CHAT_ATTACHMENT_ID_MAX_CHARS), + Schema.isPattern(/^[a-z0-9_-]+$/i), +); +export type ChatAttachmentId = typeof ChatAttachmentId.Type; + export const ChatImageAttachment = Schema.Struct({ + type: Schema.Literal("image"), + id: ChatAttachmentId, + name: TrimmedNonEmptyString.check(Schema.isMaxLength(255)), + mimeType: TrimmedNonEmptyString.check(Schema.isMaxLength(100), Schema.isPattern(/^image\//i)), + sizeBytes: NonNegativeInt.check(Schema.isLessThanOrEqualTo(PROVIDER_SEND_TURN_MAX_IMAGE_BYTES)), +}); +export type ChatImageAttachment = typeof ChatImageAttachment.Type; + +export const UploadChatImageAttachment = Schema.Struct({ type: Schema.Literal("image"), name: TrimmedNonEmptyString.check(Schema.isMaxLength(255)), mimeType: TrimmedNonEmptyString.check(Schema.isMaxLength(100), Schema.isPattern(/^image\//i)), @@ -74,10 +90,12 @@ export const ChatImageAttachment = Schema.Struct({ Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_IMAGE_DATA_URL_CHARS), ), }); -export type ChatImageAttachment = typeof ChatImageAttachment.Type; +export type UploadChatImageAttachment = typeof UploadChatImageAttachment.Type; export const ChatAttachment = Schema.Union([ChatImageAttachment]); export type ChatAttachment = typeof ChatAttachment.Type; +export const UploadChatAttachment = Schema.Union([UploadChatImageAttachment]); +export type UploadChatAttachment = typeof UploadChatAttachment.Type; export const ProjectScriptIcon = Schema.Literals([ "play", @@ -307,6 +325,24 @@ export const ThreadTurnStartCommand = Schema.Struct({ createdAt: IsoDateTime, }); +export const ClientThreadTurnStartCommand = Schema.Struct({ + type: Schema.Literal("thread.turn.start"), + commandId: CommandId, + threadId: ThreadId, + message: Schema.Struct({ + messageId: MessageId, + role: Schema.Literal("user"), + text: Schema.String, + attachments: Schema.Array(UploadChatAttachment), + }), + model: Schema.optional(TrimmedNonEmptyString), + effort: Schema.optional(TrimmedNonEmptyString), + assistantDeliveryMode: Schema.optional(AssistantDeliveryMode), + approvalPolicy: ProviderApprovalPolicy, + sandboxMode: ProviderSandboxMode, + createdAt: IsoDateTime, +}); + export const ThreadTurnInterruptCommand = Schema.Struct({ type: Schema.Literal("thread.turn.interrupt"), commandId: CommandId, @@ -339,7 +375,7 @@ export const ThreadSessionStopCommand = Schema.Struct({ createdAt: IsoDateTime, }); -export const ClientOrchestrationCommand = Schema.Union([ +export const DispatchableClientOrchestrationCommand = Schema.Union([ ProjectCreateCommand, ProjectMetaUpdateCommand, ProjectDeleteCommand, @@ -352,6 +388,22 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadCheckpointRevertCommand, ThreadSessionStopCommand, ]); +export type DispatchableClientOrchestrationCommand = + typeof DispatchableClientOrchestrationCommand.Type; + +export const ClientOrchestrationCommand = Schema.Union([ + ProjectCreateCommand, + ProjectMetaUpdateCommand, + ProjectDeleteCommand, + ThreadCreateCommand, + ThreadDeleteCommand, + ThreadMetaUpdateCommand, + ClientThreadTurnStartCommand, + ThreadTurnInterruptCommand, + ThreadApprovalRespondCommand, + ThreadCheckpointRevertCommand, + ThreadSessionStopCommand, +]); export type ClientOrchestrationCommand = typeof ClientOrchestrationCommand.Type; export const ThreadSessionSetCommand = Schema.Struct({ @@ -422,7 +474,7 @@ export const InternalOrchestrationCommand = Schema.Union([ export type InternalOrchestrationCommand = typeof InternalOrchestrationCommand.Type; export const OrchestrationCommand = Schema.Union([ - ClientOrchestrationCommand, + DispatchableClientOrchestrationCommand, InternalOrchestrationCommand, ]); export type OrchestrationCommand = typeof OrchestrationCommand.Type;