From 8f42d34db921fa623715b9524aa3894919dec8bb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 1 Mar 2026 00:38:27 -0800 Subject: [PATCH] Inject test ServerConfig across orchestration and git layers - add `ServerConfig.layerTest(cwd, stateDir)` to standardize test wiring - require `ServerConfig` in `CodexTextGeneration` attachment path resolution - update orchestration/projection tests and harness to use managed runtimes with explicit config --- .../OrchestrationEngineHarness.integration.ts | 22 +- apps/server/src/config.ts | 24 +- .../git/Layers/CodexTextGeneration.test.ts | 106 +-- .../src/git/Layers/CodexTextGeneration.ts | 12 +- .../Layers/CheckpointReactor.test.ts | 2 + .../Layers/OrchestrationEngine.test.ts | 6 + .../Layers/ProjectionPipeline.test.ts | 810 +++++++++--------- .../Layers/ProjectionPipeline.ts | 64 +- .../Layers/ProviderCommandReactor.test.ts | 48 +- .../Layers/ProviderCommandReactor.ts | 13 +- .../Layers/ProviderRuntimeIngestion.test.ts | 63 +- 11 files changed, 566 insertions(+), 604 deletions(-) diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index c99c29adf2c30c4ee17abae8d34eaf982fbdbe4e..45eb883c6e4b903c6055201523ed6fd187240b2a 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -56,6 +56,7 @@ import { makeTestProviderAdapterHarness, type TestProviderAdapterHarness, } from "./TestProviderAdapter.integration.ts"; +import { ServerConfig } from "../src/config.ts"; function runGit(cwd: string, args: ReadonlyArray) { return execFileSync("git", args, { @@ -229,19 +230,13 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(runtimeServicesLayer), ); - const gitCoreLayer = Layer.succeed( - GitCore, - { - renameBranch: (input: Parameters[0]) => - Effect.succeed({ branch: input.newBranch }), - } as unknown as GitCoreShape, - ); - const textGenerationLayer = Layer.succeed( - TextGeneration, - { - generateBranchName: () => Effect.succeed({ branch: null }), - } as unknown as TextGenerationShape, - ); + const gitCoreLayer = Layer.succeed(GitCore, { + renameBranch: (input: Parameters[0]) => + Effect.succeed({ branch: input.newBranch }), + } as unknown as GitCoreShape); + const textGenerationLayer = Layer.succeed(TextGeneration, { + generateBranchName: () => Effect.succeed({ branch: null }), + } as unknown as TextGenerationShape); const providerCommandReactorLayer = ProviderCommandReactorLive.pipe( Layer.provideMerge(runtimeServicesLayer), Layer.provideMerge(gitCoreLayer), @@ -257,6 +252,7 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { ); const layer = orchestrationReactorLayer.pipe( Layer.provide(persistenceLayer), + Layer.provideMerge(ServerConfig.layerTest(workspaceDir, stateDir)), Layer.provideMerge(NodeServices.layer), ); diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index a875e6c9562590ff980737d2bee99e94e26a54f3..7a1cb76831114ef783d564880f6d1ce3acbdd9c9 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -36,7 +36,29 @@ export interface ServerConfigShape { */ export class ServerConfig extends ServiceMap.Service()( "t3/config/ServerConfig", -) {} +) { + static readonly layerTest = (cwd: string, statedir: string) => + Layer.effect( + ServerConfig, + Effect.gen(function* () { + const path = yield* Path.Path; + return { + cwd, + stateDir: statedir, + mode: "web", + autoBootstrapProjectFromCwd: false, + logWebSocketEvents: false, + port: 0, + host: undefined, + authToken: undefined, + keybindingsConfigPath: path.join(statedir, "keybindings.json"), + staticDir: undefined, + devUrl: undefined, + noBrowser: false, + }; + }), + ); +} // Helpers diff --git a/apps/server/src/git/Layers/CodexTextGeneration.test.ts b/apps/server/src/git/Layers/CodexTextGeneration.test.ts index e8ca0303fa26e5bfa12704a16bfe0300dcf971a5..2ebde70e3f6bf80f67f1467af86daf0cb065b9b4 100644 --- a/apps/server/src/git/Layers/CodexTextGeneration.test.ts +++ b/apps/server/src/git/Layers/CodexTextGeneration.test.ts @@ -3,27 +3,16 @@ import { it } from "@effect/vitest"; import { Effect, FileSystem, Layer, Path } from "effect"; import { expect } from "vitest"; -import { type ServerConfigShape, ServerConfig } from "../../config.ts"; +import { ServerConfig } from "../../config.ts"; import { CodexTextGenerationLive } from "./CodexTextGeneration.ts"; import { TextGenerationError } from "../Errors.ts"; import { TextGeneration } from "../Services/TextGeneration.ts"; -function makeTestServerConfig(stateDir: string): ServerConfigShape { - return { - mode: "web", - port: 0, - host: undefined, - cwd: process.cwd(), - keybindingsConfigPath: "", - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - }; -} +const makeCodexTextGenerationTestLayer = (stateDir: string) => + CodexTextGenerationLive.pipe( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)), + Layer.provideMerge(NodeServices.layer), + ); function makeFakeCodexBinary(dir: string) { return Effect.gen(function* () { @@ -197,10 +186,7 @@ function withFakeCodexEnv( ); } -const CodexTextGenerationTestLayer = Layer.provideMerge( - CodexTextGenerationLive, - NodeServices.layer, -); +const CodexTextGenerationTestLayer = makeCodexTextGenerationTestLayer(process.cwd()); it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { it.effect("generates and sanitizes commit messages", () => @@ -340,14 +326,10 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; - const stateDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-state-textgen-" }); - const imagePath = path.join( - stateDir, - "attachments", - "thread-1", - "message-1-0.png", - ); - yield* fs.makeDirectory(path.join(stateDir, "attachments", "thread-1"), { + 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, }); yield* fs.writeFile(imagePath, Buffer.from("hello")); @@ -363,46 +345,52 @@ it.layer(CodexTextGenerationTestLayer)("CodexTextGenerationLive", (it) => { name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "/attachments/thread-1/message-1-0.png", + dataUrl: `/attachments/thread-1/${imageFileName}`, }, ], }) - .pipe(Effect.provideService(ServerConfig, makeTestServerConfig(stateDir))); + .pipe( + Effect.ensuring( + fs.remove(imagePath).pipe(Effect.catch(() => Effect.void)), + ), + ); expect(generated.branch).toBe("fix/ui-regression"); }), ), ); - it.effect("fails with typed TextGenerationError when codex returns wrong branch payload shape", () => - withFakeCodexEnv( - { - output: JSON.stringify({ - title: "This is not a branch payload", + it.effect( + "fails with typed TextGenerationError when codex returns wrong branch payload shape", + () => + withFakeCodexEnv( + { + output: JSON.stringify({ + title: "This is not a branch payload", + }), + }, + Effect.gen(function* () { + const textGeneration = yield* TextGeneration; + + const result = yield* textGeneration + .generateBranchName({ + cwd: process.cwd(), + message: "Fix websocket reconnect flake", + }) + .pipe( + Effect.match({ + onFailure: (error) => ({ _tag: "Left" as const, left: error }), + onSuccess: (value) => ({ _tag: "Right" as const, right: value }), + }), + ); + + expect(result._tag).toBe("Left"); + if (result._tag === "Left") { + expect(result.left).toBeInstanceOf(TextGenerationError); + expect(result.left.message).toContain("Codex returned invalid structured output"); + } }), - }, - Effect.gen(function* () { - const textGeneration = yield* TextGeneration; - - const result = yield* textGeneration - .generateBranchName({ - cwd: process.cwd(), - message: "Fix websocket reconnect flake", - }) - .pipe( - Effect.match({ - onFailure: (error) => ({ _tag: "Left" as const, left: error }), - onSuccess: (value) => ({ _tag: "Right" as const, right: value }), - }), - ); - - expect(result._tag).toBe("Left"); - if (result._tag === "Left") { - expect(result.left).toBeInstanceOf(TextGenerationError); - expect(result.left.message).toContain("Codex returned invalid structured output"); - } - }), - ), + ), ); it.effect("returns typed TextGenerationError when codex exits non-zero", () => diff --git a/apps/server/src/git/Layers/CodexTextGeneration.ts b/apps/server/src/git/Layers/CodexTextGeneration.ts index e1779f52212e6ba7357c752f6156b05742889a25..a5a161d84e1843915d9b9e737bfab0e50df02a31 100644 --- a/apps/server/src/git/Layers/CodexTextGeneration.ts +++ b/apps/server/src/git/Layers/CodexTextGeneration.ts @@ -105,6 +105,7 @@ const makeCodexTextGeneration = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const commandSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const serverConfig = yield* Effect.service(ServerConfig); const readStreamAsString = ( operation: string, @@ -180,7 +181,6 @@ const makeCodexTextGeneration = Effect.gen(function* () { if (!attachments || attachments.length === 0) { return []; } - const serverConfig = yield* Effect.serviceOption(ServerConfig); const imagePaths: string[] = []; for (const [index, attachment] of attachments.entries()) { @@ -209,12 +209,10 @@ const makeCodexTextGeneration = Effect.gen(function* () { continue; } - const persistedPath = Option.isSome(serverConfig) - ? resolveAttachmentRoutePath({ - stateDir: serverConfig.value.stateDir, - dataUrl: attachment.dataUrl, - }) - : null; + const persistedPath = resolveAttachmentRoutePath({ + stateDir: serverConfig.stateDir, + dataUrl: attachment.dataUrl, + }); const localPath = persistedPath ?? attachment.dataUrl; if (!path.isAbsolute(localPath)) { continue; diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index bbec8f62b08ce919cd2043d00f7f476491b77f5d..25ca97b8049faa031643ee70ffb85a6897ce2ac3 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -37,6 +37,7 @@ import { type ProviderServiceShape, } from "../../provider/Services/ProviderService.ts"; import { checkpointRefForThreadTurn } from "../../checkpointing/Utils.ts"; +import { ServerConfig } from "../../config.ts"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); @@ -238,6 +239,7 @@ describe("CheckpointReactor", () => { Layer.provideMerge(orchestrationLayer), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(CheckpointStoreLive), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), Layer.provideMerge(NodeServices.layer), ); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 925826bb1b43cc00eca0efd4337f0c209dd9cc98..a22b8979f27935370797592611a45fb7606b8c31 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -25,6 +25,8 @@ import { OrchestrationProjectionPipeline, type OrchestrationProjectionPipelineShape, } from "../Services/ProjectionPipeline.ts"; +import { ServerConfig } from "../../config.ts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); @@ -37,6 +39,8 @@ async function createOrchestrationSystem() { Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), ); const runtime = ManagedRuntime.make(orchestrationLayer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); @@ -310,6 +314,8 @@ describe("OrchestrationEngine", () => { Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), ), ); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 9b477f64abeb2e7f2e3fb3dbe92c1fd154fc62a8..8cbef3862b2b0fd8c24d0f26264eca9186362b4e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -12,7 +12,7 @@ import { } from "@t3tools/contracts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; -import { Effect, Layer } from "effect"; +import { Effect, Layer, ManagedRuntime } from "effect"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; @@ -32,13 +32,32 @@ import { } from "./ProjectionPipeline.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts"; -import { ServerConfig, type ServerConfigShape } from "../../config.ts"; +import { ServerConfig } from "../../config.ts"; -const projectionLayer = it.layer( +const makeProjectionPipelineTestLayer = (stateDir: string) => OrchestrationProjectionPipelineLive.pipe( Layer.provideMerge(OrchestrationEventStoreLive), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)), Layer.provideMerge(SqlitePersistenceMemory), - ), + Layer.provideMerge(NodeServices.layer), + ); + +const runWithProjectionPipelineLayer = ( + stateDir: string, + effect: Effect.Effect< + A, + E, + OrchestrationProjectionPipeline | OrchestrationEventStore | SqlClient.SqlClient + >, +) => + Effect.acquireUseRelease( + Effect.sync(() => ManagedRuntime.make(makeProjectionPipelineTestLayer(stateDir))), + (runtime) => Effect.promise(() => runtime.runPromise(effect)), + (runtime) => Effect.promise(() => runtime.dispose()), + ); + +const projectionLayer = it.layer( + makeProjectionPipelineTestLayer(process.cwd()), ); projectionLayer("OrchestrationProjectionPipeline", (it) => { @@ -160,137 +179,147 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { ); it.effect("materializes message image attachments into stateDir and stores URL references", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const sql = yield* SqlClient.SqlClient; - const now = new Date().toISOString(); - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-")); + Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe( + 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 serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-attachments"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-attachments"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-attachments"), + causationEventId: null, + correlationId: CommandId.makeUnsafe("cmd-attachments"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-attachments"), + messageId: MessageId.makeUnsafe("message-attachments"), + role: "user", + text: "Inspect this", + attachments: [ + { + type: "image", + name: "example.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,SGVsbG8=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); - yield* eventStore.append({ - type: "thread.message-sent", - eventId: EventId.makeUnsafe("evt-attachments"), - aggregateKind: "thread", - aggregateId: ThreadId.makeUnsafe("thread-attachments"), - occurredAt: now, - commandId: CommandId.makeUnsafe("cmd-attachments"), - causationEventId: null, - correlationId: CommandId.makeUnsafe("cmd-attachments"), - metadata: {}, - payload: { - threadId: ThreadId.makeUnsafe("thread-attachments"), - messageId: MessageId.makeUnsafe("message-attachments"), - role: "user", - text: "Inspect this", - attachments: [ + yield* projectionPipeline.bootstrap; + + const rows = yield* sql<{ + readonly attachmentsJson: string | null; + }>` + SELECT + attachments_json AS "attachmentsJson" + FROM projection_thread_messages + WHERE message_id = 'message-attachments' + `; + assert.equal(rows.length, 1); + assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ { type: "image", name: "example.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", + dataUrl: "/attachments/thread-attachments/message-attachments-0.png", }, - ], - turnId: null, - streaming: false, - createdAt: now, - updatedAt: now, - }, - }); + ]); - yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig)); - - const rows = yield* sql<{ - readonly attachmentsJson: string | null; - }>` - SELECT - attachments_json AS "attachmentsJson" - FROM projection_thread_messages - WHERE message_id = 'message-attachments' - `; - assert.equal(rows.length, 1); - assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ - { - type: "image", - 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")); - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + 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 }))), + ), + ), + ), ); it.effect("materializes only image data URLs and uses a safe extension whitelist", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const sql = yield* SqlClient.SqlClient; - const now = new Date().toISOString(); - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-")); + Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-"))).pipe( + 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 serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; + yield* eventStore.append({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-attachments-safe"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-attachments-safe"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-attachments-safe"), + causationEventId: null, + correlationId: CommandId.makeUnsafe("cmd-attachments-safe"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-attachments-safe"), + messageId: MessageId.makeUnsafe("message-attachments-safe"), + role: "user", + text: "Inspect this", + attachments: [ + { + type: "image", + name: "untrusted.exe", + mimeType: "image/x-unknown", + sizeBytes: 5, + dataUrl: "data:image/x-unknown;base64,SGVsbG8=", + }, + { + type: "image", + name: "not-image.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:text/plain;base64,SGVsbG8=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); - yield* eventStore.append({ - type: "thread.message-sent", - eventId: EventId.makeUnsafe("evt-attachments-safe"), - aggregateKind: "thread", - aggregateId: ThreadId.makeUnsafe("thread-attachments-safe"), - occurredAt: now, - commandId: CommandId.makeUnsafe("cmd-attachments-safe"), - causationEventId: null, - correlationId: CommandId.makeUnsafe("cmd-attachments-safe"), - metadata: {}, - payload: { - threadId: ThreadId.makeUnsafe("thread-attachments-safe"), - messageId: MessageId.makeUnsafe("message-attachments-safe"), - role: "user", - text: "Inspect this", - attachments: [ + yield* projectionPipeline.bootstrap; + + const rows = yield* sql<{ + readonly attachmentsJson: string | null; + }>` + SELECT + attachments_json AS "attachmentsJson" + FROM projection_thread_messages + WHERE message_id = 'message-attachments-safe' + `; + assert.equal(rows.length, 1); + assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ { type: "image", name: "untrusted.exe", mimeType: "image/x-unknown", sizeBytes: 5, - dataUrl: "data:image/x-unknown;base64,SGVsbG8=", + dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin", }, { type: "image", @@ -299,60 +328,30 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { sizeBytes: 5, dataUrl: "data:text/plain;base64,SGVsbG8=", }, - ], - turnId: null, - streaming: false, - createdAt: now, - updatedAt: now, - }, - }); + ]); - yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig)); + 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 rows = yield* sql<{ - readonly attachmentsJson: string | null; - }>` - SELECT - attachments_json AS "attachmentsJson" - FROM projection_thread_messages - WHERE message_id = 'message-attachments-safe' - `; - assert.equal(rows.length, 1); - assert.deepEqual(JSON.parse(rows[0]?.attachmentsJson ?? "null"), [ - { - type: "image", - name: "untrusted.exe", - mimeType: "image/x-unknown", - sizeBytes: 5, - dataUrl: "/attachments/thread-attachments-safe/message-attachments-safe-0.bin", - }, - { - type: "image", - 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); - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + 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 }))), + ), + ), + ), ); it.effect( @@ -480,31 +479,17 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { it.effect( "overwrites attachment file bytes when a message updates the same attachment index", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const stateDir = fs.mkdtempSync( - path.join(os.tmpdir(), "t3-projection-attachments-overwrite-"), - ); - const now = new Date().toISOString(); - const later = new Date(Date.now() + 1_000).toISOString(); + 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(); - const serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; - - yield* eventStore.append({ + yield* eventStore.append({ type: "project.created", eventId: EventId.makeUnsafe("evt-overwrite-1"), aggregateKind: "project", @@ -609,95 +594,86 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }, }); - yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig)); + 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")); - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + 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")); + }).pipe( + (effect) => runWithProjectionPipelineLayer(stateDir, effect), + Effect.ensuring( + Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true })), + ), + ), + ), + ), ); it.effect("does not persist attachment files when projector transaction rolls back", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const sql = yield* SqlClient.SqlClient; - const stateDir = fs.mkdtempSync( - path.join(os.tmpdir(), "t3-projection-attachments-rollback-"), - ); - const now = new Date().toISOString(); + Effect.sync(() => + fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-rollback-")), + ).pipe( + 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 serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); - const appendAndProject = (event: Parameters[0]) => - eventStore.append(event).pipe( - Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), - Effect.provideService(ServerConfig, serverConfig), - ); + yield* appendAndProject({ + type: "project.created", + eventId: EventId.makeUnsafe("evt-rollback-1"), + aggregateKind: "project", + aggregateId: ProjectId.makeUnsafe("project-rollback"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-rollback-1"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-rollback-1"), + metadata: {}, + payload: { + projectId: ProjectId.makeUnsafe("project-rollback"), + title: "Project Rollback", + workspaceRoot: "/tmp/project-rollback", + defaultModel: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); - yield* appendAndProject({ - type: "project.created", - eventId: EventId.makeUnsafe("evt-rollback-1"), - aggregateKind: "project", - aggregateId: ProjectId.makeUnsafe("project-rollback"), - occurredAt: now, - commandId: CommandId.makeUnsafe("cmd-rollback-1"), - causationEventId: null, - correlationId: CorrelationId.makeUnsafe("cmd-rollback-1"), - metadata: {}, - payload: { - projectId: ProjectId.makeUnsafe("project-rollback"), - title: "Project Rollback", - workspaceRoot: "/tmp/project-rollback", - defaultModel: null, - scripts: [], - createdAt: now, - updatedAt: now, - }, - }); + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.makeUnsafe("evt-rollback-2"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-rollback"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-rollback-2"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-rollback-2"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-rollback"), + projectId: ProjectId.makeUnsafe("project-rollback"), + title: "Thread Rollback", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }); - yield* appendAndProject({ - type: "thread.created", - eventId: EventId.makeUnsafe("evt-rollback-2"), - aggregateKind: "thread", - aggregateId: ThreadId.makeUnsafe("thread-rollback"), - occurredAt: now, - commandId: CommandId.makeUnsafe("cmd-rollback-2"), - causationEventId: null, - correlationId: CorrelationId.makeUnsafe("cmd-rollback-2"), - metadata: {}, - payload: { - threadId: ThreadId.makeUnsafe("thread-rollback"), - projectId: ProjectId.makeUnsafe("project-rollback"), - title: "Thread Rollback", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: now, - updatedAt: now, - }, - }); - - yield* sql` + yield* sql` CREATE TRIGGER fail_thread_messages_projection_state_update BEFORE UPDATE ON projection_state WHEN NEW.projector = 'projection.thread-messages' @@ -706,90 +682,81 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { END; `; - const result = yield* Effect.result( - appendAndProject({ - type: "thread.message-sent", - eventId: EventId.makeUnsafe("evt-rollback-3"), - aggregateKind: "thread", - aggregateId: ThreadId.makeUnsafe("thread-rollback"), - occurredAt: now, - commandId: CommandId.makeUnsafe("cmd-rollback-3"), - causationEventId: null, - correlationId: CorrelationId.makeUnsafe("cmd-rollback-3"), - metadata: {}, - payload: { - threadId: ThreadId.makeUnsafe("thread-rollback"), - messageId: MessageId.makeUnsafe("message-rollback"), - role: "user", - text: "Rollback me", - attachments: [ - { - type: "image", - name: "rollback.png", - mimeType: "image/png", - sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", + const result = yield* Effect.result( + appendAndProject({ + type: "thread.message-sent", + eventId: EventId.makeUnsafe("evt-rollback-3"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe("thread-rollback"), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-rollback-3"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-rollback-3"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe("thread-rollback"), + messageId: MessageId.makeUnsafe("message-rollback"), + role: "user", + text: "Rollback me", + attachments: [ + { + type: "image", + name: "rollback.png", + mimeType: "image/png", + sizeBytes: 5, + dataUrl: "data:image/png;base64,SGVsbG8=", + }, + ], + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, }, - ], - turnId: null, - streaming: false, - createdAt: now, - updatedAt: now, - }, - }), - ); - assert.equal(result._tag, "Failure"); + }), + ); + assert.equal(result._tag, "Failure"); - const rows = yield* sql<{ - readonly count: number; - }>` + const rows = yield* sql<{ + readonly count: number; + }>` SELECT COUNT(*) AS "count" FROM projection_thread_messages WHERE message_id = 'message-rollback' `; - assert.equal(rows[0]?.count ?? 0, 0); + assert.equal(rows[0]?.count ?? 0, 0); - const attachmentPath = path.join( - stateDir, - "attachments", - "thread-rollback", - "message-rollback-0.png", - ); - assert.equal(fs.existsSync(attachmentPath), false); - yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`; - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + const attachmentPath = path.join( + stateDir, + "attachments", + "thread-rollback", + "message-rollback-0.png", + ); + assert.equal(fs.existsSync(attachmentPath), false); + yield* sql`DROP TRIGGER IF EXISTS fail_thread_messages_projection_state_update`; + }).pipe( + (effect) => runWithProjectionPipelineLayer(stateDir, effect), + Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), + ), + ), + ), ); it.effect("removes unreferenced attachment files when a thread is reverted", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-revert-")); - const now = new Date().toISOString(); + Effect.sync(() => + fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-revert-")), + ).pipe( + Effect.flatMap((stateDir) => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const now = new Date().toISOString(); - const serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); - const appendAndProject = (event: Parameters[0]) => - eventStore.append(event).pipe( - Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), - Effect.provideService(ServerConfig, serverConfig), - ); - - yield* appendAndProject({ + yield* appendAndProject({ type: "project.created", eventId: EventId.makeUnsafe("evt-revert-files-1"), aggregateKind: "project", @@ -969,41 +936,32 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }, }); - assert.equal(fs.existsSync(keepPath), true); - assert.equal(fs.existsSync(removePath), false); - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + assert.equal(fs.existsSync(keepPath), true); + assert.equal(fs.existsSync(removePath), false); + }).pipe( + (effect) => runWithProjectionPipelineLayer(stateDir, effect), + Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), + ), + ), + ), ); it.effect("removes thread attachment directory when thread is deleted", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-delete-")); - const now = new Date().toISOString(); + Effect.sync(() => + fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-delete-")), + ).pipe( + Effect.flatMap((stateDir) => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const now = new Date().toISOString(); - const serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); - const appendAndProject = (event: Parameters[0]) => - eventStore.append(event).pipe( - Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)), - Effect.provideService(ServerConfig, serverConfig), - ); - - yield* appendAndProject({ + yield* appendAndProject({ type: "project.created", eventId: EventId.makeUnsafe("evt-delete-files-1"), aggregateKind: "project", @@ -1096,62 +1054,56 @@ projectionLayer("OrchestrationProjectionPipeline", (it) => { }, }); - assert.equal(fs.existsSync(threadAttachmentDir), false); - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + assert.equal(fs.existsSync(threadAttachmentDir), false); + }).pipe( + (effect) => runWithProjectionPipelineLayer(stateDir, effect), + Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), + ), + ), + ), ); it.effect("ignores unsafe thread ids for attachment cleanup paths", () => - Effect.gen(function* () { - const projectionPipeline = yield* OrchestrationProjectionPipeline; - const eventStore = yield* OrchestrationEventStore; - const stateDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-unsafe-")); - const now = new Date().toISOString(); - const attachmentsRootDir = path.join(stateDir, "attachments"); - const attachmentsSentinelPath = path.join(attachmentsRootDir, "sentinel.txt"); - const stateDirSentinelPath = path.join(stateDir, "state-sentinel.txt"); - fs.mkdirSync(attachmentsRootDir, { recursive: true }); - fs.writeFileSync(attachmentsSentinelPath, "keep-attachments-root", "utf8"); - fs.writeFileSync(stateDirSentinelPath, "keep-state-dir", "utf8"); + Effect.sync(() => fs.mkdtempSync(path.join(os.tmpdir(), "t3-projection-attachments-unsafe-"))).pipe( + Effect.flatMap((stateDir) => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const now = new Date().toISOString(); + const attachmentsRootDir = path.join(stateDir, "attachments"); + const attachmentsSentinelPath = path.join(attachmentsRootDir, "sentinel.txt"); + const stateDirSentinelPath = path.join(stateDir, "state-sentinel.txt"); + fs.mkdirSync(attachmentsRootDir, { recursive: true }); + fs.writeFileSync(attachmentsSentinelPath, "keep-attachments-root", "utf8"); + fs.writeFileSync(stateDirSentinelPath, "keep-state-dir", "utf8"); - const serverConfig = { - mode: "web", - port: 0, - host: undefined, - cwd: "/tmp/project-attachments", - keybindingsConfigPath: path.join(stateDir, "keybindings.json"), - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - } satisfies ServerConfigShape; + yield* eventStore.append({ + type: "thread.deleted", + eventId: EventId.makeUnsafe("evt-unsafe-thread-delete"), + aggregateKind: "thread", + aggregateId: ThreadId.makeUnsafe(".."), + occurredAt: now, + commandId: CommandId.makeUnsafe("cmd-unsafe-thread-delete"), + causationEventId: null, + correlationId: CorrelationId.makeUnsafe("cmd-unsafe-thread-delete"), + metadata: {}, + payload: { + threadId: ThreadId.makeUnsafe(".."), + deletedAt: now, + }, + }); - yield* eventStore.append({ - type: "thread.deleted", - eventId: EventId.makeUnsafe("evt-unsafe-thread-delete"), - aggregateKind: "thread", - aggregateId: ThreadId.makeUnsafe(".."), - occurredAt: now, - commandId: CommandId.makeUnsafe("cmd-unsafe-thread-delete"), - causationEventId: null, - correlationId: CorrelationId.makeUnsafe("cmd-unsafe-thread-delete"), - metadata: {}, - payload: { - threadId: ThreadId.makeUnsafe(".."), - deletedAt: now, - }, - }); + yield* projectionPipeline.bootstrap; - yield* projectionPipeline.bootstrap.pipe(Effect.provideService(ServerConfig, serverConfig)); - - assert.equal(fs.existsSync(attachmentsRootDir), true); - assert.equal(fs.existsSync(attachmentsSentinelPath), true); - assert.equal(fs.existsSync(stateDirSentinelPath), true); - fs.rmSync(stateDir, { recursive: true, force: true }); - }), + assert.equal(fs.existsSync(attachmentsRootDir), true); + assert.equal(fs.existsSync(attachmentsSentinelPath), true); + assert.equal(fs.existsSync(stateDirSentinelPath), true); + }).pipe( + (effect) => runWithProjectionPipelineLayer(stateDir, effect), + Effect.ensuring(Effect.sync(() => fs.rmSync(stateDir, { recursive: true, force: true }))), + ), + ), + ), ); it.effect("resumes from projector last_applied_sequence without replaying older events", () => @@ -1868,7 +1820,11 @@ it.effect("restores pending turn-start metadata across projection pipeline resta ]); fs.rmSync(tempDir, { recursive: true, force: true }); - }).pipe(Effect.provide(NodeServices.layer)), + }).pipe( + Effect.provide( + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd()), NodeServices.layer), + ), + ), ); const engineLayer = it.layer( @@ -1877,6 +1833,8 @@ const engineLayer = it.layer( Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provideMerge(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), ), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 8f8e01a512f865d29749f02a4b97261893fd9539..989b5d4dd418a235675162b144a3c84c8c3d12a3 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -96,11 +96,10 @@ const materializeAttachmentsForProjection = Effect.fn(function* (input: { }) { if (input.attachments.length === 0) return []; - const serverConfig = yield* Effect.serviceOption(ServerConfig); - if (Option.isNone(serverConfig)) return input.attachments; + const serverConfig = yield* Effect.service(ServerConfig); const path = yield* Path.Path; - const attachmentsRootDir = path.join(serverConfig.value.stateDir, "attachments"); + const attachmentsRootDir = path.join(serverConfig.stateDir, "attachments"); const threadSegment = toSafeThreadAttachmentSegment(input.threadId); if (!threadSegment) { yield* Effect.logWarning("skipping attachment materialization for unsafe thread id", { @@ -289,20 +288,15 @@ function collectThreadAttachmentRelativePaths( return relativePaths; } -const runAttachmentSideEffects = Effect.fn(function* (input: { - readonly sideEffects: AttachmentSideEffects; - readonly fileSystem: FileSystem.FileSystem; - readonly path: Path.Path; -}) { - const serverConfig = yield* Effect.serviceOption(ServerConfig); - if (Option.isNone(serverConfig)) { - return; - } +const runAttachmentSideEffects = Effect.fn(function* (sideEffects: AttachmentSideEffects) { + const serverConfig = yield* Effect.service(ServerConfig); + const fileSystem = yield* Effect.service(FileSystem.FileSystem); + const path = yield* Effect.service(Path.Path); - const attachmentsRootDir = input.path.join(serverConfig.value.stateDir, "attachments"); + const attachmentsRootDir = path.join(serverConfig.stateDir, "attachments"); yield* Effect.forEach( - input.sideEffects.deletedThreadIds, + sideEffects.deletedThreadIds, (threadId) => Effect.gen(function* () { const threadSegment = toSafeThreadAttachmentSegment(threadId); @@ -312,16 +306,16 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { }); return; } - const threadDir = input.path.join(attachmentsRootDir, threadSegment); - yield* input.fileSystem.remove(threadDir, { recursive: true, force: true }); + const threadDir = path.join(attachmentsRootDir, threadSegment); + yield* fileSystem.remove(threadDir, { recursive: true, force: true }); }), { concurrency: 1 }, ); yield* Effect.forEach( - input.sideEffects.prunedThreadRelativePaths.entries(), + sideEffects.prunedThreadRelativePaths.entries(), ([threadId, keptThreadRelativePaths]) => { - if (input.sideEffects.deletedThreadIds.has(threadId)) { + if (sideEffects.deletedThreadIds.has(threadId)) { return Effect.void; } return Effect.gen(function* () { @@ -330,8 +324,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId }); return; } - const threadDir = input.path.join(attachmentsRootDir, threadSegment); - const entries = yield* input.fileSystem + const threadDir = path.join(attachmentsRootDir, threadSegment); + const entries = yield* fileSystem .readDirectory(threadDir, { recursive: true }) .pipe(Effect.catch(() => Effect.succeed([] as Array))); yield* Effect.forEach( @@ -343,8 +337,8 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { return; } - const absolutePath = input.path.join(threadDir, threadRelativePath); - const fileInfo = yield* input.fileSystem + const absolutePath = path.join(threadDir, threadRelativePath); + const fileInfo = yield* fileSystem .stat(absolutePath) .pipe(Effect.catch(() => Effect.succeed(null))); if (!fileInfo || fileInfo.type !== "File") { @@ -352,7 +346,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { } if (!keptThreadRelativePaths.has(threadRelativePath)) { - yield* input.fileSystem.remove(absolutePath, { force: true }); + yield* fileSystem.remove(absolutePath, { force: true }); } }), { concurrency: 1 }, @@ -363,7 +357,7 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { ); const writesByPath = new Map(); - for (const pendingWrite of input.sideEffects.writes) { + 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); } @@ -371,10 +365,10 @@ const runAttachmentSideEffects = Effect.fn(function* (input: { writesByPath.entries(), ([absolutePath, bytes]) => Effect.gen(function* () { - yield* input.fileSystem.makeDirectory(input.path.dirname(absolutePath), { + yield* fileSystem.makeDirectory(path.dirname(absolutePath), { recursive: true, }); - yield* input.fileSystem.writeFile(absolutePath, bytes); + yield* fileSystem.writeFile(absolutePath, bytes); }), { concurrency: 1 }, ); @@ -394,6 +388,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; + const serverConfig = yield* ServerConfig; const applyProjectsProjection: ProjectorDefinition["apply"] = (event, _attachmentSideEffects) => Effect.gen(function* () { @@ -600,7 +595,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { stageFileWrite: (pendingWrite) => { attachmentSideEffects.writes.push(pendingWrite); }, - }).pipe(Effect.provideService(Path.Path, path)) + }).pipe( + Effect.provideService(ServerConfig, serverConfig), + Effect.provideService(Path.Path, path), + ) : existingMessage?.attachments; yield* projectionThreadMessageRepository.upsert({ messageId: event.payload.messageId, @@ -1108,11 +1106,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { ), ); - yield* runAttachmentSideEffects({ - sideEffects: attachmentSideEffects, - fileSystem, - path, - }).pipe( + yield* runAttachmentSideEffects(attachmentSideEffects).pipe( Effect.catch((cause) => Effect.logWarning("failed to apply projected attachment side-effects", { projector: projector.name, @@ -1144,6 +1138,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { Effect.forEach(projectors, (projector) => runProjectorForEvent(projector, event), { concurrency: 1, }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.provideService(ServerConfig, serverConfig), Effect.asVoid, Effect.catchTag("SqlError", (sqlError) => Effect.fail(toPersistenceSqlError("ProjectionPipeline.projectEvent:query")(sqlError)), @@ -1155,6 +1152,9 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { bootstrapProjector, { concurrency: 1 }, ).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.provideService(ServerConfig, serverConfig), Effect.asVoid, Effect.tap(() => Effect.log("orchestration projection pipeline bootstrapped").pipe( diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 8b608aa51e2a8191f29667bd4bb9c4342718b85b..65dc7f8755682d0b07225679192bc1170de7c29d 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -17,7 +17,7 @@ import { import { Effect, Exit, Layer, ManagedRuntime, PubSub, Scope, Stream } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { type ServerConfigShape, ServerConfig } from "../../config.ts"; +import { ServerConfig } from "../../config.ts"; import { TextGenerationError } from "../../git/Errors.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; @@ -33,6 +33,7 @@ import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { ProviderCommandReactorLive } from "./ProviderCommandReactor.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProviderCommandReactor } from "../Services/ProviderCommandReactor.ts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); @@ -42,23 +43,6 @@ const asApprovalRequestId = (value: string): ApprovalRequestId => const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); -function makeTestServerConfig(stateDir: string): ServerConfigShape { - return { - mode: "web", - port: 0, - host: undefined, - cwd: process.cwd(), - keybindingsConfigPath: "", - stateDir, - staticDir: undefined, - devUrl: undefined, - noBrowser: true, - authToken: undefined, - autoBootstrapProjectFromCwd: false, - logWebSocketEvents: false, - }; -} - async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise { const deadline = Date.now() + timeoutMs; const poll = async (): Promise => { @@ -81,6 +65,7 @@ describe("ProviderCommandReactor", () => { unknown > | null = null; let scope: Scope.Closeable | null = null; + const createdStateDirs = new Set(); afterEach(async () => { if (scope) { @@ -91,10 +76,16 @@ describe("ProviderCommandReactor", () => { await runtime.dispose(); } runtime = null; + for (const stateDir of createdStateDirs) { + fs.rmSync(stateDir, { recursive: true, force: true }); + } + createdStateDirs.clear(); }); async function createHarness(input?: { readonly stateDir?: string }) { const now = new Date().toISOString(); + const stateDir = input?.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "t3code-reactor-")); + createdStateDirs.add(stateDir); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); let nextSessionIndex = 1; const startSession = vi.fn((_: unknown, __: unknown) => { @@ -147,21 +138,18 @@ describe("ProviderCommandReactor", () => { Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(SqlitePersistenceMemory), ); - const liveLayer = ProviderCommandReactorLive.pipe( + const layer = ProviderCommandReactorLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(Layer.succeed(ProviderService, service)), Layer.provideMerge(Layer.succeed(GitCore, { renameBranch } as unknown as GitCoreShape)), Layer.provideMerge( Layer.succeed(TextGeneration, { generateBranchName } as unknown as TextGenerationShape), ), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), stateDir)), + Layer.provideMerge(NodeServices.layer), ); - const layer = - input?.stateDir !== undefined - ? liveLayer.pipe( - Layer.provideMerge(Layer.succeed(ServerConfig, makeTestServerConfig(input.stateDir))), - ) - : liveLayer; - runtime = ManagedRuntime.make(layer); + const runtime = ManagedRuntime.make(layer); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor)); scope = await Effect.runPromise(Scope.make("sequential")); @@ -202,6 +190,7 @@ describe("ProviderCommandReactor", () => { stopSession, renameBranch, generateBranchName, + stateDir, }; } @@ -294,7 +283,12 @@ describe("ProviderCommandReactor", () => { name: "bug.png", mimeType: "image/png", sizeBytes: 5, - dataUrl: "data:image/png;base64,SGVsbG8=", + dataUrl: path.join( + harness.stateDir, + "attachments", + "thread-1", + "user-message-worktree-rename-0.png", + ), }, ], }); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 1b5271c51ac371058fd0ff29d2c0e6c102f9e62e..a2f24607c2bf545d0fe89f4e013fd451c1402bac 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -111,7 +111,7 @@ const make = Effect.gen(function* () { const providerService = yield* ProviderService; const git = yield* GitCore; const textGeneration = yield* TextGeneration; - const serverConfig = yield* Effect.serviceOption(ServerConfig); + const serverConfig = yield* Effect.service(ServerConfig); const handledTurnStartKeys = yield* Cache.make({ capacity: HANDLED_TURN_START_KEY_MAX, timeToLive: HANDLED_TURN_START_KEY_TTL, @@ -301,20 +301,15 @@ const make = Effect.gen(function* () { return input.attachments ?? []; } - if (Option.isNone(serverConfig)) { - 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; - } + if (attachment.type !== "image") return attachment; const resolvedRoutePath = resolveAttachmentRoutePath({ - stateDir: serverConfig.value.stateDir, + stateDir: serverConfig.stateDir, dataUrl: attachment.dataUrl, }); const resolvedMaterializedPath = @@ -329,7 +324,7 @@ const make = Effect.gen(function* () { fileName: attachment.name, }); return resolveAttachmentRelativePath({ - stateDir: serverConfig.value.stateDir, + stateDir: serverConfig.stateDir, relativePath: `${threadSegment}/${messageSegment}-${index}${extension}`, }); })(); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index fa254887d8f476fd660f8447a1ac1225efee5379..55a8452be89120bb81bbe1462b91bd8ba04dbb33 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -28,11 +28,12 @@ import { type OrchestrationEngineShape, } from "../Services/OrchestrationEngine.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; +import { ServerConfig } from "../../config.ts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); -const asProviderThreadId = (value: string): ProviderThreadId => - ProviderThreadId.makeUnsafe(value); +const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); @@ -122,6 +123,8 @@ describe("ProviderRuntimeIngestion", () => { const layer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(NodeServices.layer), ); runtime = ManagedRuntime.make(layer); const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); @@ -255,7 +258,9 @@ describe("ProviderRuntimeIngestion", () => { await Effect.runPromise(Effect.sleep("40 millis")); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); - const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + const midThread = midReadModel.threads.find( + (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), + ); expect(midThread?.session?.status).toBe("running"); expect(midThread?.session?.activeTurnId).toBe("turn-primary"); @@ -308,7 +313,9 @@ describe("ProviderRuntimeIngestion", () => { await Effect.runPromise(Effect.sleep("40 millis")); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); - const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + const midThread = midReadModel.threads.find( + (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), + ); expect(midThread?.session?.status).toBe("running"); expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main"); @@ -401,7 +408,9 @@ describe("ProviderRuntimeIngestion", () => { await Effect.runPromise(Effect.sleep("30 millis")); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); - const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + const midThread = midReadModel.threads.find( + (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), + ); expect(midThread?.messages.some((message) => message.id === "assistant:item-buffered")).toBe( false, ); @@ -416,10 +425,10 @@ describe("ProviderRuntimeIngestion", () => { itemId: asItemId("item-buffered"), }); - const thread = await waitForThread( - harness.engine, - (entry) => - entry.messages.some((message) => message.id === "assistant:item-buffered" && !message.streaming), + const thread = await waitForThread(harness.engine, (entry) => + entry.messages.some( + (message) => message.id === "assistant:item-buffered" && !message.streaming, + ), ); const message = thread.messages.find((entry) => entry.id === "assistant:item-buffered"); expect(message?.text).toBe("buffer me"); @@ -475,15 +484,13 @@ describe("ProviderRuntimeIngestion", () => { delta: "hello live", }); - const liveThread = await waitForThread( - harness.engine, - (entry) => - entry.messages.some( - (message) => - message.id === "assistant:item-streaming-mode" && - message.streaming && - message.text === "hello live", - ), + const liveThread = await waitForThread(harness.engine, (entry) => + entry.messages.some( + (message) => + message.id === "assistant:item-streaming-mode" && + message.streaming && + message.text === "hello live", + ), ); const liveMessage = liveThread.messages.find( (entry) => entry.id === "assistant:item-streaming-mode", @@ -500,12 +507,10 @@ describe("ProviderRuntimeIngestion", () => { itemId: asItemId("item-streaming-mode"), }); - const finalThread = await waitForThread( - harness.engine, - (entry) => - entry.messages.some( - (message) => message.id === "assistant:item-streaming-mode" && !message.streaming, - ), + const finalThread = await waitForThread(harness.engine, (entry) => + entry.messages.some( + (message) => message.id === "assistant:item-streaming-mode" && !message.streaming, + ), ); const finalMessage = finalThread.messages.find( (entry) => entry.id === "assistant:item-streaming-mode", @@ -554,12 +559,10 @@ describe("ProviderRuntimeIngestion", () => { itemId: asItemId("item-buffer-spill"), }); - const thread = await waitForThread( - harness.engine, - (entry) => - entry.messages.some( - (message) => message.id === "assistant:item-buffer-spill" && !message.streaming, - ), + const thread = await waitForThread(harness.engine, (entry) => + entry.messages.some( + (message) => message.id === "assistant:item-buffer-spill" && !message.streaming, + ), ); const message = thread.messages.find((entry) => entry.id === "assistant:item-buffer-spill"); expect(message?.text.length).toBe(oversizedText.length); -- 2.54.0